Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Remove wait_first and extra log #240

Merged
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
17 changes: 1 addition & 16 deletions faust/transport/consumer.py
Expand Up @@ -733,13 +733,6 @@ async def getmany(self, timeout: float) -> AsyncIterator[Tuple[TP, Message]]:
# convert timestamp to seconds from int milliseconds.
yield tp, to_message(tp, record)

async def _wait_suspend(self):
"""Wrapper around self.suspend_flow.wait() with no return value.

This allows for easily
"""
await self.suspend_flow.wait()

async def _wait_next_records(
self, timeout: float
) -> Tuple[Optional[RecordMap], Optional[Set[TP]]]:
Expand All @@ -760,18 +753,10 @@ async def _wait_next_records(
# Fetch records only if active partitions to avoid the risk of
# fetching all partitions in the beginning when none of the
# partitions is paused/resumed.
_getmany = self._getmany(
records = await self._getmany(
active_partitions=active_partitions,
timeout=timeout,
)
wait_results = await self.wait_first(
_getmany,
self.suspend_flow.wait(),
)
for coro, result in zip(wait_results.done, wait_results.results):
if coro is _getmany:
records = result
break
else:
# We should still release to the event loop
await self.sleep(1)
Expand Down