rxjs file stream

get-file-stream.js

// @flow
import fs from 'fs';
import { Observable } from 'rxjs';

type Options = {
  highWaterMark: number,
  encoding: string,
}

const defaultOptions: Options = {
  // number of bits to read at a time
  highWaterMark: 1,
  encoding: 'utf8',
};

export default (path: string, options?: Options = defaultOptions) =>
  Observable.create((observer) => {
    const file$ = fs.createReadStream(path, options);

    file$.on('data', (chunk) => observer.next(chunk));
    file$.on('end', () => observer.complete());
    file$.on('close', () => observer.complete());
    file$.on('error', (error) => observer.error(error));

    // there seems to be no way to actually close the stream
    return () => file$.pause();
  });

Nice! If you create the read stream using a file descriptor instead of a path, you can call fs.close(fd). This closes the opened file descriptor and in turn closes the read stream (which will emit a close event).