Thread changes

Author: chandr-andr

README.md

@@ -24,7 +24,9 @@ You can send delayed messages and set priorities to messages using labels.
 
 To send delayed message, you have to specify
 delay label. You can do it with `task` decorator,
-or by using kicker. For example:
+or by using kicker.  
+In this type of delay we are using additional queue with `expiration` parameter and after with time message will be deleted from `delay` queue and sent to the main taskiq queue.
+For example:
 
 ```python
 broker = AioPikaBroker()
@@ -56,7 +58,10 @@ To send delayed message you can install `rabbitmq-delayed-message-exchange`
 plugin https://github.com/rabbitmq/rabbitmq-delayed-message-exchange.
 
 And you need to configure you broker.
-There is `delayed_message_exchange_plugin` `AioPikaBroker` parameter and it must be `True` to turn on delayed message functionality. For example:
+There is `delayed_message_exchange_plugin` `AioPikaBroker` parameter and it must be `True` to turn on delayed message functionality.  
+
+The delay plugin can handle tasks with different delay times well, and the delay based on dead letter queue is suitable for tasks with the same delay time.  
+For example:
 
 ```python
 broker = AioPikaBroker(

taskiq_aio_pika/broker.py

@@ -252,22 +252,21 @@ async def kick(self, message: BrokerMessage) -> None:
                 ensure=False,
             )
             await exchange.publish(rmq_message, routing_key=message.task_name)
+        elif self._delayed_message_exchange_plugin:
+            rmq_message.headers["x-delay"] = delay * 1000
+            exchange = await self.write_channel.get_exchange(
+                self._delay_plugin_exchange_name,
+            )
+            await exchange.publish(
+                rmq_message,
+                routing_key=self._routing_key,
+            )
         else:
-            if self._delayed_message_exchange_plugin:
-                rmq_message.headers["x-delay"] = delay * 1000
-                exchange = await self.write_channel.get_exchange(
-                    self._delay_plugin_exchange_name,
-                )
-                await exchange.publish(
-                    rmq_message,
-                    routing_key=self._routing_key,
-                )
-            else:
-                rmq_message.expiration = timedelta(seconds=delay)
-                await self.write_channel.default_exchange.publish(
-                    rmq_message,
-                    routing_key=self._delay_queue_name,
-                )
+            rmq_message.expiration = timedelta(seconds=delay)
+            await self.write_channel.default_exchange.publish(
+                rmq_message,
+                routing_key=self._delay_queue_name,
+            )
 
     async def listen(self) -> AsyncGenerator[bytes, None]:
         """