switchMap with array of Observables

Question

I'm trying Rxjs 5 to simplify some Node.js nested callbacks. I need to read a directory (fs.readdir), then read stats of each file (fs.stats) and parse them if they were modified since last sync.

The following code works but I find it a bit odd and not "the rxjs way" because of the first switchMap which is too big!

const fs = require('fs');
const path = require('path');
const { Observable } = require('rxjs');

const lastSync = new Date(2017, 01, 01);
const pathToFolder = '/any/path/';

Observable.bindNodeCallback(fs.readdir)(pathToFolder)
  .switchMap((files) => {
    const array = files.map((fileName) => {
      return Observable.zip(
        Observable.of(fileName),
        Observable.bindNodeCallback(fs.stat)(path.join(pathToFolder, fileName))
      );
    });
    return Observable.concat(...array);
  })
  .filter(([fileName, stats]) => stats.mtime.getTime() > lastSync.getTime())
  .subscribe(([fileName, stats]) => parseFile(fileName));

function parseFile(fileName) { /* ... */ }

How can I improve it?

Answer 1

I'd suggest that the switchMap operator be replaced with mergeMap. Only a single array of files (or an error) is going to be emitted from the bound callback, so it will never be necessary to switch; a merge is all that's required.

And the Observable.zip call could be simplified. You could use a map operator to pair the file name and the stats.

Observable.bindNodeCallback(fs.readdir)(pathToFolder)
  .mergeMap((files) => {
    const array = files.map((fileName) => Observable
      .bindNodeCallback(fs.stat)(path.join(pathToFolder, fileName))
      .map((stats) => [fileName, stats])
    );
    return Observable.concat(...array);
  })
  .filter(([fileName, stats]) => stats.mtime.getTime() > lastSync.getTime())
  .subscribe(([fileName, stats]) => parseFile(fileName));

It's possible to further simplify the composed observable by taking advantage of the fact that the operators in the concact and merge families also support arrays. So, concatAll can be used to flatten the array, emitting the file names from the readdir callback. And those file names can then be mapped to stats with a concatMap operator:

Observable.bindNodeCallback(fs.readdir)(pathToFolder)
  .concatAll()
  .concatMap((fileName) => Observable
    .bindNodeCallback(fs.stat)(path.join(pathToFolder, fileName))
    .map((stats) => [fileName, stats])
  )
  .filter(([fileName, stats]) => stats.mtime.getTime() > lastSync.getTime())
  .subscribe(([fileName, stats]) => parseFile(fileName));