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 <[email protected]> Co-authored-by: William Barnhart <[email protected]>
faust-streaming · Jan 4, 2023 · 89b82ab · 89b82ab
1 parent a48a7c4
commit 89b82ab
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 3 deletions.
diff --git a/faust/transport/_cython/conductor.pyx b/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:

diff --git a/faust/transport/conductor.py b/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,