async-deco

async-deco

async-deco is a collection of decorators. It allows to add features such as timeout, retry, dedupe, limit and much more! They can be combined together using the "compose" function (included).

Here is the list of the decorators (available for callback/promise functions):

• Log
• Memoize
• Cache
• Proxy
• Fallback
• Fallback value
• Fallback cache
• Timeout
• Retry
• Limit
• Dedupe
• parallel
• waterfall
• race
• balance
• debounce
• throttle

Callback and promises

All decorators are designed to work with functions using a callback or returning a promise. In case of callbacks, it must follow the node convention: the callback should be the last argument and its arguments should be, an error instance and the output of the function.

Every decorator is available in two different flavours:

• callback based:

var logDecorator = require('async-deco/callback/log');

This should be applied to functions with the node callback convention:

var decoratedFunction = logDecorator(logger)(function (a, b, c, next) {
  ...
  next(undefined, result); // or next(error);
});

• and promise based:

var logDecorator = require('async-deco/promise/log');

This should be used for function returning promises:

var decoratedFunction = logDecorator(logger)(function (a, b, c) {
  return new Promise(function (resolve, reject) {
    ...
    resolve(result); // or reject(error);    
  });
});

Then you can run the decorated function.

Logging

All decorators uses a "logging context" added using the decorator returned by "addLogger" (it works the same for callback/promises):

var addLogger = require('async-deco/utils/add-logger');

var logger = addLogger(function (event, payload, ts) {
  // log here
});

This methods is called with the following arguments:

• evt: the name of the event
• payload: an object with additional information about this event
• ts: the time stamp for this event (in ms)

The resulting decorator can wrap a function:

var defaultLogger = require('async-deco/utils/default-logger');

var f = logger(function () {
  var log = defaultLogger.call(this);
  log('event-name', { ... data ... })
});

The defaultLogger function extracts the logger function from the context. This is a very simple case but this pattern is really useful to share the log function between decorators:

var addLogger = require('async-deco/utils/add-logger');
var logDecorator = require('async-deco/callback/log');
var timeoutDecorator = require('async-deco/callback/timeout');
var retryDecorator = require('async-deco/callback/retry');

var decorator = compose(
  addLogger(function (evt, payload, ts) {
    console.log(ts, evt, payload);
  }),
  logDecorator(),
  retryDecorator(2, undefined, Error),
  timeoutDecorator(20)
);

var f = decorator(... func ...);

Then when you execute the function "f":

f(...args..., function (err, res) {
  ...
});

You can get something similar to:

1459770371655, "start", undefined
1459770371675, "timeout", { ms: 20 }
1459770371675, "retry", { times: 1 }
1459770371695, "timeout", { ms: 20 }
1459770371700, "end", { result: ... }

To make this work, the addLogger decorator extends the context (this) with a new method __log. The context attributes and methods are still available through the prototype chain. For this reason inspecting "this" using Object.keys and using this.hasOwnProperty('prop') can return an unexpected result.

Requiring the library

You can either:

var memoizeDecorator = require('async-deco/callback/memoize');

or

var memoizeDecorator = require('async-deco').callback.memoize;

or

var callbackDecorators = require('async-deco');
var memoizeDecorator = callbackDecorators.callback.memoize;

I strongly advice to use the first method, especially when using browserify. It allows to import only the functions you are actually using.

Decorators

The examples are related to the callback version. Just import the promise version in case of decorating promise based functions.

AddLogger

It enables the logging for the whole chain of decorators (if combined together).

var addLogger = require('async-deco/utils/add-logger');

var logger = addLogger(function (event, payload, ts) {
  // log here
});
var f = logger(function () { /* decorated function */ });

The logger takes these arguments:

• evt: the name of the event
• payload: an object with additional information about this event
• ts: the timestamp for this event

You can use this decorator multiple times to add multiple loggers.

Log

It logs when a function start, end and fail.

var logDecorator = require('async-deco/callback/log');

var addLogs = logDecorator();
var myfunc = addLogs(function (..., cb) { .... });

Memoize

It caches the result. At any subsequent calls it will return the cached result.

var memoizeDecorator = require('async-deco/callback/memoize');

var simpleMemoize = memoizeDecorator(getKey);
var myfunc = simpleMemoize(function (..., cb) { .... });

It takes 1 argument:

• a getKey function [optional]: it runs against the original arguments and returns the key used for the caching. If it is missing, only one result will be memoized.

It logs "memoize-hit" with {key: cache key, result: cache result}

Cache

It is a more sophisticated version of the memoize decorator. It can be used for caching in a db/file etc using memoize-cache (https://github.com/sithmel/memoize-cache).

var cacheDecorator = require('async-deco/callback/cache');

var cached = cacheDecorator(cache);
var myfunc = cached(function (..., cb) { .... });

It takes 2 arguments:

• a cache object [mandatory]. The interface should be compatible with memoize-cache (https://github.com/sithmel/memoize-cache)
• an "options" object [optional]:

It logs:

• "cache-hit" with {key: cache key, result: cache result}
• "cache-error" (when the cache fails) with {cacheErr: error object from the cache}
• "cache-miss" (when the item is not in the cache) with {key: cache key}
• "cache-set" with {args: arguments for caching, res: result to cache}

Proxy

It executes a "guard" function before the original one. If it returns an error it will use this error as the return value of the original function. It is useful if you want to run a function only if it passes some condition (access control).

var proxyDecorator = require('async-deco/callback/proxy');

var proxy = cacheDecorator(function (..., cb) {
  // calls cb(errorInstance) if the access is denied
  // calls cb() if I can procede calling the function
});
var myfunc = proxy(function (..., cb) { .... });

It takes 1 argument:

• a guard function [mandatory]. It takes the same arguments of the original function. If it returns an error (using the callback) the original function won't be called.

It logs "proxy-denied" with { err: error returned by the guard function}

Fallback

If a function fails, calls another one

var fallbackDecorator = require('async-deco/callback/fallback');

var fallback = fallbackDecorator(function (a, b, c, func) {
  func(undefined, 'giving up');
}, Error);
var myfunc = fallback(function (..., cb) { .... });

It takes 2 arguments:

• fallback function [mandatory]. It takes the same arguments of the original function (and a callback, even in the promise case).
• error instance for deciding to fallback, or a function taking error and result (if it returns true it'll trigger the fallback) [optional, it falls back on any error by default]

It logs "fallback" with {actualResult: {err: error returned, res: result returned}}

Fallback value

If a function fails, returns a value

var fallbackValueDecorator = require('async-deco/callback/fallback-value');

var fallback = fallbackValueDecorator('giving up', Error);
var myfunc = fallback(function (..., cb) { .... });

It takes 2 arguments:

• fallback value [mandatory]
• error instance for deciding to fallback, or a function taking error and result (if it returns true it'll trigger the fallback) [optional, it falls back on any error by default]

It logs "fallback" with {actualResult: {err: error returned, res: result returned}}

Fallback cache

If a function fails, it tries to use a previous cached result.

var fallbackCacheDecorator = require('async-deco/callback/fallback-cache');

var fallback = fallbackCacheDecorator(cache, options);
var myfunc = fallback(function (..., cb) { .... });

It takes 2 arguments:

• a cache object [mandatory]. The interface should be compatible with memoize-cache (https://github.com/sithmel/memoize-cache)
• an options object with this optional attributes:
  • error: the error instance for deciding to fallback, or a function taking the error and result (if it returns true it'll trigger the fallback) [optional, it falls back on any error by default]
  • useStale: if true it will use "stale" cache items as valid [optional, defaults to false]
  • noPush: it true it won't put anything in the cache [optional, defaults to false]

It logs:

• "fallback-cache-hit" with {key: cache key, result: cache object, actualResult: {err: error returned, res: result returned}}
• "fallback-cache-error" with {err: error returned by the function, cacheErr: error returned by the cache}
• "fallback-cache-miss" with {key: cache key, actualResult: {err: error returned, res: result returned}}
• "fallback-cache-set" with {args: arguments for caching, res: result to cache}

Timeout

If a function takes to much, returns a timeout exception.

var timeoutDecorator = require('async-deco/callback/timeout');

var timeout20 = timeoutDecorator(20);
var myfunc = timeout20(function (..., cb) { .... });

This will wait 20 ms before returning a TimeoutError. It takes 1 argument:

• time in ms [mandatory]

It logs "timeout" with { ms: ms passed since the last execution}

Retry

If a function fails, it retry running it again

var retryDecorator = require('async-deco/callback/retry');

var retryTenTimes = retryDecorator(10, 0, Error);
var myfunc = retryTenTimes(function (..., cb) { .... });

You can initialise the decorator with 2 arguments:

• number of retries [optional, it defaults to Infinity]
• interval for trying again (number of a function based on the number of times) [optional, it defaults to 0]
• error instance for deciding to retry, or function taking error and result (if it returns true it'll trigger the retry) [optional, it falls back on any error by default]

It logs "retry" with {times: number of attempts, actualResult: {err: original error, res: original result}}

Limit

Limit the concurrency of a function. Every function call that excedees the limit will be queued. If the queue size is reached the function call will return an error.

var limitDecorator = require('async-deco/callback/limit');

var limitToTwo = limitDecorator(2, getKey);
var myfunc = limitToTwo(function (..., cb) { .... });

You can initialise the decorator with 1 argument:

• number of parallel execution [default to 1]. It can also be an object: {max: number, queueSize: number}.
  • "limit" will be the number of parallel execution
  • "queueSize" is the size of the queue (default to Infinity). If the queue reaches this size any further function call will return an error without calling the original function
• a getKey function [optional]: it runs against the original arguments and returns the key used for creating different queues of execution. If it is missing there will be only one execution queue. If it returns null or undefined, the limit will be ignored.
• a getPriority function [optional]: it runs against the original arguments and returns a number that represent the priority of this function when queued (less == prioritary).

It logs "limit-queue" when a function gets queued or "limit-drop" when a function gets rejected (queue full). It'll also log these data: { queueSize: number of function queued, key: cache key, parallel: number of functions currently running }

Dedupe

It executes the original function, while is waiting for the output it doesn't call the function anymore but instead it collects the callbacks. After getting the result, it dispatches the same to all callbacks. It may use the "getKey" function to group the callbacks into queues.

var dedupeDecorator = require('async-deco/callback/dedupe');

var dedupe = dedupeDecorator(getKey);
var myfunc = dedupe(function (..., cb) { .... });

The argument:

• getKey function [optional]: it runs against the original arguments and returns the key used for creating different queues of execution. If it is missing there will be only one execution queue.

It logs "dedupe-queue" when a function is queued waiting for the result from another function. {key: cache key}

Parallel

"parallel" executes every function in parallel. If a function returns an error the execution stops immediatly returning the error. The functions will get the same arguments and the result will be an array of all the results.

var parallel = require('async-deco/callback/parallel');

var func = parallel([
  function (x, cb) {
    cb(null, x + 1);
  },
  function (x, cb) {
    cb(null, x + 2);
  }
]);

func(3, function (err, values) {
  // values contains  [4, 5]
});

Waterfall

"waterfall" executes the functions in series. The first function will get the arguments and the others will use the arguments passed by the previous one:

var waterfall = require('async-deco/callback/waterfall');

var func = waterfall([
  function (x, cb) {
    cb(null, x + ' world');
  },
  function (x, cb) {
    cb(null, x + '!');
  }
]);

func('hello', function (err, value) {
  // value === 'hello world!'
});

Race

"race" will execute all functions in parallel but it will return the first valid result.

var race = require('async-deco/callback/race');

var func = race([
  function (x, cb) {
    setTimeout(function () {
      cb(null, x + 1);
    }, 20)
  },
  function (x, cb) {
    setTimeout(function () {
      cb(null, x + 2);
    }, 10)
  }
]);

func(3, function (err, values) {
  // values contains  5 (fastest)
});

Parallel - Waterfall - Race

It is very easy to combine these functions to create a more complex flow:

var func = waterfall([
  parallel([
    function (x, cb) {
      cb(null, x * 2);
    },
    function (x, cb) {
      cb(null, x * 3);
    }
  ]),
  function (numbers, cb) {
    cb(null, numbers.reduce(function (acc, item) {
      return acc + item;
    }, 0));
  },
  function (x, cb) {
    cb(null, x - 5);
  }
]);

func(5, function (err, value) {
  // value === 20;
});

Although these functions are also available for promise, I suggest to use the native promise API, unless you have a better reason for doing differently.

• parallel: Promise.all
• race: Promise.race
• waterfall: just chain different promises

Balance

This decorator allows to distribute the load between a group of functions. The functions should take the same arguments.

var balance = require('async-deco/callback/balance');

var balanceDecorator = balance();

var func = balanceDecorator([...list of functions]);
func(...args, function (err, res) {
  // ...
});

You can initialise the decorator with different policies:

var balance = require('async-deco/callback/balance');
var balancePolicies = require('async-deco/utils/balance-policies');

var balanceDecorator = balance(balancePolicies.roundRobin);
...

There are 3 policies available in the "balance-policies" package:

• roundRobin: it rotates the execution between the functions
• random: it picks up a random function
• idlest (default): it tracks the load of each function and use the idlest

You can also define your own policy:

var balance = require('async-deco/callback/balance');

var balanceDecorator = balance(function (counter, loads, args) {
  // "counter" is the number of times I have called the function
  // "loads" is an array with length equal to the number of functions
  //         it contains how many calls are currently running for that function
  // "args" is an array containing the arguments of the current function call
  // this function should return the index of the function I want to run
});
...

Debounce

This decorator is a pretty sophisticated version of debounce. In a few words, when a debounced function is called many times within a time interval, it gets executed only once. It uses the same options of lodash debounce (that is used internally), but also allows to have multiple "debounce" contexts. The decorators takes these arguments:

• wait (mandatory): it is the time interval to debounce
• debounceOpts (optional): the debounce options used by lodash debounce:
  • leading: Specify invoking on the leading edge of the timeout. (default false)
  • trailing: Specify invoking on the trailing edge of the timeout. (default true)
  • maxWait: The maximum time the decorated function is allowed to be delayed before it’s invoked
• getKey (optional): it runs against the original arguments and returns the key used for creating different "debounce" context. If is undefined there will be a single debouncing context. If it returns null there won't be debouncing. Two functions in different contexts aren't influenced each other and are executed independently.
• cacheOpts (optional): the contexts are cached, in this object you can define a maxLen (maximum number of context) and a defaultTTL (contexts last only for this amount of ms).

Example:

var debounce = require('async-deco/callback/debounce');

var debounceDecorator = debounce(1000, { maxWait: 500 }, function (key) { return key; }, { maxLen: 100 });

var func = debounceDecorator(function (key, cb) {
  // this is the function I want to debounce
});

func('r', function (err, res) {
  // the callback is not guaranteed to be called
  // for every execution, being debounced.
});

Throttle

This decorator is a pretty sophisticated version of throttle. In a few words, a throttled function can be called only a certain amount of times within a time interval. It uses the same options of lodash throttle (that is used internally), but also allows to have multiple "throttle" contexts. The decorators takes these arguments:

• wait (mandatory): it is the time interval to throttle
• throttleOpts (optional): the throttle options used by lodash throttle:
  • leading: Specify invoking on the leading edge of the timeout. (default true)
  • trailing: Specify invoking on the trailing edge of the timeout. (default true)
• getKey (optional): it runs against the original arguments and returns the key used for creating different "throttle" context. If is undefined there will be a single debouncing context. If it returns null there won't be throttling. Two functions in different contexts aren't influenced each other and are executed independently.
• cacheOpts (optional): the contexts are cached, in this object you can define a maxLen (maximum number of context) and a defaultTTL (contexts last only for this amount of ms).

Example:

var throttle = require('async-deco/callback/throttle');

var throttleDecorator = throttle(1000, { maxWait: 500 }, function (key) { return key; }, { maxLen: 100 });

var func = throttleDecorator(function (key, cb) {
  // this is the function I want to throttle
});

func('r', function (err, res) {
  // the callback is not guaranteed to be called
  // for every execution, being throttled.
});

A note about throttle/debounce. A function that uses these decorators is not guaranteed to be executed every time is called. And the same is true for their callback and promise (yes, there is a promise version!). If you don't need to return a promise, I advice to use the simple callback version.

Utilities

Callbackify

Convert a synchronous/promise based function to a plain callback.

var callbackify = require('async-deco/utils/callbackify');

var func = callbackify(function (a, b){
  return a + b;
});
func(4, 6, function (err, result){
  ... // result === 10 here
})

sanitizeAsyncFunction

This special decorator tries to take care of some nasty common cases when you work with "callback based" asynchronous functions.

• if a function calls the callback more than once, the second time it will throw an exception instead
• if a function throws an exception, it will instead call the callback with the error
• if the callback itself throws an exception, it propagates the exception to the calling function

var sanitizeAsyncFunction = require('async-deco/utils/sanitizeAsyncFunction');

var func = sanitizeAsyncFunction(function () {
  throw new Error('generic error');
});

func(function (err, out) {
  // err will be the error
});

and

var func = sanitizeAsyncFunction(function () {
  cb(null, 'hello');
  cb(null, 'hello');
});

func(function (err, out) {
  // this will throw an exception instead of being called the second time
});

Promisify

Convert a callback based function to a function returning a promise. It is a reference to es6-promisify, exposed here for convenience.

var promisify = require('async-deco/utils/promisify');

var func = promisify(function (a, b, next){
  return next(undefined, a + b);
});
func(4, 6).then(function (result){
  ... // result === 10 here
})

You can also use it in an environment where standard "Promise" object is not supported. Just use a polyfill like (https://www.npmjs.com/package/es6-promise).

var Promise = require('es6-promise').Promise;

(function (global) {
global.  Promise = Promise;
}(this));

Compose

It can combine more than one decorators. You can pass either an array or using multiple arguments. "undefined" items are ignored.

var compose = require('async-deco/utils/compose');

var decorator = compose(
  retryDecorator(10, Error, logger),
  timeoutDecorator(20, logger));

var newfunc = decorator(function (..., cb) { .... });

Timeout after 20 ms and then retry 10 times before giving up. You should consider the last function is the one happen first! The order in which you compose the decorator changes the way it works, so plan it carefully! I suggest to:

• put the log decorator as first
• put the fallback decorators before the "timeout" and "retry"
• put the "retry" before the "timeout"
• put "limit" and/or "proxy" as the last one
• put "dedupe", "memoize" or "cache" as last, just before limit/proxy

Decorate

This is a shortcut that allows to compose and decorate in a single instruction:

var decorate = require('async-deco/utils/decorate');

var newfunc = decorate(
  retryDecorator(10, Error, logger),
  timeoutDecorator(20, logger),
  function (..., cb) { .... });

The function to decorate has to be the last argument.

Examples and use cases

Smart memoize/cache

Using memoize(or cache) on an asynchronous function has a conceptual flaw. Let's say for example I have a function with 100ms latency. I call this function every 10 ms:

executed            ⬇⬇⬇⬇⬇⬇⬇⬇⬇⬇          
          ------------------------------
requested ⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆

What happen is that while I am still waiting for the first result (to cache) I regularly execute other 9 functions. What if I compose memoize with dedupe?

var decorator = compose(dedupeDecorator(), memoizeDecorator());

var newfunc = decorator(function (..., cb) { .... });

dedupe should fill the gap:

executed            ⬇
          ------------------------------
requested ⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆⬆

Reliable function

Imagine a case in which you want to be sure you did everything to get a result, and in case is not possible you want to return a sane fallback:

var decorator = compose(
  fallbackDecorator(getFallbackResult), // last resort fallback
  fallbackCacheDecorator(cache),        // try to use a previous cached output
  retryDecorator(3),                    // it retry 3 times
  timeoutDecorator(5000));              // it times out after 5 seconds

var newfunc = decorator(function (..., cb) { .... });

Queue

In some case you may want to preserve the sequence used to call a function. For example, sending commands to a db being sure they are executed in the right order.

var limitDecorator = require('async-deco/callback/limit');

var queue = limitDecorator(1);
var myfunc = queue(function (..., cb) { .... });