From b483b0a61316f303a9d5d5df782f6efc354961f5 Mon Sep 17 00:00:00 2001 From: Noley Holland Date: Thu, 16 Jul 2026 11:02:25 -0700 Subject: [PATCH 1/3] Add attachClient/detachClient so subscribers can share one managed client per key --- frontend/src/services/mqtt.service.ts | 190 ++++++++++++------ frontend/src/types/services/mqtt.types.ts | 14 +- .../frontend/services/mqtt.service.spec.js | 76 +++++++ 3 files changed, 217 insertions(+), 63 deletions(-) diff --git a/frontend/src/services/mqtt.service.ts b/frontend/src/services/mqtt.service.ts index 27d56c4824..f17d86dea6 100644 --- a/frontend/src/services/mqtt.service.ts +++ b/frontend/src/services/mqtt.service.ts @@ -14,7 +14,10 @@ import { BaseService } from './service.contract' import { Maybe } from '@/types/common/types' import type { BinaryPayload, + ManagedMqttAttachment, ManagedMqttClient, + MqttAttachmentHandle, + MqttConnectionHandlers, MqttConnectionOptions, MqttConnectionWaiter, MqttCredentialProvider, @@ -48,6 +51,8 @@ class MqttService extends BaseService implements MqttServiceI { protected $destroyed: boolean + protected $attachmentSeq: number + constructor ({ app, router, services }: CreateServiceOptions) { super({ name: 'mqtt', @@ -60,6 +65,7 @@ class MqttService extends BaseService implements MqttServiceI { this.$clients = new Map() this.$clientOperations = new Map() this.$destroyed = false + this.$attachmentSeq = 0 this.init() } @@ -99,75 +105,135 @@ class MqttService extends BaseService implements MqttServiceI { } async createClient (key: string, options: Partial = {}): Promise { - const { - reconnectPeriod = 0, - getCredentials, - reconnect, - onConnect, - onClose, - onOffline, - onError, - onMessage - } = options - if (!key || typeof key !== 'string') { throw new Error('MQTT connection key is required') } - if (typeof getCredentials !== 'function') { + if (typeof options.getCredentials !== 'function') { throw new Error(`MQTT connection "${key}" requires a getCredentials callback`) } - return this.runClientOperation(key, async () => { - if (this.$destroyed) { - throw new Error('MqttService has been destroyed') - } + const attachment = this._buildAttachment(options) + return this.runClientOperation(key, async () => { if (this.hasClient(key)) { await this._destroyClientUnlocked(key) } + return this._createAndConnect(key, options, attachment) + }) + } + + async attachClient (key: string, options: Partial = {}): Promise { + if (!key || typeof key !== 'string') { + throw new Error('MQTT connection key is required') + } - const managed: ManagedMqttClient = { - key, - client: null, - listeners: new Set(), - destroyed: false, - intentionalDisconnect: false, - status: 'idle', - getCredentials: this._buildCredentialsProvider(key, getCredentials), - reconnectPolicy: this._normalizeReconnectPolicy({ - reconnect, - reconnectPeriod, - hasDynamicCredentials: true - }), - reconnectAttempt: 0, - reconnectGeneration: 0, - reconnectTimer: null, - subscriptions: new Map(), - connectionWaiters: new Set(), - terminalFailure: false, - lastError: null, - connectHandlerPromise: null, - handlers: { - onConnect, - onClose, - onOffline, - onError, - onMessage + const attachment = this._buildAttachment(options) + + await this.runClientOperation(key, async () => { + const existing = this.$clients.get(key) + if (existing && !existing.destroyed) { + existing.attachments.add(attachment) + if (existing.status === 'connected' && existing.client?.connected) { + attachment.handlers.onConnect?.({} as IConnackPacket, existing.client) } + return } - this.$clients.set(key, managed) + if (typeof options.getCredentials !== 'function') { + throw new Error(`MQTT connection "${key}" requires a getCredentials callback`) + } + await this._createAndConnect(key, options, attachment) + }) - try { - return await this.connectManagedClient(managed, false) - } catch (error) { - this.$clients.delete(key) - throw error + return { key, id: attachment.id } + } + + async detachClient (handle: MqttAttachmentHandle): Promise { + const { key, id } = handle + await this.runClientOperation(key, async () => { + const managed = this.$clients.get(key) + if (!managed) return + + for (const attachment of managed.attachments) { + if (attachment.id === id) { + managed.attachments.delete(attachment) + break + } + } + + if (managed.attachments.size === 0) { + await this._destroyClientUnlocked(key) } }) } + private _buildAttachment (options: Partial): ManagedMqttAttachment { + const { onConnect, onClose, onOffline, onError, onMessage } = options + return { + id: this.$attachmentSeq++, + handlers: { onConnect, onClose, onOffline, onError, onMessage } + } + } + + private _dispatch ( + managed: ManagedMqttClient, + event: K, + ...args: Parameters> + ): void { + for (const attachment of managed.attachments) { + const handler = attachment.handlers[event] + if (handler) { + ;(handler as (...a: unknown[]) => void)(...args) + } + } + } + + private async _createAndConnect ( + key: string, + options: Partial, + attachment: ManagedMqttAttachment + ): Promise { + if (this.$destroyed) { + throw new Error('MqttService has been destroyed') + } + + const { reconnectPeriod = 0, getCredentials, reconnect } = options + + const managed: ManagedMqttClient = { + key, + client: null, + listeners: new Set(), + destroyed: false, + intentionalDisconnect: false, + status: 'idle', + getCredentials: this._buildCredentialsProvider(key, getCredentials as MqttCredentialProvider), + reconnectPolicy: this._normalizeReconnectPolicy({ + reconnect, + reconnectPeriod, + hasDynamicCredentials: true + }), + reconnectAttempt: 0, + reconnectGeneration: 0, + reconnectTimer: null, + subscriptions: new Map(), + attachments: new Set([attachment]), + connectionWaiters: new Set(), + terminalFailure: false, + lastError: null, + connectHandlerPromise: null + } + + this.$clients.set(key, managed) + + try { + return await this.connectManagedClient(managed, false) + } catch (error) { + this.$clients.delete(key) + throw error + } + } + async publishMessage (key: string, options?: MqttPublishRequest): Promise { const managed = this.getManagedClient(key) @@ -505,11 +571,11 @@ class MqttService extends BaseService implements MqttServiceI { await this.replaySubscriptions(managed, client) } catch (error) { if (error instanceof Error) { - managed.handlers.onError?.(error, client) + this._dispatch(managed, 'onError', error, client) } } - managed.handlers.onConnect?.(connack, client) + this._dispatch(managed, 'onConnect', connack, client) } managed.connectHandlerPromise = handleAsync().finally(() => { @@ -519,46 +585,46 @@ class MqttService extends BaseService implements MqttServiceI { register('close', () => { managed.status = 'disconnected' - managed.handlers.onClose?.(client) + this._dispatch(managed, 'onClose', client) this.scheduleReconnect(managed) }) register('disconnect', (packet: IDisconnectPacket) => { - managed.handlers.onDisconnect?.(packet, client) + this._dispatch(managed, 'onDisconnect', packet, client) }) register('offline', () => { managed.status = 'disconnected' - managed.handlers.onOffline?.(client) + this._dispatch(managed, 'onOffline', client) this.scheduleReconnect(managed) }) register('end', () => { - managed.handlers.onEnd?.(client) + this._dispatch(managed, 'onEnd', client) }) register('reconnect', () => { - managed.handlers.onReconnect?.(client) + this._dispatch(managed, 'onReconnect', client) }) register('error', (error: Error | ErrorWithReasonCode) => { - managed.handlers.onError?.(error, client) + this._dispatch(managed, 'onError', error, client) }) register('message', (topic: string, message: Buffer, packet: IPublishPacket) => { - managed.handlers.onMessage?.(topic, message, packet, client) + this._dispatch(managed, 'onMessage', topic, message, packet, client) }) register('packetsend', (packet: Packet) => { - managed.handlers.onPacketSend?.(packet, client) + this._dispatch(managed, 'onPacketSend', packet, client) }) register('packetreceive', (packet: Packet) => { - managed.handlers.onPacketReceive?.(packet, client) + this._dispatch(managed, 'onPacketReceive', packet, client) }) register('outgoingEmpty', () => { - managed.handlers.onOutgoingEmpty?.(client) + this._dispatch(managed, 'onOutgoingEmpty', client) }) } @@ -619,7 +685,7 @@ class MqttService extends BaseService implements MqttServiceI { this.clearReconnectTimer(managed) this.rejectConnectionWaiters(managed, normalizedError) } - managed.handlers.onError?.(normalizedError, managed.client) + this._dispatch(managed, 'onError', normalizedError, managed.client) if (!managed.terminalFailure) { this.scheduleReconnect(managed) } diff --git a/frontend/src/types/services/mqtt.types.ts b/frontend/src/types/services/mqtt.types.ts index e14dfbb082..9a42c816a8 100644 --- a/frontend/src/types/services/mqtt.types.ts +++ b/frontend/src/types/services/mqtt.types.ts @@ -87,6 +87,16 @@ export interface MqttConnectionWaiter { timer: ReturnType | null } +export interface ManagedMqttAttachment { + id: number + handlers: MqttConnectionHandlers +} + +export interface MqttAttachmentHandle { + key: string + id: number +} + export interface ManagedMqttClient { key: string client: MqttClient | null @@ -100,7 +110,7 @@ export interface ManagedMqttClient { reconnectGeneration: number reconnectTimer: ReturnType | null subscriptions: Map - handlers: MqttConnectionHandlers + attachments: Set connectionWaiters: Set terminalFailure: boolean lastError: Error | null @@ -110,6 +120,8 @@ export interface ManagedMqttClient { export interface MqttServiceI extends AppService { createClient(key: string, options?: Partial): Promise destroyClient(key: string): Promise + attachClient(key: string, options?: Partial): Promise + detachClient(handle: MqttAttachmentHandle): Promise getManagedClient(key: string): ManagedMqttClient | null hasClient(key: string): boolean publishMessage(key: string, options: MqttPublishRequest): Promise diff --git a/test/unit/frontend/services/mqtt.service.spec.js b/test/unit/frontend/services/mqtt.service.spec.js index 4465344d09..c2e9ee39e9 100644 --- a/test/unit/frontend/services/mqtt.service.spec.js +++ b/test/unit/frontend/services/mqtt.service.spec.js @@ -944,4 +944,80 @@ describe('MqttService', async () => { await destroyMqttService() await destroyMqttService() }) + + describe('attachClient / detachClient (shared refcounted connections)', () => { + const creds = async () => ({ url: 'mqtt://example.com', username: 'user', password: 'pass' }) + + test('a second attach on the same key reuses the existing client', async () => { + const service = createMqttService({ app: {}, services: {}, router: {} }) + const client = createMockClient() + mockConnect.mockReturnValue(client) + + const first = await service.attachClient('team:team-1', { getCredentials: creds }) + const second = await service.attachClient('team:team-1', { getCredentials: creds }) + + expect(mockConnect).toHaveBeenCalledTimes(1) + expect(first.id).not.toBe(second.id) + expect(service.getManagedClient('team:team-1').attachments.size).toBe(2) + }) + + test('detaching a non-last attachment keeps the connection alive', async () => { + const service = createMqttService({ app: {}, services: {}, router: {} }) + mockConnect.mockReturnValue(createMockClient()) + + const first = await service.attachClient('team:team-1', { getCredentials: creds }) + const second = await service.attachClient('team:team-1', { getCredentials: creds }) + + await service.detachClient(first) + expect(service.hasClient('team:team-1')).toBe(true) + expect(service.getManagedClient('team:team-1').attachments.size).toBe(1) + + await service.detachClient(second) + expect(service.hasClient('team:team-1')).toBe(false) + }) + + test('detaching an unknown handle is a no-op', async () => { + const service = createMqttService({ app: {}, services: {}, router: {} }) + mockConnect.mockReturnValue(createMockClient()) + await service.attachClient('team:team-1', { getCredentials: creds }) + + await service.detachClient({ key: 'team:team-1', id: 999 }) + expect(service.hasClient('team:team-1')).toBe(true) + expect(service.getManagedClient('team:team-1').attachments.size).toBe(1) + }) + + test('a late attach onto an already-connected client fires its onConnect immediately', async () => { + const service = createMqttService({ app: {}, services: {}, router: {} }) + const client = createMockClient() + mockConnect.mockReturnValue(client) + + await service.attachClient('team:team-1', { getCredentials: creds }) + client.emit('connect', { cmd: 'connack' }) + + const onConnect = vi.fn() + await service.attachClient('team:team-1', { getCredentials: creds, onConnect }) + expect(onConnect).toHaveBeenCalledTimes(1) + }) + + test('messages fan out to every attachment', async () => { + const service = createMqttService({ app: {}, services: {}, router: {} }) + const client = createMockClient() + mockConnect.mockReturnValue(client) + + const onMessageA = vi.fn() + const onMessageB = vi.fn() + await service.attachClient('team:team-1', { getCredentials: creds, onMessage: onMessageA }) + await service.attachClient('team:team-1', { getCredentials: creds, onMessage: onMessageB }) + + client.emit('message', 'ff/v1/team-1/t/updated', Buffer.from('{}'), { cmd: 'publish' }) + + expect(onMessageA).toHaveBeenCalledTimes(1) + expect(onMessageB).toHaveBeenCalledTimes(1) + }) + + test('requires a credential provider when creating a fresh connection', async () => { + const service = createMqttService({ app: {}, services: {}, router: {} }) + await expect(service.attachClient('team:team-1', {})).rejects.toThrow('requires a getCredentials callback') + }) + }) }) From b60fd4f268c751667fb6a799d60c533ba53bee54 Mon Sep 17 00:00:00 2001 From: Noley Holland Date: Fri, 17 Jul 2026 08:35:55 -0700 Subject: [PATCH 2/3] Rename to Observers for methods & use clientKey as prop name --- frontend/src/services/mqtt.service.ts | 72 +++++++++---------- frontend/src/types/services/mqtt.types.ts | 10 +-- .../frontend/services/mqtt.service.spec.js | 36 +++++----- 3 files changed, 59 insertions(+), 59 deletions(-) diff --git a/frontend/src/services/mqtt.service.ts b/frontend/src/services/mqtt.service.ts index f17d86dea6..537ecfe944 100644 --- a/frontend/src/services/mqtt.service.ts +++ b/frontend/src/services/mqtt.service.ts @@ -14,14 +14,14 @@ import { BaseService } from './service.contract' import { Maybe } from '@/types/common/types' import type { BinaryPayload, - ManagedMqttAttachment, ManagedMqttClient, - MqttAttachmentHandle, + ManagedMqttObserver, MqttConnectionHandlers, MqttConnectionOptions, MqttConnectionWaiter, MqttCredentialProvider, MqttCredentials, + MqttObserverHandle, MqttPublishRequest, MqttReconnectOptions, MqttReconnectPolicy, @@ -51,7 +51,7 @@ class MqttService extends BaseService implements MqttServiceI { protected $destroyed: boolean - protected $attachmentSeq: number + protected $observerSeq: number constructor ({ app, router, services }: CreateServiceOptions) { super({ @@ -65,7 +65,7 @@ class MqttService extends BaseService implements MqttServiceI { this.$clients = new Map() this.$clientOperations = new Map() this.$destroyed = false - this.$attachmentSeq = 0 + this.$observerSeq = 0 this.init() } @@ -113,65 +113,65 @@ class MqttService extends BaseService implements MqttServiceI { throw new Error(`MQTT connection "${key}" requires a getCredentials callback`) } - const attachment = this._buildAttachment(options) + const observer = this._buildObserver(options) return this.runClientOperation(key, async () => { if (this.hasClient(key)) { await this._destroyClientUnlocked(key) } - return this._createAndConnect(key, options, attachment) + return this._createAndConnect(key, options, observer) }) } - async attachClient (key: string, options: Partial = {}): Promise { - if (!key || typeof key !== 'string') { + async attachClientObserver (clientKey: string, options: Partial = {}): Promise { + if (!clientKey || typeof clientKey !== 'string') { throw new Error('MQTT connection key is required') } - const attachment = this._buildAttachment(options) + const observer = this._buildObserver(options) - await this.runClientOperation(key, async () => { - const existing = this.$clients.get(key) + await this.runClientOperation(clientKey, async () => { + const existing = this.$clients.get(clientKey) if (existing && !existing.destroyed) { - existing.attachments.add(attachment) + existing.observers.add(observer) if (existing.status === 'connected' && existing.client?.connected) { - attachment.handlers.onConnect?.({} as IConnackPacket, existing.client) + observer.handlers.onConnect?.({} as IConnackPacket, existing.client) } return } if (typeof options.getCredentials !== 'function') { - throw new Error(`MQTT connection "${key}" requires a getCredentials callback`) + throw new Error(`MQTT connection "${clientKey}" requires a getCredentials callback`) } - await this._createAndConnect(key, options, attachment) + await this._createAndConnect(clientKey, options, observer) }) - return { key, id: attachment.id } + return { key: clientKey, id: observer.id } } - async detachClient (handle: MqttAttachmentHandle): Promise { - const { key, id } = handle - await this.runClientOperation(key, async () => { - const managed = this.$clients.get(key) + async detachClientObserver (handle: MqttObserverHandle): Promise { + const { key: clientKey, id } = handle + await this.runClientOperation(clientKey, async () => { + const managed = this.$clients.get(clientKey) if (!managed) return - for (const attachment of managed.attachments) { - if (attachment.id === id) { - managed.attachments.delete(attachment) + for (const observer of managed.observers) { + if (observer.id === id) { + managed.observers.delete(observer) break } } - if (managed.attachments.size === 0) { - await this._destroyClientUnlocked(key) + if (managed.observers.size === 0) { + await this._destroyClientUnlocked(clientKey) } }) } - private _buildAttachment (options: Partial): ManagedMqttAttachment { + private _buildObserver (options: Partial): ManagedMqttObserver { const { onConnect, onClose, onOffline, onError, onMessage } = options return { - id: this.$attachmentSeq++, + id: this.$observerSeq++, handlers: { onConnect, onClose, onOffline, onError, onMessage } } } @@ -181,8 +181,8 @@ class MqttService extends BaseService implements MqttServiceI { event: K, ...args: Parameters> ): void { - for (const attachment of managed.attachments) { - const handler = attachment.handlers[event] + for (const observer of managed.observers) { + const handler = observer.handlers[event] if (handler) { ;(handler as (...a: unknown[]) => void)(...args) } @@ -190,9 +190,9 @@ class MqttService extends BaseService implements MqttServiceI { } private async _createAndConnect ( - key: string, + clientKey: string, options: Partial, - attachment: ManagedMqttAttachment + observer: ManagedMqttObserver ): Promise { if (this.$destroyed) { throw new Error('MqttService has been destroyed') @@ -201,13 +201,13 @@ class MqttService extends BaseService implements MqttServiceI { const { reconnectPeriod = 0, getCredentials, reconnect } = options const managed: ManagedMqttClient = { - key, + key: clientKey, client: null, listeners: new Set(), destroyed: false, intentionalDisconnect: false, status: 'idle', - getCredentials: this._buildCredentialsProvider(key, getCredentials as MqttCredentialProvider), + getCredentials: this._buildCredentialsProvider(clientKey, getCredentials as MqttCredentialProvider), reconnectPolicy: this._normalizeReconnectPolicy({ reconnect, reconnectPeriod, @@ -217,19 +217,19 @@ class MqttService extends BaseService implements MqttServiceI { reconnectGeneration: 0, reconnectTimer: null, subscriptions: new Map(), - attachments: new Set([attachment]), + observers: new Set([observer]), connectionWaiters: new Set(), terminalFailure: false, lastError: null, connectHandlerPromise: null } - this.$clients.set(key, managed) + this.$clients.set(clientKey, managed) try { return await this.connectManagedClient(managed, false) } catch (error) { - this.$clients.delete(key) + this.$clients.delete(clientKey) throw error } } diff --git a/frontend/src/types/services/mqtt.types.ts b/frontend/src/types/services/mqtt.types.ts index 9a42c816a8..85c298e202 100644 --- a/frontend/src/types/services/mqtt.types.ts +++ b/frontend/src/types/services/mqtt.types.ts @@ -87,12 +87,12 @@ export interface MqttConnectionWaiter { timer: ReturnType | null } -export interface ManagedMqttAttachment { +export interface ManagedMqttObserver { id: number handlers: MqttConnectionHandlers } -export interface MqttAttachmentHandle { +export interface MqttObserverHandle { key: string id: number } @@ -110,7 +110,7 @@ export interface ManagedMqttClient { reconnectGeneration: number reconnectTimer: ReturnType | null subscriptions: Map - attachments: Set + observers: Set connectionWaiters: Set terminalFailure: boolean lastError: Error | null @@ -120,8 +120,8 @@ export interface ManagedMqttClient { export interface MqttServiceI extends AppService { createClient(key: string, options?: Partial): Promise destroyClient(key: string): Promise - attachClient(key: string, options?: Partial): Promise - detachClient(handle: MqttAttachmentHandle): Promise + attachClientObserver(clientKey: string, options?: Partial): Promise + detachClientObserver(handle: MqttObserverHandle): Promise getManagedClient(key: string): ManagedMqttClient | null hasClient(key: string): boolean publishMessage(key: string, options: MqttPublishRequest): Promise diff --git a/test/unit/frontend/services/mqtt.service.spec.js b/test/unit/frontend/services/mqtt.service.spec.js index c2e9ee39e9..a77cbcf264 100644 --- a/test/unit/frontend/services/mqtt.service.spec.js +++ b/test/unit/frontend/services/mqtt.service.spec.js @@ -945,7 +945,7 @@ describe('MqttService', async () => { await destroyMqttService() }) - describe('attachClient / detachClient (shared refcounted connections)', () => { + describe('attachClientObserver / detachClientObserver (shared refcounted connections)', () => { const creds = async () => ({ url: 'mqtt://example.com', username: 'user', password: 'pass' }) test('a second attach on the same key reuses the existing client', async () => { @@ -953,37 +953,37 @@ describe('MqttService', async () => { const client = createMockClient() mockConnect.mockReturnValue(client) - const first = await service.attachClient('team:team-1', { getCredentials: creds }) - const second = await service.attachClient('team:team-1', { getCredentials: creds }) + const first = await service.attachClientObserver('team:team-1', { getCredentials: creds }) + const second = await service.attachClientObserver('team:team-1', { getCredentials: creds }) expect(mockConnect).toHaveBeenCalledTimes(1) expect(first.id).not.toBe(second.id) - expect(service.getManagedClient('team:team-1').attachments.size).toBe(2) + expect(service.getManagedClient('team:team-1').observers.size).toBe(2) }) test('detaching a non-last attachment keeps the connection alive', async () => { const service = createMqttService({ app: {}, services: {}, router: {} }) mockConnect.mockReturnValue(createMockClient()) - const first = await service.attachClient('team:team-1', { getCredentials: creds }) - const second = await service.attachClient('team:team-1', { getCredentials: creds }) + const first = await service.attachClientObserver('team:team-1', { getCredentials: creds }) + const second = await service.attachClientObserver('team:team-1', { getCredentials: creds }) - await service.detachClient(first) + await service.detachClientObserver(first) expect(service.hasClient('team:team-1')).toBe(true) - expect(service.getManagedClient('team:team-1').attachments.size).toBe(1) + expect(service.getManagedClient('team:team-1').observers.size).toBe(1) - await service.detachClient(second) + await service.detachClientObserver(second) expect(service.hasClient('team:team-1')).toBe(false) }) test('detaching an unknown handle is a no-op', async () => { const service = createMqttService({ app: {}, services: {}, router: {} }) mockConnect.mockReturnValue(createMockClient()) - await service.attachClient('team:team-1', { getCredentials: creds }) + await service.attachClientObserver('team:team-1', { getCredentials: creds }) - await service.detachClient({ key: 'team:team-1', id: 999 }) + await service.detachClientObserver({ key: 'team:team-1', id: 999 }) expect(service.hasClient('team:team-1')).toBe(true) - expect(service.getManagedClient('team:team-1').attachments.size).toBe(1) + expect(service.getManagedClient('team:team-1').observers.size).toBe(1) }) test('a late attach onto an already-connected client fires its onConnect immediately', async () => { @@ -991,23 +991,23 @@ describe('MqttService', async () => { const client = createMockClient() mockConnect.mockReturnValue(client) - await service.attachClient('team:team-1', { getCredentials: creds }) + await service.attachClientObserver('team:team-1', { getCredentials: creds }) client.emit('connect', { cmd: 'connack' }) const onConnect = vi.fn() - await service.attachClient('team:team-1', { getCredentials: creds, onConnect }) + await service.attachClientObserver('team:team-1', { getCredentials: creds, onConnect }) expect(onConnect).toHaveBeenCalledTimes(1) }) - test('messages fan out to every attachment', async () => { + test('messages fan out to every observer', async () => { const service = createMqttService({ app: {}, services: {}, router: {} }) const client = createMockClient() mockConnect.mockReturnValue(client) const onMessageA = vi.fn() const onMessageB = vi.fn() - await service.attachClient('team:team-1', { getCredentials: creds, onMessage: onMessageA }) - await service.attachClient('team:team-1', { getCredentials: creds, onMessage: onMessageB }) + await service.attachClientObserver('team:team-1', { getCredentials: creds, onMessage: onMessageA }) + await service.attachClientObserver('team:team-1', { getCredentials: creds, onMessage: onMessageB }) client.emit('message', 'ff/v1/team-1/t/updated', Buffer.from('{}'), { cmd: 'publish' }) @@ -1017,7 +1017,7 @@ describe('MqttService', async () => { test('requires a credential provider when creating a fresh connection', async () => { const service = createMqttService({ app: {}, services: {}, router: {} }) - await expect(service.attachClient('team:team-1', {})).rejects.toThrow('requires a getCredentials callback') + await expect(service.attachClientObserver('team:team-1', {})).rejects.toThrow('requires a getCredentials callback') }) }) }) From 82d57701f35074e13088405fdb3480a4f81af307 Mon Sep 17 00:00:00 2001 From: Noley Holland Date: Fri, 17 Jul 2026 08:47:15 -0700 Subject: [PATCH 3/3] Rename key -> clientKey for entire mqtt service file --- frontend/src/services/mqtt.service.ts | 114 +++++++++++----------- frontend/src/types/services/mqtt.types.ts | 18 ++-- 2 files changed, 66 insertions(+), 66 deletions(-) diff --git a/frontend/src/services/mqtt.service.ts b/frontend/src/services/mqtt.service.ts index 537ecfe944..2bb58b812e 100644 --- a/frontend/src/services/mqtt.service.ts +++ b/frontend/src/services/mqtt.service.ts @@ -74,52 +74,52 @@ class MqttService extends BaseService implements MqttServiceI { this.$mqtt = Mqtt } - getManagedClient (key: string): ManagedMqttClient | null { - return this.$clients.get(key) ?? null + getManagedClient (clientKey: string): ManagedMqttClient | null { + return this.$clients.get(clientKey) ?? null } - hasClient (key: string): boolean { - return this.$clients.has(key) + hasClient (clientKey: string): boolean { + return this.$clients.has(clientKey) } - async runClientOperation (key: string, operation: () => Promise | T): Promise { - const previous = this.$clientOperations.get(key) ?? Promise.resolve() + async runClientOperation (clientKey: string, operation: () => Promise | T): Promise { + const previous = this.$clientOperations.get(clientKey) ?? Promise.resolve() const current = previous .catch(() => {}) .then(operation) const tracked = current.finally(() => { - if (this.$clientOperations.get(key) === tracked) { - this.$clientOperations.delete(key) + if (this.$clientOperations.get(clientKey) === tracked) { + this.$clientOperations.delete(clientKey) } }) - this.$clientOperations.set(key, tracked) + this.$clientOperations.set(clientKey, tracked) return tracked } - async destroyClient (key: string): Promise { - await this.runClientOperation(key, async () => { - await this._destroyClientUnlocked(key) + async destroyClient (clientKey: string): Promise { + await this.runClientOperation(clientKey, async () => { + await this._destroyClientUnlocked(clientKey) }) } - async createClient (key: string, options: Partial = {}): Promise { - if (!key || typeof key !== 'string') { + async createClient (clientKey: string, options: Partial = {}): Promise { + if (!clientKey || typeof clientKey !== 'string') { throw new Error('MQTT connection key is required') } if (typeof options.getCredentials !== 'function') { - throw new Error(`MQTT connection "${key}" requires a getCredentials callback`) + throw new Error(`MQTT connection "${clientKey}" requires a getCredentials callback`) } const observer = this._buildObserver(options) - return this.runClientOperation(key, async () => { - if (this.hasClient(key)) { - await this._destroyClientUnlocked(key) + return this.runClientOperation(clientKey, async () => { + if (this.hasClient(clientKey)) { + await this._destroyClientUnlocked(clientKey) } - return this._createAndConnect(key, options, observer) + return this._createAndConnect(clientKey, options, observer) }) } @@ -234,11 +234,11 @@ class MqttService extends BaseService implements MqttServiceI { } } - async publishMessage (key: string, options?: MqttPublishRequest): Promise { - const managed = this.getManagedClient(key) + async publishMessage (clientKey: string, options?: MqttPublishRequest): Promise { + const managed = this.getManagedClient(clientKey) if (!managed || managed.destroyed) { - return Promise.reject(new Error(`MQTT connection "${key}" does not exist`)) + return Promise.reject(new Error(`MQTT connection "${clientKey}" does not exist`)) } const { @@ -309,7 +309,7 @@ class MqttService extends BaseService implements MqttServiceI { if (waitForConnection) { try { - await this.waitForConnection(key, { timeout: connectionTimeout }) + await this.waitForConnection(clientKey, { timeout: connectionTimeout }) } catch (error) { if (error instanceof Error && typeof onError === 'function') onError(error) return Promise.reject(error) @@ -319,7 +319,7 @@ class MqttService extends BaseService implements MqttServiceI { const { client } = managed if (managed.status === 'reconnecting' || !client?.connected) { - return Promise.reject(new Error(`MQTT connection "${key}" is not connected`)) + return Promise.reject(new Error(`MQTT connection "${clientKey}" is not connected`)) } return new Promise((resolve, reject) => { @@ -334,14 +334,14 @@ class MqttService extends BaseService implements MqttServiceI { }) } - subscribe (key: string, topic: string | string[], options: MqttSubscribeOptions = {}): Promise { - const managed = this.$clients.get(key) + subscribe (clientKey: string, topic: string | string[], options: MqttSubscribeOptions = {}): Promise { + const managed = this.$clients.get(clientKey) if (!managed || managed.destroyed) { - return Promise.reject(new Error(`MQTT connection "${key}" does not exist`)) + return Promise.reject(new Error(`MQTT connection "${clientKey}" does not exist`)) } if (!managed.client?.connected) { - return Promise.reject(new Error(`MQTT connection "${key}" is not connected`)) + return Promise.reject(new Error(`MQTT connection "${clientKey}" is not connected`)) } return new Promise((resolve, reject) => { @@ -356,14 +356,14 @@ class MqttService extends BaseService implements MqttServiceI { }) } - unsubscribe (key: string, topic: string | string[]): Promise { - const managed = this.getManagedClient(key) + unsubscribe (clientKey: string, topic: string | string[]): Promise { + const managed = this.getManagedClient(clientKey) if (!managed || managed.destroyed) { return Promise.resolve() } if (!managed.client?.connected) { - return Promise.reject(new Error(`MQTT connection "${key}" is not connected`)) + return Promise.reject(new Error(`MQTT connection "${clientKey}" is not connected`)) } return new Promise((resolve, reject) => { @@ -378,15 +378,15 @@ class MqttService extends BaseService implements MqttServiceI { }) } - async endConnection (key: string): Promise { - await this.destroyClient(key) + async endConnection (clientKey: string): Promise { + await this.destroyClient(clientKey) } async destroy (): Promise { this.$destroyed = true const keys = [...this.$clients.keys()] - await Promise.all(keys.map(async (key) => this.destroyClient(key))) + await Promise.all(keys.map(async (clientKey) => this.destroyClient(clientKey))) this.$clients.clear() this.$mqtt = null @@ -458,10 +458,10 @@ class MqttService extends BaseService implements MqttServiceI { managed.connectionWaiters.clear() } - waitForConnection (key: string, { timeout = 5000 }: MqttWaitForConnectionOptions = {}): Promise { - const managed = this.getManagedClient(key) + waitForConnection (clientKey: string, { timeout = 5000 }: MqttWaitForConnectionOptions = {}): Promise { + const managed = this.getManagedClient(clientKey) if (!managed || managed.destroyed) { - return Promise.reject(new Error(`MQTT connection "${key}" does not exist`)) + return Promise.reject(new Error(`MQTT connection "${clientKey}" does not exist`)) } if (!Number.isInteger(timeout) || timeout < 0) { @@ -490,7 +490,7 @@ class MqttService extends BaseService implements MqttServiceI { } waiter.timer = setTimeout(() => { - waiter.reject(new Error(`MQTT connection "${key}" did not connect before the timeout elapsed`)) + waiter.reject(new Error(`MQTT connection "${clientKey}" did not connect before the timeout elapsed`)) }, timeout) managed.connectionWaiters.add(waiter) @@ -659,9 +659,9 @@ class MqttService extends BaseService implements MqttServiceI { }, delay) } - async reconnectClient (key: string): Promise { - await this.runClientOperation(key, async () => { - const managed = this.$clients.get(key) + async reconnectClient (clientKey: string): Promise { + await this.runClientOperation(clientKey, async () => { + const managed = this.$clients.get(clientKey) if (!managed || managed.destroyed || managed.intentionalDisconnect || this.$destroyed) { return } @@ -729,7 +729,7 @@ class MqttService extends BaseService implements MqttServiceI { })) } - private _buildCredentialsProvider (key: string, getCredentials: MqttCredentialProvider): MqttCredentialProvider { + private _buildCredentialsProvider (clientKey: string, getCredentials: MqttCredentialProvider): MqttCredentialProvider { return async () => { let credentials: MqttCredentials try { @@ -737,28 +737,28 @@ class MqttService extends BaseService implements MqttServiceI { } catch (error) { throw this._asTerminalConnectionError( error, - `MQTT credential provider for connection "${key}" failed` + `MQTT credential provider for connection "${clientKey}" failed` ) } if (!credentials || typeof credentials !== 'object') { - throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must resolve to an object`) + throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must resolve to an object`) } if (!this._isValidUrl(credentials.url)) { - throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must return a valid url`) + throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must return a valid url`) } if (typeof credentials.username !== 'string' || !credentials.username.trim()) { - throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must return a non-empty username`) + throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must return a non-empty username`) } if (typeof credentials.password !== 'string' || !credentials.password.trim()) { - throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must return a non-empty password`) + throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must return a non-empty password`) } if (credentials.clientId !== undefined && (typeof credentials.clientId !== 'string' || !credentials.clientId.trim())) { - throw this._createTerminalConnectionError(`MQTT credential provider for connection "${key}" must return a non-empty clientId when provided`) + throw this._createTerminalConnectionError(`MQTT credential provider for connection "${clientKey}" must return a non-empty clientId when provided`) } if (!credentials.clientId) { @@ -785,8 +785,8 @@ class MqttService extends BaseService implements MqttServiceI { return error instanceof Error && (error as TerminalConnectionError).code === 'MQTT_TERMINAL_CONNECTION_ERROR' } - private async _destroyClientUnlocked (key: string): Promise { - const managed = this.$clients.get(key) + private async _destroyClientUnlocked (clientKey: string): Promise { + const managed = this.$clients.get(clientKey) if (!managed) return managed.destroyed = true @@ -794,7 +794,7 @@ class MqttService extends BaseService implements MqttServiceI { managed.status = 'disconnected' this.clearReconnectTimer(managed) - this.rejectConnectionWaiters(managed, new Error(`MQTT connection "${key}" has been closed`)) + this.rejectConnectionWaiters(managed, new Error(`MQTT connection "${clientKey}" has been closed`)) for (const off of managed.listeners) { try { @@ -816,7 +816,7 @@ class MqttService extends BaseService implements MqttServiceI { this._killStaleClient(client) } - this.$clients.delete(key) + this.$clients.delete(clientKey) } private _killStaleClient (client: MqttClient) { @@ -975,22 +975,22 @@ class MqttService extends BaseService implements MqttServiceI { } const normalized: Record = {} - for (const [key, value] of Object.entries(userProperties)) { - if (!key.trim()) { + for (const [clientKey, value] of Object.entries(userProperties)) { + if (!clientKey.trim()) { throw new TypeError('MQTT publish userProperties keys must be non-empty strings') } if (typeof value === 'string') { - normalized[key] = value + normalized[clientKey] = value continue } if (Array.isArray(value) && value.every(item => typeof item === 'string')) { - normalized[key] = value + normalized[clientKey] = value continue } - throw new TypeError(`MQTT publish userProperties["${key}"] must be a string or string[]`) + throw new TypeError(`MQTT publish userProperties["${clientKey}"] must be a string or string[]`) } return Object.keys(normalized).length ? normalized : undefined diff --git a/frontend/src/types/services/mqtt.types.ts b/frontend/src/types/services/mqtt.types.ts index 85c298e202..2e74a0672f 100644 --- a/frontend/src/types/services/mqtt.types.ts +++ b/frontend/src/types/services/mqtt.types.ts @@ -118,16 +118,16 @@ export interface ManagedMqttClient { } export interface MqttServiceI extends AppService { - createClient(key: string, options?: Partial): Promise - destroyClient(key: string): Promise + createClient(clientKey: string, options?: Partial): Promise + destroyClient(clientKey: string): Promise attachClientObserver(clientKey: string, options?: Partial): Promise detachClientObserver(handle: MqttObserverHandle): Promise - getManagedClient(key: string): ManagedMqttClient | null - hasClient(key: string): boolean - publishMessage(key: string, options: MqttPublishRequest): Promise - subscribe(key: string, topic: string | string[], options?: MqttSubscribeOptions): Promise - unsubscribe(key: string, topic: string | string[]): Promise - endConnection(key: string): Promise - waitForConnection(key: string, options?: MqttWaitForConnectionOptions): Promise + getManagedClient(clientKey: string): ManagedMqttClient | null + hasClient(clientKey: string): boolean + publishMessage(clientKey: string, options: MqttPublishRequest): Promise + subscribe(clientKey: string, topic: string | string[], options?: MqttSubscribeOptions): Promise + unsubscribe(clientKey: string, topic: string | string[]): Promise + endConnection(clientKey: string): Promise + waitForConnection(clientKey: string, options?: MqttWaitForConnectionOptions): Promise reset(): Promise }