Pass loop attribute to asyncio.ensure_future

[max-k]

If we specify a custom event loop at worker instantiation, it has to also be used by asyncio instead of default one.

Description

In some situation (for example in integration tests of Faust agents), I have to retry worker startup because kafka is not yet available.

But the first time I run execute_from_commandline, the default event loop is closed so I have to create a new one before calling execute_from_commandline again.

Unfortunately, custom event loop is not properly transmitted to asyncio, so it automatically makes use of default event loop and raise a RuntimeError('Event loop is closed').

With this simple fix, user-specified event loop will be properly transmitted to asyncio.

[patkivikram]

How do we plan on releasing this in pypi? Maybe call it mode-steaming?

[max-k]

Hi.

It's exactly the question I asked in the issue related to mode on faust-streaming project :-)
faust-streaming/faust#43 (comment)
mode-streaming is okay for me.