Event Bus Communication — Centralized Event Routing for Microservices

# event_bus_arch.py
# Event bus architecture

def event_bus_arch():
    print("Event Bus Architecture")
    print("=" * 40)
    print()
    
    components = [
        {
            "component": "Topics",
            "desc": "Named channels for events. Producers publish to topics. Consumers subscribe to topics.",
            "examples": "user.events, order.events, file.events, system.alerts"
        },
        {
            "component": "Subscriptions",
            "desc": "Consumer registrations that define which events to receive and how to deliver them.",
            "example": "email-service subscribes to user.events with filter event_type=registered"
        },
        {
            "component": "Filters",
            "desc": "Rules that determine which events reach which subscribers, based on event attributes.",
            "example": "Only forward events where event.metadata.environment == 'production'"
        },
        {
            "component": "Transformers",
            "desc": "Modify events before delivery, adding headers, enriching data, or changing format.",
            "example": "Add tenant_id from event metadata before delivering to multi-tenant consumers"
        },
        {
            "component": "Dead Letter Queue",
            "desc": "Storage for events that could not be delivered after all retry attempts.",
            "example": "Event moved to DLQ after 3 delivery failures with reason logged"
        },
    ]
    
    for c in components:
        print(f"{c['component']:25s}")
        print(f"  {c['desc']}")
        print(f"  Examples: {c['examples']}")
        print()

event_bus_arch()

# event_bus_impl.py
# Event bus implementation

def event_bus_impl():
    print("Event Bus Implementation")
    print("=" * 40)
    print()
    
    code = """
import asyncio
import json
import time
from collections import defaultdict
import uuid

class Event:
    def __init__(self, topic, event_type, data, 
                 source=None, trace_id=None):
        self.id = str(uuid.uuid4())
        self.topic = topic
        self.type = event_type
        self.data = data
        self.source = source
        self.trace_id = trace_id or str(uuid.uuid4())
        self.timestamp = time.time()
        self.metadata = {}

class Subscription:
    def __init__(self, topic, handler, filter_func=None,
                 transformer=None, name=None):
        self.id = str(uuid.uuid4())
        self.topic = topic
        self.handler = handler
        self.filter = filter_func or (lambda e: True)
        self.transformer = transformer or (lambda e: e)
        self.name = name or f"sub_{self.id[:8]}"

class EventBus:
    def __init__(self, name="default"):
        self.name = name
        self.subscriptions = defaultdict(list)  # topic -> [Subscription]
        self.dlq = []
    
    def subscribe(self, topic, handler, filter_func=None,
                  transformer=None, name=None):
        sub = Subscription(topic, handler, filter_func, 
                          transformer, name)
        self.subscriptions[topic].append(sub)
        print(f"Subscribed: {sub.name} to {topic}")
        return sub
    
    def unsubscribe(self, subscription_id):
        for topic in self.subscriptions:
            self.subscriptions[topic] = [
                s for s in self.subscriptions[topic]
                if s.id != subscription_id
            ]
    
    def publish(self, event):
        print(f"Published: {event.type} to {event.topic}")
        
        deliveries = []
        for sub in self.subscriptions.get(event.topic, []):
            if sub.filter(event):
                transformed = sub.transformer(event)
                deliveries.append((sub, transformed))
        
        return deliveries
    
    async def publish_async(self, event):
        deliveries = self.publish(event)
        
        results = []
        for sub, event_to_deliver in deliveries:
            try:
                await sub.handler(event_to_deliver)
                results.append((sub.id, True, None))
            except Exception as e:
                self.dlq.append({
                    "event_id": event.id,
                    "subscription": sub.id,
                    "error": str(e),
                    "timestamp": time.time()
                })
                results.append((sub.id, False, str(e)))
        
        return results
    
    def get_dlq(self):
        return list(self.dlq)
"""
    print(code)

event_bus_impl()

# routing_filters.py
# Event routing and filtering

def routing_filters():
    print("Event Bus Routing and Filtering")
    print("=" * 40)
    print()
    
    code = """
# Event routing with filters

import re

def build_subscription_routes():
    bus = EventBus("application-bus")
    
    # Route 1: All user events to audit service
    bus.subscribe(
        "user.events",
        audit_handler,
        name="audit-all-users"
    )
    
    # Route 2: Only user.registered events to email service
    bus.subscribe(
        "user.events",
        email_welcome_handler,
        filter_func=lambda e: e.type == "user.registered",
        name="email-on-register"
    )
    
    # Route 3: High-priority events with transformation
    bus.subscribe(
        "system.alerts",
        pagerduty_handler,
        filter_func=lambda e: e.data.get("severity") in ["critical", "high"],
        transformer=lambda e: enrich_with_oncall_info(e),
        name="pagerduty-critical"
    )
    
    # Route 4: Events matching a pattern
    bus.subscribe(
        "file.events",
        analytics_handler,
        filter_func=lambda e: re.match(
            r"file\.(uploaded|downloaded|deleted)", e.type
        ),
        name="analytics-file-ops"
    )
    
    # Route 5: Sampling for expensive processing
    import random
    bus.subscribe(
        "order.events",
        ml_recommendation_handler,
        filter_func=lambda e: random.random() < 0.1,  # 10% sample
        name="ml-sampled"
    )

# Example events
events = [
    Event("user.events", "user.registered", 
          {"user_id": "u1", "email": "a@b.com"}),
    Event("user.events", "user.deleted",
          {"user_id": "u1", "reason": "requested"}),
    Event("system.alerts", "alert.critical",
          {"message": "DB connection pool exhausted", "severity": "critical"}),
    Event("system.alerts", "alert.info",
          {"message": "Cache hit ratio: 95%", "severity": "info"}),
]

# Events with severity=info would be filtered out from Route 3
"""
    print(code)

routing_filters()

# delivery_guarantees.py
# Event delivery guarantees

def delivery_guarantees():
    print("Event Delivery Guarantees")
    print("=" * 40)
    print()
    
    code = """
import asyncio
import time

class ReliableEventBus(EventBus):
    """Event bus with at-least-once delivery."""
    
    def __init__(self, name="reliable", max_retries=3):
        super().__init__(name)
        self.max_retries = max_retries
        self.retry_backoff = [1, 5, 30]  # seconds
    
    async def publish_with_retry(self, event):
        deliveries = self.publish(event)
        
        for sub, event_to_deliver in deliveries:
            last_error = None
            for attempt in range(self.max_retries):
                try:
                    await sub.handler(event_to_deliver)
                    print(f"Delivered to {sub.name} (attempt {attempt + 1})")
                    break
                except Exception as e:
                    last_error = e
                    if attempt < self.max_retries - 1:
                        backoff = self.retry_backoff[
                            min(attempt, len(self.retry_backoff) - 1)
                        ]
                        print(f"Retry {sub.name} in {backoff}s "
                              f"(attempt {attempt + 1} failed)")
                        await asyncio.sleep(backoff)
            
            if last_error:
                self.dlq.append({
                    "event_id": event.id,
                    "subscription": sub.id,
                    "error": str(last_error),
                    "attempts": self.max_retries,
                    "timestamp": time.time()
                })
                print(f"Failed after {self.max_retries} retries: "
                      f"{sub.name}")

class DeliveryMode:
    # At-most-once: fire and forget
    AT_MOST_ONCE = "at_most_once"
    
    # At-least-once: retry until acknowledged
    AT_LEAST_ONCE = "at_least_once"
    
    # Exactly-once: deduplication + at-least-once
    EXACTLY_ONCE = "exactly_once"

class DeduplicatingBus(ReliableEventBus):
    """Bus with exactly-once delivery semantics."""
    
    def __init__(self, name="deduplicated", 
                 dedup_store=None):
        super().__init__(name)
        self.processed_events = dedup_store or set()
    
    def is_duplicate(self, event_id, subscription_id):
        key = f"{event_id}:{subscription_id}"
        return key in self.processed_events
    
    def mark_processed(self, event_id, subscription_id):
        key = f"{event_id}:{subscription_id}"
        self.processed_events.add(key)
    
    async def publish_exactly_once(self, event):
        deliveries = self.publish(event)
        
        for sub, event_to_deliver in deliveries:
            if self.is_duplicate(event.id, sub.id):
                print(f"Skipping duplicate: {event.id} -> {sub.name}")
                continue
            
            try:
                await sub.handler(event_to_deliver)
                self.mark_processed(event.id, sub.id)
            except Exception as e:
                # Retry without marking processed
                raise
"""
    print(code)

delivery_guarantees()

def devops_event_bus():
    print("DevOps Platform Event Bus Design")
    print("=" * 45)
    print()
    print("Topics:")
    print("  deploy.events              - Deployment lifecycle")
    print("  monitoring.alerts           - System alerts")
    print("  infrastructure.changes      - Infra state changes")
    print()
    print("Subscriptions:")
    print()
    print("  Slack Notifier:")
    print("    Topic: deploy.events")
    print("    Filter: All deploy events")
    print()
    print("  PagerDuty:")
    print("    Topic: monitoring.alerts")
    print("    Filter: severity in [critical, high]")
    print()
    print("  Auto-Scaler:")
    print("    Topic: infrastructure.changes")
    print("    Filter: type == cpu_utilization_high")
    print()
    print("  Auditor:")
    print("    Topic: ALL topics")
    print("    Filter: None (all events logged)")
    print()
    print("Delivery: At-least-once with 3 retries")
    print("DLQ: All failed events stored for manual inspection")

devops_event_bus()