MJPEGConsumer.js

Filename	Percent	Threshold	Total	Covered	Uncovered
src/consumers/MJPEGConsumer.js	93.88%	80%	441	414	27

// @ts-check
/*
** Copyright (C) 2021 Gabriele De Blasi
**
** This software is provided 'as-is', without any express or implied
** warranty.  In no event will the authors be held liable for any damages
** arising from the use of this software.
**
** Permission is granted to anyone to use this software for any purpose,
** including commercial applications, and to alter it and redistribute it
** freely, subject to the following restrictions:
**
** 1. The origin of this software must not be misrepresented; you must not
**    claim that you wrote the original software. If you use this software
**    in a product, an acknowledgment in the product documentation would be
**    appreciated but is not required.
** 2. Altered source versions must be plainly marked as such, and must not be
**    misrepresented as being the original software.
** 3. This notice may not be removed or altered from any source distribution.
**
** Created on: 2021-11-30T09:30:00+01:00
**     Author: Gabriele De Blasi <gabriele.de.blasi@cern.ch>
**
*/

import { EventEmitter } from "node:events";
import http from "node:http";
import { parse } from "content-type";
import VarBuffer from "./VarBuffer.js";

/**
 * @typedef {import('http').ClientRequest} Request
 * @typedef {import('http').IncomingMessage} Response
 *
 * @typedef {{ user: string, password: string }} Authentication
 * @typedef {{ src: string, auth?: Authentication, contentLength?: boolean, jpegMagicBytes: boolean,
 *  headerDelimiter: string, boundary?: string, throttle?: number }} MJPEGConsumerOptions
 */

const SearchStates = { HEADER_END: 1, CONTENT_END: 2, NEXT_BOUNDARY: 3 };
const CRLFx2 = "\r\n\r\n";
const JPEG_MAGIC_BYTES = Buffer.from("ffd8ff", "hex");

/**
 * @param  {function} fun
 */
function frameSkip(fun) {
  const DELTA = 10; // ms
  let deadline = 0;

  return function(/** @type {any[]} */ ...args) {
    const now = Date.now();
    if (now + DELTA < deadline) { return; }
    deadline = now + this.throttle;
    fun(...args);
  };
}

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

  /**
   * @param {MJPEGConsumerOptions} options
   */
  constructor(options) {
    super();
    this.src = options?.src ?? "";
    this.user = options?.auth?.user;
    this.pass = options?.auth?.password;
    this.contentLength = options?.contentLength ?? true;
    this.boundary = options?.boundary;
    this.headerDelimiter = options?.headerDelimiter;
    this.jpegMagicBytes = options?.jpegMagicBytes ?? false;
    this._connected = false;
    this._buffer = new VarBuffer();

    this._initialThrottle = options?.throttle;

    this.updateThrottle(options?.throttle);

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

    this._state = SearchStates.HEADER_END;
    this._offset = 0; // search start byte in buffer
    this._contentStart = 0;
    this._frameEnd = 0;

    /** @type {Buffer|null} */
    this._delimiter = null;

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

    this.emitFrame = this.throttle ?
      frameSkip(this.emit.bind(this, "frame")) : // limit the frame rate
      this.emit.bind(this, "frame"); // no throttle => direct emit to reduce frame loss
  }

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

    this.throttle = Number(throttle);
  }

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

    this._connected = true;
    this._buffer.reset();

    this._delimiter = this.jpegMagicBytes ? JPEG_MAGIC_BYTES :
      Buffer.from(this.headerDelimiter || CRLFx2);

    const opts = {
      auth: (this.user && this.pass) ? `${this.user}:${this.pass}` : undefined
    };

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

    req
    .on("error", this.emitError)
    .once("close", () => {
      req.removeAllListeners();
      // disconnect if the current connection has been terminated by the source
      if (req === this._req) {
        this.disconnect();
      }
    })
    .on("response", (res) => {
      res
      .on("error", this.emitError)
      .once("close", () => res.removeAllListeners());

      if ((res.headers["content-type"] ?? "").includes("multipart/x-mixed-replace")) {
        try {
          this.parse(res);
        }
        catch (err) {
          this.emitError(err);
        }
      }
      else {
        this.emitError(new Error("No MJPEG stream detected"));
        this.disconnect();
      }
    })
    .end();
  }

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

    this.emit("close");
    this._req = null;
  }

  /**
   * @param {Response} res
   */
  parse(res) {
    /** @type {Buffer} */ let boundary;
    try {
      boundary = Buffer.from(this.boundary ??
        parse(res.headers["content-type"] ?? "")?.parameters?.boundary);
    }
    catch (err) {
      // retrieve boundary param by regex
      // (see https://datatracker.ietf.org/doc/html/rfc2046#section-5.1.1)
      const match = (res.headers["content-type"] ?? "")
      .match(/boundary="?([\w\s'()+,\-./:=?]{1,70})"?;?/);

      if (!match) { throw err; } // boundary missing or invalid

      boundary = Buffer.from(match[1].trimEnd());
    }

    // minimum parsable header length
    const minHeaderLen = boundary.length + this._delimiter.length;

    res
    .on("data", (chunk) => { // eslint-disable-line complexity, max-statements
      this._buffer.add(chunk);

      let index = 0;

      while (this._buffer.length > minHeaderLen) {

        switch (this._state) {
        case SearchStates.HEADER_END:
          index = this._buffer.buffer().indexOf(this._delimiter, this._offset);

          if (index >= 0) {
            this._contentStart = this._offset =
              index + (this.jpegMagicBytes ? 0 : this._delimiter.length);
          }
          else if (this._buffer.length > MJPEGConsumer.MAX_HEADER_LENGTH) {

            if (!this.jpegMagicBytes) {
              // fallback solution: search for JPEG magic numbers
              index = this._buffer.buffer().indexOf(JPEG_MAGIC_BYTES);
            }

            if (index >= 0) {
              this._contentStart = this._offset = index;
            }
            else { // drop data
              this._buffer.shift(MJPEGConsumer.MAX_HEADER_LENGTH);
              this._offset = 0;
              console.warn(
                `Discarded the first ${MJPEGConsumer.MAX_HEADER_LENGTH} ` +
                "bytes as the maximum header length has been reached.");
              return;
            }
          }
          else {
            // set starting byte for next search considering the worst case:
            // e.g. last 3 bytes equal to '\r\n\r' (with CRLFx2 as delimiter)
            this._offset =
              this._buffer.length - Math.abs(this._delimiter.length - 1);

            if (this._offset < 0) { this._offset = 0; }
            return;
          }

          // choice between content-Length and boundary
          {
            const match = this.contentLength ?
              this._buffer.slice(0, index).toString("ascii")
              .match(/Content-Length:\s*(\d+)/i) : null;

            if (!match) {
              this._state = SearchStates.NEXT_BOUNDARY;
              break;
            }

            this._frameEnd = this._contentStart + Number(match[1]);
            this._state = SearchStates.CONTENT_END;
          }
          /* falls through */
        case SearchStates.CONTENT_END:
          if (this._buffer.length < this._frameEnd) { return; }
          this.emitFrame(
            this._buffer.slice(this._contentStart, this._frameEnd));

          this._buffer.shift(this._frameEnd);
          this._offset = 0;
          this._state = SearchStates.HEADER_END;
          break;
        case SearchStates.NEXT_BOUNDARY:
          index = this._buffer.buffer().indexOf(boundary, this._offset);

          if (index === -1) {
            // same as above
            this._offset = this._buffer.length - (boundary.length - 1);
            if (this._offset < 0) {
              this._offset = 0;
            }
            else if (this._buffer.length > MJPEGConsumer.MAX_CONTENT_LENGTH) {
              this._buffer.shift(this._offset);
              this._offset = 0;
            }
            return;
          }
          this.emitFrame(this._buffer.slice(this._contentStart, index));

          this._buffer.shift(index);
          this._offset = boundary.length; // re-start from after the boundary
          /* falls through */
        default:
          this._state = SearchStates.HEADER_END;
          break;
        }
      }
    })
    .on("end", this.emit.bind(this, "ended"));
  }
}
MJPEGConsumer.MAX_HEADER_LENGTH = 2048; // bytes
MJPEGConsumer.MAX_CONTENT_LENGTH = 20971520; // bytes (= 20 MiB)
MJPEGConsumer.MIN_THROTTLE_VALUE = 17; // ms  (~ 60 fps)

export default MJPEGConsumer;

[{"file":"src/consumers/MJPEGConsumer.js","line":53,"character":21,"text":"this","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":53,"character":26,"text":"throttle","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":74,"character":9,"text":"src","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":75,"character":9,"text":"user","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":76,"character":9,"text":"pass","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":77,"character":9,"text":"contentLength","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":78,"character":9,"text":"boundary","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":79,"character":9,"text":"headerDelimiter","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":80,"character":9,"text":"jpegMagicBytes","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":81,"character":9,"text":"_connected","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":82,"character":9,"text":"_buffer","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":84,"character":9,"text":"_initialThrottle","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":91,"character":9,"text":"_state","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":92,"character":9,"text":"_offset","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":93,"character":9,"text":"_contentStart","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":94,"character":9,"text":"_frameEnd","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":99,"character":9,"text":"emitError","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":99,"character":22,"text":"err","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":101,"character":27,"text":"err","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":105,"character":9,"text":"emitFrame","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":110,"character":17,"text":"throttle","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":111,"character":8,"text":"throttle","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":111,"character":20,"text":"throttle","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":112,"character":6,"text":"throttle","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":114,"character":24,"text":"throttle","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":119,"character":27,"text":"throttle","kind":1},{"file":"src/consumers/MJPEGConsumer.js","line":200,"character":17,"text":"chunk","kind":1}]

TypeScript coverage report