Three variations on async

observer_async_options.js

// based on https://github.com/deanius/async-gen-vs-events/blob/3ad02c4af7ab7c56178e4614bb1585220ccec8db/walk_dir_tree_with_events.js
// three different algorithms on async, with a fake file system that's easy to modify

const { agent } = require("rx-helper");

// Timer (useful for understanding how output gets processed)
const start = new Date()
const log = (whatevz) => console.log(new Date() - start, whatevz)

// Fake slow file system:
const pType = Symbol.for("type")
const pDur  = Symbol.for("duration")
const fs    = {
  file: (dur)           => ({ [pType]: "file", [pDur]: dur }),
  dir:  (dur, contents) => ({ [pType]: "dir",  [pDur]: dur, ...contents }),
  ls:   (path) => fs.find(path).then(dir => Object.keys(dir)),
  stat: (path) => fs.find(path).then(node => ({ type: node[pType] })),
  get:  (path) => path.split("/").reduce((cwd, name) => cwd[name], fs.root),
  find: (path) => new Promise(resolve => {
    const node = fs.get(path)
    setTimeout(resolve, node[pDur], node)
  }),
}

// Two file systems to choose from:

// fs.root = fs.dir(1000, {
//   'a': fs.dir(1000, {
//     'b1': fs.dir(500, { 'c1': fs.file(500), 'c2': fs.file(500), 'c3': fs.file(500) }),
//     'b2': fs.dir(500, { 'c1': fs.file(500), 'c2': fs.file(500), 'c3': fs.file(500) }),
//     'b3': fs.dir(500, { 'c1': fs.file(500), 'c2': fs.file(500), 'c3': fs.file(500) }),
//   })
// })

fs.root = fs.dir(100, {
  'a': fs.dir(100, {
    'b1': fs.dir(150, { 'c1': fs.file(200), 'c2': fs.file(300), 'c3': fs.file(500) }),
    'b2': fs.dir(250, { 'c1': fs.file(100), 'c2': fs.file(400), 'c3': fs.file(100) }),
    'b3': fs.dir( 10, { 'c1': fs.file(350), 'c2': fs.file(200), 'c3': fs.file( 10) }),
  })
})



// // ALGORITHM 1:
// // Processes each entry asynchronously
// agent.on("dir", ({ action: { payload: dir } }) => {
//   fs.ls(dir).then(listings => {
//     for (const name of listings) {
//       const path = `${dir}/${name}`
//       fs.stat(path).then(({ type }) => agent.process({ type, payload: path }))
//     }
//   })
// });


// ALGORITHM 2:
// Processes each entry asynchronously, but then syncs them back up.
// It takes the same amount of time as the above one,
// but releases them in order (the order it traverses the subdirectories is
// dependent on how long it takes to ls them, though)
agent.on("dir", async ({ action: { payload: dir } }) => {
  const names    = await fs.ls(dir)
  const paths    = names.map(name => `${dir}/${name}`)
  const promises = paths.map(path => fs.stat(path).then(stat => [path, stat.type]))
  for (const [path, type] of await Promise.all(promises))
    agent.process({ type: type, payload: path })
})


// // ALGORITHM 3:
// // processes each entry synchronously
// agent.on("dir", ({ action: { payload: dir } }) => {
//   fs.ls(dir).then(async listings => {
//     for (const name of listings) {
//       const path = `${dir}/${name}`
//       const { type } = await fs.stat(path)
//       agent.process({ type, payload: path })
//     }
//   })
// });


// our subscriber prints it
agent.filter(/file|dir/, ({ type, action: { payload } }) => log({ payload }));

// Kick off the starting node
agent.process({ type: "dir", payload: "a" });