Support telemetry batching and move WebSocket handling to worker

src/api/telemetry/BatchingWebSocketProvider.js

@@ -0,0 +1,111 @@
+/*****************************************************************************
+ * Open MCT Web, Copyright (c) 2014-2024, United States Government
+ * as represented by the Administrator of the National Aeronautics and Space
+ * Administration. All rights reserved.
+ *
+ * Open MCT Web is licensed under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * Open MCT Web includes source code licensed under additional open source
+ * licenses. See the Open Source Licenses file (LICENSES.md) included with
+ * this source code distribution or the Licensing information page available
+ * at runtime from the About dialog for additional information.
+ *****************************************************************************/
+
+import installWorker from './WebSocketWorker.js';
+
+class BatchingWebSocketProvider extends EventTarget {
+  #worker;
+  #openmct;
+  #showingRateLimitNotification;
+
+  constructor(openmct) {
+    super();
+    // Install worker, register listeners etc.
+    const workerFunction = `(${installWorker.toString()})()`;
+    const workerBlob = new Blob([workerFunction]);
+    const workerUrl = URL.createObjectURL(workerBlob, { type: 'application/javascript' });
+    this.#worker = new Worker(workerUrl);
+    this.#openmct = openmct;
+    this.#showingRateLimitNotification = false;
+
+    this.routeMessageToHandler = this.routeMessageToHandler.bind(this);
+    this.#worker.addEventListener('message', this.routeMessageToHandler);
+  }
+
+  connect(url) {
+    this.#worker.postMessage({
+      type: 'connect',
+      url
+    });
+  }
+
+  disconnect() {
+    this.#worker.postMessage({ type: 'disconnect' });
+  }
+
+  sendMessage(message) {
+    this.#worker.postMessage({
+      type: 'message',
+      message
+    });
+  }
+
+  setBatchingStrategy(strategy) {
+    const serializedStrategy = {
+      shouldBatchMessage: strategy.shouldBatchMessage.toString(),
+      getBatchIdFromMessage: strategy.getBatchIdFromMessage.toString()
+    };
+
+    this.#worker.postMessage({
+      type: 'setBatchingStrategy',
+      serializedStrategy
+    });
+  }
+
+  setRate(rate) {
+    this.#worker.postMessage({
+      type: 'setRate',
+      rate
+    });
+  }
+
+  setMaxBatchSize(maxBatchSize) {
+    this.#worker.postMessage({
+      type: 'setMaxBatchSize',
+      maxBatchSize
+    });
+  }
+
+  routeMessageToHandler(message) {
+    // Batch message would need to be handle differently here
+    if (message.data.type === 'batch') {
+      if (message.data.batch.dropped === true && !this.#showingRateLimitNotification) {
+        const notification = this.#openmct.notifications.alert(
+          'Telemetry dropped due to client rate limiting.',
+          { hint: 'Refresh individual telemetry views to retrieve dropped telemetry if needed.' }
+        );
+        this.#showingRateLimitNotification = true;
+        notification.once('minimized', () => {
+          this.#showingRateLimitNotification = false;
+        });
+        return;
+      }
+      this.dispatchEvent(new CustomEvent('batch', { detail: message.data.batch }));
+    } else if (message.data.type === 'message') {
+      this.dispatchEvent(new CustomEvent('message', { detail: message.data.message }));
+    } else {
+      throw new Error(`Unknown message type: ${message.data.type}`);
+    }
+  }
+}
+
+export default BatchingWebSocketProvider;

src/api/telemetry/TelemetryAPI.js

@@ -23,6 +23,7 @@
 import objectUtils from 'objectUtils';
 
 import CustomStringFormatter from '../../plugins/displayLayout/CustomStringFormatter.js';
+import BatchingWebSocketProvider from './BatchingWebSocketProvider.js';
 import DefaultMetadataProvider from './DefaultMetadataProvider.js';
 import TelemetryCollection from './TelemetryCollection.js';
 import TelemetryMetadataManager from './TelemetryMetadataManager.js';
@@ -54,6 +55,23 @@
  * @memberof module:openmct.TelemetryAPI~
  */
 
+/**
+ * Describes and bounds requests for telemetry data.
+ *
+ * @typedef TelemetrySubscriptionOptions
+ * @property {String} [strategy] symbolic identifier directing providers on how
+ * to handle telemetry subscriptions. The default behavior is 'latest' which will
+ * always return a single telemetry value with each callback, and in the event
+ * of throttling will always prioritize the latest data, meaning intermediate
+ * data will be skipped. Alternatively, the `batch` strategy can be used, which
+ * will return all telemetry values since the last callback. This strategy is
+ * useful for cases where intermediate data is important, such as when
+ * rendering a telemetry plot or table. If `batch` is specified, the subscription
+ * callback will be invoked with an Array.
+ *
+ * @memberof module:openmct.TelemetryAPI~
+ */
+
 /**
  * Utilities for telemetry
  * @interface TelemetryAPI
@@ -78,6 +96,7 @@
     this.valueFormatterCache = new WeakMap();
     this.requestInterceptorRegistry = new TelemetryRequestInterceptorRegistry();
     this.#isGreedyLAD = true;
+    this.BatchingWebSocketProvider = BatchingWebSocketProvider;
   }
 
   abortAllRequests() {
@@ -378,18 +397,23 @@
    * @memberof module:openmct.TelemetryAPI~TelemetryProvider#
    * @param {module:openmct.DomainObject} domainObject the object
    *        which has associated telemetry
-   * @param {TelemetryRequestOptions} options configuration items for subscription
+   * @param {TelemetrySubscriptionOptions} options configuration items for subscription
    * @param {Function} callback the callback to invoke with new data, as
    *        it becomes available
    * @returns {Function} a function which may be called to terminate
    *          the subscription
    */
-  subscribe(domainObject, callback, options) {
+  subscribe(domainObject, callback, options = { strategy: 'latest' }) {
     if (domainObject.type === 'unknown') {
       return () => {};
     }
 
-    const provider = this.findSubscriptionProvider(domainObject);
+    // Default behavior is to use the latest strategy, as opposed to the new "batch" strategy
+    if (options.strategy === undefined) {
+      options.strategy = 'latest';
+    }
+
+    const provider = this.findSubscriptionProvider(domainObject, options);
 
     if (!this.subscribeCache) {
       this.subscribeCache = {};
@@ -405,11 +429,9 @@
       if (provider) {
         subscriber.unsubscribe = provider.subscribe(
           domainObject,
-          function (value) {
-            subscriber.callbacks.forEach(function (cb) {
-              cb(value);
-            });
-          },
+          options.strategy === 'batch'
+            ? subscriptionCallbackForArray
+            : subscriptionCallbackForSingleValue,
           options
         );
       } else {
@@ -419,6 +441,26 @@
       subscriber.callbacks.push(callback);
     }
 
+    function subscriptionCallbackForArray(value) {
+      if (!Array.isArray(value)) {
+        value = [value];
+      }
+
+      subscriber.callbacks.forEach(function (cb) {
+        cb(value);
+      });
+    }
+
+    function subscriptionCallbackForSingleValue(value) {
+      if (Array.isArray(value)) {
+        value = value[value.length - 1];
+      }
+
+      subscriber.callbacks.forEach(function (cb) {
+        cb(value);
+      });
+    }
+
     return function unsubscribe() {
       subscriber.callbacks = subscriber.callbacks.filter(function (cb) {
         return cb !== callback;

src/api/telemetry/TelemetryCollection.js

@@ -180,11 +180,14 @@ export default class TelemetryCollection extends EventEmitter {
     if (this.unsubscribe) {
       this.unsubscribe();
     }
+    const options = { ...this.options };
+    //We always want to receive all available values in telemetry tables.
+    options.strategy = 'batch';
 
     this.unsubscribe = this.openmct.telemetry.subscribe(
       this.domainObject,
       (datum) => this._processNewTelemetry(datum),
-      this.options
+      options
     );
   }