From bdcec7a472ef41f4abb3aeed68433589f0b41bc2 Mon Sep 17 00:00:00 2001 From: Alex Christoffer Rasmussen Date: Wed, 7 May 2025 11:40:30 +0200 Subject: [PATCH] Refactor: WebSocket from flush queue to one patch (#1595) * change flush to one patch * remove unused types * create batch * merge patch into eventStore Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- apps/client/src/common/stores/runtime.ts | 12 ---- apps/client/src/common/utils/socket.ts | 69 +++---------------- .../runtime-service/RuntimeService.ts | 15 ++-- apps/server/src/stores/EventStore.ts | 33 ++++----- 4 files changed, 34 insertions(+), 95 deletions(-) diff --git a/apps/client/src/common/stores/runtime.ts b/apps/client/src/common/stores/runtime.ts index 70ac4c993..e8a291932 100644 --- a/apps/client/src/common/stores/runtime.ts +++ b/apps/client/src/common/stores/runtime.ts @@ -14,18 +14,6 @@ export const runtimeStore = createWithEqualityFn( export const useRuntimeStore = (selector: (state: RuntimeStore) => T) => useStoreWithEqualityFn(runtimeStore, selector, deepCompare); -let batchStore: Partial = {}; - -export function addToBatchUpdates(key: K, value: RuntimeStore[K]) { - batchStore[key] = value; -} - -export function flushBatchUpdates() { - const state = runtimeStore.getState(); - runtimeStore.setState({ ...state, ...batchStore }); - batchStore = {}; -} - /** * Allows patching a property of the runtime store */ diff --git a/apps/client/src/common/utils/socket.ts b/apps/client/src/common/utils/socket.ts index cde2d9c8a..e6e809593 100644 --- a/apps/client/src/common/utils/socket.ts +++ b/apps/client/src/common/utils/socket.ts @@ -14,7 +14,7 @@ import { } from '../stores/clientStore'; import { addDialog } from '../stores/dialogStore'; import { addLog } from '../stores/logger'; -import { addToBatchUpdates, flushBatchUpdates, patchRuntime, patchRuntimeProperty } from '../stores/runtime'; +import { patchRuntime, patchRuntimeProperty } from '../stores/runtime'; let websocket: WebSocket | null = null; let reconnectTimeout: NodeJS.Timeout | null = null; @@ -50,9 +50,9 @@ export const connectSocket = () => { // we decide to allows reconnect reconnectTimeout = setTimeout(() => { - if (reconnectAttempts > 2) { - setOnlineStatus(false); - } + if (reconnectAttempts > 2) { + setOnlineStatus(false); + } console.warn('WebSocket: attempting reconnect'); if (websocket && websocket.readyState === WebSocket.CLOSED) { reconnectAttempts += 1; @@ -141,59 +141,10 @@ export const connectSocket = () => { updateDevTools(serverPayload); break; } - case 'ontime-clock': { - addToBatchUpdates('clock', payload); - updateDevTools({ clock: payload }); - break; - } - case 'ontime-timer': { - addToBatchUpdates('timer', payload); - updateDevTools({ timer: payload }); - break; - } - case 'ontime-onAir': { - addToBatchUpdates('onAir', payload); - updateDevTools({ onAir: payload }); - break; - } - case 'ontime-message': { - addToBatchUpdates('message', payload); - updateDevTools({ message: payload }); - break; - } - case 'ontime-runtime': { - addToBatchUpdates('runtime', payload); - updateDevTools({ runtime: payload }); - break; - } - case 'ontime-eventNow': { - addToBatchUpdates('eventNow', payload); - updateDevTools({ eventNow: payload }); - break; - } - case 'ontime-currentBlock': { - addToBatchUpdates('currentBlock', payload); - updateDevTools({ currentBlock: payload }); - break; - } - case 'ontime-publicEventNow': { - addToBatchUpdates('publicEventNow', payload); - updateDevTools({ publicEventNow: payload }); - break; - } - case 'ontime-eventNext': { - addToBatchUpdates('eventNext', payload); - updateDevTools({ eventNext: payload }); - break; - } - case 'ontime-publicEventNext': { - addToBatchUpdates('publicEventNext', payload); - updateDevTools({ publicEventNext: payload }); - break; - } - case 'ontime-auxtimer1': { - addToBatchUpdates('auxtimer1', payload); - updateDevTools({ auxtimer1: payload }); + case 'ontime-patch': { + const patch = payload as Partial; + patchRuntime(patch); + updateDevTools(patch); break; } case 'ontime-refetch': { @@ -213,8 +164,8 @@ export const connectSocket = () => { } break; } - case 'ontime-flush': { - flushBatchUpdates(); + default: { + console.log('unknown WS message', type); break; } } diff --git a/apps/server/src/services/runtime-service/RuntimeService.ts b/apps/server/src/services/runtime-service/RuntimeService.ts index 3bda4acde..cb6d09191 100644 --- a/apps/server/src/services/runtime-service/RuntimeService.ts +++ b/apps/server/src/services/runtime-service/RuntimeService.ts @@ -696,6 +696,8 @@ function broadcastResult(_target: any, _propertyKey: string, descriptor: Propert const result = originalMethod.apply(this, args); const state = runtimeState.getState(); + const batch = eventStore.createBatch(); + // we do the comparison by explicitly for each property // to apply custom logic for different datasets @@ -758,24 +760,24 @@ function broadcastResult(_target: any, _propertyKey: string, descriptor: Propert //Now we set all the updates on the eventstore and update the previous value if (hasChangedPlayback) { - eventStore.set('onAir', state.timer.playback !== Playback.Stop); + batch.add('onAir', state.timer.playback !== Playback.Stop); } if (shouldUpdateTimer) { - eventStore.set('timer', state.timer); + batch.add('timer', state.timer); RuntimeService.previousTimerUpdate = state.clock; RuntimeService.previousTimerValue = state.timer.current; RuntimeService.previousState.timer = { ...state.timer }; } if (shouldRuntimeUpdate) { - eventStore.set('runtime', state.runtime); + batch.add('runtime', state.runtime); RuntimeService.previousRuntimeUpdate = state.clock; RuntimeService.previousState.runtime = { ...state.runtime }; } if (shouldBlockUpdate) { - eventStore.set('currentBlock', state.currentBlock); + batch.add('currentBlock', state.currentBlock); RuntimeService.previousState.currentBlock = { ...state.currentBlock }; } @@ -785,7 +787,7 @@ function broadcastResult(_target: any, _propertyKey: string, descriptor: Propert if (shouldUpdateClock) { RuntimeService.previousClockUpdate = state.clock; - eventStore.set('clock', state.clock); + batch.add('clock', state.clock); } // Update the events if they have changed @@ -817,7 +819,7 @@ function broadcastResult(_target: any, _propertyKey: string, descriptor: Propert } function storeKey(eventKey: RuntimeStateEventKeys) { - eventStore.set(eventKey, state[eventKey]); + batch.add(eventKey, state[eventKey]); // @ts-expect-error -- not sure how to type this in a sane way RuntimeService.previousState[eventKey] = { ...state[eventKey] }; } @@ -836,6 +838,7 @@ function broadcastResult(_target: any, _propertyKey: string, descriptor: Propert }); } + batch.send(); return result; }; diff --git a/apps/server/src/stores/EventStore.ts b/apps/server/src/stores/EventStore.ts index c168e6369..8ffc2abd1 100644 --- a/apps/server/src/stores/EventStore.ts +++ b/apps/server/src/stores/EventStore.ts @@ -1,18 +1,16 @@ import { RuntimeStore } from 'ontime-types'; import { socket } from '../adapters/WebsocketAdapter.js'; +import { isEmptyObject } from '../utils/parserUtils.js'; export type PublishFn = (key: T, value: RuntimeStore[T]) => void; export type StoreGetter = (key: T) => Partial[T]; let store: Partial = {}; -const changedKeys = new Set(); -let isUpdatePending: NodeJS.Immediate | null = null; /** * A runtime store that broadcasts its payload * - init: allows for adding an initial payload to the store - * - batchSet: allows setting several keys with a single broadcast * - poll: utility to return state * - broadcast: send its payload as json object */ @@ -25,21 +23,20 @@ export const eventStore = { }, set(key: T, value: RuntimeStore[T]) { store[key] = value; - - // check if the key is already marked for and update otherwise push it onto the update array - changedKeys.add(key); - - //if there is already and update pending we don't need to schedule another one - if (!isUpdatePending) { - isUpdatePending = setImmediate(() => { - for (const dataKey of changedKeys) { - socket.sendAsJson({ type: `ontime-${dataKey}`, payload: store[dataKey] }); - } - socket.sendAsJson({ type: 'ontime-flush' }); - isUpdatePending = null; - changedKeys.clear(); - }); - } + socket.sendAsJson({ type: 'ontime-patch', payload: { [key]: value } }); + }, + createBatch() { + const patch: Partial = {}; + return { + add(key: T, value: RuntimeStore[T]) { + patch[key] = value; + }, + send() { + if (isEmptyObject(patch)) return; + store = { ...store, ...patch }; + socket.sendAsJson({ type: 'ontime-patch', payload: patch }); + }, + }; }, poll() { return store as RuntimeStore;