Writable stream with options to cap in-memory store on internal buffer. Stores in array, then concats last.

memory-store.ts

import { WritableOptions, Writable } from "stream";

export interface MemoryStoreOptions extends WritableOptions {
    objectMode?: false | undefined;
    maxBufsize?: number;
}

export const DEFAULT_MAX_BUFSIZE = 1024 ** 2;

export class MemoryStore extends Writable {
    private _chunks: Buffer[] = [];
    private _buffer?: Buffer;
    private _bufsize: number = 0;
    private _maxBufsize: number = DEFAULT_MAX_BUFSIZE;

    get maxBufsize() {
        return this._maxBufsize;
    }

    get bufsize() {
        return this._bufsize;
    }

    get buffer() {
        if (this._buffer) return this._buffer;
        throw new Error("MemoryStore stream has not finished");
    }

    constructor(options?: MemoryStoreOptions) {
        super(options);
        const { maxBufsize, objectMode } = options || {};

        if (typeof maxBufsize == "number") {
            this._maxBufsize = maxBufsize;
        }

        if (objectMode)
            throw new Error("MemoryStore cannot operate in objectMode");
        if (this._maxBufsize <= 0)
            throw new Error("Invalid maxBufsize in MemoryStoreOptions");
    }

    _write(
        chunk: string | Buffer,
        encoding: BufferEncoding,
        callback: (err?: Error | null) => void
    ) {
        try {
            const buf =
                typeof chunk == "string"
                    ? Buffer.from(chunk, encoding)
                    : Buffer.from(chunk);

            this._chunks.push(buf);
            this._bufsize += buf.byteLength;

            if (this._bufsize > this._maxBufsize) {
                throw new Error("Maximum buffer length exceeded");
            }

            callback();
        } catch (err) {
            if (err instanceof Error) {
                callback(err);
            } else {
                callback(new Error("Could not write to MemoryStore"));
            }
        }
    }

    _final(callback: (err?: Error | null) => void) {
        try {
            this._buffer = Buffer.concat(this._chunks);
            this._chunks.length = 0;
            callback();
        } catch (err) {
            if (err instanceof Error) {
                callback(err);
            } else {
                callback(new Error("Could not finalize MemoryStore"));
            }
        }
    }
}