eventstream.ts

import {PassThrough, Stream} from "stream";
import {parseJSON} from "./index";

export class WriteEventStream {
    public stream: PassThrough;

    constructor() {
        this.stream = new PassThrough();
    }

    write(event: string, data: string) {
        if (!this.stream.closed) {
            this.stream.write(`event: ${event}\n`);
            this.stream.write(`data: ${JSON.stringify(data)}\n\n`);
        }
    }

    end(cb?: () => void) {
        this.stream?.end(cb);
    }
}

export class ReadEventStream {
    private readonly stream: Stream;

    constructor(stream: Stream) {
        this.stream = stream;
    }

    read(dataCB: ({event, data}: { event: string, data: string }) => void, doneCB: () => void) {
        let buffer = '';
        this.stream.on('data', data => {
            buffer += data.toString();
            let index = buffer.indexOf('\n\n');
            while (index !== -1) {
                const v = buffer.slice(0, index).trim();
                buffer = buffer.slice(index + 2);

                const lines = v.split('\n');
                const lineEvent = lines[0].replace('event: ', '');
                const lineData = lines[1].replace('data: ', '');
                dataCB({event: lineEvent, data: JSON.parse(lineData)});
                index = buffer.indexOf('\n\n');
            }
        });
        this.stream.on('close', doneCB);
    }
}