openmct/example/exampleStalenessProvider/ExampleStalenessProvider.js

/*****************************************************************************
 * Open MCT, Copyright (c) 2014-2025, United States Government
 * as represented by the Administrator of the National Aeronautics and Space
 * Administration. All rights reserved.
 *
 * Open MCT 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 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.
 *****************************************************************************/

/**
 * @implements {import('src/api/telemetry/TelemetryAPI').StalenessProvider}
 */
export default class ExampleStalenessProvider {
  #intervalId;
  constructor(openmct, config = { stalenessInterval: 3000, reportStalenessInterval: 300 }) {
    this.openmct = openmct;
    this.stalenessInterval = config.stalenessInterval;
    this.reportStalenessInterval = config.reportStalenessInterval;
    this.observingStaleness = {};
    this.latestReceivedTelemetry = {};

    this.#observeTimeSystem();
    this.#observeStaleness();
  }

  #observeTimeSystem() {
    this.openmct.time.on('timeSystemChanged', () => {
      this.timeSystem = this.openmct.time.getTimeSystem();
    });
  }

  supportsStaleness(domainObject) {
    return this.openmct.telemetry.isTelemetryObject(domainObject);
  }

  subscribeToStaleness(domainObject, callback) {
    const keyString = this.openmct.objects.makeKeyString(domainObject.identifier);

    this.observingStaleness[keyString] = { callback };
    const unsubscribe = this.openmct.telemetry.subscribe(domainObject, (datum) => {
      this.#updateLatestReceivedTelemetry(domainObject, datum);
    });

    return () => {
      delete this.observingStaleness[keyString];
      unsubscribe?.();
      if (Object.keys(this.observingStaleness).length === 0) {
        clearInterval(this.#intervalId);
      }
    };
  }

  #observeStaleness() {
    this.#intervalId = setInterval(() => {
      if (!this.timeSystem) {
        return;
      }

      Object.entries(this.observingStaleness).forEach(([keyString, observer]) => {
        if (!this.latestReceivedTelemetry[keyString]) {
          return;
        }

        const now = this.openmct.time.now();
        const isStale = now - this.latestReceivedTelemetry[keyString] >= this.stalenessInterval;

        // Overly reports when not stale because of generated telemetry flake
        if (!isStale || !observer.response || isStale !== observer.response.isStale) {
          const stalenessResponseObject = {
            isStale,
            [this.timeSystem.key]: now
          };

          observer.response = stalenessResponseObject;
          observer.callback(stalenessResponseObject);
        }
      });
    }, this.reportStalenessInterval);
  }

  /**
   * @param {*} domainObject
   * @returns {import('src/api/telemetry/TelemetryAPI').StalenessResponseObject}
   */
  async isStale(domainObject) {
    if (!this.timeSystem) {
      this.timeSystem = this.openmct.time.getTimeSystem();
    }

    const keyString = this.openmct.objects.makeKeyString(domainObject.identifier);

    if (!this.latestReceivedTelemetry[keyString]) {
      // Naively assumes sorted request response so uses last datum in array
      const response = await this.openmct.telemetry.request(domainObject, { strategy: 'latest' });
      const lastDatum = response?.length ? response[response.length - 1] : undefined;
      this.#updateLatestReceivedTelemetry(domainObject, lastDatum);
    }

    const timestamp = this.latestReceivedTelemetry[keyString];
    if (timestamp) {
      const isStale = this.openmct.time.now() - timestamp >= this.stalenessInterval;

      const stalenessResponseObject = { isStale };
      stalenessResponseObject[this.timeSystem.key] = timestamp;

      return stalenessResponseObject;
    }
  }

  #updateLatestReceivedTelemetry(domainObject, datum) {
    const keyString = this.openmct.objects.makeKeyString(domainObject.identifier);
    const metadata = this.openmct.telemetry.getMetadata(domainObject);
    const metadataValue = metadata.value(this.timeSystem.key) || { format: this.timeSystem.key };
    const valueFormatter = this.openmct.telemetry.getValueFormatter(metadataValue);
    const timestamp = valueFormatter.parse(datum);

    if (timestamp) {
      this.latestReceivedTelemetry[keyString] = timestamp;
    } else {
      console.warn('Could not parse timestamp for staleness check');
    }
  }
}