Retry queue for inbound queue

Author: harishkannarao

app/config.py

@@ -40,6 +40,10 @@ class Settings(BaseSettings):
     app_rabbit_inbound_exchange: str = "in-bound-ex"
     app_rabbit_inbound_routing_key: str = "in-bound-rk"
 
+    app_rabbit_inbound_retry_queue: str = "in-bound-retry-queue"
+    app_rabbit_inbound_retry_exchange: str = "in-bound-retry-ex"
+    app_rabbit_inbound_retry_routing_key: str = "in-bound-retry-rk"
+
     app_rabbit_outbound_queue: str = "out-bound-queue"
     app_rabbit_outbound_exchange: str = "out-bound-ex"
     app_rabbit_outbound_routing_key: str = "out-bound-rk"

app/rabbit_mq_setup/rabbit_mq_initialisation.py

@@ -37,7 +37,7 @@ async def configure_rabbitmq():
         )
 
         await inbound_queue.bind(
-            inbound_exchange, routing_key=settings.app_rabbit_inbound_exchange
+            inbound_exchange, routing_key=settings.app_rabbit_inbound_routing_key
         )
 
         inbound_dead_queue = await channel.declare_queue(
@@ -56,6 +56,42 @@ async def configure_rabbitmq():
             routing_key=f"{settings.app_rabbit_inbound_queue}.dlq.rk",
         )
 
+        inbound_retry_queue = await channel.declare_queue(
+            settings.app_rabbit_inbound_retry_queue,
+            durable=settings.app_rabbit_mq_durable,
+            arguments={
+                "x-dead-letter-exchange": f"{settings.app_rabbit_inbound_retry_queue}.dlx",
+                "x-dead-letter-routing-key": f"{settings.app_rabbit_inbound_retry_queue}.dlq.rk",
+            },
+        )
+
+        inbound_retry_exchange = await channel.declare_exchange(
+            settings.app_rabbit_inbound_retry_exchange,
+            type=aio_pika.ExchangeType.TOPIC,
+            durable=settings.app_rabbit_mq_durable,
+        )
+
+        await inbound_retry_queue.bind(
+            inbound_retry_exchange,
+            routing_key=settings.app_rabbit_inbound_retry_exchange,
+        )
+
+        inbound_retry_dead_queue = await channel.declare_queue(
+            f"{settings.app_rabbit_inbound_retry_queue}.dlq",
+            durable=settings.app_rabbit_mq_durable,
+        )
+
+        inbound_retry_dead_exchange = await channel.declare_exchange(
+            f"{settings.app_rabbit_inbound_retry_queue}.dlx",
+            type=aio_pika.ExchangeType.TOPIC,
+            durable=settings.app_rabbit_mq_durable,
+        )
+
+        await inbound_retry_dead_queue.bind(
+            inbound_retry_dead_exchange,
+            routing_key=f"{settings.app_rabbit_inbound_retry_queue}.dlq.rk",
+        )
+
         outbound_queue = await channel.declare_queue(
             settings.app_rabbit_outbound_queue,
             durable=settings.app_rabbit_mq_durable,
@@ -72,7 +108,7 @@ async def configure_rabbitmq():
         )
 
         await outbound_queue.bind(
-            outbound_exchange, routing_key=settings.app_rabbit_outbound_exchange
+            outbound_exchange, routing_key=settings.app_rabbit_outbound_routing_key
         )
 
         outbound_dead_queue = await channel.declare_queue(