feat: implement asynchronous Kafka and Redis pub/sub mechanisms with message handling

Author: npv2k1

src/modules/transporter/kafka.py

@@ -0,0 +1,73 @@
+import asyncio
+from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
+from functools import wraps
+from typing import Callable, Dict
+
+
+class AsyncKafkaPubSub:
+    def __init__(self, bootstrap_servers="ubuntu:9092", group_id="default-py-group"):
+        self.bootstrap_servers = bootstrap_servers
+        self.group_id = group_id
+        self.handlers: Dict[str, Callable[[str], None]] = {}
+        self._consumer_started = False
+        self._lock = asyncio.Lock()
+        self._producer = None
+
+    def subscribe(self, topic: str):
+        print(f"Subscribing to topic: {topic}")
+        def decorator(func: Callable[[str], None]):
+            self.handlers[topic] = func
+            asyncio.create_task(self._ensure_consumer())
+            return func
+        return decorator
+
+    async def publish(self, topic: str, message: str):
+        if not self._producer:
+            self._producer = AIOKafkaProducer(bootstrap_servers=self.bootstrap_servers)
+            await self._producer.start()
+        await self._producer.send_and_wait(topic, message.encode())
+
+    async def _ensure_consumer(self):
+        async with self._lock:
+            if self._consumer_started:
+                return
+            self._consumer_started = True
+            await self._start_consumer()
+
+    async def _start_consumer(self):
+        consumer = AIOKafkaConsumer(
+            *self.handlers.keys(),
+            bootstrap_servers=self.bootstrap_servers,
+            group_id=self.group_id,
+            auto_offset_reset='latest',
+        )
+        await consumer.start()
+        print(f"Subscribed to: {', '.join(self.handlers.keys())}")
+
+        async def listen():
+            try:
+                async for msg in consumer:
+                    print(f"Received message on topic '{msg.topic}': {msg.value.decode()}")
+                    topic = msg.topic
+                    data = msg.value.decode()
+                    handler = self.handlers.get(topic)
+                    if handler:
+                        await self._maybe_async(handler, data)
+            finally:
+                await consumer.stop()
+
+        asyncio.create_task(listen())
+
+    async def _maybe_async(self, func: Callable, *args, **kwargs):
+        result = func(*args, **kwargs)
+        if asyncio.iscoroutine(result):
+            await result
+
+
+kafka_pubsub = AsyncKafkaPubSub()
+
+
+@kafka_pubsub.subscribe("chat")
+def handle_chat(msg):
+    print(f"[chat kafka] {msg}")
+

src/modules/transporter/rabbitmq.py

@@ -2,6 +2,7 @@
 import threading
 
 credentials = pika.PlainCredentials('guest', 'guest')
+
 parameters = pika.ConnectionParameters(
     host='ubuntu',
     port=5672,  # default RabbitMQ port
@@ -39,6 +40,12 @@ def process_message(msg):
     print(f"[x] Received: {msg}")
 
 
+@queue_listener("test")
+def test_queue_listener(msg):
+    # This function is just for testing the queue listener
+    print(f"[test] Received message: {msg}")
+
+
 # publish a message to the queue
 def publish_message(queue_name, message):
     connection = pika.BlockingConnection(parameters)

src/modules/transporter/redis_client.py

@@ -0,0 +1,79 @@
+
+import redis
+
+# Connect to Redis
+r = redis.Redis(host='ubuntu', port=6379, decode_responses=True)
+
+# Set a value
+r.set('foo', 'bar')
+
+# Get a value
+value = r.get('foo')
+print(value)  # Output: bar
+
+
+print("Redis client is running...", r.ping())
+
+
+import asyncio
+from redis.asyncio import Redis
+from functools import wraps
+from typing import Callable, Dict
+
+
+class AsyncRedisPubSub:
+    def __init__(self, host="ubuntu", port=6379, db=0):
+        self.redis = Redis(host=host, port=port, db=db)
+        self.channel_handlers: Dict[str, Callable[[str], None]] = {}
+        self._subscriber_started = False
+        self._lock = asyncio.Lock()
+
+    def subscribe(self, channel: str):
+        def decorator(func: Callable[[str], None]):
+            self.channel_handlers[channel] = func
+            asyncio.create_task(self._ensure_subscriber())
+            return func
+        return decorator
+
+    async def publish(self, channel: str, message: str):
+        await self.redis.publish(channel, message)
+
+    async def _ensure_subscriber(self):
+        async with self._lock:
+            if self._subscriber_started:
+                return
+            self._subscriber_started = True
+            await self._start_subscriber()
+
+    async def _start_subscriber(self):
+        pubsub = self.redis.pubsub()
+        await pubsub.subscribe(*self.channel_handlers.keys())
+        print(f"Subscribed to: {', '.join(self.channel_handlers.keys())}")
+
+        async def listen():
+            async for message in pubsub.listen():
+                if message["type"] != "message":
+                    continue
+                channel = message["channel"].decode()
+                data = message["data"].decode()
+                handler = self.channel_handlers.get(channel)
+                if handler:
+                    await self._maybe_async(handler, data)
+
+        asyncio.create_task(listen())
+
+    async def _maybe_async(self, func: Callable, *args, **kwargs):
+        result = func(*args, **kwargs)
+        if asyncio.iscoroutine(result):
+            await result
+
+
+pubsub = AsyncRedisPubSub()
+
+@pubsub.subscribe("chat")
+def handle_chat(msg):
+    print(f"[chat redis] Received: {msg}")
+
+@pubsub.subscribe("news")
+def handle_news(msg):
+    print(f"[news] Breaking: {msg}")