test_MJPEGConsumer.js

Filename	Percent	Threshold	Total	Covered	Uncovered
test/test_MJPEGConsumer.js	96.34%	80%	573	552	21

// @ts-check
// SPDX-License-Identifier: Zlib
// SPDX-FileCopyrightText: 2024 CERN (home.cern)

/* eslint-disable max-lines */

import { expect } from "chai";
import { afterEach, beforeEach, describe, it } from "mocha";
import { isNil } from "@cern/nodash";

import { MJPEGServer, waitFor, waitForEvent, waitForValue } from "./utils/index.js";
import config from "./config-stub.js";
import { MJPEGConsumer } from "../src/index.js";


describe("MJPEG Consumer", function() {
  let env = {};

  beforeEach(async function() {
    env.mjpegServer = new MJPEGServer(config.stubs);
    await env.mjpegServer.listen();

    const port = env.mjpegServer.address().port;

    const cameras = Object.keys(config.stubs ?? {});
    const cam = cameras[0];

    env.streamer = env.mjpegServer.streamers[cam];

    env.consumerOpts = {
      src: `http://localhost:${port}` + config.stubs[cam]?.path || "",
      auth: config.stubs[cam]?.auth
    };

    env.consumer = undefined;
  });

  afterEach(function() {
    env.mjpegServer?.close();
    if (env.timer) {
      clearInterval(env.timer);
      env.timer = null;
    }

    env?.consumer?.disconnect();

    env = {};
  });

  it("can parse frames using 'content-length' or the 'boundary delimiter'",
    async function() {
      const boundary = env.streamer.boundary;
      let received = "";

      env.consumer = new MJPEGConsumer(env.consumerOpts);
      env.consumer.on("frame", (frame) => { received = frame?.toString(); });

      const subscription = waitForEvent(env.streamer, "subscribed");
      env.consumer.connect();
      await subscription;

      // using Content-Length
      let prom = waitFor(() => received.includes("FIRST FRAME"));
      env.streamer.sendRaw(Buffer.from(
        `--${boundary}\r\n` +
        "Content-Length: 11\r\n" +
        "\r\n" +
        "FIRST FRAME\r\n"));
      await prom;

      // using boundary delimiter
      received = "";
      prom = waitFor(() => received.includes("SECOND FRAME"));
      env.streamer.sendRaw(Buffer.from(
        `--${boundary}\r\n` +
        "\r\n" +
        "SECOND FRAME\r\n"));

      env.streamer.sendRaw(Buffer.from(`--${boundary}`));
      await prom;
    });

  it("can parse frame using a custom header delimiter", async function() {
    const boundary = env.streamer.boundary;
    const headerDelimiter = env.consumerOpts.headerDelimiter = "\n\n";
    let received = "";

    env.consumer = new MJPEGConsumer(env.consumerOpts);
    env.consumer.on("frame", (frame) => { received = frame?.toString(); });

    const subscription = waitForEvent(env.streamer, "subscribed");
    env.consumer.connect();
    await subscription;

    const prom = waitFor(() => received.includes("FRAME"));
    env.streamer.sendRaw(Buffer.from(`--${boundary}\r\nContent-Length: 5`));
    env.streamer.sendRaw(Buffer.from(headerDelimiter));
    env.streamer.sendRaw(Buffer.from("FRAME\r\n"));
    await prom;
  });

  it("can parse frame using JPEG magic bytes with an invalid header delimiter",
    async function() {
      const boundary = env.streamer.boundary;
      env.consumerOpts.jpegMagicBytes = true;
      env.consumerOpts.headerDelimiter = "\n\r\n";
      const headerDelimiter = "\n\n";
      const JFIFexample = "FFD8FFE000104A46494600010100000100";
      let received = "";

      env.consumer = new MJPEGConsumer(env.consumerOpts);
      env.consumer.on("frame", (frame) => { received = frame?.toString(); });

      const subscription = waitForEvent(env.streamer, "subscribed");
      env.consumer.connect();
      await subscription;

      const prom = waitFor(
        () => received.includes(`${Buffer.from(JFIFexample, "hex")}`));
      env.streamer.sendRaw(Buffer.from(`--${boundary}\r\n`));
      env.streamer.sendRaw(Buffer.from(headerDelimiter));
      env.streamer.sendRaw(Buffer.from(JFIFexample, "hex"));
      env.streamer.sendRaw(Buffer.from(`\r\n--${boundary}\r\n`));
      await prom;
    });

  it("can parse frame using JPEG magic bytes with as fallback solution",
    async function() {
      const boundary = env.streamer.boundary;
      const headerDelimiter = "\n\n"; // delimiter different from '\r\n\r\n'
      const JFIFexample = "FFD8FFE000104A46494600010100000100";
      let received = "";

      env.consumer = new MJPEGConsumer(env.consumerOpts);
      env.consumer.on("frame", (frame) => { received = frame?.toString(); });

      const subscription = waitForEvent(env.streamer, "subscribed");
      env.consumer.connect();
      await subscription;

      const prom = waitFor(
        () => received.includes(`${Buffer.from(JFIFexample, "hex")}`));
      env.streamer.sendRaw(Buffer.from(`--${boundary}\r\n`));
      env.streamer.sendRaw(Buffer.from(headerDelimiter));
      env.streamer.sendRaw(Buffer.from(JFIFexample, "hex"));
      env.streamer.sendRaw(Buffer.from("\r\n"));

      // this fallback solution is triggered when the MAX HEADER LENGTH is reached
      // so let us send a big fragment
      env.streamer.sendRaw(Buffer.from(`--${boundary}\r\n`));
      env.streamer.sendRaw(Buffer.allocUnsafe(MJPEGConsumer.MAX_HEADER_LENGTH));

      await prom;
    });

  it("can parse chunks containing serveral frames", async function() {
    const boundary = env.streamer.boundary;
    const received = [];

    env.consumer = new MJPEGConsumer(env.consumerOpts);
    env.consumer.on("frame", (frame) => { received.push(frame?.toString()); });

    const subscription = waitForEvent(env.streamer, "subscribed");
    env.consumer.connect();
    await subscription;

    // using Content-Length
    const prom =
      waitForValue(() => received, [ "FIRST FRAME", "SECOND FRAME" ]);

    env.streamer.sendRaw(Buffer.from(
      `--${boundary}\r\n` +
      "Content-Length: 11\r\n" +
      "\r\n" +
      "FIRST FRAME\r\n" +
      `--${boundary}\r\n` +
      "Content-Length: 12\r\n" +
      "\r\n" +
      "SECOND FRAME\r\n" +
      `--${boundary}\r\n`));

    await prom;
  });

  it("can detect the end of the stream", async function() {
    const subscription = waitForEvent(env.streamer, "subscribed");

    env.consumer = new MJPEGConsumer(env.consumerOpts);
    env.consumer.connect();
    await subscription;

    const endStream = waitForEvent(env.consumer, "ended");
    env.streamer.stop();
    await endStream;
  });

  it("avoids indefinite buffering when parsing using boundary delimiter",
    async function() {
      const boundary = env.streamer.boundary;
      const content = Buffer.allocUnsafe(MJPEGConsumer.MAX_CONTENT_LENGTH + 1);
      let received = "";

      env.consumer = new MJPEGConsumer(env.consumerOpts);
      env.consumer.on("frame", (frame) => { received = frame?.toString(); });

      const subscription = waitForEvent(env.streamer, "subscribed");
      env.consumer.connect();
      await subscription;

      const prom = waitFor(() => received.includes("FRAME"));
      // send frames without content-length
      env.streamer.sendFrame(content, false);
      env.streamer.sendFrame("FRAME", false);
      env.streamer.sendRaw(Buffer.from(`--${boundary}`));
      await prom;
    });

  it("can handle stream with invalid Content-Type field", async function() {
    // set an invalid boundary parameter in Content-Type (quotes needed)
    // (see https://datatracker.ietf.org/doc/html/rfc2046#section-5.1.1)
    const boundary = env.streamer.boundary = "gc0pJq0M:08jU534c0p";

    let received = "";

    env.consumer = new MJPEGConsumer(env.consumerOpts);
    env.consumer.on("frame", (frame) => { received = frame?.toString(); });

    const subscription = waitForEvent(env.streamer, "subscribed");
    env.consumer.connect();
    await subscription;

    const prom = waitFor(() => received.includes("FRAME"));
    env.streamer.sendRaw(Buffer.from(
      `--${boundary}\r\n` +
      "\r\n" +
      "FRAME\r\n"));

    env.streamer.sendRaw(Buffer.from(`--${boundary}`));
    await prom;

  });

  it("raises an error if the boundary is missing", async function() {
    const boundary = env.streamer.boundary = "";

    let errMsg = "";

    env.consumer = new MJPEGConsumer(env.consumerOpts);
    env.consumer.on("error", (err) => { errMsg = err.message; });

    const subscription = waitForEvent(env.streamer, "subscribed");
    env.consumer.connect();
    await subscription;

    const prom = waitFor(() => errMsg.includes("invalid"));
    env.streamer.sendRaw(Buffer.from(
      `--${boundary}\r\n` +
      "\r\n" +
      "FRAME\r\n"));

    env.streamer.sendRaw(Buffer.from(`--${boundary}`));
    await prom;
  });

  it("closes connection upon disconnection", async function() {
    const subscription = waitForEvent(env.streamer, "subscribed");

    env.consumer = new MJPEGConsumer(env.consumerOpts);
    env.consumer.connect();
    await subscription;

    const req = env.consumer._req;

    expect(isNil(req)).to.be.equal(false);

    const closeEvent = waitForEvent(req, "close");
    env.consumer.disconnect();
    await closeEvent;

    expect(req.socket.destroyed).to.be.equal(true);
  });

  it("raises an error if the response content-type is not " +
     "'multipart/x-mixed-replace'",
  async function() {

    env.consumer = new MJPEGConsumer({
      src: `http://localhost:${env.mjpegServer.address().port}/plain-text`
    });

    const prom = waitForEvent(env.consumer, "error");

    env.consumer.connect();

    const err = await prom;

    expect(err).to.be.an("error");
    expect(err.message).to.be.equal("No MJPEG stream detected");
  });
});

[{"file":"test/test_MJPEGConsumer.js","line":55,"character":32,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":55,"character":55,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":88,"character":30,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":88,"character":53,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":111,"character":32,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":111,"character":55,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":134,"character":32,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":134,"character":55,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":160,"character":30,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":160,"character":56,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":203,"character":32,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":203,"character":55,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":225,"character":30,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":225,"character":53,"text":"frame","kind":1},{"file":"test/test_MJPEGConsumer.js","line":248,"character":30,"text":"err","kind":1},{"file":"test/test_MJPEGConsumer.js","line":248,"character":49,"text":"err","kind":1},{"file":"test/test_MJPEGConsumer.js","line":248,"character":53,"text":"message","kind":1},{"file":"test/test_MJPEGConsumer.js","line":294,"character":10,"text":"err","kind":1},{"file":"test/test_MJPEGConsumer.js","line":296,"character":11,"text":"err","kind":1},{"file":"test/test_MJPEGConsumer.js","line":297,"character":11,"text":"err","kind":1},{"file":"test/test_MJPEGConsumer.js","line":297,"character":15,"text":"message","kind":1}]

TypeScript coverage report