Event-Bus Pattern

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107

from typing import Dict, List, Callable
from dataclasses import dataclass
from datetime import datetime
import threading
import queue
import uuid

# 이벤트 기본 클래스
@dataclass
class Event:
    event_type: str
    timestamp: datetime
    data: dict
    event_id: str = str(uuid.uuid4())

# 이벤트 버스 구현
class EventBus:
    def __init__(self):
        # 이벤트 타입별 구독자 목록
        self._subscribers: Dict[str, List[Callable]] = {}
        # 이벤트 큐
        self._event_queue = queue.Queue()
        # 이벤트 처리 스레드
        self._processing_thread = threading.Thread(target=self._process_events)
        self._is_running = True
        self._processing_thread.start()

    def subscribe(self, event_type: str, handler: Callable):
        """이벤트 타입에 대한 핸들러 등록"""
        if event_type not in self._subscribers:
            self._subscribers[event_type] = []
        self._subscribers[event_type].append(handler)
    
    def publish(self, event: Event):
        """이벤트 발행"""
        self._event_queue.put(event)

    def _process_events(self):
        """이벤트 처리 루프"""
        while self._is_running:
            try:
                event = self._event_queue.get(timeout=1.0)
                self._dispatch_event(event)
            except queue.Empty:
                continue

    def _dispatch_event(self, event: Event):
        """등록된 핸들러들에게 이벤트 전달"""
        if event.event_type in self._subscribers:
            for handler in self._subscribers[event.event_type]:
                try:
                    handler(event)
                except Exception as e:
                    print(f"Error handling event {event.event_id}: {e}")

    def shutdown(self):
        """이벤트 버스 종료"""
        self._is_running = False
        self._processing_thread.join()

# 예시: 주문 시스템에서의 사용
class OrderService:
    def __init__(self, event_bus: EventBus):
        self.event_bus = event_bus

    def place_order(self, order_data: dict):
        # 주문 처리 로직
        print(f"Processing order: {order_data}")
        
        # 주문 완료 이벤트 발행
        event = Event(
            event_type="OrderPlaced",
            timestamp=datetime.now(),
            data=order_data
        )
        self.event_bus.publish(event)

class InventoryService:
    def __init__(self, event_bus: EventBus):
        self.event_bus = event_bus
        # OrderPlaced 이벤트 구독
        self.event_bus.subscribe("OrderPlaced", self.handle_order_placed)

    def handle_order_placed(self, event: Event):
        print(f"Updating inventory for order: {event.data}")
        # 재고 업데이트 로직

class NotificationService:
    def __init__(self, event_bus: EventBus):
        self.event_bus = event_bus
        # OrderPlaced 이벤트 구독
        self.event_bus.subscribe("OrderPlaced", self.handle_order_placed)

    def handle_order_placed(self, event: Event):
        print(f"Sending notification for order: {event.data}")
        # 알림 발송 로직

# 사용 예시
if __name__ == "__main__":
    event_bus = EventBus()
    
    order_service = OrderService(event_bus)
    inventory_service = InventoryService(event_bus)
    notification_service = NotificationService(event_bus)

    # 주문 처리
    order_service.place_order({"order_id": "123", "product": "laptop"})