| 1 | import type { |
| 2 | RuntimeEventEnvelope, |
| 3 | RuntimeEventFilter, |
| 4 | RuntimeEventMap, |
| 5 | RuntimeEventType |
| 6 | } from './envelope' |
| 7 | |
| 8 | type Listener = (event: RuntimeEventEnvelope) => void |
| 9 | |
| 10 | type Subscriber = { |
| 11 | filter: RuntimeEventFilter |
| 12 | listener: Listener |
| 13 | } |
| 14 | |
| 15 | export type TypedEventBusOptions = { |
| 16 | onListenerError?: (error: unknown, event: RuntimeEventEnvelope) => void |
| 17 | } |
| 18 | |
| 19 | export class TypedEventBus { |
| 20 | private readonly subscribers = new Map<number, Subscriber>() |
| 21 | private nextSubscriberId = 1 |
| 22 | |
| 23 | constructor(private readonly options: TypedEventBusOptions = {}) {} |
| 24 | |
| 25 | emit<K extends RuntimeEventType>(event: RuntimeEventEnvelope<K>): void { |
| 26 | for (const subscriber of this.subscribers.values()) { |
| 27 | if (!this.matches(subscriber.filter, event)) continue |
| 28 | try { |
| 29 | subscriber.listener(event) |
| 30 | } catch (error) { |
| 31 | try { |
| 32 | this.options.onListenerError?.(error, event) |
| 33 | } catch { |
| 34 | // A diagnostic hook must never make one broken listener affect the job that emitted. |
| 35 | } |
| 36 | } |
| 37 | } |
| 38 | } |
| 39 | |
| 40 | subscribe(filter: RuntimeEventFilter, listener: Listener): () => void { |
| 41 | const subscriberId = this.nextSubscriberId |
| 42 | this.nextSubscriberId += 1 |
| 43 | this.subscribers.set(subscriberId, { filter, listener }) |
| 44 | return () => this.subscribers.delete(subscriberId) |
| 45 | } |
| 46 | |
| 47 | private matches(filter: RuntimeEventFilter, event: RuntimeEventEnvelope): boolean { |
| 48 | if (filter.domain && event.domain !== filter.domain) return false |
| 49 | if ( |
| 50 | filter.owner && |
| 51 | Object.entries(filter.owner).some(([key, value]) => { |
| 52 | const ownerKey = key as keyof typeof event.owner |
| 53 | return event.owner[ownerKey] !== value |
| 54 | }) |
| 55 | ) { |
| 56 | return false |
| 57 | } |
| 58 | if (event.audience.kind === 'broadcast') return true |
| 59 | if (event.audience.kind === 'requester') { |
| 60 | return event.audience.subscriberId === filter.subscriberId |
| 61 | } |
| 62 | |
| 63 | // Owner delivery is meaningful only for an explicit matching owner filter. |
| 64 | // The owner comparison above already established that this subscriber matches. |
| 65 | return Boolean(filter.owner && Object.keys(filter.owner).length > 0) |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | export type { RuntimeEventEnvelope, RuntimeEventFilter, RuntimeEventMap, RuntimeEventType } |
| 70 |