# Простая реализация шины событий
class EventBus:
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, event_type, handler):
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(handler)

    def publish(self, event):
        event_type = type(event)
        if event_type in self.subscribers:
            for handler in self.subscribers[event_type]:
                handler(event)

# Пример использования
class UserRegisteredEvent:
    def __init__(self, user_id, email):
        self.user_id = user_id
        self.email = email

def send_welcome_email(event):
    print(f"Sending email to {event.email}")

bus = EventBus()
bus.subscribe(UserRegisteredEvent, send_welcome_email)

# Эмуляция события
bus.publish(UserRegisteredEvent(1, "user@example.com"))

import asyncio

class AsyncEventBus:
    def __init__(self):
        self.subscribers = {}
        self.queue = asyncio.Queue()

    async def publish(self, event):
        await self.queue.put(event)

    async def run(self):
        while True:
            event = await self.queue.get()
            event_type = type(event)
            if event_type in self.subscribers:
                await asyncio.gather(
                    *[handler(event) for handler in self.subscribers[event_type]]
                )

    def subscribe(self, event_type, handler):
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(handler)

# Запуск шины
async def main():
    bus = AsyncEventBus()
    bus.subscribe(UserRegisteredEvent, send_welcome_email)
    asyncio.create_task(bus.run())
    await bus.publish(UserRegisteredEvent(2, "async@example.com"))

import pika

def publish_event(event):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='user_events')
    channel.basic_publish(
        exchange='',
        routing_key='user_events',
        body=json.dumps({'event_type': 'UserRegistered', 'data': event.__dict__})
    )
    connection.close()

def consume_events():
    def callback(ch, method, properties, body):
        event_data = json.loads(body)
        if event_data['event_type'] == 'UserRegistered':
            handle_user_registered(event_data['data'])

    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='user_events')
    channel.basic_consume(queue='user_events', on_message_callback=callback, auto_ack=True)
    channel.start_consuming()

from confluent_kafka import Producer, Consumer

producer = Producer({'bootstrap.servers': 'localhost:9092'})
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'user-service',
    'auto.offset.reset': 'earliest'
})

def publish_kafka_event(event):
    producer.produce('user-events', json.dumps(event.__dict__))
    producer.flush()

def consume_kafka_events():
    consumer.subscribe(['user-events'])
    while True:
        msg = consumer.poll(1.0)
        if msg is None: continue
        event_data = json.loads(msg.value())
        handle_event(event_data)

import faust

app = faust.App('user-events', broker='kafka://localhost')

class UserRegistered(faust.Record):
    user_id: int
    email: str

user_topic = app.topic('user_events', value_type=UserRegistered)

@app.agent(user_topic)
async def process_user_registrations(events):
    async for event in events:
        print(f"Sending welcome email to {event.email}")
        # Логика обработки

from nameko.rpc import rpc
from nameko.events import EventDispatcher, event_handler

class UserService:
    name = "user_service"
    dispatch = EventDispatcher()

    @rpc
    def register_user(self, email):
        user_id = create_user_in_db(email)
        self.dispatch("user_registered", {"user_id": user_id, "email": email})
        return user_id

class EmailService:
    name = "email_service"

    @event_handler("user_service", "user_registered")
    def send_welcome_email(self, payload):
        send_email(payload["email"], "Welcome!")

# Пример версионирования события
class UserRegisteredV1:
    version = 1
    fields = ['user_id', 'email']

class UserRegisteredV2:
    version = 2
    fields = ['user_id', 'email', 'registration_date']

graph LR
    A[Order Service] -->|OrderCreated| B[Message Broker]
    B --> C[Payment Service]
    B --> D[Inventory Service]
    B --> E[Notification Service]
    C -->|PaymentProcessed| B
    D -->|InventoryReserved| B

# Реализация на FastAPI + RabbitMQ
@app.post("/orders")
async def create_order(order: Order):
    # Сохраняем заказ в БД
    order_id = save_order(order)

    # Публикуем событие
    message = {
        "event_type": "OrderCreated",
        "data": {
            "order_id": order_id,
            "items": order.items,
            "total": order.total
        }
    }
    channel.basic_publish(
        exchange='orders',
        routing_key='',
        body=json.dumps(message)
    )
    return {"order_id": order_id}

# Обработчик в Payment Service
def process_order_created(ch, method, properties, body):
    data = json.loads(body)['data']
    process_payment(data['order_id'], data['total'])
    # Публикация PaymentProcessed...