import events from "node:events"; import logger from "./util/logger"; type ListenerKey = object; interface EventBusMap { adapterDisconnected: []; permitJoinChanged: [data: eventdata.PermitJoinChanged]; publishAvailability: []; entityRenamed: [data: eventdata.EntityRenamed]; entityRemoved: [data: eventdata.EntityRemoved]; lastSeenChanged: [data: eventdata.LastSeenChanged]; deviceNetworkAddressChanged: [data: eventdata.DeviceNetworkAddressChanged]; deviceAnnounce: [data: eventdata.DeviceAnnounce]; deviceInterview: [data: eventdata.DeviceInterview]; deviceJoined: [data: eventdata.DeviceJoined]; entityOptionsChanged: [data: eventdata.EntityOptionsChanged]; exposesChanged: [data: eventdata.ExposesChanged]; deviceLeave: [data: eventdata.DeviceLeave]; deviceMessage: [data: eventdata.DeviceMessage]; mqttMessage: [data: eventdata.MQTTMessage]; mqttMessagePublished: [data: eventdata.MQTTMessagePublished]; publishEntityState: [data: eventdata.PublishEntityState]; groupMembersChanged: [data: eventdata.GroupMembersChanged]; devicesChanged: []; scenesChanged: [data: eventdata.ScenesChanged]; reconfigure: [data: eventdata.Reconfigure]; stateChange: [data: eventdata.StateChange]; } type EventBusListener = K extends keyof EventBusMap ? EventBusMap[K] extends unknown[] ? (...args: EventBusMap[K]) => Promise | void : never : never; type Stats = { devices: Map< string, // IEEE address { lastSeenChanges?: {messages: number; first: number}; leaveCounts: number; networkAddressChanges: number; } >; mqtt: { published: number; received: number; }; }; export default class EventBus { private callbacksByExtension = new Map}[]>(); private emitter = new events.EventEmitter(); readonly stats: Stats = {devices: new Map(), mqtt: {published: 0, received: 0}}; constructor() { this.emitter.setMaxListeners(100); } public emitAdapterDisconnected(): void { this.emitter.emit("adapterDisconnected"); } public onAdapterDisconnected(key: ListenerKey, callback: () => Promise): void { this.on("adapterDisconnected", callback, key); } public emitPermitJoinChanged(data: eventdata.PermitJoinChanged): void { this.emitter.emit("permitJoinChanged", data); } public onPermitJoinChanged(key: ListenerKey, callback: (data: eventdata.PermitJoinChanged) => void): void { this.on("permitJoinChanged", callback, key); } public emitEntityRenamed(data: eventdata.EntityRenamed): void { this.emitter.emit("entityRenamed", data); } public onEntityRenamed(key: ListenerKey, callback: (data: eventdata.EntityRenamed) => void): void { this.on("entityRenamed", callback, key); } public emitEntityRemoved(data: eventdata.EntityRemoved): void { this.emitter.emit("entityRemoved", data); } public onEntityRemoved(key: ListenerKey, callback: (data: eventdata.EntityRemoved) => void): void { this.on("entityRemoved", callback, key); } public emitLastSeenChanged(data: eventdata.LastSeenChanged): void { this.emitter.emit("lastSeenChanged", data); const device = this.stats.devices.get(data.device.ieeeAddr); if (device?.lastSeenChanges) { device.lastSeenChanges.messages += 1; } else { this.stats.devices.set(data.device.ieeeAddr, { lastSeenChanges: {messages: 1, first: Date.now()}, leaveCounts: 0, networkAddressChanges: 0, }); } } public onLastSeenChanged(key: ListenerKey, callback: (data: eventdata.LastSeenChanged) => void): void { this.on("lastSeenChanged", callback, key); } public emitDeviceNetworkAddressChanged(data: eventdata.DeviceNetworkAddressChanged): void { this.emitter.emit("deviceNetworkAddressChanged", data); const device = this.stats.devices.get(data.device.ieeeAddr); if (device) { device.networkAddressChanges += 1; } else { this.stats.devices.set(data.device.ieeeAddr, {leaveCounts: 0, networkAddressChanges: 1}); } } public onDeviceNetworkAddressChanged(key: ListenerKey, callback: (data: eventdata.DeviceNetworkAddressChanged) => void): void { this.on("deviceNetworkAddressChanged", callback, key); } public emitDeviceAnnounce(data: eventdata.DeviceAnnounce): void { this.emitter.emit("deviceAnnounce", data); } public onDeviceAnnounce(key: ListenerKey, callback: (data: eventdata.DeviceAnnounce) => void): void { this.on("deviceAnnounce", callback, key); } public emitDeviceInterview(data: eventdata.DeviceInterview): void { this.emitter.emit("deviceInterview", data); } public onDeviceInterview(key: ListenerKey, callback: (data: eventdata.DeviceInterview) => void): void { this.on("deviceInterview", callback, key); } public emitDeviceJoined(data: eventdata.DeviceJoined): void { this.emitter.emit("deviceJoined", data); } public onDeviceJoined(key: ListenerKey, callback: (data: eventdata.DeviceJoined) => void): void { this.on("deviceJoined", callback, key); } public emitEntityOptionsChanged(data: eventdata.EntityOptionsChanged): void { this.emitter.emit("entityOptionsChanged", data); } public onEntityOptionsChanged(key: ListenerKey, callback: (data: eventdata.EntityOptionsChanged) => void): void { this.on("entityOptionsChanged", callback, key); } public emitExposesChanged(data: eventdata.ExposesChanged): void { this.emitter.emit("exposesChanged", data); } public onExposesChanged(key: ListenerKey, callback: (data: eventdata.ExposesChanged) => void): void { this.on("exposesChanged", callback, key); } public emitDeviceLeave(data: eventdata.DeviceLeave): void { this.emitter.emit("deviceLeave", data); const device = this.stats.devices.get(data.ieeeAddr); if (device) { device.leaveCounts += 1; } else { this.stats.devices.set(data.ieeeAddr, {leaveCounts: 1, networkAddressChanges: 0}); } } public onDeviceLeave(key: ListenerKey, callback: (data: eventdata.DeviceLeave) => void): void { this.on("deviceLeave", callback, key); } public emitDeviceMessage(data: eventdata.DeviceMessage): void { this.emitter.emit("deviceMessage", data); } public onDeviceMessage(key: ListenerKey, callback: (data: eventdata.DeviceMessage) => void): void { this.on("deviceMessage", callback, key); } public emitMQTTMessage(data: eventdata.MQTTMessage): void { this.emitter.emit("mqttMessage", data); this.stats.mqtt.received += 1; } public onMQTTMessage(key: ListenerKey, callback: (data: eventdata.MQTTMessage) => void): void { this.on("mqttMessage", callback, key); } public emitMQTTMessagePublished(data: eventdata.MQTTMessagePublished): void { this.emitter.emit("mqttMessagePublished", data); this.stats.mqtt.published += 1; } public onMQTTMessagePublished(key: ListenerKey, callback: (data: eventdata.MQTTMessagePublished) => void): void { this.on("mqttMessagePublished", callback, key); } public emitPublishEntityState(data: eventdata.PublishEntityState): void { this.emitter.emit("publishEntityState", data); } public onPublishEntityState(key: ListenerKey, callback: (data: eventdata.PublishEntityState) => void): void { this.on("publishEntityState", callback, key); } public emitGroupMembersChanged(data: eventdata.GroupMembersChanged): void { this.emitter.emit("groupMembersChanged", data); } public onGroupMembersChanged(key: ListenerKey, callback: (data: eventdata.GroupMembersChanged) => void): void { this.on("groupMembersChanged", callback, key); } public emitDevicesChanged(): void { this.emitter.emit("devicesChanged"); } public onDevicesChanged(key: ListenerKey, callback: () => void): void { this.on("devicesChanged", callback, key); } public emitScenesChanged(data: eventdata.ScenesChanged): void { this.emitter.emit("scenesChanged", data); } public onScenesChanged(key: ListenerKey, callback: (data: eventdata.ScenesChanged) => void): void { this.on("scenesChanged", callback, key); } public emitReconfigure(data: eventdata.Reconfigure): void { this.emitter.emit("reconfigure", data); } public onReconfigure(key: ListenerKey, callback: (data: eventdata.Reconfigure) => void): void { this.on("reconfigure", callback, key); } public emitStateChange(data: eventdata.StateChange): void { this.emitter.emit("stateChange", data); } public onStateChange(key: ListenerKey, callback: (data: eventdata.StateChange) => void): void { this.on("stateChange", callback, key); } public emitExposesAndDevicesChanged(device: Device): void { this.emitDevicesChanged(); this.emitExposesChanged({device}); } private on(event: K, callback: EventBusListener, key: ListenerKey): void { if (!this.callbacksByExtension.has(key.constructor.name)) { this.callbacksByExtension.set(key.constructor.name, []); } const wrappedCallback = async (...args: never[]): Promise => { try { await callback(...args); } catch (error) { logger.error(`EventBus error '${key.constructor.name}/${event}': ${(error as Error).message}`); // biome-ignore lint/style/noNonNullAssertion: always Error logger.debug((error as Error).stack!); } }; // biome-ignore lint/style/noNonNullAssertion: just created if wasn't valid this.callbacksByExtension.get(key.constructor.name)!.push({event, callback: wrappedCallback}); this.emitter.on(event, wrappedCallback as EventBusListener); } public removeListeners(key: ListenerKey): void { const callbacks = this.callbacksByExtension.get(key.constructor.name); if (callbacks) { for (const cb of callbacks) { this.emitter.removeListener(cb.event, cb.callback); } } } }