diff --git a/taskiq/funcs.py b/taskiq/funcs.py index 9245c4de..13fe76e2 100644 --- a/taskiq/funcs.py +++ b/taskiq/funcs.py @@ -52,6 +52,7 @@ async def check_task(task: AsyncTaskiqTask[Any]) -> None: for task in tasks: check_tasks.append(loop.create_task(check_task(task))) await asyncio.gather(*check_tasks) - await asyncio.sleep(periodicity) + if task_ids: + await asyncio.sleep(periodicity) return tuple(results[task_id] for task_id in ordered_ids) diff --git a/tests/test_gather_ready.py b/tests/test_gather_ready.py new file mode 100644 index 00000000..08faf1b9 --- /dev/null +++ b/tests/test_gather_ready.py @@ -0,0 +1,61 @@ +import asyncio +from typing import Any + +from taskiq import AsyncTaskiqTask, TaskiqResult +from taskiq.brokers.inmemory_broker import InmemoryResultBackend +from taskiq.funcs import gather + + +async def test_ready_results_do_not_wait_for_polling_interval() -> None: + backend: InmemoryResultBackend[Any] = InmemoryResultBackend() + await backend.set_result( + "ready", + TaskiqResult(is_err=False, return_value=42, execution_time=0), + ) + task: AsyncTaskiqTask[Any] = AsyncTaskiqTask( + task_id="ready", + result_backend=backend, + ) + results = await asyncio.wait_for(gather(task, periodicity=5), timeout=0.5) + assert results[0].return_value == 42 + + +async def test_pending_results_are_still_polled() -> None: + backend: InmemoryResultBackend[Any] = InmemoryResultBackend() + task: AsyncTaskiqTask[Any] = AsyncTaskiqTask( + task_id="later", + result_backend=backend, + ) + + async def publish() -> None: + await asyncio.sleep(0) + await backend.set_result( + "later", + TaskiqResult(is_err=False, return_value=7, execution_time=0), + ) + + publisher = asyncio.create_task(publish()) + try: + results = await asyncio.wait_for(gather(task, periodicity=0.01), timeout=1) + assert results[0].return_value == 7 + finally: + await publisher + + +async def test_empty_gather_still_returns_an_empty_tuple() -> None: + assert await gather(periodicity=5) == () + + +async def test_gather_preserves_order_and_duplicate_tasks() -> None: + backend: InmemoryResultBackend[Any] = InmemoryResultBackend() + tasks = [] + for task_id, value in [("first", 1), ("second", 2)]: + await backend.set_result( + task_id, + TaskiqResult(is_err=False, return_value=value, execution_time=0), + ) + tasks.append( + AsyncTaskiqTask[Any](task_id=task_id, result_backend=backend), + ) + results = await gather(tasks[1], tasks[0], tasks[1], periodicity=0) + assert [result.return_value for result in results] == [2, 1, 2]