From ffcc7cd0aa96768538ca84ccf57591d5e502b925 Mon Sep 17 00:00:00 2001 From: Tom Butcher Date: Tue, 1 Sep 2026 14:14:18 +0100 Subject: [PATCH] Add support for subscribing and unsubscribing to all object updates - Implemented `subscribeToAllObjectUpdates` and `unsubscribeAllObjectUpdates` methods in the `UpdateManager` to handle bulk updates for object types. - Updated `SocketHost` and `SocketUser` classes to listen for new socket events related to all object updates. - Enhanced test coverage for the new subscription methods to ensure correct functionality and integration with the socket client. --- src/socket/__tests__/socketuser.test.js | 2 + src/socket/sockethost.js | 16 ++++++ src/socket/socketuser.js | 21 +++++++ src/updates/__tests__/updatemanager.test.js | 43 ++++++++++++++ src/updates/updatemanager.js | 62 +++++++++++++++++++++ 5 files changed, 144 insertions(+) diff --git a/src/socket/__tests__/socketuser.test.js b/src/socket/__tests__/socketuser.test.js index a8c8e60..f812907 100644 --- a/src/socket/__tests__/socketuser.test.js +++ b/src/socket/__tests__/socketuser.test.js @@ -34,9 +34,11 @@ jest.unstable_mockModule('../../updates/updatemanager.js', () => ({ subscribeToObjectNew: jest.fn(), subscribeToObjectDelete: jest.fn(), subscribeToObjectUpdate: jest.fn(), + subscribeToAllObjectUpdates: jest.fn(), removeObjectNewListener: jest.fn(), removeObjectDeleteListener: jest.fn(), removeObjectUpdateListener: jest.fn(), + removeAllObjectUpdatesListener: jest.fn(), removeAllListeners: jest.fn() })) })); diff --git a/src/socket/sockethost.js b/src/socket/sockethost.js index 82bd034..ef625eb 100644 --- a/src/socket/sockethost.js +++ b/src/socket/sockethost.js @@ -46,10 +46,18 @@ export class SocketHost { 'subscribeToObjectUpdates', this.handleSubscribeToObjectUpdatesEvent.bind(this) ); + this.socket.on( + 'subscribeToAllObjectUpdates', + this.handleSubscribeToAllObjectUpdatesEvent.bind(this) + ); this.socket.on( 'unsubscribeToObjectUpdates', this.handleUnsubscribeToObjectUpdatesEvent.bind(this) ); + this.socket.on( + 'unsubscribeAllObjectUpdates', + this.handleUnsubscribeAllObjectUpdatesEvent.bind(this) + ); this.socket.on( 'subscribeToObjectActions', this.handleSubscribeToObjectActions.bind(this) @@ -217,6 +225,10 @@ export class SocketHost { ); } + async handleSubscribeToAllObjectUpdatesEvent(data) { + await this.updateManager.subscribeToAllObjectUpdates(data.objectType); + } + async handleSubscribeToObjectActions(data) { await this.actionManager.subscribeToObjectActions( data._id, @@ -247,6 +259,10 @@ export class SocketHost { ); } + async handleUnsubscribeAllObjectUpdatesEvent(data) { + await this.updateManager.removeAllObjectUpdatesListener(data.objectType); + } + async setDevicesState(state, online, connectedAt) { logger.info('Setting devices state to', state, 'and online to', online); diff --git a/src/socket/socketuser.js b/src/socket/socketuser.js index 33c01f0..f6b4c5f 100644 --- a/src/socket/socketuser.js +++ b/src/socket/socketuser.js @@ -60,6 +60,10 @@ export class SocketUser { 'subscribeToObjectUpdate', this.handleSubscribeToObjectUpdateEvent.bind(this) ); + this.socket.on( + 'subscribeToAllObjectUpdates', + this.handleSubscribeToAllObjectUpdatesEvent.bind(this) + ); this.socket.on( 'subscribeToObjectEvent', this.handleSubscribeToObjectEventEvent.bind(this) @@ -72,6 +76,10 @@ export class SocketUser { 'unsubscribeObjectUpdate', this.handleUnsubscribeToObjectUpdateEvent.bind(this) ); + this.socket.on( + 'unsubscribeAllObjectUpdates', + this.handleUnsubscribeAllObjectUpdatesEvent.bind(this) + ); this.socket.on( 'unsubscribeObjectEvent', this.handleUnsubscribeObjectEventEvent.bind(this) @@ -253,6 +261,15 @@ export class SocketUser { } } + async handleSubscribeToAllObjectUpdatesEvent(data, callback) { + const result = await this.updateManager.subscribeToAllObjectUpdates( + data.objectType + ); + if (typeof callback === 'function') { + callback(result); + } + } + async handleSubscribeToObjectEventEvent(data) { await this.eventManager.subscribeToObjectEvent( data._id, @@ -279,6 +296,10 @@ export class SocketUser { ); } + async handleUnsubscribeAllObjectUpdatesEvent(data) { + await this.updateManager.removeAllObjectUpdatesListener(data.objectType); + } + async handleUnsubscribeObjectEventEvent(data) { await this.eventManager.removeObjectEventsListener( data._id, diff --git a/src/updates/__tests__/updatemanager.test.js b/src/updates/__tests__/updatemanager.test.js index b92cdab..00d95d4 100644 --- a/src/updates/__tests__/updatemanager.test.js +++ b/src/updates/__tests__/updatemanager.test.js @@ -198,6 +198,41 @@ describe('UpdateManager', () => { }); }); + describe('subscribeToAllObjectUpdates', () => { + it('should subscribe to all update events for an object type', async () => { + await updateManager.subscribeToAllObjectUpdates('printer'); + + expect(natsServer.subscribe).toHaveBeenCalledWith( + 'printers.*.object', + 'test-socket-id', + expect.any(Function) + ); + + const natsCallback = natsServer.subscribe.mock.calls[0][2]; + const data = { status: 'idle' }; + natsCallback('printers.456.object', data); + + expect(mockSocketClient.socket.emit).toHaveBeenCalledWith( + 'objectUpdate', + { + _id: '456', + objectType: 'printer', + object: data + } + ); + }); + + it('should skip all-object updates when a specific subscription exists', async () => { + await updateManager.subscribeToObjectUpdate('123', 'printer'); + await updateManager.subscribeToAllObjectUpdates('printer'); + + const allUpdatesCallback = natsServer.subscribe.mock.calls[1][2]; + allUpdatesCallback('printers.123.object', { status: 'idle' }); + + expect(mockSocketClient.socket.emit).toHaveBeenCalledTimes(0); + }); + }); + describe('remove methods', () => { it('should remove new listener', async () => { await updateManager.removeObjectNewListener('printer'); @@ -222,6 +257,14 @@ describe('UpdateManager', () => { 'test-socket-id' ); }); + + it('should remove all object updates listener', async () => { + await updateManager.removeAllObjectUpdatesListener('printer'); + expect(natsServer.removeSubscription).toHaveBeenCalledWith( + 'printers.*.object', + 'test-socket-id' + ); + }); }); describe('removeAllListeners', () => { diff --git a/src/updates/updatemanager.js b/src/updates/updatemanager.js index 8bfdb20..ae0669e 100644 --- a/src/updates/updatemanager.js +++ b/src/updates/updatemanager.js @@ -202,6 +202,11 @@ export class UpdateManager { constructor(socketClient) { this.socketClient = socketClient; this.subscriptions = new Set(); + this.objectUpdateSubscriptions = new Set(); + } + + getObjectUpdateSubscriptionKey(objectType, id) { + return `${objectType}:${id}`; } matchesObjectTypeFilter(objectType, filter, value) { @@ -307,6 +312,50 @@ export class UpdateManager { }); }); + this.objectUpdateSubscriptions.add( + this.getObjectUpdateSubscriptionKey(objectType, id) + ); + this.subscriptions.add(getSubscriptionKey(subject, owner)); + return { success: true }; + } + + extractIdFromUpdateSubject(subject) { + const parts = subject.split('.'); + if (parts.length < 3 || parts[parts.length - 1] !== 'object') { + return null; + } + return parts[parts.length - 2]; + } + + async subscribeToAllObjectUpdates(objectType) { + logger.debug('Subscribing to all object updates...', objectType); + const subject = `${objectType}s.*.object`; + const owner = this.socketClient.socketId; + + await natsServer.subscribe(subject, owner, (key, value) => { + const id = this.extractIdFromUpdateSubject(key); + if (!id) { + logger.warn('Unable to extract id from update subject:', key); + return; + } + + if ( + this.objectUpdateSubscriptions.has( + this.getObjectUpdateSubscriptionKey(objectType, id) + ) + ) { + return; + } + + const expandedValue = expandObjectIds(value); + logger.trace('All object update event:', id, objectType); + this.socketClient.socket.emit('objectUpdate', { + _id: id, + objectType: objectType, + object: { ...expandedValue } + }); + }); + this.subscriptions.add(getSubscriptionKey(subject, owner)); return { success: true }; } @@ -333,6 +382,18 @@ export class UpdateManager { const subject = `${objectType}s.${id}.object`; const owner = this.socketClient.socketId; + await natsServer.removeSubscription(subject, owner); + this.objectUpdateSubscriptions.delete( + this.getObjectUpdateSubscriptionKey(objectType, id) + ); + this.subscriptions.delete(getSubscriptionKey(subject, owner)); + return { success: true }; + } + + async removeAllObjectUpdatesListener(objectType) { + const subject = `${objectType}s.*.object`; + const owner = this.socketClient.socketId; + await natsServer.removeSubscription(subject, owner); this.subscriptions.delete(getSubscriptionKey(subject, owner)); return { success: true }; @@ -351,6 +412,7 @@ export class UpdateManager { await Promise.all(removePromises); this.subscriptions.clear(); + this.objectUpdateSubscriptions.clear(); logger.debug(`Removed ${removePromises.length} update listener(s)`); return { success: true }; }