fix for issue 419 (#420)

* fix for issue 419

* Revert "fix for issue 419"

This reverts commit e68821e.

* throw in ensure_future to make sure coroutines get converted

* lint conductor.pyx

Co-authored-by: William Barnhart <william.barnhart@he360.com>
Co-authored-by: William Barnhart <williambbarnhart@gmail.com>

Author: patkivikram

faust/transport/_cython/conductor.pyx

@@ -1,5 +1,5 @@
 # cython: language_level=3
-from asyncio import ALL_COMPLETED, wait
+from asyncio import ALL_COMPLETED, ensure_future, wait
 
 from faust.exceptions import KeyDecodeError, ValueDecodeError
 
@@ -75,7 +75,7 @@ cdef class ConductorHandler:
                         continue
                     delivered.add(chan)
                 if full:
-                    await wait([self._handle_full(event, chan, delivered)
+                    await wait([ensure_future(self._handle_full(event, chan, delivered))
                                         for event, chan in full],
                                         return_when=ALL_COMPLETED)
             except KeyDecodeError as exc:

faust/transport/conductor.py

@@ -168,7 +168,7 @@ async def on_message(message: Message) -> None:
                             on_topic_buffer_full(dest_chan)
                         await asyncio.wait(
                             [
-                                dest_chan.put(dest_event)
+                                asyncio.ensure_future(dest_chan.put(dest_event))
                                 for dest_event, dest_chan in full
                             ],
                             return_when=asyncio.ALL_COMPLETED,