TypeScript coverage report

// @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;

​