fix: replace with simple acquire

Author: Anton

taskiq/receiver/receiver.py

@@ -14,7 +14,6 @@
 from taskiq.message import TaskiqMessage
 from taskiq.receiver.params_parser import parse_params
 from taskiq.result import TaskiqResult
-from taskiq.semaphore import DequeSemaphore
 from taskiq.state import TaskiqState
 from taskiq.utils import DequeQueue, maybe_awaitable
 
@@ -59,15 +58,15 @@ def __init__(  # noqa: WPS211
             self.task_signatures[task.task_name] = inspect.signature(task.original_func)
             self.task_hints[task.task_name] = get_type_hints(task.original_func)
             self.dependency_graphs[task.task_name] = DependencyGraph(task.original_func)
-        self.sem: "Optional[DequeSemaphore]" = None
+        self.sem: "Optional[asyncio.Semaphore]" = None
         if max_async_tasks is not None and max_async_tasks > 0:
-            self.sem = DequeSemaphore(max_async_tasks)
+            self.sem = asyncio.Semaphore(max_async_tasks)
         else:
             logger.warning(
                 "Setting unlimited number of async tasks "
                 + "can result in undefined behavior",
             )
-        self.sem_prefetch = DequeSemaphore(max_prefetch)
+        self.sem_prefetch = asyncio.Semaphore(max_prefetch)
         self.queue: DequeQueue[bytes] = DequeQueue()
 
         self.sem_idle: Optional[asyncio.Semaphore] = None
@@ -310,7 +309,7 @@ def task_cb(task: "asyncio.Task[Any]") -> None:
                 break
             if message is QUEUE_SKIP:
                 # Decrease max_prefetch
-                prefetch_dec = asyncio.create_task(self.sem_prefetch.acquire_first())
+                prefetch_dec = asyncio.create_task(self.sem_prefetch.acquire())
                 prefetch_dec.add_done_callback(tasks.discard)
                 tasks.add(prefetch_dec)
 
@@ -357,5 +356,5 @@ async def task_idler(self, wait: float) -> None:
             # Decrease max_prefetch in runner
             task = asyncio.create_task(self.queue.put_first(QUEUE_SKIP))
             # Decrease max_tasks
-            await self.sem.acquire_first()
+            await self.sem.acquire()
             await task

taskiq/semaphore.py

@@ -308,7 +308,6 @@ async def task_map(vals: List[int]) -> List[int]:
 
         return [r.return_value for r in resps]
 
-    await broker.startup()
     receiver = get_receiver(broker, max_async_tasks=1)
     listen_task = asyncio.create_task(receiver.listen())
 
@@ -338,7 +337,6 @@ async def task_map(vals: List[int], ctx: Context = Depends()) -> List[int]:
         res = [r.return_value for r in resps]
         return res
 
-    await broker.startup()
     receiver = get_receiver(broker, max_async_tasks=1, max_idle_tasks=1)
     listen_task = asyncio.create_task(receiver.listen())