🚀 Smart Queue Processor

A high-performance, backend-agnostic asynchronous queue processing engine built in Python.

This project demonstrates advanced resiliency patterns, including non-blocking exponential backoff, dead-letter routing, and graceful shutdowns, all visualizable through a stunning real-time WebSocket dashboard.

---

🏗️ Architecture

The system is built on an Adapter Factory Pattern, decoupling the core processing logic from the underlying message broker. This allows you to hot-swap between Redis Streams and RabbitMQ seamlessly.

graph TD
    %% Producers
    UI[FastAPI Web Dashboard] -->|REST / WebSockets| Producer[API Producer]
    Producer -->|Backend=Redis| RedisAdapter(Redis Adapter)
    Producer -->|Backend=RabbitMQ| RabbitMQAdapter(RabbitMQ Adapter)

    %% Brokers
    subgraph Message Brokers
        Redis[(Redis Streams)]
        RMQ[(RabbitMQ)]
    end

    RedisAdapter -.->|Publish| Redis
    RabbitMQAdapter -.->|Publish| RMQ

    %% Consumers
    subgraph Consumer Worker 1
        R_Adapter[Redis Adapter] -->|Subscribe| StreamProcessor1{Stream Processor}
    end
    
    subgraph Consumer Worker 2
        RMQ_Adapter[RabbitMQ Adapter] -->|Subscribe| StreamProcessor2{Stream Processor}
    end

    Redis -.->|Consume| R_Adapter
    RMQ -.->|Consume| RMQ_Adapter

    %% Resiliency Logic
    StreamProcessor1 -->|Success| R_Ack((ACK))
    StreamProcessor1 -->|Fail| R_Retry([Asyncio Sleep / Retry])
    R_Retry -->|Max Retries Exceeded| R_DLQ[(Redis DLQ Stream)]
    
    StreamProcessor2 -->|Success| RMQ_Ack((ACK))
    StreamProcessor2 -->|Fail| RMQ_Retry([Asyncio Sleep / Retry])
    RMQ_Retry -->|Max Retries Exceeded| RMQ_DLQ[(RabbitMQ DLQ)]
    
    %% Metrics
    StreamProcessor1 ==>|Live Metrics| RedisMetrics[(Redis Hash)]
    StreamProcessor2 ==>|Live Metrics| RedisMetrics
    RedisMetrics -.->|ws:// stream| UI

Loading

---

✨ Key Features

• Backend Agnosticism: Standardized BaseQueueAdapter interface. Easily switch between RabbitMQ and Redis Streams.
• Non-Blocking Processing: Uses asyncio.create_task to handle retries asynchronously, preventing a single failing message from halting the entire consumption loop.
• Application-Level DLQ: A unified Dead-Letter Queue implementation that behaves identically across all backends.
• Real-Time Observability: A FastAPI-served dashboard connects via WebSockets to stream metrics (Received, Success, Retries, DLQ) live.
• Interactive Chaos Testing: UI controls allow you to pump high-throughput traffic, inject explicit "Poison Pills" to trigger failures, and safely purge entire queue states.

---

🛠️ Quickstart Guide

1. Prerequisites

Ensure you have Docker and Python installed. Spin up the underlying infrastructure using the provided compose file:

docker-compose up -d

This starts Redis on port 6379 and RabbitMQ on port 5672 (Management UI on 15672).

2. Environment Setup

Create a virtual environment and install dependencies:

python3 -m venv myvenv
source myvenv/bin/activate
pip install -r requirements.txt

3. Run the Demo

You will need 3 terminal windows to see the full dual-backend demo.

Terminal 1: Start the Dashboard & Producer API

export PYTHONPATH=$(pwd)
fastapi dev smart_queue_processor/app/producer/main.py

Terminal 2: Start the Redis Consumer

export PYTHONPATH=$(pwd)
QUEUE_BACKEND=redis python3 smart_queue_processor/app/consumer/main.py

Terminal 3: Start the RabbitMQ Consumer

export PYTHONPATH=$(pwd)
QUEUE_BACKEND=rabbitmq python3 smart_queue_processor/app/consumer/main.py

---

🎮 Using the Dashboard

Once all services are running, open your browser to http://127.0.0.1:8000.

1. Start Traffic: Click this to unleash 10 messages per second into the respective backend.
2. Send Poison Pill: Injects a single payload with "simulate_crash": true. Watch your terminal logs for the exponential backoff (2s, 4s, 8s), and watch the Retries and DLQ cards light up.
3. Reset Entire State: Instantly purges all messages from both main queues and DLQs, and zeroes out the Redis metrics.

🗂️ Project Structure

smart_queue_processor/
├── app/
│   ├── adapters/
│   │   ├── base.py            # Abstract adapter interface
│   │   ├── factory.py         # Dynamic backend loader
│   │   ├── rabbitmq_adapter.py
│   │   └── redis_adapter.py
│   ├── consumer/              # Graceful shutdown worker script
│   ├── producer/              # FastAPI app, Dashboard, WebSockets
│   ├── config.py              # Pydantic environment configuration
│   ├── metrics.py             # Redis-backed live counters
│   ├── models.py              # Pydantic schemas (QueueMessage)
│   └── processor.py           # Core StreamProcessor & Resiliency Logic
├── docker-compose.yml         # Infrastructure setup
└── requirements.txt