Messaging-Oriented Architecture

 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

// RabbitMQ를 사용한 메시지 큐 구현
const amqp = require('amqplib');

class MessageQueueManager {
    constructor(connectionUrl) {
        this.connectionUrl = connectionUrl;
        this.connection = null;
        this.channel = null;
    }
    
    async connect() {
        this.connection = await amqp.connect(this.connectionUrl);
        this.channel = await this.connection.createChannel();
    }
    
    async createQueue(queueName, options = {}) {
        const queueOptions = {
            durable: true,      // 큐 지속성
            exclusive: false,   // 연결 전용 큐 여부
            autoDelete: false,  // 자동 삭제 여부
            …options
        };
        
        await this.channel.assertQueue(queueName, queueOptions);
        console.log(`Queue ${queueName} created successfully`);
    }
    
    async sendMessage(queueName, message) {
        const messageBuffer = Buffer.from(JSON.stringify(message));
        const messageOptions = {
            persistent: true,   // 메시지 지속성
            timestamp: Date.now(),
            messageId: this.generateMessageId()
        };
        
        await this.channel.sendToQueue(queueName, messageBuffer, messageOptions);
        console.log(`Message sent to ${queueName}:`, message);
    }
    
    async consumeMessages(queueName, processor) {
        await this.channel.consume(queueName, async (msg) => {
            if (msg) {
                try {
                    const message = JSON.parse(msg.content.toString());
                    await processor(message);
                    this.channel.ack(msg);  // 메시지 처리 확인
                } catch (error) {
                    console.error('Message processing failed:', error);
                    this.channel.nack(msg, false, false);  // Dead Letter Queue로 이동
                }
            }
        });
    }
    
    generateMessageId() {
        return Date.now().toString(36) + Math.random().toString(36).substr(2);
    }
}

 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

// Apache Kafka를 사용한 Pub-Sub 구현
const kafka = require('kafkajs');

class PubSubManager {
    constructor(brokers) {
        this.kafka = kafka({
            clientId: 'messaging-app',
            brokers: brokers
        });
        this.producer = this.kafka.producer();
        this.admin = this.kafka.admin();
    }
    
    async initialize() {
        await this.producer.connect();
        await this.admin.connect();
    }
    
    async createTopic(topicName, partitions = 3, replicas = 1) {
        await this.admin.createTopics({
            topics: [{
                topic: topicName,
                numPartitions: partitions,
                replicationFactor: replicas,
                configEntries: [
                    {
                        name: 'cleanup.policy',
                        value: 'delete'
                    },
                    {
                        name: 'retention.ms',
                        value: '86400000'  // 24시간
                    }
                ]
            }]
        });
        console.log(`Topic ${topicName} created`);
    }
    
    async publishMessage(topicName, message, key = null) {
        await this.producer.send({
            topic: topicName,
            messages: [{
                key: key,
                value: JSON.stringify(message),
                timestamp: Date.now(),
                headers: {
                    'content-type': 'application/json',
                    'message-id': this.generateMessageId()
                }
            }]
        });
        console.log(`Message published to ${topicName}`);
    }
    
    async createSubscriber(topicName, groupId, messageHandler) {
        const consumer = this.kafka.consumer({ groupId });
        await consumer.connect();
        await consumer.subscribe({ topic: topicName });
        
        await consumer.run({
            eachMessage: async ({ topic, partition, message }) => {
                try {
                    const data = JSON.parse(message.value.toString());
                    await messageHandler(data, {
                        topic,
                        partition,
                        offset: message.offset,
                        timestamp: message.timestamp
                    });
                } catch (error) {
                    console.error(`Error processing message from ${topic}:`, error);
                    // 에러 처리 로직 (Dead Letter Topic으로 전송 등)
                }
            }
        });
        
        return consumer;
    }
    
    generateMessageId() {
        return `${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
    }
}

  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

// Request-Reply 패턴 구현
class RequestReplyManager {
    constructor(messageQueueManager) {
        this.queueManager = messageQueueManager;
        this.pendingRequests = new Map();
        this.replyQueueName = 'reply_queue_' + Math.random().toString(36).substr(2, 9);
    }
    
    async initialize() {
        await this.queueManager.createQueue(this.replyQueueName, {
            exclusive: true,
            autoDelete: true
        });
        
        // 응답 메시지 처리
        await this.queueManager.consumeMessages(this.replyQueueName, (message) => {
            this.handleReply(message);
        });
    }
    
    async sendRequest(targetQueue, requestData, timeout = 30000) {
        const correlationId = this.generateCorrelationId();
        const requestMessage = {
            data: requestData,
            replyTo: this.replyQueueName,
            correlationId: correlationId,
            timestamp: Date.now()
        };
        
        // 응답 대기 프로미스 생성
        const responsePromise = new Promise((resolve, reject) => {
            const timer = setTimeout(() => {
                this.pendingRequests.delete(correlationId);
                reject(new Error('Request timeout'));
            }, timeout);
            
            this.pendingRequests.set(correlationId, {
                resolve,
                reject,
                timer,
                timestamp: Date.now()
            });
        });
        
        // 요청 메시지 전송
        await this.queueManager.sendMessage(targetQueue, requestMessage);
        console.log(`Request sent with correlation ID: ${correlationId}`);
        
        return responsePromise;
    }
    
    handleReply(replyMessage) {
        const correlationId = replyMessage.correlationId;
        const pendingRequest = this.pendingRequests.get(correlationId);
        
        if (pendingRequest) {
            clearTimeout(pendingRequest.timer);
            this.pendingRequests.delete(correlationId);
            
            if (replyMessage.error) {
                pendingRequest.reject(new Error(replyMessage.error));
            } else {
                pendingRequest.resolve(replyMessage.data);
            }
        } else {
            console.warn(`Received reply for unknown correlation ID: ${correlationId}`);
        }
    }
    
    // 서버 측 요청 처리기 등록
    async registerRequestHandler(requestQueue, handler) {
        await this.queueManager.createQueue(requestQueue);
        
        await this.queueManager.consumeMessages(requestQueue, async (requestMessage) => {
            try {
                const result = await handler(requestMessage.data);
                
                const replyMessage = {
                    data: result,
                    correlationId: requestMessage.correlationId,
                    timestamp: Date.now()
                };
                
                await this.queueManager.sendMessage(requestMessage.replyTo, replyMessage);
            } catch (error) {
                const errorReply = {
                    error: error.message,
                    correlationId: requestMessage.correlationId,
                    timestamp: Date.now()
                };
                
                await this.queueManager.sendMessage(requestMessage.replyTo, errorReply);
            }
        });
    }
    
    generateCorrelationId() {
        return Date.now().toString(36) + Math.random().toString(36).substr(2);
    }
}

  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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

# 메시지 라우팅 구현 예시
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum

class RoutingStrategy(ABC):
    """라우팅 전략 추상 클래스"""
    
    @abstractmethod
    def determine_route(self, message: Dict[str, Any]) -> List[str]:
        """라우팅 대상 결정"""
        pass

class ContentBasedRouter(RoutingStrategy):
    """콘텐츠 기반 라우팅 전략"""
    
    def __init__(self):
        self.routing_rules = {
            "order_amount": {
                "high_value": {"threshold": 1000, "routes": ["premium_processing", "finance_review"]},
                "standard": {"threshold": 100, "routes": ["standard_processing"]},
                "low_value": {"threshold": 0, "routes": ["automated_processing"]}
            },
            "customer_tier": {
                "VIP": {"routes": ["vip_processing", "dedicated_support"]},
                "PREMIUM": {"routes": ["premium_processing"]},
                "STANDARD": {"routes": ["standard_processing"]}
            },
            "region": {
                "US": {"routes": ["us_processing_center"]},
                "EU": {"routes": ["eu_processing_center"]},
                "ASIA": {"routes": ["asia_processing_center"]}
            }
        }
    
    def determine_route(self, message: Dict[str, Any]) -> List[str]:
        """메시지 내용 기반 라우팅 대상 결정"""
        routes = set()
        
        # 주문 금액 기반 라우팅
        order_amount = message.get("order_amount", 0)
        amount_routes = self._get_amount_based_routes(order_amount)
        routes.update(amount_routes)
        
        # 고객 등급 기반 라우팅
        customer_tier = message.get("customer_tier", "STANDARD")
        tier_routes = self.routing_rules["customer_tier"].get(customer_tier, {}).get("routes", [])
        routes.update(tier_routes)
        
        # 지역 기반 라우팅
        region = message.get("region", "US")
        region_routes = self.routing_rules["region"].get(region, {}).get("routes", [])
        routes.update(region_routes)
        
        return list(routes)
    
    def _get_amount_based_routes(self, amount: float) -> List[str]:
        """주문 금액 기반 라우팅 규칙"""
        if amount >= 1000:
            return self.routing_rules["order_amount"]["high_value"]["routes"]
        elif amount >= 100:
            return self.routing_rules["order_amount"]["standard"]["routes"]
        else:
            return self.routing_rules["order_amount"]["low_value"]["routes"]

class MessageRouter:
    """메시지 라우터"""
    
    def __init__(self, routing_strategy: RoutingStrategy):
        self.routing_strategy = routing_strategy
        self.route_handlers: Dict[str, callable] = {}
        self.routing_statistics = {}
    
    def register_route_handler(self, route_name: str, handler: callable):
        """라우팅 핸들러 등록"""
        self.route_handlers[route_name] = handler
        self.routing_statistics[route_name] = {"message_count": 0, "success_count": 0}
        print(f"Registered handler for route: {route_name}")
    
    async def route_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
        """메시지 라우팅 수행"""
        try:
            # 라우팅 대상 결정
            target_routes = self.routing_strategy.determine_route(message)
            
            if not target_routes:
                raise RoutingError("No valid routes found for message")
            
            # 각 라우팅 대상에 메시지 전달
            routing_results = {}
            for route in target_routes:
                result = await self._deliver_to_route(route, message)
                routing_results[route] = result
                
                # 통계 업데이트
                self.routing_statistics[route]["message_count"] += 1
                if result.get("success", False):
                    self.routing_statistics[route]["success_count"] += 1
            
            return {
                "message_id": message.get("message_id"),
                "routes_used": target_routes,
                "routing_results": routing_results,
                "routing_timestamp": asyncio.get_event_loop().time()
            }
            
        except Exception as e:
            raise RoutingError(f"Message routing failed: {str(e)}")
    
    async def _deliver_to_route(self, route_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
        """특정 라우트로 메시지 전달"""
        if route_name not in self.route_handlers:
            return {"success": False, "error": f"No handler for route: {route_name}"}
        
        try:
            handler = self.route_handlers[route_name]
            result = await handler(message)
            print(f"Message routed to {route_name}: {message.get('message_id')}")
            return {"success": True, "result": result}
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def get_routing_statistics(self) -> Dict[str, Dict]:
        """라우팅 통계 반환"""
        return self.routing_statistics.copy()

class RoutingError(Exception):
    """라우팅 오류 예외 클래스"""
    pass

# 라우팅 핸들러 예시
async def premium_processing_handler(message: Dict[str, Any]) -> Dict[str, Any]:
    """프리미엄 처리 핸들러"""
    print(f"Premium processing: {message.get('order_id')}")
    await asyncio.sleep(0.2)  # 프리미엄 처리 시뮬레이션
    return {"status": "premium_processed", "processing_time": 0.2}

async def standard_processing_handler(message: Dict[str, Any]) -> Dict[str, Any]:
    """표준 처리 핸들러"""
    print(f"Standard processing: {message.get('order_id')}")
    await asyncio.sleep(0.5)  # 표준 처리 시뮬레이션
    return {"status": "standard_processed", "processing_time": 0.5}

 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

# 메시지 변환 구현 예시
from abc import ABC, abstractmethod
import json
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from typing import Dict, Any

class MessageTransformer(ABC):
    """메시지 변환기 추상 클래스"""
    
    @abstractmethod
    def transform(self, message: Any) -> Any:
        """메시지 변환 추상 메서드"""
        pass

class ERPToCRMTransformer(MessageTransformer):
    """ERP에서 CRM으로 고객 데이터 변환"""
    
    def transform(self, erp_message: Dict) -> Dict:
        """ERP 포맷을 CRM 포맷으로 변환"""
        try:
            # ERP 포맷 (XML 기반) → CRM 포맷 (JSON 기반) 변환
            crm_message = {
                "customer_id": erp_message.get("client_code"),
                "personal_info": {
                    "full_name": f"{erp_message.get('first_name', '')} {erp_message.get('last_name', '')}".strip(),
                    "email": erp_message.get("email_address"),
                    "phone": erp_message.get("contact_number")
                },
                "business_info": {
                    "company": erp_message.get("organization"),
                    "industry": erp_message.get("business_sector"),
                    "annual_revenue": erp_message.get("yearly_revenue")
                },
                "account_status": self._map_status(erp_message.get("status")),
                "created_date": erp_message.get("registration_date"),
                "source_system": "ERP"
            }
            
            return crm_message
            
        except Exception as e:
            raise TransformationError(f"ERP to CRM transformation failed: {str(e)}")
    
    def _map_status(self, erp_status: str) -> str:
        """ERP 상태를 CRM 상태로 매핑"""
        status_mapping = {
            "ACTIVE": "active",
            "INACTIVE": "dormant",
            "SUSPENDED": "suspended",
            "TERMINATED": "closed"
        }
        return status_mapping.get(erp_status, "unknown")

class TransformationError(Exception):
    """변환 오류 예외 클래스"""
    pass

class MessageTransformationEngine:
    """메시지 변환 엔진"""
    
    def __init__(self):
        self.transformers: Dict[str, MessageTransformer] = {}
        self.transformation_rules: Dict[str, str] = {}
    
    def register_transformer(self, source_format: str, target_format: str, transformer: MessageTransformer):
        """변환기 등록"""
        key = f"{source_format}_to_{target_format}"
        self.transformers[key] = transformer
        self.transformation_rules[key] = f"{source_format} → {target_format}"
        print(f"Registered transformer: {self.transformation_rules[key]}")
    
    def transform_message(self, message: Any, source_format: str, target_format: str) -> Any:
        """메시지 변환 수행"""
        key = f"{source_format}_to_{target_format}"
        
        if key not in self.transformers:
            raise TransformationError(f"No transformer found for {source_format} to {target_format}")
        
        transformer = self.transformers[key]
        
        try:
            transformed_message = transformer.transform(message)
            print(f"Successfully transformed message from {source_format} to {target_format}")
            return transformed_message
            
        except Exception as e:
            raise TransformationError(f"Transformation failed: {str(e)}")
    
    def get_supported_transformations(self) -> List[str]:
        """지원되는 변환 목록 반환"""
        return list(self.transformation_rules.values())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import pika

# 메시지 생산자
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='order_queue')
channel.basic_publish(exchange='', routing_key='order_queue', body='{"order_id": "ORD-12345"}')
connection.close()

# 메시지 소비자
def callback(ch, method, properties, body):
    print(f"주문 처리: {body}")

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

from kafka import KafkaProducer, KafkaConsumer
import json

# Producer
producer = KafkaProducer(bootstrap_servers='kafka:9092',
                         value_serializer=lambda v: json.dumps(v).encode('utf-8'))
producer.send('user.events', {'user_id':123, 'action':'login'})

# Consumer
consumer = KafkaConsumer('user.events',
                         bootstrap_servers='kafka:9092',
                         value_deserializer=lambda m: json.loads(m.decode('utf-8')),
                         group_id='analytics-group',
                         auto_offset_reset='earliest')

for msg in consumer:
    data = msg.value
    process_login_event(data)  # 이벤트 처리 함수

  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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393

// 전자상거래 주문 처리 시스템 - Messaging-Oriented Styles 구현

const EventEmitter = require('events');

// 메시지 브로커 클래스
class MessageBroker extends EventEmitter {
    constructor() {
        super();
        this.topics = new Map();
        this.subscribers = new Map();
        this.messageHistory = [];
    }
    
    // 토픽 생성
    createTopic(topicName) {
        if (!this.topics.has(topicName)) {
            this.topics.set(topicName, []);
            this.subscribers.set(topicName, new Set());
            console.log(`Topic '${topicName}' created`);
        }
    }
    
    // 구독자 등록
    subscribe(topicName, subscriberId, handler) {
        this.createTopic(topicName);
        this.subscribers.get(topicName).add({ subscriberId, handler });
        console.log(`${subscriberId} subscribed to '${topicName}'`);
    }
    
    // 메시지 발행
    async publish(topicName, message) {
        if (!this.topics.has(topicName)) {
            throw new Error(`Topic '${topicName}' does not exist`);
        }
        
        const enrichedMessage = {
            …message,
            messageId: this.generateMessageId(),
            topic: topicName,
            timestamp: Date.now()
        };
        
        // 메시지 히스토리 저장
        this.messageHistory.push(enrichedMessage);
        
        console.log(`Message published to '${topicName}':`, enrichedMessage);
        
        // 구독자들에게 메시지 전달
        const subscribers = this.subscribers.get(topicName);
        for (const subscriber of subscribers) {
            try {
                await subscriber.handler(enrichedMessage);
            } catch (error) {
                console.error(`Error in subscriber ${subscriber.subscriberId}:`, error);
                // Dead Letter Queue로 이동 (실제 구현에서는 별도 처리)
                await this.handleFailedMessage(enrichedMessage, subscriber, error);
            }
        }
    }
    
    async handleFailedMessage(message, subscriber, error) {
        console.log(`Moving message to DLQ for subscriber ${subscriber.subscriberId}`);
        // Dead Letter Queue 처리 로직
    }
    
    generateMessageId() {
        return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
}

// 주문 서비스
class OrderService {
    constructor(messageBroker) {
        this.broker = messageBroker;
        this.orders = new Map();
        
        // 이벤트 구독
        this.broker.subscribe('payment.completed', 'OrderService', this.handlePaymentCompleted.bind(this));
        this.broker.subscribe('inventory.reserved', 'OrderService', this.handleInventoryReserved.bind(this));
        this.broker.subscribe('shipping.started', 'OrderService', this.handleShippingStarted.bind(this));
    }
    
    async createOrder(customerId, items) {
        const orderId = this.generateOrderId();
        const order = {
            orderId,
            customerId,
            items,
            status: 'created',
            totalAmount: this.calculateTotal(items),
            createdAt: Date.now()
        };
        
        this.orders.set(orderId, order);
        
        // Order Created 이벤트 발행
        await this.broker.publish('order.created', {
            eventType: 'OrderCreated',
            orderId,
            customerId,
            items,
            totalAmount: order.totalAmount
        });
        
        console.log(`Order ${orderId} created for customer ${customerId}`);
        return { orderId, status: 'processing' };
    }
    
    async handlePaymentCompleted(message) {
        const { orderId } = message;
        if (this.orders.has(orderId)) {
            const order = this.orders.get(orderId);
            order.status = 'payment_completed';
            order.paymentCompletedAt = Date.now();
            console.log(`Payment completed for order ${orderId}`);
        }
    }
    
    async handleInventoryReserved(message) {
        const { orderId } = message;
        if (this.orders.has(orderId)) {
            const order = this.orders.get(orderId);
            order.status = 'inventory_reserved';
            order.inventoryReservedAt = Date.now();
            console.log(`Inventory reserved for order ${orderId}`);
        }
    }
    
    async handleShippingStarted(message) {
        const { orderId, trackingNumber } = message;
        if (this.orders.has(orderId)) {
            const order = this.orders.get(orderId);
            order.status = 'shipped';
            order.trackingNumber = trackingNumber;
            order.shippedAt = Date.now();
            console.log(`Order ${orderId} shipped with tracking ${trackingNumber}`);
        }
    }
    
    calculateTotal(items) {
        return items.reduce((total, item) => total + (item.price * item.quantity), 0);
    }
    
    generateOrderId() {
        return `ORDER_${Date.now()}_${Math.random().toString(36).substr(2, 6).toUpperCase()}`;
    }
    
    getOrder(orderId) {
        return this.orders.get(orderId);
    }
}

// 결제 서비스
class PaymentService {
    constructor(messageBroker) {
        this.broker = messageBroker;
        this.payments = new Map();
        
        // Order Created 이벤트 구독
        this.broker.subscribe('order.created', 'PaymentService', this.handleOrderCreated.bind(this));
    }
    
    async handleOrderCreated(message) {
        const { orderId, customerId, totalAmount } = message;
        
        console.log(`Processing payment for order ${orderId}, amount: $${totalAmount}`);
        
        // 결제 처리 시뮬레이션 (비동기 처리)
        setTimeout(async () => {
            try {
                const paymentResult = await this.processPayment(customerId, totalAmount);
                
                this.payments.set(orderId, paymentResult);
                
                // Payment Completed 이벤트 발행
                await this.broker.publish('payment.completed', {
                    eventType: 'PaymentCompleted',
                    orderId,
                    customerId,
                    paymentId: paymentResult.paymentId,
                    amount: totalAmount,
                    paymentMethod: paymentResult.paymentMethod
                });
                
            } catch (error) {
                // Payment Failed 이벤트 발행
                await this.broker.publish('payment.failed', {
                    eventType: 'PaymentFailed',
                    orderId,
                    customerId,
                    error: error.message
                });
            }
        }, 1000); // 1초 후 결제 완료 시뮬레이션
    }
    
    async processPayment(customerId, amount) {
        // 실제 결제 처리 로직 시뮬레이션
        if (Math.random() > 0.1) { // 90% 성공률
            return {
                paymentId: `PAY_${Date.now()}`,
                paymentMethod: 'credit_card',
                status: 'completed',
                processedAt: Date.now()
            };
        } else {
            throw new Error('Payment processing failed');
        }
    }
}

// 재고 서비스
class InventoryService {
    constructor(messageBroker) {
        this.broker = messageBroker;
        this.inventory = new Map([
            ['ITEM_001', { quantity: 100, reserved: 0 }],
            ['ITEM_002', { quantity: 50, reserved: 0 }],
            ['ITEM_003', { quantity: 200, reserved: 0 }]
        ]);
        
        // Order Created 이벤트 구독
        this.broker.subscribe('order.created', 'InventoryService', this.handleOrderCreated.bind(this));
    }
    
    async handleOrderCreated(message) {
        const { orderId, items } = message;
        
        console.log(`Checking inventory for order ${orderId}`);
        
        try {
            // 재고 확인 및 예약
            for (const item of items) {
                await this.reserveInventory(item.itemId, item.quantity);
            }
            
            // Inventory Reserved 이벤트 발행
            await this.broker.publish('inventory.reserved', {
                eventType: 'InventoryReserved',
                orderId,
                items,
                reservedAt: Date.now()
            });
            
        } catch (error) {
            // Inventory Insufficient 이벤트 발행
            await this.broker.publish('inventory.insufficient', {
                eventType: 'InventoryInsufficient',
                orderId,
                error: error.message
            });
        }
    }
    
    async reserveInventory(itemId, quantity) {
        const item = this.inventory.get(itemId);
        if (!item) {
            throw new Error(`Item ${itemId} not found`);
        }
        
        if (item.quantity - item.reserved < quantity) {
            throw new Error(`Insufficient inventory for item ${itemId}`);
        }
        
        item.reserved += quantity;
        console.log(`Reserved ${quantity} units of ${itemId}`);
    }
}

// 배송 서비스
class ShippingService {
    constructor(messageBroker) {
        this.broker = messageBroker;
        this.shipments = new Map();
        
        // Payment Completed 이벤트 구독
        this.broker.subscribe('payment.completed', 'ShippingService', this.handlePaymentCompleted.bind(this));
    }
    
    async handlePaymentCompleted(message) {
        const { orderId, customerId } = message;
        
        console.log(`Preparing shipment for order ${orderId}`);
        
        // 배송 준비 시뮬레이션 (비동기 처리)
        setTimeout(async () => {
            const trackingNumber = this.generateTrackingNumber();
            
            this.shipments.set(orderId, {
                trackingNumber,
                status: 'shipped',
                shippedAt: Date.now()
            });
            
            // Shipping Started 이벤트 발행
            await this.broker.publish('shipping.started', {
                eventType: 'ShippingStarted',
                orderId,
                customerId,
                trackingNumber,
                estimatedDelivery: Date.now() + (3 * 24 * 60 * 60 * 1000) // 3일 후
            });
            
        }, 2000); // 2초 후 배송 시작 시뮬레이션
    }
    
    generateTrackingNumber() {
        return `TRK_${Date.now().toString(36).toUpperCase()}`;
    }
}

// 알림 서비스
class NotificationService {
    constructor(messageBroker) {
        this.broker = messageBroker;
        
        // 다양한 이벤트 구독
        this.broker.subscribe('order.created', 'NotificationService', this.handleOrderCreated.bind(this));
        this.broker.subscribe('payment.completed', 'NotificationService', this.handlePaymentCompleted.bind(this));
        this.broker.subscribe('shipping.started', 'NotificationService', this.handleShippingStarted.bind(this));
    }
    
    async handleOrderCreated(message) {
        const { orderId, customerId } = message;
        await this.sendNotification(customerId, 'order_created', 
            `Your order ${orderId} has been received and is being processed.`);
    }
    
    async handlePaymentCompleted(message) {
        const { orderId, customerId } = message;
        await this.sendNotification(customerId, 'payment_completed',
            `Payment for order ${orderId} has been successfully processed.`);
    }
    
    async handleShippingStarted(message) {
        const { orderId, customerId, trackingNumber } = message;
        await this.sendNotification(customerId, 'shipping_started',
            `Your order ${orderId} has been shipped. Tracking number: ${trackingNumber}`);
    }
    
    async sendNotification(customerId, type, message) {
        console.log(`📧 NOTIFICATION [${type}] to customer ${customerId}: ${message}`);
        // 실제 구현에서는 이메일, SMS, 푸시 알림 등을 전송
    }
}

// 시스템 시뮬레이션
async function runECommerceSimulation() {
    console.log('🚀 Starting E-commerce Order Processing System with Messaging-Oriented Styles\n');
    
    // 메시지 브로커 초기화
    const broker = new MessageBroker();
    
    // 토픽 생성
    broker.createTopic('order.created');
    broker.createTopic('payment.completed');
    broker.createTopic('payment.failed');
    broker.createTopic('inventory.reserved');
    broker.createTopic('inventory.insufficient');
    broker.createTopic('shipping.started');
    
    // 서비스 초기화
    const orderService = new OrderService(broker);
    const paymentService = new PaymentService(broker);
    const inventoryService = new InventoryService(broker);
    const shippingService = new ShippingService(broker);
    const notificationService = new NotificationService(broker);
    
    console.log('\n✅ All services initialized and subscribed to events\n');
    
    // 주문 생성 시뮬레이션
    const customerOrder = await orderService.createOrder('CUSTOMER_001', [
        { itemId: 'ITEM_001', quantity: 2, price: 29.99 },
        { itemId: 'ITEM_002', quantity: 1, price: 49.99 }
    ]);
    
    console.log('📦 Order created:', customerOrder);
    
    // 시스템 상태 확인 (5초 후)
    setTimeout(() => {
        console.log('\n📊 Final Order Status:');
        const finalOrder = orderService.getOrder(customerOrder.orderId);
        console.log(JSON.stringify(finalOrder, null, 2));
        
        console.log('\n📈 Message History:');
        broker.messageHistory.forEach((msg, index) => {
            console.log(`${index + 1}. [${msg.topic}] ${msg.eventType} at ${new Date(msg.timestamp).toISOString()}`);
        });
    }, 5000);
}

// 시뮬레이션 실행
runECommerceSimulation().catch(console.error);