8 implementations of the same pipeline in JavaScript

flow-4.png

vanilla.js

/*
Task template:

const task6 = function (value, callback) {
  console.log(6, value)
  setImmediate(() => callback(null, value+'.6'))
}
*/

const process = (value, callback) => {
  let done3 = false
  let done4 = false
  let done5 = false
  let value3
  let value4
  let value5
  let alreadyCallbacked = false

  const callbackErr = (err) {
    if (! alreadyCallbacked) {
      alreadyCallbacked = true
      callback(err)
    }
  }

  task1(value, (err, value1) => {
    if (err) return callbackErr(err)
    task2(value1, (err, value2) => {
      if (err) return callbackErr(err)
      task3(value2, (err, _value3) => {
        if (err) return callbackErr(err)
        done3 = true
        value3 = _value3
        if (done3 && done4 && done5) {
          task6([[value3, value4], value5], callback)
        }
      })
      task4(value2, (err, _value4) => {
        if (err) return callbackErr(err)
        done4 = true
        value4 = _value4
        if (done3 && done4 && done5) {
          task6([[value3, value4], value5], callback)
        }
      })
    })
    task5(value1, (err, _value5) => {
      if (err) return callbackErr(err)
      done5 = true
      value5 = _value5
      if (done3 && done4 && done5) {
        task6([[value3, value4], value5], callback)
      }
    })
  })
}

process('foo', (err, value) => console.log('done', value))

vanilla_pointfree.js

/*
Task template:

const task6 = function (value, callback) {
  console.log(6, value)
  setImmediate(() => callback(null, value+'.6'))
}
*/

const once = (f) => {
  let ran = false
  return (...args) => {
    if (ran) return
    ran = true
    return f(...args)
  }
}

const parallel = (fs) =>
  (value, callback) => {
    const done = once(callback)
    const results = []
    let callCount = 0

    const join = (resultIdx) => (err, result) => {
      results[resultIdx] = result
      callCount += 1
      if (err) {
        done(err)
      } else if (callCount === fs.length) {
        done(null, results)
      }
    }

    fs.forEach((f, idx) => f(value, join(idx)))
  }

const compose = (f, g) =>
  (value, callback) =>
    f(value, (err, result) =>
      err ? callback(err) : g(result, callback))

const sequence = (fs) => fs.reduce(compose)

const group34 = parallel([ task3, task4 ])
const group234 = sequence([ task2, group34 ])
const group2345 = parallel([ group234, task5 ])
const process = sequence([ task1, group2345, task6 ])

process('foo', (err, value) => console.log('done', value))

with_async_auto.js

/*
Task template:

const task6 = function ({ value3, value4, value5 }, callback) {
  console.log(6, value3, value4, value5)
  setImmediate(() => callback(null, value3+value4+value5+'.6'))
}
*/

const async = require('async')

const process = (value, callback) =>
  async.auto({
    value1: async.apply(task1, value),
    value2: [ 'value1', task2 ],
    value5: [ 'value1', task5 ],
    value3: [ 'value2', task3 ],
    value4: [ 'value2', task4 ],
    value6: [ 'value3', 'value4', 'value5', task6 ],
  }, callback)

process('foo', (err, { value6 }) => console.log('done', value6))

with_async_await.js

/*
Task template:

const task6 = function (value) {
  console.log(6, value)
  return Promise.resolve(value+'.6')
}
*/

const group234 = async (value) => {
  const value2 = await task2(value)
  return Promise.all([ task3(value2), task4(value2) ])
}

const process = async (value) => {
  const value1 = await task1(value)
  const value2345 = await Promise.all([ group234(value1), task5(value1) ])
  return await task6(value2345)
}

const main = async () => {
  const result = await process('foo')
  console.log('done', result)
}
main()

with_async_parallel_waterfall.js

/*
Task template:

const task6 = function (value, callback) {
  console.log(6, value)
  setImmediate(() => callback(null, value+'.6'))
}
*/

const async = require('async')

const group34 = (value, callback) =>
  async.parallel([ async.apply(task3, value), async.apply(task4, value) ], callback)

const group234 = (value, callback) =>
  async.waterfall([ async.apply(task2, value), group34 ], callback)

const group2345 = (value, callback) =>
  async.parallel([ async.apply(group234, value), async.apply(task5, value) ], callback)

const process = (value, callback) =>
  async.waterfall([ async.apply(task1, value), group2345, task6 ], callback)
  
process('foo', (err, value) => console.log('done', value))

with_async_parallel_waterfall_pointfree.js

/*
Task template:

const task6 = function (value, callback) {
  console.log(6, value)
  setImmediate(() => callback(null, value+'.6'))
}
*/

const async = require('async')

const parallel = (f, g) =>
  (value, callback) =>
    async.parallel([ async.apply(f, value), async.apply(g, value) ], callback)

const group34 = parallel(task3, task4)
const group234 = async.seq(task2, group34)
const group2345 = parallel(group234, task5)
const process = async.seq(task1, group2345, task6)
  
process('foo', (err, value) => console.log('done', value))

with_promises.js

/*
Task template:

const task6 = function (value) {
  console.log(6, value)
  return Promise.resolve(value+'.6')
}
*/

const group34 = (value) =>
  Promise.all([ task3(value), task4(value) ])

const group234 = (value) =>
  task2(value).then(group34)

const group2345 = (value) =>
  Promise.all([ group234(value), task5(value) ])

const process = (value) =>
  task1(value).then(group2345).then(task6)

process('foo').then((value) => console.log('done', value))

with_promises_pointfree.js

/*
Task template:

const task6 = function (value) {
  console.log(6, value)
  return Promise.resolve(value+'.6')
}
*/

const parallel = (f, g) =>
  (value) => Promise.all([ f(value), g(value) ])

const sequence = (f, g) =>
  (value) => f(value).then(g)

const group34 = parallel(task3, task4)
const group234 = sequence(task2, group34)
const group2345 = parallel(group234, task5)
const process = sequence(sequence(task1, group2345), task6)

process('foo').then((value) => console.log('done', value))