update files 4 lesson

Author: RivanDev

Diff for: emit_log.py

@@ -11,13 +11,14 @@ async def main() -> None:
     async with connection:
         channel = await connection.channel()
 
-        logs_exchange = await channel.declare_exchange(
-            "logs",
-            ExchangeType.FANOUT,
+        topic_logs_exchange = await channel.declare_exchange(
+            "topic_logs",
+            ExchangeType.TOPIC,
         )
+        routing_key = sys.argv[1] if len(sys.argv) > 2 else "anonymous.info"
 
         message_body = (
-            b" ".join(arg.encode() for arg in sys.argv[1:]) or b"Hello World!"
+            b" ".join(arg.encode() for arg in sys.argv[2:]) or b"Hello World!"
         )
 
         message = Message(
@@ -26,7 +27,7 @@ async def main() -> None:
         )
 
         # Sending the message
-        await logs_exchange.publish(message, routing_key="info")
+        await topic_logs_exchange.publish(message, routing_key=routing_key)
 
         print(f" [x] Sent {message!r}")
 

Diff for: receive_logs.py

@@ -1,39 +1,43 @@
 import asyncio
+import sys
 
 from aio_pika import ExchangeType, connect
 from aio_pika.abc import AbstractIncomingMessage
 
 
-async def on_message(message: AbstractIncomingMessage) -> None:
-    async with message.process():
-        print(f"[x] {message.body!r}")
-
-
 async def main() -> None:
     # Perform connection
     connection = await connect("amqp://guest:guest@localhost/")
 
-    async with connection:
-        # Creating a channel
-        channel = await connection.channel()
-        await channel.set_qos(prefetch_count=1)
+    channel = await connection.channel()
+    await channel.set_qos(prefetch_count=1)
+
+    topic_logs_exchange = await channel.declare_exchange(
+        "topic_logs",
+        ExchangeType.TOPIC,
+    )
+
+    queue = await channel.declare_queue(
+        "task_queue",
+        durable=True,
+    )
 
-        logs_exchange = await channel.declare_exchange(
-            "logs",
-            ExchangeType.FANOUT,
-        )
+    binding_keys = sys.argv[1:]
 
-        # Declaring queue
-        queue = await channel.declare_queue(exclusive=True)
+    if not binding_keys:
+        sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
+        sys.exit(1)
 
-        # Binding the queue to the exchange
-        await queue.bind(logs_exchange)
+    for binding_key in binding_keys:
+        await queue.bind(topic_logs_exchange, routing_key=binding_key)
 
-        # Start listening the queue
-        await queue.consume(on_message)
+    print(" [*] Waiting for messages. To exit press CTRL+C")
 
-        print(" [*] Waiting for logs. To exit press CTRL+C")
-        await asyncio.Future()
+    async with queue.iterator() as iterator:
+        message: AbstractIncomingMessage
+        async for message in iterator:
+            async with message.process():
+                print(f" [x] {message.routing_key!r}:{message.body!r}")
 
 
 if __name__ == "__main__":