HTTPConsumer.js

Filename	Percent	Threshold	Total	Covered	Uncovered
src/consumers/HTTPConsumer.js	90.71%	80%	226	205	21

// @ts-check

import { EventEmitter } from "node:events";
import http from "node:http";
import { isNil } from "@cern/nodash";
import VarBuffer from "./VarBuffer.js";
import Agent from "agentkeepalive";

const httpAgent = new Agent({ timeout: 10000 });
const httpsAgent = new Agent.HttpsAgent({ timeout: 10000 });

/**
 * @typedef {import('http').ClientRequest} Request
 * @typedef {import('http').IncomingMessage} Response
 *
 * @typedef {{ user: string, password: string }} Authentication
 * @typedef {{ src: string, auth?: Authentication, throttle?: number }} HTTPConsumerOptions
 */

/**
 * @implements {CamExpress.Consumer}
 * @extends {EventEmitter}
 * @emits :
 *  - 'frame' when a new jpeg frame is received from source
 *  - 'error' when an error occurs
 *  - 'close' once the connection with source is closed
 */
class HTTPConsumer extends EventEmitter {

  /**
   * @param {HTTPConsumerOptions} options
   */
  constructor(options) {
    super();
    this.src = options?.src ?? "";
    this.user = options?.auth?.user;
    this.pass = options?.auth?.password;

    this._initialThrottle = options?.throttle;

    this.updateThrottle(options?.throttle);

    this._connected = false;
    this._buffer = new VarBuffer();

    /** @type {Request|null} */
    this._req = null;

    /** @type {NodeJS.Timeout|undefined} */
    this._pollingTimer = undefined;

    this.emitError = (err) => {
      if (this._connected) {
        this.emit("error", err);
      }
    };

    this.emitFrame = this.emit.bind(this, "frame");
  }

  updateThrottle(throttle) {
    throttle = throttle ?? HTTPConsumer.DEFAULT_FRAME_THROTTLE; // Assert non null value

    if (throttle < HTTPConsumer.MIN_THROTTLE_VALUE) {
      throttle = HTTPConsumer.MIN_THROTTLE_VALUE; // Assert throttle min value
      console.warn(
        `The throttle ${throttle} is too low and` +
        ` was replaced with ${HTTPConsumer.MIN_THROTTLE_VALUE} ms`
      );
    }

    this.throttle = throttle;
  }

  connect() {
    if (this._connected) { return; }

    this._connected = true;

    const opts = {
      agent: this.src.includes("https") ? httpsAgent : httpAgent,
      auth: (this.user && this.pass) ? `${this.user}:${this.pass}` : undefined
    };

    const httpPolling = () => {

      if (!this._connected) {
        return; // Avoid keeping recalling itself if the connexion was closed
      }

      const startTime = Date.now(); // time at the request queuing

      this._buffer.reset();
      const req = this._req = http.get(this.src, opts);

      req
      .on("error", this.emitError)
      .once("close", () => {
        req.removeAllListeners();
        // disconnect if http polling has been stopped due to some errors
        if (req === this._req && isNil(this._pollingTimer)) {
          this.disconnect();
        }
      })
      .on("response", (res) => {
        res
        .on("error", this.emitError)
        .once("close", () => res.removeAllListeners());

        if ((res.headers["content-type"] ?? "").includes("image/jpeg")) {

          res
          .on("data", (chunk) => this._buffer.add(chunk))
          .on("end", () => {
            if (req !== this._req || !this._connected) { return; }

            const endTime = Date.now(); // time at the request end
            const elapsedTime = endTime - startTime;
            const diffThrottle = this.throttle - elapsedTime;

            const toWait = diffThrottle > HTTPConsumer.MIN_FRAME_THROTTLE ?
              diffThrottle :
              HTTPConsumer.MIN_FRAME_THROTTLE;

            if (diffThrottle < 0) {
              console.warn(
                `The throttle ${this.throttle} is too low and was not applied`
              );
            }

            this.emitFrame(this._buffer.buffer());
            // stop polling if disconnection occurs on 'frame' event
            if (!this._connected) { return; }

            // When the previous frame received, we fetch the next one
            // if still connected
            this._pollingTimer = setTimeout(() => {
              this._pollingTimer = undefined;
              httpPolling(); // Recursive call
            }, toWait);
          });
        }
        else {
          this.emitError(new Error("No JPEG image detected"));
          this.disconnect();
        }
      })
      .end();
    };

    httpPolling(); // Begin polling loop

  }

  disconnect() {
    if (!this._connected) { return; }
    this._connected = false;

    clearTimeout(this._pollingTimer);
    this._pollingTimer = undefined;

    this.emit("close");
    this._req = null;
  }
}
// default interval between a frame receival and the next frame request
HTTPConsumer.DEFAULT_FRAME_THROTTLE = 1000; // ms
// minimum throttle that will be applied between two frames
HTTPConsumer.MIN_FRAME_THROTTLE = 0; // ms
// minimum input value for the throttle property
HTTPConsumer.MIN_THROTTLE_VALUE = 200; // ms (~ 5 fps)

export default HTTPConsumer;

[{"file":"src/consumers/HTTPConsumer.js","line":34,"character":9,"text":"src","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":35,"character":9,"text":"user","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":36,"character":9,"text":"pass","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":38,"character":9,"text":"_initialThrottle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":42,"character":9,"text":"_connected","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":43,"character":9,"text":"_buffer","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":51,"character":9,"text":"emitError","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":51,"character":22,"text":"err","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":53,"character":27,"text":"err","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":57,"character":9,"text":"emitFrame","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":60,"character":17,"text":"throttle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":61,"character":4,"text":"throttle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":61,"character":15,"text":"throttle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":63,"character":8,"text":"throttle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":64,"character":6,"text":"throttle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":66,"character":24,"text":"throttle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":71,"character":9,"text":"throttle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":71,"character":20,"text":"throttle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":112,"character":23,"text":"chunk","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":118,"character":38,"text":"throttle","kind":1},{"file":"src/consumers/HTTPConsumer.js","line":126,"character":37,"text":"throttle","kind":1}]

TypeScript coverage report