RxJS fromStream Adapter

Buffer based fromStream

export default function fromStream<T = Uint8Array>(
  stream: Readable,
  finishEventName = "end",
  dataEventName = "data"
) {
  stream.pause();
  const headerSize = 24;
  const recordSize = 52;
  let gotHeader: boolean = false;
  let backStream: Buffer = Buffer.from([]);

  return new Observable<Buffer>(observer => {
    function dataHandler(data: Uint8Array) {
      let size = gotHeader ? recordSize : headerSize;
      backStream = Buffer.concat([backStream, data]);
      while (backStream.length >= size) {
        observer.next(backStream.subarray(0, size));
        backStream = backStream.subarray(size);
        if (!gotHeader) {
          gotHeader = true;
          size = recordSize;
        }
      }
    }

    function errorHandler(err: Error) {
      observer.error(err);
    }

    function endHandler() {
      observer.complete();
    }

    stream.addListener(dataEventName, dataHandler);
    stream.addListener("error", errorHandler);
    stream.addListener(finishEventName, endHandler);

    stream.resume();

    return () => {
      stream.removeListener(dataEventName, dataHandler);
      stream.removeListener("error", errorHandler);
      stream.removeListener(finishEventName, endHandler);
    };
  }).pipe(share());
}

byte by byte fromStream.ts

import { Observable } from "rxjs";
import { share } from "rxjs/operators";
import { Readable } from "stream";

// Borrowed this excellent function from https://stackoverflow.com/a/42174935
// This function fromStream() should have been provided by default in RxJS library,
// but for some reason its not included.

export default function fromStream<T = Uint8Array>(
  stream: Readable,
  finishEventName = "end",
  dataEventName = "data"
) {
  stream.pause();

  return new Observable<number>(observer => {
    function dataHandler(data: Uint8Array) {
      // make sure we send data one byte at a time so its easier to handle buffering with byte counts
      data.forEach(byte => observer.next(byte));
    }

    function errorHandler(err: Error) {
      observer.error(err);
    }

    function endHandler() {
      observer.complete();
    }

    stream.addListener(dataEventName, dataHandler);
    stream.addListener("error", errorHandler);
    stream.addListener(finishEventName, endHandler);

    stream.resume();

    return () => {
      stream.removeListener(dataEventName, dataHandler);
      stream.removeListener("error", errorHandler);
      stream.removeListener(finishEventName, endHandler);
    };
  }).pipe(share());
}

Elegant buffer based `toRecordStream`

import { Observable } from "rxjs";
import { share } from "rxjs/operators";
import { Readable } from "stream";
import { Buffer } from "buffer";

// Borrowed this excellent function from https://stackoverflow.com/a/42174935
// This function fromStream() should have been provided by default in RxJS library,
// but for some reason its not included.
export function fromStream<T = Uint8Array>(
  stream: Readable,
  finishEventName = "end",
  dataEventName = "data"
) {
  stream.pause();

  return new Observable<Uint8Array>(observer => {
    function dataHandler(data: Uint8Array) {
      observer.next(data);
    }

    function errorHandler(err: Error) {
      observer.error(err);
    }

    function endHandler() {
      observer.complete();
    }

    stream.addListener(dataEventName, dataHandler);
    stream.addListener("error", errorHandler);
    stream.addListener(finishEventName, endHandler);

    stream.resume();

    return () => {
      stream.removeListener(dataEventName, dataHandler);
      stream.removeListener("error", errorHandler);
      stream.removeListener(finishEventName, endHandler);
    };
  }).pipe(share());
}

export const toRecordStream = (
  stream: Observable<Uint8Array>,
  recordSize: number,
  headerSize?: number
) => {
  let backStream: Buffer = Buffer.of();
  let gotHeader = headerSize == null ? true : false;

  return new Observable<Buffer>(observer => {
    stream.subscribe(data => {
      let size = gotHeader || headerSize == null ? recordSize : headerSize;
      backStream = Buffer.concat([backStream, data]);
      while (backStream.length >= size) {
        observer.next(backStream.subarray(0, size));
        backStream = backStream.subarray(size);
        if (!gotHeader) {
          gotHeader = true;
          size = recordSize;
        }
      }
    });
  }).pipe(share());
};