Arya Pub-Sub System

🚀 Features

🏗️ Architecture

┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ WebSocket │ │ PubSubManager │ │ REST API │
│ Handler │◄──►│ (Thread-Safe) │◄──►│ Endpoints │
│ │ │ │ │ │
│ • Connection │ │ • Topic Storage │ │ • Topic CRUD │
│ • Message Route │ │ • Subscriptions │ │ • Health Check │
│ • Broadcasting │ │ • Message Queue │ │ • Statistics │
└─────────────────┘ └──────────────────┘ └─────────────────┘

Core Components

Project Structure

arya/
├── src/arya/ # Main application code
│ ├── api/ # REST API endpoints
│ ├── core/ # Core business logic
│ │ ├── models.py # Pydantic data models
│ │ └── pubsub.py # PubSub manager
│ ├── websocket/ # WebSocket handlers
│ ├── config/ # Configuration management
│ ├── exceptions/ # Custom exceptions
│ └── utils/ # Utility functions
├── tests/ # Test suite
├── simple_test.py # Contract compliance test
├── test_client.py # Interactive demo client
├── docker-compose.yml # Docker setup
└── requirements.txt # Dependencies

🚦 Quick Start

Using Docker (Recommended)

# Build and run with Docker
docker build -t arya-pubsub .
docker run -p 8000:8000 arya-pubsub
# Or use docker-compose
docker-compose up --build

Local Development

# Install dependencies
pip install -r requirements.txt
# Run the server
PYTHONPATH=src python src/main.py
# Or with uvicorn directly (development with auto-reload)
PYTHONPATH=src uvicorn arya.api.main:create_app --factory --host 0.0.0.0 --port 8000 --reload

📡 API Documentation

REST Endpoints

Create Topic

curl -X POST "http://localhost:8000/topics" \
 -H "Content-Type: application/json" \
 -d '{"name": "orders"}'
# Response: {"status": "created", "topic": "orders"}

Delete Topic

curl -X DELETE "http://localhost:8000/topics/orders"
# Response: {"status": "deleted", "topic": "orders"}

List Topics

curl "http://localhost:8000/topics"
# Response: {
# "topics": [
# {"name": "orders", "subscribers": 3, "messages": 42}
# ]
# }

Health Check

curl "http://localhost:8000/health"
# Response: {
# "uptime_sec": 3600,
# "topics": 5,
# "subscribers": 15
# }

System Statistics

curl "http://localhost:8000/stats"
# Response: {
# "topics": {
# "orders": {"messages": 42, "subscribers": 3},
# "notifications": {"messages": 15, "subscribers": 7}
# }
# }

WebSocket Protocol

Message Format

{
 "type": "message_type",
 "request_id": "optional_correlation_id",
 // ... additional fields based on message type
}

Subscribe to Topic

{
 "type": "subscribe",
 "topic": "orders",
 "client_id": "unique-client-123",
 "last_n": 5,
 "request_id": "sub-1"
}

{
 "type": "ack",
 "message": "Subscribed to topic 'orders'",
 "request_id": "sub-1",
 "timestamp": "2024年01月15日T10:30:00Z"
}

Unsubscribe from Topic

{
 "type": "unsubscribe",
 "topic": "orders",
 "client_id": "unique-client-123",
 "request_id": "unsub-1"
}

Publish Message

{
 "type": "publish",
 "topic": "orders",
 "message": {
 "order_id": 12345,
 "customer_id": "cust-789",
 "amount": 99.99,
 "items": ["item1", "item2"]
 },
 "request_id": "pub-1"
}

Receive Published Messages

{
 "type": "event",
 "topic": "orders",
 "message": {
 "id": "msg-uuid-here",
 "payload": {
 "order_id": 12345,
 "customer_id": "cust-789",
 "amount": 99.99
 },
 "timestamp": "2024年01月15日T10:30:00Z"
 }
}

Heartbeat (Ping/Pong)

// Send ping
{"type": "ping", "request_id": "ping-1"}
// Receive pong
{"type": "pong", "request_id": "ping-1", "timestamp": "2024年01月15日T10:30:00Z"}

Error Responses

{
 "type": "error",
 "error_code": "TOPIC_NOT_FOUND",
 "error_message": "Topic 'nonexistent' not found",
 "request_id": "sub-1",
 "timestamp": "2024年01月15日T10:30:00Z"
}

🧪 Testing

Run Tests

# Create and activate virtual environment (if not already done)
python3 -m venv venv
source venv/bin/activate
# Install dependencies
pip install -r requirements.txt
# Run all tests
PYTHONPATH=src python -m pytest tests/ -v
# Run with coverage (if pytest-cov is installed)
PYTHONPATH=src python -m pytest tests/ --cov=src --cov-report=html
# Run specific test categories
PYTHONPATH=src python -m pytest tests/test_models.py tests/test_pubsub.py -v # Unit tests
PYTHONPATH=src python -m pytest tests/test_api.py tests/test_websocket.py -v # Integration tests

Test Categories

Manual Testing

Basic Contract Compliance Test

# Start the service first (in one terminal):
source venv/bin/activate
PYTHONPATH=src python src/main.py
# Then run the test (in another terminal):
source venv/bin/activate
python simple_test.py

Interactive WebSocket Demo

# Start the service first (in one terminal):
source venv/bin/activate
PYTHONPATH=src python src/main.py
# Then run the test client (in another terminal):
source venv/bin/activate
python test_client.py

🔧 Configuration

Environment Variables

Performance Tuning

🚀 Production Deployment

Docker Production Setup

# Use multi-stage build for smaller image
FROM python:3.11-slim as builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --user -r requirements.txt
FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY . .
RUN useradd --create-home --shell /bin/bash arya
USER arya
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Kubernetes Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
 name: arya-pubsub
spec:
 replicas: 3
 selector:
 matchLabels:
 app: arya-pubsub
 template:
 metadata:
 labels:
 app: arya-pubsub
 spec:
 containers:
 - name: arya-pubsub
 image: arya-pubsub:latest
 ports:
 - containerPort: 8000
 env:
 - name: LOG_LEVEL
 value: "INFO"
 resources:
 requests:
 memory: "256Mi"
 cpu: "250m"
 limits:
 memory: "512Mi"
 cpu: "500m"

Load Balancing

📊 Monitoring

Health Checks

# Basic health check
curl http://localhost:8000/health
# Detailed statistics
curl http://localhost:8000/stats

Metrics to Monitor

Logging

# Critical operations
logger.info("Client subscribed to topic")
# Debug information 
logger.debug("Message added to topic history")
# Errors and warnings
logger.error("Failed to broadcast message")

🔒 Security Considerations

Current Implementation

Production Recommendations

🚨 Limitations & Design Decisions

Current Limitations

Design Decisions

🛠️ Development

Project Structure

arya/
├── main.py # FastAPI application
├── models.py # Pydantic data models
├── pubsub.py # Core pub-sub manager
├── websocket_handler.py # WebSocket connection management
├── requirements.txt # Python dependencies
├── Dockerfile # Container configuration
├── docker-compose.yml # Local development setup
├── pytest.ini # Test configuration
├── tests/ # Test suite
│ ├── conftest.py # Test fixtures
│ ├── test_models.py # Model validation tests
│ ├── test_pubsub.py # Core logic tests
│ ├── test_api.py # REST API tests
│ └── test_websocket.py # WebSocket tests
└── README.md # This file

Adding Features

Contributing

📝 Example Usage

Python WebSocket Client

import asyncio
import websockets
import json
async def client_example():
 uri = "ws://localhost:8000/ws"
 
 async with websockets.connect(uri) as websocket:
 # Subscribe to topic
 subscribe_msg = {
 "type": "subscribe",
 "topic": "orders",
 "client_id": "python-client-1",
 "last_n": 5
 }
 await websocket.send(json.dumps(subscribe_msg))
 
 # Receive acknowledgment
 response = await websocket.recv()
 print(f"Subscribe response: {response}")
 
 # Publish a message
 publish_msg = {
 "type": "publish",
 "topic": "orders",
 "message": {"order_id": 123, "amount": 99.99}
 }
 await websocket.send(json.dumps(publish_msg))
 
 # Listen for events
 while True:
 message = await websocket.recv()
 data = json.loads(message)
 if data["type"] == "event":
 print(f"Received: {data['message']['payload']}")
# Run the client
asyncio.run(client_example())

JavaScript WebSocket Client

const ws = new WebSocket('ws://localhost:8000/ws');
ws.onopen = function() {
 // Subscribe to topic
 const subscribeMsg = {
 type: 'subscribe',
 topic: 'orders',
 client_id: 'js-client-1',
 last_n: 10
 };
 ws.send(JSON.stringify(subscribeMsg));
};
ws.onmessage = function(event) {
 const data = JSON.parse(event.data);
 
 if (data.type === 'event') {
 console.log('Received message:', data.message.payload);
 } else if (data.type === 'ack') {
 console.log('Operation acknowledged:', data.message);
 } else if (data.type === 'error') {
 console.error('Error:', data.error_message);
 }
};
// Publish message
function publishMessage(orderData) {
 const publishMsg = {
 type: 'publish',
 topic: 'orders',
 message: orderData
 };
 ws.send(JSON.stringify(publishMsg));
}
// Example usage
publishMessage({order_id: 456, customer: 'john@example.com', amount: 149.99});

📄 License

🤝 Support