Line by line file reading with RxJS on Node.js

line-by-line-rxjs-node.js

var Rx = require('rx');
var readline = require('readline');
var fs = require('fs');

var rl = readline.createInterface({
  input: fs.createReadStream('lines.txt')
});

var lines = Rx.Observable.fromEvent(rl, 'line')
  .takeUntil(Rx.Observable.fromEvent(rl, 'close'))
  .subscribe(
    console.log,
    err => console.log("Error: %s", err),
    () => console.log("Completed"));

readline's Interface implements AsyncIterable<string> to iterate over all lines, and RxJS supports from(AsyncIterable<T>) since version 7. Therefore, this is now a simple one liner:

import readline from "readline"
import { from } from 'rxjs'

const input = fs.createReadStream('lines.txt')
const lines = from(readline.createInterface({input}))
lines.subscribe({
  next: console.log,
  error: console.error,
  complete: () => console.log("DONE!")
})