async.applyEachSeries converted to highland.js

Question

I have this snippet working fine using async.applyEachSeries.

var async = require("async");

function firstThing(state, next) {
  state.firstThingDone = true;
  setImmediate(next);
}

function secondThing(state, next) {
  state.secondThingDone = true;
  setImmediate(next);
}

var state = {};
async.applyEachSeries([
  firstThing,
  secondThing
], state, function (error) {
  console.log(error, state);
});

I have made several attempts to convert it to highland.js but I'm not grokking the plumbing there. I'm pretty sure I need to do _.wrapCallback(firstThing) for both firstThing and secondThing but not sure whether I need _.pipeline or .series() or what.

Answer 1

There isn't a nice 1:1 translation at the moment, since Highland lacks an 'applyEach', but by changing (or wrapping) the firstThing and lastThing functions you might get a nice enough result:

var _ = require('highland');

/**
 * These functions now return the state object,
 * and are wrapped with _.wrapCallback
 */

var firstThing = _.wrapCallback(function (state, next) {
  state.firstThingDone = true;
  setImmediate(function () {
    next(null, state);
  });
});

var secondThing = _.wrapCallback(function (state, next) {
  state.secondThingDone = true;
  setImmediate(function () {
    next(null, state);
  });
});

var state = {};

_([state])
  .flatMap(firstThing)
  .flatMap(secondThing)
  .apply(function (state) {
    // updated state
  });

Answer 2

In my own attempts to abandon async for Highland, I've come to use .map(_.wrapCallback).invoke('call') with some regularity. We can use it here:

var _ = require('highland');

var state = {};

function firstThing(state, next) {
  state.firstThingDone = true;
  setImmediate(next);
}

function secondThing(state, next) {
  state.secondThingDone = true;
  setImmediate(next);
}

_([firstThing, secondThing])
  .map(_.wrapCallback)
  .invoke('call', [null, state])
  .series().toArray(function() {
    console.log(state)
  });

This lets us keep your functions as they were, it expands nicely to more than two 'things', and I think it feels more like a direct breakdown of applyEachSeries.

If we need to bind this or pass different arguments to each function, we can just use .bind when constructing the stream and omit the call arguments:

_([firstThing.bind(firstThing, state),
  secondThing.bind(secondThing, state)])
  .map(_.wrapCallback)
  .invoke('call')
  .series().toArray(function() {
    console.log(state)
  });

On the other hand, this approach is more side-effecty; the stuff in our stream is no longer what we're ultimately transforming and making use of.

Finally, having to use .toArray at the end feels odd, although maybe that's just an argument for adding .done(cb) to Highland.