concatStreams.ts

import internal, { PassThrough, Transform } from 'stream';

export function concatStreams({
  streams,
  delimiter,
}: {
  delimiter?: string;
  streams: (() => internal.Readable)[];
}) {
  const destination = new PassThrough({ objectMode: true });
  if (!Array.isArray(streams) || !streams.length) {
    destination.end();
    console.log('No streams to concat');
    return destination;
  }
    
  const next = () => {
    if (!streams.length) {
      destination.end();
      return;
    }

    if (destination.destroyed || destination.closed) return;

    const stream = streams.shift()!();

    if (delimiter) {
      const delimiterTransform = new Transform({
        transform(chunk, encoding, callback) {
          this.push(chunk);
          callback();
        },
        flush(callback) {
          if (streams.length) {
            this.push(delimiter);
            callback();
          }
        },
      });
      stream.pipe(delimiterTransform).pipe(destination, { end: false });
    } else {
      stream.pipe(destination, { end: false });
    }

    stream.on('end', next);
    stream.on('error', (error) => destination.emit('error', error));
    destination.once('close', () => {
      if (!stream.destroyed) {
        stream.destroy();
      }
    });
  };

  next();

  return destination;
}