MultiPart.js

Filename	Percent	Threshold	Total	Covered	Uncovered
test/MultiPart.js	99.21%	80%	252	250	2

// @ts-check
// SPDX-License-Identifier: Zlib
// SPDX-FileCopyrightText: 2024 CERN (home.cern)
// SPDX-Created: 2024-02-02T15:09:02
// SPDX-FileContributor: Author: Sylvain Fargier <sylvain.fargier@cern.ch>

import { EventEmitter } from "node:events";
import VarBuffer from "../src/consumers/VarBuffer.js";
import { makeDeferred } from "@cern/nodash";
import { Readable } from "node:stream";
import d from "debug";

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

const SearchStates = { HEADER_END: 1, CONTENT_END: 2, NEXT_BOUNDARY: 3 };
const CRLFx2 = "\r\n\r\n";

/**
 * @typedef {import('axios').AxiosResponse} Response
 */

/**
 * @details emitted buffer is not copied, it must be either duplicated or
 * used in the callback.
 */
class MultiPart extends EventEmitter {
  #buffer = new VarBuffer();

  #state = SearchStates.HEADER_END;

  MaxHeaderLength = 1024;

  #offset = 0;

  #frameEnd = 0;

  #boundary;

  #contentType = "";

  #deferred = makeDeferred();

  get promise() { return this.#deferred.promise; }

  /**
   * @param {Response} res
   */
  constructor(res) {
    super();

    this.res = res;
    if (!(res?.data instanceof Readable)) {
      throw new Error("MultiPart must be used on a `stream` request from axios");
    }

    this.#boundary = this.#getBoundary(res?.headers?.["content-type"] ?? "") ??
      "--";
    debug("part parser constructed");

    const addfunc = this.#add.bind(this);
    const readable = res.data;
    readable
    .on("data", addfunc)
    .once("error", (err) => {
      if (this.#deferred.isPending) {
        this.#deferred.reject(err);
      }
      debug("part reader error");
    })
    .once("close", () => {
      /* nothing is emitted after close anyway */
      readable.removeAllListeners();
      if (this.#deferred.isPending) {
        this.#deferred.resolve(null);
      }
      debug("part reader closed");
    });
  }

  /**
   *
   * @param {string} contentType
   */
  #getBoundary(contentType) {
    const match = contentType
    .match(/boundary="?([\w\s'()+,\-./:=?]{1,70})"?;?/);

    if (!match) { return undefined; } // boundary missing or invalid

    return match[1].trimEnd();
  }

  /**
   * @param {Buffer} data
   */
  // eslint-disable-next-line complexity
  #add(data) {
    debug("adding data");
    this.#buffer.add(data);
    let index = 0;

    while (this.#buffer.length - this.#offset > this.#boundary.length + 2) {
      switch (this.#state) {
      case SearchStates.HEADER_END:
        index = this.#buffer.buffer().indexOf(CRLFx2, this.#offset);

        if (index >= 0) {
          const header = this.#buffer.slice(0, index).toString("ascii");
          const match = header.match(/Content-Length:\s*(\d+)/i);
          this.#offset = index + CRLFx2.length;

          if (match) {
            this.#contentType =
              (header.match(/Content-Type:(.*)/i)?.[1] ?? "").trim();
            this.#frameEnd = this.#offset + Number(match[1]);
            this.#state = SearchStates.CONTENT_END;
          }
          else {
            this.#state = SearchStates.NEXT_BOUNDARY;
          }
        }
        else if (this.#buffer.length > this.MaxHeaderLength) {
          // drop data
          this.#buffer.shift(this.#buffer.length - (CRLFx2.length - 1));
          this.#offset = 0;
        }
        else {
          this.#offset = this.#buffer.length;
        }
        break;
      case SearchStates.CONTENT_END:
        if (this.#buffer.length < this.#frameEnd) {
          return;
        }
        this.emit("part", this.#contentType,
          this.#buffer.slice(this.#offset, this.#frameEnd));

        this.#buffer.shift(this.#frameEnd);
        this.#offset = 0;
        this.#state = SearchStates.HEADER_END;
        break;
      case SearchStates.NEXT_BOUNDARY:
        /* skip until next boundary */
        index = this.#buffer.buffer().indexOf(this.#boundary, this.#offset);

        if (index > 0) {
          /* found */
          this.#buffer.shift(index + this.#boundary.length);
          this.#offset = 0;
          this.#state = SearchStates.HEADER_END;
        }
        else if (this.#buffer.length > this.MaxHeaderLength) {
          this.#buffer.shift(this.#buffer.length - (this.#boundary.length - 1));
          this.#offset = 0;
        }
        else {
          this.#offset = Math.max(0,
            this.#buffer.length - (this.#boundary.length - 1));
        }
        break;
      default:
        this.#state = SearchStates.HEADER_END;
        break;
      }
    }
  }
}

export default MultiPart;

[{"file":"test/MultiPart.js","line":50,"character":9,"text":"res","kind":1},{"file":"test/MultiPart.js","line":51,"character":15,"text":"data","kind":1}]

TypeScript coverage report