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

@@ -14,18 +14,6 @@ export const runtimeStore = createWithEqualityFn<RuntimeStore>(
 export const useRuntimeStore = <T>(selector: (state: RuntimeStore) => T) =>
   useStoreWithEqualityFn(runtimeStore, selector, deepCompare);
 
-let batchStore: Partial<RuntimeStore> = {};
-
-export function addToBatchUpdates<K extends keyof RuntimeStore>(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
  */

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) {
+      if (reconnectAttempts > 2) {
-          setOnlineStatus(false);
+        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': {
+        case 'ontime-patch': {
-          addToBatchUpdates('clock', payload);
+          const patch = payload as Partial<RuntimeStore>;
-          updateDevTools({ clock: payload });
+          patchRuntime(patch);
-          break;
+          updateDevTools(patch);
-        }
-        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 });
           break;
         }
         case 'ontime-refetch': {
@@ -213,8 +164,8 @@ export const connectSocket = () => {
           }
           break;
         }
-        case 'ontime-flush': {
+        default: {
-          flushBatchUpdates();
+          console.log('unknown WS message', type);
           break;
         }
       }

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;
   };
 

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 = <T extends keyof RuntimeStore>(key: T, value: RuntimeStore[T]) => void;
 export type StoreGetter = <T extends keyof RuntimeStore>(key: T) => Partial<RuntimeStore>[T];
 
 let store: Partial<RuntimeStore> = {};
 
-const changedKeys = new Set<keyof RuntimeStore>();
-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<T extends keyof RuntimeStore>(key: T, value: RuntimeStore[T]) {
     store[key] = value;
-
+    socket.sendAsJson({ type: 'ontime-patch', payload: { [key]: value } });
-    // check if the key is already marked for and update otherwise push it onto the update array
+  },
-    changedKeys.add(key);
+  createBatch() {
-
+    const patch: Partial<RuntimeStore> = {};
-    //if there is already and update pending we don't need to schedule another one
+    return {
-    if (!isUpdatePending) {
+      add<T extends keyof RuntimeStore>(key: T, value: RuntimeStore[T]) {
-      isUpdatePending = setImmediate(() => {
+        patch[key] = value;
-        for (const dataKey of changedKeys) {
+      },
-          socket.sendAsJson({ type: `ontime-${dataKey}`, payload: store[dataKey] });
+      send() {
-        }
+        if (isEmptyObject(patch)) return;
-        socket.sendAsJson({ type: 'ontime-flush' });
+        store = { ...store, ...patch };
-        isUpdatePending = null;
+        socket.sendAsJson({ type: 'ontime-patch', payload: patch });
-        changedKeys.clear();
+      },
-      });
+    };
-    }
   },
   poll() {
     return store as RuntimeStore;