bach

Compose your async functions with elegance.

Usage

With bach, it is very easy to compose async functions to run in series or parallel.

1var bach = require('bach');
2
3function fn1(cb) {
4  cb(null, 1);
5}
6
7function fn2(cb) {
8  cb(null, 2);
9}
10
11function fn3(cb) {
12  cb(null, 3);
13}
14
15var seriesFn = bach.series(fn1, fn2, fn3);
16// fn1, fn2, and fn3 will be run in series
17seriesFn(function (err, res) {
18  if (err) {
19    // in this example, err is undefined
20    // handle error
21  }
22  // handle results
23  // in this example, res is [1, 2, 3]
24});
25
26var parallelFn = bach.parallel(fn1, fn2, fn3);
27// fn1, fn2, and fn3 will be run in parallel
28parallelFn(function (err, res) {
29  if (err) {
30    // in this example, err is undefined
31    // handle error
32  }
33  // handle results
34  // in this example, res is [1, 2, 3]
35});

Since the composer functions return a function, you can combine them.

1var combinedFn = bach.series(fn1, bach.parallel(fn2, fn3));
2// fn1 will be executed before fn2 and fn3 are run in parallel
3combinedFn(function (err, res) {
4  if (err) {
5    // in this example, err is undefined
6    // handle error
7  }
8  // handle results
9  // in this example, res is [1, [2, 3]]
10});

Functions are called with async-done, so you can return a stream, promise, observable or child process. See async-done completion and error resolution for more detail.

1// streams
2var fs = require('fs');
3
4function streamFn1() {
5  return fs
6    .createReadStream('./example')
7    .pipe(fs.createWriteStream('./example'));
8}
9
10function streamFn2() {
11  return fs
12    .createReadStream('./example')
13    .pipe(fs.createWriteStream('./example'));
14}
15
16var parallelStreams = bach.parallel(streamFn1, streamFn2);
17parallelStreams(function (err) {
18  if (err) {
19    // in this example, err is undefined
20    // handle error
21  }
22  // all streams have emitted an 'end' or 'close' event
23});

1// promises
2function promiseFn1() {
3  return Promise.resolve(1);
4}
5
6function promiseFn2() {
7  return Promise.resolve(2);
8}
9
10var parallelPromises = bach.parallel(promiseFn1, promiseFn2);
11parallelPromises(function (err, res) {
12  if (err) {
13    // in this example, err is undefined
14    // handle error
15  }
16  // handle results
17  // in this example, res is [1, 2]
18});

All errors are caught in a domain and passed to the final callback as the first argument.

1function success(cb) {
2  setTimeout(function () {
3    cb(null, 1);
4  }, 500);
5}
6
7function error() {
8  throw new Error('Thrown Error');
9}
10
11var errorThrownFn = bach.parallel(error, success);
12errorThrownFn(function (err, res) {
13  if (err) {
14    // handle error
15    // in this example, err is an error caught by the domain
16  }
17  // handle results
18  // in this example, res is [undefined]
19});

When an error happens in a parallel composition, the callback will be called as soon as the error happens. If you want to continue on error and wait until all functions have finished before calling the callback, use settleSeries or settleParallel.

1function success(cb) {
2  setTimeout(function () {
3    cb(null, 1);
4  }, 500);
5}
6
7function error(cb) {
8  cb(new Error('Async Error'));
9}
10
11var parallelSettlingFn = bach.settleParallel(success, error);
12parallelSettlingFn(function (err, res) {
13  // all functions have finished executing
14  if (err) {
15    // handle error
16    // in this example, err is an error passed to the callback
17  }
18  // handle results
19  // in this example, res is [1]
20});

API

series(fns..., [options])

Takes a variable amount of functions (fns) to be called in series when the returned function is called. Optionally, takes an options object as the last argument.

Returns an invoker(cb) function to be called to start the serial execution. The invoker function takes a callback (cb) with the function(error, results) signature.

If all functions complete successfully, the callback function will be called with all results as the second argument.

If an error occurs, execution will stop and the error will be passed to the callback function as the first parameter. The error parameter will always be a single error.

parallel(fns..., [options])

Takes a variable amount of functions (fns) to be called in parallel when the returned function is called. Optionally, takes an options object as the last argument.

Returns an invoker(cb) function to be called to start the parallel execution. The invoker function takes a callback (cb) with the function(error, results) signature.

If all functions complete successfully, the callback function will be called with all results as the second argument.

If an error occurs, the callback function will be called with the error as the first parameter. Any async functions that have not completed, will still complete, but their results will not be available. The error parameter will always be a single error.

settleSeries(fns..., [options])

Takes a variable amount of functions (fns) to be called in series when the returned function is called. Optionally, takes an options object as the last argument.

Returns an invoker(cb) function to be called to start the serial execution. The invoker function takes a callback (cb) with the function(error, results) signature.

All functions will always be called and the callback will receive all settled errors and results. If any errors occur, the error parameter will be an array of errors.

settleParallel(fns..., [options])

Takes a variable amount of functions (fns) to be called in parallel when the returned function is called. Optionally, takes an options object as the last argument.

Returns an invoker(cb) function to be called to start the parallel execution. The invoker function takes a callback (cb) with the function(error, results) signature.

All functions will always be called and the callback will receive all settled errors and results. If any errors occur, the error parameter will be an array of errors.

options

The options object is primarily used for specifying functions that give insight into the lifecycle of each function call. The possible extension points are create, before, after and error. If an extension point is not specified, it defaults to a no-op function.

The options object for parallel and settleParallel also allows specifying concurrency in which to run your functions. By default, your functions will run at maximum concurrency.

options.concurrency

Limits the amount of functions allowed to run at a given time.

options.create(fn, index)

Called at the very beginning of each function call with the function (fn) being executed and the index from the array/arguments. If create returns a value (storage), it is passed to the before, after and error extension points.

If a value is not returned, an empty object is used as storage for each other extension point.

This is useful for tracking information across an iteration.

options.before(storage)

Called immediately before each function call with the storage value returned from the create extension point.

options.after(result, storage)

Called immediately after each function call with the result of the function and the storage value returned from the create extension point.

options.error(error, storage)

Called immediately after a failed function call with the error of the function and the storage value returned from the create extension point.

License

MIT