initial files

README.md

@@ -1,11 +1,74 @@
----
-title: Async Framework Test
-emoji: 📈
-colorFrom: gray
-colorTo: purple
-sdk: docker
-pinned: false
-short_description: to test the pub/sub
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Redis Message Queue Test Project
+
+This project demonstrates the usage of the async-framework library with multiple queues, publishers, and consumers.
+
+## Project Structure
+- `config.py` - Configuration for Redis and broker settings
+- `message_generators.py` - Message generators for different queue types
+- `publishers.py` - Publishers for different queues
+- `consumers.py` - Consumers with message handlers
+
+## Setup
+
+1. Install Redis:
+   - Windows: Download from [Redis Windows](https://github.com/microsoftarchive/redis/releases)
+   - Linux: `sudo apt-get install redis-server`
+   - MacOS: `brew install redis`
+
+2. Create and activate virtual environment:
+```bash
+python -m venv venv
+.\venv\Scripts\activate  # Windows
+source venv/bin/activate  # Linux/MacOS
+```
+
+3. Install dependencies:
+```bash
+pip install async-framework structlog
+```
+
+4. Start Redis server:
+```bash
+redis-server
+```
+
+## Running the Test
+
+1. Start the consumers:
+```bash
+python consumers.py
+```
+
+2. In a new terminal, start the publishers:
+```bash
+python publishers.py
+```
+
+## Message Types
+
+1. Orders Queue:
+   - Order details with ID, customer, amount
+   - Simulated processing time: 0.5s
+   - Fails for orders > $900
+
+2. Notifications Queue:
+   - Notification details with type and priority
+   - Simulated processing time: 0.3s
+   - High priority notifications fail twice before success
+
+3. Analytics Queue:
+   - Event tracking data
+   - Simulated processing time: 0.1s
+   - No simulated failures
+
+## Monitoring
+
+Access Prometheus metrics at:
+```
+http://localhost:8000
+```
+
+Available metrics:
+- messages_published_total
+- messages_processed_total
+- message_processing_seconds

consumers.py

@@ -0,0 +1,83 @@
+from async_framework import MessageConsumer, Message
+import time
+import structlog
+from config import BROKER_CONFIG
+
+logger = structlog.get_logger()
+
+def process_order(message: Message):
+    logger.info(
+        "Processing order",
+        message_id=message.id,
+        order_id=message.payload.get("order_id"),
+        amount=message.payload.get("amount")
+    )
+    # Simulate processing time
+    time.sleep(0.5)
+    # Simulate occasional failures
+    if message.payload.get("amount", 0) > 900:
+        raise ValueError("Order amount too high")
+
+def process_notification(message: Message):
+    logger.info(
+        "Processing notification",
+        message_id=message.id,
+        notification_type=message.payload.get("type"),
+        priority=message.payload.get("priority")
+    )
+    # Simulate processing time
+    time.sleep(0.3)
+    # Simulate occasional failures for high priority notifications
+    if message.payload.get("priority") == "high":
+        if message.retry_count < 2:  # Will succeed on third try
+            raise ValueError("Failed to send high priority notification")
+
+def process_analytics(message: Message):
+    logger.info(
+        "Processing analytics event",
+        message_id=message.id,
+        event_type=message.payload.get("event_type"),
+        user_id=message.payload.get("user_id")
+    )
+    # Simulate processing time
+    time.sleep(0.1)
+
+def main():
+    # Create consumers for different queues
+    consumers = [
+        MessageConsumer(
+            config=BROKER_CONFIG,
+            queue="orders",
+            handler=process_order
+        ),
+        MessageConsumer(
+            config=BROKER_CONFIG,
+            queue="notifications",
+            handler=process_notification
+        ),
+        MessageConsumer(
+            config=BROKER_CONFIG,
+            queue="analytics",
+            handler=process_analytics
+        )
+    ]
+
+    try:
+        # Start all consumers
+        for consumer in consumers:
+            consumer.start()
+            logger.info(
+                "Started consumer",
+                queue=consumer.queue
+            )
+
+        # Keep the main thread running
+        while True:
+            time.sleep(1)
+    except KeyboardInterrupt:
+        logger.info("Shutting down consumers")
+        for consumer in consumers:
+            consumer.stop()
+
+if __name__ == "__main__":
+    main()

message_generators.py

@@ -0,0 +1,40 @@
+from abc import ABC, abstractmethod
+from typing import Dict, Any
+import random
+import time
+
+class MessageGenerator(ABC):
+    @abstractmethod
+    def generate_message(self) -> Dict[str, Any]:
+        pass
+
+class OrderMessageGenerator(MessageGenerator):
+    def generate_message(self) -> Dict[str, Any]:
+        order_id = f"ORD-{random.randint(1000, 9999)}"
+        return {
+            "order_id": order_id,
+            "customer_id": f"CUST-{random.randint(100, 999)}",
+            "amount": round(random.uniform(10.0, 1000.0), 2),
+            "items": random.randint(1, 10),
+            "timestamp": time.time()
+        }
+
+class NotificationMessageGenerator(MessageGenerator):
+    def generate_message(self) -> Dict[str, Any]:
+        return {
+            "notification_id": f"NOTIF-{random.randint(1000, 9999)}",
+            "user_id": f"USER-{random.randint(100, 999)}",
+            "type": random.choice(["email", "sms", "push"]),
+            "priority": random.choice(["high", "medium", "low"]),
+            "timestamp": time.time()
+        }
+
+class AnalyticsMessageGenerator(MessageGenerator):
+    def generate_message(self) -> Dict[str, Any]:
+        return {
+            "event_id": f"EVT-{random.randint(1000, 9999)}",
+            "event_type": random.choice(["page_view", "click", "purchase", "login"]),
+            "user_id": f"USER-{random.randint(100, 999)}",
+            "session_id": f"SESSION-{random.randint(1000, 9999)}",
+            "timestamp": time.time()
+        }

publishers.py

@@ -0,0 +1,77 @@
+from async_framework import MessageBroker, Message
+import time
+import random
+from typing import Optional
+import structlog
+from config import BROKER_CONFIG
+from message_generators import (
+    OrderMessageGenerator,
+    NotificationMessageGenerator,
+    AnalyticsMessageGenerator
+)
+
+logger = structlog.get_logger()
+
+class Publisher:
+    def __init__(self, queue_name: str, message_generator, publish_interval: float = 1.0):
+        self.broker = MessageBroker(BROKER_CONFIG)
+        self.queue_name = queue_name
+        self.message_generator = message_generator
+        self.publish_interval = publish_interval
+        logger.info("Publisher initialized", queue=queue_name)
+
+    def run(self, num_messages: Optional[int] = None):
+        count = 0
+        try:
+            while num_messages is None or count < num_messages:
+                message_data = self.message_generator.generate_message()
+                message = self.broker.publish(
+                    queue=self.queue_name,
+                    payload=message_data
+                )
+                logger.info(
+                    "Message published",
+                    queue=self.queue_name,
+                    message_id=message.id
+                )
+                count += 1
+                # Random delay between 0.5x and 1.5x of publish_interval
+                time.sleep(self.publish_interval * random.uniform(0.5, 1.5))
+        except KeyboardInterrupt:
+            logger.info("Publisher stopped", queue=self.queue_name)
+        except Exception as e:
+            logger.error("Error in publisher", queue=self.queue_name, error=str(e))
+
+def main():
+    # Create publishers for different queues
+    publishers = [
+        Publisher("orders", OrderMessageGenerator(), 2.0),
+        Publisher("notifications", NotificationMessageGenerator(), 1.0),
+        Publisher("analytics", AnalyticsMessageGenerator(), 0.5)
+    ]
+
+    try:
+        # Run publishers in parallel using threads
+        import threading
+        threads = []
+        for publisher in publishers:
+            thread = threading.Thread(
+                target=publisher.run,
+                name=f"publisher-{publisher.queue_name}"
+            )
+            thread.daemon = True
+            thread.start()
+            threads.append(thread)
+            logger.info(
+                "Started publisher thread",
+                queue=publisher.queue_name
+            )
+
+        # Keep the main thread running
+        while True:
+            time.sleep(1)
+    except KeyboardInterrupt:
+        logger.info("Shutting down publishers")
+
+if __name__ == "__main__":
+    main()