Readline transform stream

readline-transform-stream.ts

import * as readline from 'readline';
import * as stream from 'stream';

export async function* asyncIterableToGenerator<T>(iter: AsyncIterable<T>) {
  for await (const entry of iter) {
    yield entry;
  }
}

export class LineReaderStream extends stream.Duplex {
  asyncGen: AsyncGenerator<string, void, unknown>;
  readlineInstance: readline.Interface;
  passthrough: stream.Duplex;
  constructor(opts?: stream.DuplexOptions) {
    super({ readableObjectMode: true, ...opts });
    this.passthrough = new stream.PassThrough();
    this.readlineInstance = readline.createInterface({
      input: this.passthrough,
      crlfDelay: Infinity,
    });
    this.asyncGen = asyncIterableToGenerator(this.readlineInstance);
  }
  async _read(size: number) {
    for (let i = 0; i < size; i++) {
      const chunk = await this.asyncGen.next();
      if (!this.push(chunk.done ? null : chunk.value)) {
        break;
      }
    }
  }
  _write(chunk: any, encoding: string, callback: (error?: Error | null) => void) {
    this.passthrough.write(chunk, encoding, callback);
  }
  _destroy(error: any, callback: (error: Error | null) => void) {
    this.passthrough.destroy(error);
    this.readlineInstance.close();
    callback(error);
  }
  _final(callback: (error?: Error | null) => void) {
    this.passthrough.end(() => callback());
  }
}