I have the following code:

import { Duplex } from "stream";

export class JsonDuplex extends Duplex {

    private _timeout: NodeJS.Timeout;
    private _timeMS: number;

    constructor(timeMS?: number) {
        super({
            objectMode: true,
            highWaterMark: 50,
        });
        this._timeMS= timeMS || 5000;
        this._timeout = this.newTimeout();
    }

    _read(size: number): void {

    }

    _write(chunk: any, encoding: BufferEncoding, callback: (error?: Error | null) => void): void {
        if (this.readable) {
            try {
                this.push(JSON.stringify(chunk));
                callback();
                this._timeout = this.newTimeout();
            } catch (e: any) {
                callback(e);
            }
        }
    }

    newTimeout(): NodeJS.Timeout {
        clearTimeout(this._timeout);
        return setTimeout(() => {
            this.emit('end');
        }, this._timeMS);
    }
}

It transforms any object (_write method) to json, just to pipe to response:

stream.pipe(new JsonDuplex()).pipe(response);

It's parsing the data, the only problem is that the response doens't know when to end(), so the browser keeps loading forever, event if all the data has been flushed. I have put a workaround on the class with the timeout, it re-init every _write and if does complete, emits 'end' event.

There's another way to accomplish this?