Publish-Subscribe 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

/**
 * Event Bus Interface
 * 
 * The contract for any event bus implementation.
 * Keeps the API simple while hiding complexity.
 */
interface IEventBus {
    /**
     * Publishes an event to all interested subscribers.
     * 
     * @param event The event to publish
     * @returns Promise that resolves when publishing completes
     *          (what "completes" means depends on delivery semantics)
     */
    publish<TEvent>(event: TEvent): Promise<void>;
    
    /**
     * Registers a subscriber to receive events.
     * 
     * @param subscriber The subscriber to register
     * @returns A function to unsubscribe (for cleanup)
     */
    subscribe(subscriber: ISubscriber<unknown>): () => void;
    
    /**
     * Unregisters a subscriber.
     * 
     * @param subscriber The subscriber to remove
     */
    unsubscribe(subscriber: ISubscriber<unknown>): void;
    
    /**
     * Gracefully shuts down the event bus.
     * Waits for in-flight events to complete.
     */
    shutdown(): Promise<void>;
}
 
/**
 * Extended interface for observable event buses
 */
interface IObservableEventBus extends IEventBus {
    /**
     * Returns current subscriber count for an event type.
     * Useful for monitoring and debugging.
     */
    subscriberCount(eventType: string): number;
    
    /**
     * Returns all registered event types.
     */
    registeredEventTypes(): string[];
    
    /**
     * Checks if any subscriber is registered for an event type.
     */
    hasSubscribers(eventType: string): boolean;
}

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

/**
 * In-Process Event Bus Implementation
 * 
 * A production-quality event bus for single-process applications.
 * Features:
 * - Type-safe event routing
 * - Parallel subscriber execution
 * - Error isolation between subscribers
 * - Priority-based ordering
 * - Graceful shutdown
 */
class InProcessEventBus implements IObservableEventBus {
    // Maps event types to their subscribers
    private readonly subscriptions: Map<string, Set<ISubscriber<unknown>>> = new Map();
    
    // Pending operations for graceful shutdown
    private readonly pendingOperations: Set<Promise<void>> = new Set();
    
    // Shutdown state
    private isShuttingDown = false;
    
    // Logger for debugging and monitoring
    private readonly logger: ILogger;
    
    constructor(logger: ILogger = console) {
        this.logger = logger;
    }
    
    /**
     * Registers a subscriber for its declared event types.
     */
    subscribe(subscriber: ISubscriber<unknown>): () => void {
        const eventTypes = subscriber.subscribedEventTypes();
        
        for (const eventType of eventTypes) {
            if (!this.subscriptions.has(eventType)) {
                this.subscriptions.set(eventType, new Set());
            }
            this.subscriptions.get(eventType)!.add(subscriber);
            
            this.logger.debug?.(`Subscriber registered for ${eventType}`);
        }
        
        // Return unsubscribe function for cleanup
        return () => this.unsubscribe(subscriber);
    }
    
    /**
     * Removes a subscriber from all its event types.
     */
    unsubscribe(subscriber: ISubscriber<unknown>): void {
        const eventTypes = subscriber.subscribedEventTypes();
        
        for (const eventType of eventTypes) {
            this.subscriptions.get(eventType)?.delete(subscriber);
            
            // Clean up empty sets
            if (this.subscriptions.get(eventType)?.size === 0) {
                this.subscriptions.delete(eventType);
            }
        }
    }
    
    /**
     * Publishes an event to all interested subscribers.
     * Executes subscribers in parallel for performance.
     */
    async publish<TEvent extends { eventType: string }>(event: TEvent): Promise<void> {
        if (this.isShuttingDown) {
            throw new Error('Event bus is shutting down, cannot publish');
        }
        
        const eventType = event.eventType;
        const subscribers = this.subscriptions.get(eventType);
        
        if (!subscribers || subscribers.size === 0) {
            this.logger.debug?.(`No subscribers for event type: ${eventType}`);
            return;
        }
        
        // Sort by priority (lower = higher priority)
        const sortedSubscribers = [...subscribers].sort(
            (a, b) => (a.priority || 100) - (b.priority || 100)
        );
        
        // Execute all subscribers in parallel
        const operation = this.executeSubscribers(sortedSubscribers, event);
        this.pendingOperations.add(operation);
        
        try {
            await operation;
        } finally {
            this.pendingOperations.delete(operation);
        }
    }
    
    /**
     * Executes subscribers with error isolation.
     * One subscriber's failure doesn't affect others.
     */
    private async executeSubscribers<TEvent>(
        subscribers: ISubscriber<TEvent>[],
        event: TEvent
    ): Promise<void> {
        const results = await Promise.allSettled(
            subscribers.map(subscriber => 
                this.executeWithErrorHandling(subscriber, event)
            )
        );
        
        // Log any failures
        for (let i = 0; i < results.length; i++) {
            const result = results[i];
            if (result.status === 'rejected') {
                this.logger.error(
                    `Subscriber ${subscribers[i].constructor.name} failed:`,
                    result.reason
                );
            }
        }
    }
    
    /**
     * Wraps subscriber execution with error handling.
     */
    private async executeWithErrorHandling<TEvent>(
        subscriber: ISubscriber<TEvent>,
        event: TEvent
    ): Promise<void> {
        try {
            await subscriber.handle(event);
        } catch (error) {
            // Error is caught and isolated - other subscribers continue
            // Re-throw to be caught by Promise.allSettled
            throw error;
        }
    }
    
    /**
     * Graceful shutdown: wait for pending operations.
     */
    async shutdown(): Promise<void> {
        this.isShuttingDown = true;
        
        if (this.pendingOperations.size > 0) {
            this.logger.info(`Waiting for ${this.pendingOperations.size} pending operations...`);
            await Promise.all(this.pendingOperations);
        }
        
        this.subscriptions.clear();
        this.logger.info('Event bus shut down successfully');
    }
    
    // Observable methods
    subscriberCount(eventType: string): number {
        return this.subscriptions.get(eventType)?.size || 0;
    }
    
    registeredEventTypes(): string[] {
        return [...this.subscriptions.keys()];
    }
    
    hasSubscribers(eventType: string): boolean {
        return this.subscriberCount(eventType) > 0;
    }
}

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

/**
 * Synchronous Event Bus
 * 
 * publish() awaits all subscriber completions.
 * Publisher blocks until all reactions finish.
 */
class SynchronousEventBus implements IEventBus {
    async publish<TEvent>(event: TEvent): Promise<void> {
        const subscribers = this.getSubscribers(event);
        
        // Publisher waits for ALL subscribers to complete
        // This means publish() can take a long time
        await Promise.all(
            subscribers.map(s => s.handle(event))
        );
    }
}
 
// Usage: Publisher blocks until all handlers complete
async function placeOrder(command: PlaceOrderCommand): Promise<Order> {
    const order = await createOrder(command);
    
    // This line blocks until:
    // - Email is sent
    // - Inventory is reserved
    // - Analytics are tracked
    // - All other subscribers complete
    await eventBus.publish(orderPlacedEvent);
    
    return order;  // Only returns after all reactions
}
 
 
/**
 * Asynchronous Event Bus
 * 
 * publish() returns immediately after queuing.
 * Handlers execute in background.
 */
class AsynchronousEventBus implements IEventBus {
    private readonly queue: AsyncQueue<{event: unknown, subscribers: ISubscriber<unknown>[]}>;
    
    constructor() {
        this.queue = new AsyncQueue();
        this.startBackgroundProcessor();
    }
    
    async publish<TEvent>(event: TEvent): Promise<void> {
        const subscribers = this.getSubscribers(event);
        
        // Queue the event and return immediately
        // Publisher is not blocked
        await this.queue.enqueue({ event, subscribers });
        
        // Returns here without waiting for handlers
    }
    
    private async startBackgroundProcessor(): Promise<void> {
        while (true) {
            const { event, subscribers } = await this.queue.dequeue();
            
            // Process in background - not blocking any publisher
            await Promise.allSettled(
                subscribers.map(s => s.handle(event))
            );
        }
    }
}
 
// Usage: Publisher returns immediately
async function placeOrder(command: PlaceOrderCommand): Promise<Order> {
    const order = await createOrder(command);
    
    // This returns IMMEDIATELY
    // Email, inventory, analytics happen in background
    await eventBus.publish(orderPlacedEvent);
    
    return order;  // Returns before reactions complete
}
 
 
/**
 * Hybrid Event Bus
 * 
 * Supports both modes based on configuration or event properties.
 * Best of both worlds for mixed requirements.
 */
class HybridEventBus implements IEventBus {
    async publish<TEvent extends { async?: boolean }>(event: TEvent): Promise<void> {
        const subscribers = this.getSubscribers(event);
        
        if (event.async) {
            // Fire-and-forget for async events
            setImmediate(() => this.executeSubscribers(subscribers, event));
        } else {
            // Wait for completion for sync events
            await this.executeSubscribers(subscribers, event);
        }
    }
    
    private async executeSubscribers<TEvent>(
        subscribers: ISubscriber<TEvent>[],
        event: TEvent
    ): Promise<void> {
        await Promise.allSettled(
            subscribers.map(s => s.handle(event))
        );
    }
}

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

/**
 * Event bus with content-based filtering
 * 
 * Subscribers can specify predicates beyond event type.
 */
interface IFilterableSubscriber<TEvent> extends ISubscriber<TEvent> {
    /**
     * Optional filter predicate.
     * Returns true if this subscriber should handle the event.
     */
    filter?(event: TEvent): boolean;
}
 
class FilterableEventBus implements IEventBus {
    async publish<TEvent>(event: TEvent): Promise<void> {
        const subscribers = this.getSubscribers(event);
        
        // Filter subscribers based on their predicates
        const eligibleSubscribers = subscribers.filter(sub => {
            const filterableSub = sub as IFilterableSubscriber<TEvent>;
            if (filterableSub.filter) {
                return filterableSub.filter(event);
            }
            return true;  // No filter = receive all
        });
        
        await this.executeSubscribers(eligibleSubscribers, event);
    }
}
 
// Example: Subscriber that only handles high-value orders
class HighValueOrderSubscriber implements IFilterableSubscriber<OrderPlacedEvent> {
    subscribedEventTypes(): string[] {
        return ['OrderPlaced'];
    }
    
    // Only handle orders over $1000
    filter(event: OrderPlacedEvent): boolean {
        return event.payload.totalAmount > 1000;
    }
    
    async handle(event: OrderPlacedEvent): Promise<void> {
        // This only runs for high-value orders
        await this.notifyVIPTeam(event);
        await this.schedulePriorityFulfillment(event);
    }
}
 
// Example: Subscriber filtered by customer segment
class EnterpriseCustomerSubscriber implements IFilterableSubscriber<OrderPlacedEvent> {
    subscribedEventTypes(): string[] {
        return ['OrderPlaced'];
    }
    
    filter(event: OrderPlacedEvent): boolean {
        return event.payload.customerSegment === 'enterprise';
    }
    
    async handle(event: OrderPlacedEvent): Promise<void> {
        await this.assignAccountManager(event);
    }
}

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

/**
 * Event Middleware: Cross-cutting concerns for all events
 */
type EventMiddleware = (
    event: unknown,
    next: () => Promise<void>
) => Promise<void>;
 
class MiddlewareEventBus implements IEventBus {
    private readonly middlewares: EventMiddleware[] = [];
    
    use(middleware: EventMiddleware): void {
        this.middlewares.push(middleware);
    }
    
    async publish<TEvent>(event: TEvent): Promise<void> {
        // Build middleware chain
        const executeSubscribers = async () => {
            const subscribers = this.getSubscribers(event);
            await Promise.allSettled(
                subscribers.map(s => s.handle(event))
            );
        };
        
        // Execute middleware chain
        const chain = this.buildMiddlewareChain(event, executeSubscribers);
        await chain();
    }
    
    private buildMiddlewareChain(
        event: unknown,
        finalHandler: () => Promise<void>
    ): () => Promise<void> {
        let current = finalHandler;
        
        // Build chain from end to start
        for (let i = this.middlewares.length - 1; i >= 0; i--) {
            const middleware = this.middlewares[i];
            const next = current;
            current = () => middleware(event, next);
        }
        
        return current;
    }
}
 
// Example middlewares
const loggingMiddleware: EventMiddleware = async (event, next) => {
    console.log(`[EVENT] Publishing: ${(event as any).eventType}`);
    const start = Date.now();
    
    try {
        await next();
        console.log(`[EVENT] Completed: ${(event as any).eventType} in ${Date.now() - start}ms`);
    } catch (error) {
        console.error(`[EVENT] Failed: ${(event as any).eventType}`, error);
        throw error;
    }
};
 
const metricsMiddleware: EventMiddleware = async (event, next) => {
    metrics.increment(`events.published.${(event as any).eventType}`);
    const timer = metrics.timer(`events.duration.${(event as any).eventType}`);
    
    try {
        await next();
        timer.stop();
    } catch (error) {
        metrics.increment(`events.failed.${(event as any).eventType}`);
        throw error;
    }
};
 
const tracingMiddleware: EventMiddleware = async (event, next) => {
    const span = tracer.startSpan(`event:${(event as any).eventType}`);
    span.setTag('event.id', (event as any).eventId);
    
    try {
        await next();
        span.setStatus('ok');
    } catch (error) {
        span.setStatus('error');
        span.setTag('error.message', (error as Error).message);
        throw error;
    } finally {
        span.finish();
    }
};
 
// Usage
const eventBus = new MiddlewareEventBus();
eventBus.use(loggingMiddleware);
eventBus.use(metricsMiddleware);
eventBus.use(tracingMiddleware);

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

/**
 * Dead Letter Queue Interface
 * 
 * Stores failed events for investigation and replay.
 */
interface IDeadLetterQueue {
    /**
     * Stores a failed event with failure context.
     */
    store(entry: DeadLetterEntry): Promise<void>;
    
    /**
     * Retrieves failed events for investigation.
     */
    list(options: ListOptions): Promise<DeadLetterEntry[]>;
    
    /**
     * Replays a dead letter entry.
     */
    replay(entryId: string): Promise<void>;
    
    /**
     * Removes an entry after investigation.
     */
    remove(entryId: string): Promise<void>;
}
 
interface DeadLetterEntry {
    id: string;
    event: unknown;
    eventType: string;
    subscriberName: string;
    failedAt: Date;
    errorMessage: string;
    errorStack?: string;
    attemptCount: number;
    originalPublishTime: Date;
}
 
class EventBusWithDLQ implements IEventBus {
    private readonly maxRetries = 3;
    
    constructor(
        private readonly deadLetterQueue: IDeadLetterQueue,
        private readonly logger: ILogger
    ) {}
    
    async publish<TEvent extends DomainEvent>(event: TEvent): Promise<void> {
        const subscribers = this.getSubscribers(event);
        
        await Promise.allSettled(
            subscribers.map(sub => 
                this.executeWithRetryAndDLQ(sub, event)
            )
        );
    }
    
    private async executeWithRetryAndDLQ<TEvent extends DomainEvent>(
        subscriber: ISubscriber<TEvent>,
        event: TEvent
    ): Promise<void> {
        let lastError: Error | null = null;
        
        for (let attempt = 1; attempt <= this.maxRetries; attempt++) {
            try {
                await subscriber.handle(event);
                return;  // Success
            } catch (error) {
                lastError = error as Error;
                this.logger.warn(
                    `Attempt ${attempt}/${this.maxRetries} failed for ${subscriber.constructor.name}`,
                    error
                );
                
                if (attempt < this.maxRetries) {
                    await this.sleep(Math.pow(2, attempt) * 100);  // Exponential backoff
                }
            }
        }
        
        // All retries exhausted - send to DLQ
        await this.deadLetterQueue.store({
            id: generateUUID(),
            event: event,
            eventType: event.eventType,
            subscriberName: subscriber.constructor.name,
            failedAt: new Date(),
            errorMessage: lastError?.message || 'Unknown error',
            errorStack: lastError?.stack,
            attemptCount: this.maxRetries,
            originalPublishTime: event.occurredAt
        });
        
        this.logger.error(
            `Event ${event.eventId} sent to DLQ for subscriber ${subscriber.constructor.name}`
        );
    }
    
    private sleep(ms: number): Promise<void> {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}

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

/**
 * Event Bus Abstraction
 * 
 * Same interface works for both in-process and distributed.
 * Swap implementations without changing business code.
 */
 
// Same interface for all implementations
interface IEventBus {
    publish<TEvent>(event: TEvent): Promise<void>;
    subscribe(subscriber: ISubscriber<unknown>): () => void;
    shutdown(): Promise<void>;
}
 
// In-Process Implementation
class InProcessEventBus implements IEventBus { /* ... */ }
 
// RabbitMQ Implementation
class RabbitMQEventBus implements IEventBus {
    constructor(private readonly connection: amqp.Connection) {}
    
    async publish<TEvent extends DomainEvent>(event: TEvent): Promise<void> {
        const channel = await this.connection.createChannel();
        
        // Serialize event for network transport
        const message = JSON.stringify(event);
        
        // Publish to exchange
        await channel.publish(
            'domain-events',      // Exchange name
            event.eventType,      // Routing key
            Buffer.from(message),
            {
                persistent: true,  // Survive broker restart
                contentType: 'application/json',
                messageId: event.eventId,
                timestamp: event.occurredAt.getTime()
            }
        );
    }
    
    subscribe(subscriber: ISubscriber<unknown>): () => void {
        // Create queue bound to relevant event types
        const eventTypes = subscriber.subscribedEventTypes();
        
        this.setupConsumer(subscriber, eventTypes);
        
        return () => this.teardownConsumer(subscriber);
    }
    
    private async setupConsumer(
        subscriber: ISubscriber<unknown>,
        eventTypes: string[]
    ): Promise<void> {
        const channel = await this.connection.createChannel();
        const queueName = `${subscriber.constructor.name}-queue`;
        
        // Create durable queue
        await channel.assertQueue(queueName, { durable: true });
        
        // Bind to each event type
        for (const eventType of eventTypes) {
            await channel.bindQueue(queueName, 'domain-events', eventType);
        }
        
        // Consume messages
        await channel.consume(queueName, async (msg) => {
            if (!msg) return;
            
            try {
                const event = JSON.parse(msg.content.toString());
                await subscriber.handle(event);
                channel.ack(msg);  // Acknowledge success
            } catch (error) {
                channel.nack(msg, false, true);  // Requeue for retry
            }
        });
    }
    
    async shutdown(): Promise<void> {
        await this.connection.close();
    }
}
 
// Kafka Implementation
class KafkaEventBus implements IEventBus {
    constructor(
        private readonly producer: KafkaProducer,
        private readonly consumerGroup: string
    ) {}
    
    async publish<TEvent extends DomainEvent>(event: TEvent): Promise<void> {
        await this.producer.send({
            topic: 'domain-events',
            messages: [{
                key: event.aggregateId,  // Partition by aggregate for ordering
                value: JSON.stringify(event),
                headers: {
                    'event-type': event.eventType,
                    'event-id': event.eventId
                }
            }]
        });
    }
    
    // Kafka consumer implementation...
}
 
// Dependency Injection allows swapping implementations
function configureEventBus(config: Config): IEventBus {
    switch (config.eventBusType) {
        case 'in-process':
            return new InProcessEventBus();
        case 'rabbitmq':
            return new RabbitMQEventBus(config.rabbitConnection);
        case 'kafka':
            return new KafkaEventBus(config.kafkaProducer, config.consumerGroup);
        default:
            throw new Error(`Unknown event bus type: ${config.eventBusType}`);
    }
}

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

/**
 * Test Double Event Bus
 * 
 * Captures published events for assertions.
 * Optionally executes real subscribers for integration tests.
 */
class TestEventBus implements IEventBus {
    private readonly publishedEvents: unknown[] = [];
    private readonly subscribers: Map<string, Set<ISubscriber<unknown>>> = new Map();
    private executeSubscribers = false;
    
    /**
     * Enable subscriber execution for integration tests.
     */
    withSubscriberExecution(): this {
        this.executeSubscribers = true;
        return this;
    }
    
    async publish<TEvent extends { eventType: string }>(event: TEvent): Promise<void> {
        // Capture the event
        this.publishedEvents.push(event);
        
        // Optionally execute subscribers
        if (this.executeSubscribers) {
            const subs = this.subscribers.get(event.eventType) || new Set();
            await Promise.allSettled(
                [...subs].map(s => s.handle(event))
            );
        }
    }
    
    subscribe(subscriber: ISubscriber<unknown>): () => void {
        for (const eventType of subscriber.subscribedEventTypes()) {
            if (!this.subscribers.has(eventType)) {
                this.subscribers.set(eventType, new Set());
            }
            this.subscribers.get(eventType)!.add(subscriber);
        }
        return () => this.unsubscribe(subscriber);
    }
    
    unsubscribe(subscriber: ISubscriber<unknown>): void {
        for (const eventType of subscriber.subscribedEventTypes()) {
            this.subscribers.get(eventType)?.delete(subscriber);
        }
    }
    
    async shutdown(): Promise<void> {
        // No-op for test double
    }
    
    // ============ Test Assertions ============
    
    /**
     * Returns all published events.
     */
    getPublishedEvents(): unknown[] {
        return [...this.publishedEvents];
    }
    
    /**
     * Returns published events of a specific type.
     */
    getEventsOfType<TEvent>(eventType: string): TEvent[] {
        return this.publishedEvents.filter(
            e => (e as any).eventType === eventType
        ) as TEvent[];
    }
    
    /**
     * Asserts that an event of the given type was published.
     */
    expectEventPublished(eventType: string): void {
        const found = this.publishedEvents.some(
            e => (e as any).eventType === eventType
        );
        if (!found) {
            throw new Error(`Expected event '${eventType}' to be published, but it wasn't`);
        }
    }
    
    /**
     * Asserts that no event of the given type was published.
     */
    expectEventNotPublished(eventType: string): void {
        const found = this.publishedEvents.some(
            e => (e as any).eventType === eventType
        );
        if (found) {
            throw new Error(`Expected event '${eventType}' NOT to be published, but it was`);
        }
    }
    
    /**
     * Asserts the exact number of events of a type were published.
     */
    expectEventCount(eventType: string, count: number): void {
        const actual = this.getEventsOfType(eventType).length;
        if (actual !== count) {
            throw new Error(
                `Expected ${count} '${eventType}' events, but found ${actual}`
            );
        }
    }
    
    /**
     * Clears all captured events (for test isolation).
     */
    reset(): void {
        this.publishedEvents.length = 0;
    }
}
 
// ============ Example Tests ============
 
describe('OrderService', () => {
    let eventBus: TestEventBus;
    let orderService: OrderService;
    
    beforeEach(() => {
        eventBus = new TestEventBus();
        orderService = new OrderService(eventBus, mockOrderRepository);
    });
    
    afterEach(() => {
        eventBus.reset();
    });
    
    it('publishes OrderPlaced event when order is created', async () => {
        // Act
        const order = await orderService.placeOrder({
            customerId: 'cust-123',
            items: [{ productId: 'prod-1', quantity: 2 }]
        });
        
        // Assert
        eventBus.expectEventPublished('OrderPlaced');
        
        const events = eventBus.getEventsOfType<OrderPlacedEvent>('OrderPlaced');
        expect(events).toHaveLength(1);
        expect(events[0].payload.orderId).toBe(order.id);
        expect(events[0].payload.customerId).toBe('cust-123');
    });
    
    it('does not publish OrderPlaced when validation fails', async () => {
        // Act & Assert
        await expect(
            orderService.placeOrder({ customerId: '', items: [] })
        ).rejects.toThrow('Invalid order');
        
        eventBus.expectEventNotPublished('OrderPlaced');
    });
});
 
describe('EmailSubscriber Integration', () => {
    let eventBus: TestEventBus;
    let emailService: MockEmailService;
    
    beforeEach(() => {
        eventBus = new TestEventBus().withSubscriberExecution();
        emailService = new MockEmailService();
        
        // Register real subscriber for integration test
        eventBus.subscribe(new EmailNotificationSubscriber(emailService));
    });
    
    it('sends email when OrderPlaced event is published', async () => {
        // Arrange
        const event = createOrderPlacedEvent({
            orderId: 'order-123',
            customerEmail: 'test@example.com'
        });
        
        // Act
        await eventBus.publish(event);
        
        // Assert
        expect(emailService.sentEmails).toHaveLength(1);
        expect(emailService.sentEmails[0].to).toBe('test@example.com');
    });
});