CamService.js

Filename	Percent	Threshold	Total	Covered	Uncovered
src/CamService.js	98.93%	80%	843	834	9

/* eslint-disable max-lines */
// @ts-check
// SPDX-License-Identifier: Zlib
// SPDX-FileCopyrightText: 2024 CERN (home.cern)

import { isEmpty, isNil, set } from "@cern/nodash";
import d from "debug";
import cors from "cors";
import ConsumerFactory from "./consumers/ConsumerFactory.js";
import StreamClient from "./StreamClient.js";
import CamSwagger from "./CamSwagger.js";
import PTZService from "./ptz/PTZService.js";
import SShotService from "./screenshots/SShotService.js";

const debug = d("cam:service");

/**
 * @typedef {import('express').Request} Request
 * @typedef {import('http').OutgoingMessage} HttpResponse
 * @typedef {{ [id: number]: StreamClient }} StreamClients
 *
 * @typedef {import('express').IRouter} expressIRouter
 * @typedef {import('express-ws').WithWebsocketMethod & expressIRouter} IRouter
 * @typedef {import('ws').default} WebSocket
 */

/**
 * @swagger
 * definitions:
 *   Config:
 *     description: 'List of cameras divided by group'
 *     type: object
 *     additionalProperties:
 *       type: object
 *       properties:
 *         description:
 *           type: string
 *         cameras:
 *           type: array
 *           items:
 *             type: object
 *             properties:
 *               name: { type: string}
 *               description: { type: string }
 *               ptz: { type: boolean }
 *             required: [ name ]
 *     example:
 *       default:
 *         cameras:
 *           - name: "cam1"
 *             description: "camera 1"
 *           - name: "cam2"
 *             description: "camera 2"
 *         description: "default group"
 */

const WS_GOING_AWAY = 1001;
const WS_INTERNAL_ERROR = 1011;

function exponentialTimeout(i) {
  const max = 10; // -> max timeout ~1h
  return CamService.CONN_RETRY << ((i < max) ? i : max);
}

class CamService {

  /**
   * @param {CamExpress.Config} config groups and cameras configuration
   * @param {CamExpress.CamSwagger.Options} [options] filling swagger base config
   * @param {CamExpress.CamSwagger.Info} [extraDoc] merged with dynamic swagger doc
   */
  constructor(config, options, extraDoc) {
    /** @type {number} */
    this._id = 0; // subscription id
    this._camId = 0; // CameraConfig id
    this.hasPTZ = false;
    this.hasSShot = false;

    /** @type {PTZService|null} */
    this.ptz = null;

    /** @type {CamExpress.InternalGroupConfig} */
    this.groups = this._prepareGroups(config?.groups);

    /** @type {CamExpress.Credentials} */
    this.creds = config?.credentials ?? {};

    /** @type {CamExpress.CameraList} */
    this.camConfig = {};

    /** @type {number} */
    this.unsubscribeTimeout = (config?.unsubscribeTimeout ?? 0) * 1000; // ms

    /** @type {CamExpress.CameraMap} */
    this._camMap = {};

    /** @type {CamExpress.CamSwagger.Options|undefined} */
    this._swaggerOptions = options;

    /** @type {CamExpress.CamSwagger.Info|undefined} */
    this._swaggerExtraDoc = extraDoc;
  }

  /**
   *
   * @param {{ [group: string]: CamExpress.GroupConfig }} groups
   * @return {CamExpress.InternalGroupConfig}
   */
  _prepareGroups(groups) {
    Object.values(groups ?? {}).forEach((g) => {
      Object.values(g?.cameras ?? []).forEach((c) => {
        // @ts-ignore
        if (isNil(c.id)) {
          // @ts-ignore
          c.id = this._camId++;
        }
        this.hasPTZ = this.hasPTZ || !!c.ptz;
        this.hasSShot = this.hasSShot || !!c.screenshot;
      });
      // @ts-ignore
      g.cameras = Object.values(g?.cameras ?? []).reduce(
        (r, x) => { r[x["name"]] = x; return r; }, {});
    });
    return /** @type {CamExpress.InternalGroupConfig} */ (/** @type {unknown}*/ (groups));
  }

  /**
   * @param {Request} req
   * @return {CamExpress.Camera|undefined}
   */
  getCamera(req) {
    const group = req?.params?.group;
    const camera = req?.params?.camera;
    return this.groups?.[group]?.cameras?.[camera];
  }

  /**
   * @param {Request} req
   * @return {CamExpress.CameraConfig}
   */
  getMultiplexedCamera(req) {
    // cameras as array of string => "group/camera"
    /** @type {string[]} */
    const paths = req.query.cameras?.toString().split(",") ?? [];
    const durations = req.query.durations?.toString().split(",") ?? [];
    const cameras = paths?.map((path, index) => {
      const [ group, camera ] = path.split("/");
      const duration = Number(durations?.[index]);

      if (isNil(group) || isNil(camera)) {
        throw new Error(
          'Bad query (cameras path should be as "group/camera")'
        );
      }

      return { group, camera, duration };
    }) ?? [];

    return { name: JSON.stringify(cameras), multiplex: cameras };
  }

  /**
   * @param {string|number} camId
   * @param {Buffer|Error} data
   */
  _broadcast(camId, data) {
    Object.values(this._camMap[camId]?.clients ?? {})
    .forEach((c) => c.throttledCB(data));
  }

  /**
   * @param {IRouter} router
   */
  register(router) {
    this.release();

    this.camConfig = this.getConfig();

    /**
     * @swagger
     * /config:
     *   get:
     *     tags: ['config']
     *     summary: list current cameras config
     *     description: list current cameras config
     *     produces:
     *       - application/json
     *     responses:
     *       200:
     *         description: list current cameras config
     *         schema:
     *           $ref: '#/definitions/Config'
     *       500:
     *         description: camera configuration is missing
     */
    router.get("/config", cors(), (req, res) => {
      if (!isEmpty(this.camConfig)) {
        res.status(200).json(this.camConfig);
      }
      else {
        res.status(500).send("Camera configuration missing");
      }
    });

    /**
     * @swagger
     * /cameras/{group}/{camera}:
     *   get:
     *     tags: ['cameras']
     *     summary: access the camera frames
     *     description: |
     *       Exposes to clients two routes (streaming via HTTP and WebSocket) and sends the JPEG images
     *       extracted from the MJPEG fragments or fetched through http polling.
     *     parameters:
     *      - in: path
     *        name: group
     *        schema:
     *           type: string
     *        description: name of the group defined in the config
     *        required: true
     *      - in: path
     *        name: camera
     *        schema:
     *           type: string
     *        description: name of the camera defined in the config
     *        required: true
     *      - in: query
     *        name: throttle
     *        schema:
     *           type: int
     *        description: asked throttle for framerate regulation from client
     *        required: false
     *     produces:
     *       - application/json
     *     responses:
     *       200:
     *         description: (GET) open a frame stream via http
     *       101:
     *         description: (WS) open a frame stream via websockets
     *       500:
     *         description: error message
     */
    router.get("/cameras/:group/:camera", (req, res) => {
      if (isEmpty(this.camConfig)) {
        res.status(500).send("No cameras in configuration");
        return;
      }
      const throttle = req.query.throttle; // Throttle from client
      const closeOnErrorArg = (req.query.closeOnError ?? "")
      .toString().toLowerCase();
      const closeOnError =
        !([ "false", "0" ].some((v) => (closeOnErrorArg === v)));

      const cam = this.getCamera(req);

      if (isNil(cam)) {
        res.status(500).send("Camera not found");
        return;
      }
      this.subscribeHTTP(cam, throttle, res, closeOnError);
    });

    /**
     * @swagger
     * /multiplex:
     *   get:
     *     tags: ['config']
     *     summary: rotate between the client query specified cameras stream
     *     description: |
     *       Exposes to clients two routes (streaming via HTTP and WebSocket) and
     *       rotates between the client query specified cameras stream
     *     parameters:
     *      - in: query
     *        name: cameras
     *        schema:
     *          type: array
     *          items: string
     *          example: 'default/cam1, default/cam2'
     *      - in: query
     *        name: durations
     *        schema:
     *          type: array
     *          items: string
     *          example: '2000,4000'
     *     produces:
     *       - application/json
     *     responses:
     *       200:
     *         description: (GET) open a frame stream via http
     *       101:
     *         description: (WS) open a frame stream via websockets
     *       500:
     *         description: error message
     */
    router.get("/multiplex", (req, res) => {
      if (isEmpty(this.camConfig)) {
        res.status(500).send("No cameras in configuration");
        return;
      }
      const closeOnErrorArg = (req.query.closeOnError ?? "")
      .toString().toLowerCase();
      const closeOnError =
        !([ "false", "0" ].some((v) => (closeOnErrorArg === v)));

      try {
        // virtual camera config for referencing a temporary consumer
        const virtual = this.getMultiplexedCamera(req);

        this.subscribeHTTP(virtual, undefined, res, closeOnError);
      }
      catch (err) {
        res.status(500).send(err.message);
      }
    });

    if (this.hasPTZ) {
      this.ptzService = new PTZService(this.groups, this.creds);
      this.ptzService.register(router);
    }
    if (this.hasSShot) {
      this.sshotService = new SShotService(this.groups, this.creds);
      this.sshotService.register(router);
    }

    if (router.ws) {
      router.ws("/cameras/:group/:camera", (ws, req) => {
        if (isEmpty(this.camConfig)) {
          ws.close(WS_INTERNAL_ERROR, "No cameras in configuration");
          return;
        }

        const throttle = req.query.throttle; // Throttle from client
        const closeOnErrorArg = (req.query.closeOnError ?? "")
        .toString().toLowerCase();
        const closeOnError =
          !([ "false", "0" ].some((v) => (closeOnErrorArg === v)));

        const cam = this.getCamera(req);

        if (isNil(cam)) {
          ws.close(WS_INTERNAL_ERROR, "Camera not found");
          return;
        }

        this.subscribeWS(cam, throttle, ws, closeOnError);
      });

      router.ws("/multiplex", (ws, req) => {
        if (isEmpty(this.camConfig)) {
          ws.close(WS_INTERNAL_ERROR, "No cameras in configuration");
          return;
        }
        const closeOnErrorArg = (req.query.closeOnError ?? "")
        .toString().toLowerCase();
        const closeOnError =
          !([ "false", "0" ].some((v) => (closeOnErrorArg === v)));

        try {
          // virtual camera config for referencing a temporary consumer
          const virtual = this.getMultiplexedCamera(req);

          this.subscribeWS(virtual, undefined, ws, closeOnError);
        }
        catch (err) {
          ws.close(WS_INTERNAL_ERROR, err.message);
        }
      });
    }

    const swagger = new CamSwagger(this._swaggerOptions, this._swaggerExtraDoc);

    swagger.swag(this);

    swagger.register(router);
  }

  /**
   * @brief provides configuration to retrieve video streams from cameras
   * @return {CamExpress.ExposedConfig}
   */
  getConfig() {
    return Object.entries(this.groups ?? {}).reduce((ret, [ key, group ]) => {
      ret[key] = {
        description: group?.description ?? "",
        cameras: Object.values(group?.cameras ?? {}).map((camera) => {
          const ret = {
            name: camera.name, description: camera?.description ?? ""
          };
          if (camera.ptz) { ret.ptz = true; }
          if (camera.screenshot) { ret.screenshot = true; }
          if (camera.extra) { Object.assign(ret, camera.extra); }
          return ret;
        })
      };
      return ret;
    }, {});
  }

  /**
   * @param {CamExpress.Camera} cam
   * @param {number|undefined} throttle
   * @param {HttpResponse} res
   * @param {boolean} [closeOnError=true]
   */
  subscribeHTTP(cam, throttle, res, closeOnError = true) {

    const id = this.subscribe(cam, throttle, (data) => {

      if (!data) {
        res.end();
        return;
      }

      try {
        if (Buffer.isBuffer(data)) {
          res.write("--frame\r\n");
          res.write("Content-Type: image/jpeg\r\n");
          res.write(`Content-Length: ${data.length}\r\n\r\n`);
          res.write(data);
          res.write("\r\n");
        }
        else if (data instanceof Error) {
          if (closeOnError) { throw data; }
          res.write("--frame\r\n");
          const payload = JSON.stringify({ type: "error", data: data.message });
          res.write("Content-Type: application/json\r\n");
          res.write(`Content-Length: ${payload.length}\r\n\r\n`);
          res.write(payload);
          res.write("\r\n");
        }
      }
      catch (err) {
        res.write("--frame\r\n");
        const payload = JSON.stringify({ type: "error", data: data.message });
        res.write("Content-Type: application/json\r\n");
        res.write(`Content-Length: ${payload.length}\r\n\r\n`);
        res.end(payload);
      }
    });

    res
    .on("error", (err) => {
      if (!res.writableEnded) {
        res.write("--frame\r\n");
        res.write("Content-Type: application/json\r\n");
        res.end(JSON.stringify({ type: "error", data: err.message }));
      }
    })
    .once("close", () => {
      res.removeAllListeners();
      this.unsubscribe(cam, id);
    })
    .status(200)
    .set("Cache-Control", "no-store")
    .set("Content-Type", "multipart/x-mixed-replace; boundary=frame")
    .set("Pragma", "no-cache")
    .flushHeaders();
  }

  /**
   * @param {CamExpress.Camera} cam
   * @param {number|undefined} throttle
   * @param {WebSocket} ws
   * @param {boolean} [closeOnError=true]
   */
  subscribeWS(cam, throttle, ws, closeOnError = true) {

    const id = this.subscribe(cam, throttle, (data) => {

      // Check if we want to close the socket
      if (!data) {
        ws.close(WS_GOING_AWAY);
        return;
      }

      // Otherwise, we send the datas
      try {
        if (Buffer.isBuffer(data)) {
          ws.send(data);
        }
        else if (data instanceof Error) {
          if (closeOnError) { throw data; }
          ws.send(JSON.stringify({ type: "error", data: data.message }));
        }
      }
      catch (err) {
        ws.close(WS_INTERNAL_ERROR, err.message);
      }
    });

    ws.on("error", (err) => {
      ws.close(WS_INTERNAL_ERROR, err.message);
    })
    .once("close", () => {
      ws.removeAllListeners();
      this.unsubscribe(cam, id);
    });
  }

  /**
   * @param {CamExpress.Camera} cam
   * @param {number|undefined} throttle
   * @param {(data?: Buffer|Error) => void} cb
   * @return {number}
   */
  subscribe(cam, throttle, cb) {

    const clients = this._camMap[cam.id]?.clients || {};

    // If first client, init the fetching of camera frames
    if (isEmpty(clients)) {
      set(this._camMap, [ cam.id, "clients" ], clients);
      // try to resume camera stream
      const timeout = this._camMap[cam.id]?.unSubTimeout;
      if (timeout) {
        clearTimeout(timeout);
        this._camMap[cam.id].unSubTimeout = null;
      }
      else {
        // Create, store and add event handling for the camera's consumer
        this.startStream(cam);
      }
    }
    else if (!isNil(this._camMap[cam.id]?.error)) {
      cb(/** @type {Error} */ (this._camMap[cam.id].error)); // eslint-disable-line callback-return
    }

    const id = this._id++;

    // Instanciate a new client
    clients[id] = new StreamClient({ throttle, cb });

    return id;
  }

  /**
   * @param {CamExpress.Camera} cam
   */
  startStream(cam) {

    // Determine type of stream
    const consumer = ConsumerFactory.createConsumer(cam, this);

    // Add events handling
    consumer
    .on("frame", (frame) => {
      set(this._camMap, [ cam.id, "retryAttempt" ], 0);
      set(this._camMap, [ cam.id, "error" ], null);
      this._broadcast(cam.id, frame);
    })
    .on("error", (err) => {
      set(this._camMap, [ cam.id, "error" ], err);
      this._broadcast(cam.id, err);
    })
    .on("close", () => {
      const camera = this._camMap[cam.id];

      if (!isEmpty(camera?.clients) && isNil(camera.retryTimer)) {
        // retry connection
        camera.retryTimer = setTimeout(() => {
          if (this._camMap[cam.id]?.consumer === consumer) {
            this._camMap[cam.id].retryTimer = null;
            consumer.connect();
          }
          else {
            consumer.removeAllListeners();
          }
        }, exponentialTimeout(camera.retryAttempt++));
      }
      else {
        consumer.removeAllListeners();
      }
    })
    .connect();

    // Store the camera context
    set(this._camMap, [ cam.id, "consumer" ], consumer);
    set(this._camMap, [ cam.id, "retryTimer" ], null);
    set(this._camMap, [ cam.id, "retryAttempt" ], 0);
    set(this._camMap, [ cam.id, "error" ], null);
  }

  /**
   * @param {string|number} camId
   */
  stopStream(camId) {
    const camera = this._camMap[camId];

    if (camera) {
      camera.consumer?.removeAllListeners();
      camera.consumer?.disconnect();
      camera.consumer = null;
      clearTimeout(camera.retryTimer);
      camera.retryTimer = null;
      camera.retryAttempt = 0;
      camera.error = null;
    }
  }

  /**
   * @param {CamExpress.Camera} cam
   * @param {number} id
   */
  unsubscribe(cam, id) {
    const clients = this._camMap[cam.id]?.clients;
    if (isNil(clients)) { return; }

    delete clients[id];

    if (isEmpty(clients)) {
      this._camMap[cam.id].unSubTimeout = setTimeout(() => {

        this.stopStream(cam.id);

        set(this._camMap, [ cam.id, "clients" ], {});
        set(this._camMap, [ cam.id, "unSubTimeout" ], null);
      },
      this.unsubscribeTimeout);
    }
  }

  release() {
    Object.values(this._camMap ?? {}).forEach((cam, camId) => {
      clearTimeout(cam.unSubTimeout ?? undefined);
      cam.unSubTimeout = null;
      Object.values(cam.clients ?? {}).forEach((c) => c.cb()); // close WS

      this.stopStream(camId);
    });

    this._camMap = {};
    this.camConfig = {};
    this._id = 0;
    this.ptzService?.release();
    this.sshotService?.release();

    debug("CamService released");
  }

}
CamService.CONN_RETRY = 3500; // ms

export default CamService;

[{"file":"src/CamService.js","line":59,"character":28,"text":"i","kind":1},{"file":"src/CamService.js","line":61,"character":36,"text":"i","kind":1},{"file":"src/CamService.js","line":61,"character":47,"text":"i","kind":1},{"file":"src/CamService.js","line":74,"character":9,"text":"_camId","kind":1},{"file":"src/CamService.js","line":75,"character":9,"text":"hasPTZ","kind":1},{"file":"src/CamService.js","line":76,"character":9,"text":"hasSShot","kind":1},{"file":"src/CamService.js","line":545,"character":18,"text":"frame","kind":1},{"file":"src/CamService.js","line":550,"character":18,"text":"err","kind":1},{"file":"src/CamService.js","line":551,"character":45,"text":"err","kind":1}]

TypeScript coverage report