Message Broker Communication — Async Microservices Messaging

# brokers_comparison.py
# Message broker comparison

def compare_brokers():
    print("Message Broker Comparison")
    print("=" * 40)
    print()
    
    brokers = [
        {
            "name": "RabbitMQ",
            "model": "Queue / Exchange",
            "persistence": "Disk + memory",
            "ordering": "Per-queue",
            "best_for": "Task queues, RPC, complex routing"
        },
        {
            "name": "Apache Kafka",
            "model": "Log / Topic",
            "persistence": "Disk (configurable retention)",
            "ordering": "Per-partition",
            "best_for": "Event streaming, log aggregation, data pipelines"
        },
        {
            "name": "AWS SQS",
            "model": "Queue",
            "persistence": "Disk (up to 14 days)",
            "ordering": "Best-effort (FIFO available)",
            "best_for": "Simple queues, AWS ecosystem, serverless"
        },
        {
            "name": "Redis Pub/Sub",
            "model": "Pub/Sub channels",
            "persistence": "None (in-memory)",
            "ordering": "FIFO per subscriber",
            "best_for": "Real-time notifications, ephemeral messages"
        },
    ]
    
    for b in brokers:
        print(f"{b['name']:20s}")
        print(f"  Model:         {b['model']}")
        print(f"  Persistence:   {b['persistence']}")
        print(f"  Ordering:      {b['ordering']}")
        print(f"  Best for:      {b['best_for']}")
        print()

compare_brokers()

# work_queue.py
# Work queue pattern with RabbitMQ

def work_queue():
    print("Work Queue Pattern (RabbitMQ)")
    print("=" * 40)
    print()
    
    producer = """
import pika

# Producer: publishes jobs to a queue
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# Declare queue (idempotent)
channel.queue_declare(queue='compression_jobs', durable=True)

# Publish a job
channel.basic_publish(
    exchange='',
    routing_key='compression_jobs',
    body='{"file_id": "abc123", "format": "zip"}',
    properties=pika.BasicProperties(
        delivery_mode=2,  # persistent message
    )
)
print("Job published")
connection.close()
"""
    print("Producer:")
    print(producer)
    
    consumer = """
import pika
import time

# Consumer: processes jobs from the queue
def callback(ch, method, properties, body):
    print(f"Processing job: {body}")
    time.sleep(2)  # simulate work
    print("Job complete")
    ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()
channel.queue_declare(queue='compression_jobs', durable=True)

# Only fetch one message at a time
channel.basic_qos(prefetch_count=1)
channel.basic_consume(
    queue='compression_jobs',
    on_message_callback=callback
)
print("Waiting for jobs...")
channel.start_consuming()
"""
    print("Consumer:")
    print(consumer)

work_queue()

# pub_sub.py
# Publish/subscribe pattern

def pub_sub():
    print("Publish/Subscribe Pattern")
    print("=" * 40)
    print()
    
    pub_sub_code = """
import pika

# Publisher: sends event to exchange
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# Fanout exchange broadcasts to all bound queues
channel.exchange_declare(exchange='events', exchange_type='fanout')

message = '{"event": "file_uploaded", "file_id": "abc123"}'
channel.basic_publish(
    exchange='events',
    routing_key='',
    body=message
)
print(f"Event published: {message}")
connection.close()

# Subscriber 1: audit log service
# Each consumer creates its own queue bound to the exchange
ch1 = connection.channel()
ch1.exchange_declare(exchange='events', exchange_type='fanout')
result = ch1.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
ch1.queue_bind(exchange='events', queue=queue_name)
print(f"Audit service listening on: {queue_name}")

# Subscriber 2: notification service works the same way
# Each subscriber gets ALL messages published to the exchange
"""
    print(pub_sub_code)
    print()
    print("Key points:")
    print("- Fanout exchange sends to ALL bound queues")
    print("- Each consumer gets its own exclusive queue")
    print("- Perfect for event broadcast scenarios")

pub_sub()

def video_pipeline_design():
    print("Video Processing Pipeline - Broker Topology")
    print("=" * 45)
    print()
    print("Exchanges:")
    print("  video.uploaded  - Fanout (all services notified)")
    print("  video.processed - Direct (routing by output type)")
    print()
    print("Queues:")
    print("  transcoding.jobs     - Workers transcode video")
    print("  thumbnail.jobs       - Workers generate thumbnails")
    print("  notification.emails  - Email service")
    print("  notification.webhooks - Webhook service")
    print()
    print("Flow:")
    print("  1. Upload -> video.uploaded (fanout)")
    print("  2. transcoding.jobs consumer -> transcode")
    print("  3. thumbnail.jobs consumer -> generate thumb")
    print("  4. On complete -> video.processed")
    print("  5. notification.* consumers -> send alerts")
    print()
    print("Dead Letter:")
    print("  All queues have DLQ after 3 retries")

video_pipeline_design()