metarhia/metasync

Asynchronous Programming Library for JavaScript & Node.js

metasync's Issues

Proposal: rebranding and functionality change

@tshemsedinov, what do you think about turning MetaSync (under some new name) into a full-blown generic library of utility functions for JavaScript with asynchronous utilities being only a part of it? We definitely need some sort of common library since I've encountered situations when two analogous functions have to be fixed twice in Impress and JSTP.

Refs: metarhia/Metarhia#2

Change DataCollector interface

// expect 4 data keys
var dc1 = new metasync.DataCollector(4);

// expect 4 data keys within 5 second timeout
var dc2 = new metasync.DataCollector(4, 5000);

dc1.on('error', function(err, key) { ... });
dc2.on('timeout', function(err, data) { ... });
dc2.on('done', function(errs, data) {
  // errs - hash of errors
  // data - hash of successfully received data
});

FYI: @aqrln, @belochub, @DzyubSpirit

Implement .map(items, callback, done)

Use case:

metasync.map(['Beijing', 'Roma', 'Kiev'], (item, callback) => {
  getCityPopulation(item, (count) => {
    callback(count);
  });
}, (err, arr) => {
  let total = arr.reduce((prev, cur) => (prev + cur), 0);
  console.log('Total population: ' + total);
});

Collect same key twice

In 88#discussion_r115333601 @aqrln noticed that collector now increments counter even on same keys. On the one hand it's lightweight but on the other it may lead to errors in applied code. So I'd prefer to have both implementation:

• metasync.collect(3)
• metasync.collect(3).distinct()

Callbacks in metasync.filter and metasync.find

Now callbacks in metasync.filter and metasync.find have single argument (result or data). I think it will be good to use common convention with (err, data). FYI: @aqrln

clearTimeout wrong usage

https://github.com/metarhia/MetaSync/blob/master/metasync.js#L141

I think should be simply:
clearTimeout(this.timer);

Rename metasync.composition to flow

Actually metasync.composition not returning composed function so this name is not the best choice. I'd prefer to rename it to something related with flow control or function commutation. For ecample:

metasync.flow([...])(data, (err, data) => {});

FYI @aqrln @nechaido @belochub

Implement Throttle

var t1 = metasync.throttle(5000, func);
t1(); t1(); t1(); // single call
setTimeout(t1, 7000); // another call

Add callback to fetch to resume chain after fetching

Callback will be optional

metasync
.for(data)
.map(fn1)
.fetch((err, data, [cb]) => cb(err, data)) // call cb to resume chain
.map(fn2)
.fetch((err, data) => console.log(data));

Rewrite duplicate code for collectors

KeyCollector's constructor and collect method have pretty same code with DataCollector. I think it's good to abstract it

Tests for metasync.queue

• Performance tests
• Unittests
• Cases and docs

Callback wrapper

Here is an idea of small but useful callback wrapper.
After applying metasync.cb:

• no need to check like this: if (callback) return callback(err, data);
• wrapper prevents to call it more then once
• it may optionally add debugging wraper to trace async code (not for production)
  Usage:

const doSomethingAsync = (par1, parN, callback) => {
  const cb = metasync.cb(callback); // apply wrapper
  // ... some code here ...
  // No need to check: if (cb) return ....
  cb(err, data); // first call will be passed to callback
  return cb(err, data); // second will write warning to console or log
};

Rewrite metasync.find to iterate array in parallel

Implement .reduce(items, callback, done, initial)

Use case:

metasync.reduce(['Beijing', 'Roma', 'Kiev'], (prev, cur, callback) => {
  getCityPopulation(cur, (count) => {
    callback(prev + count);
  });
}, (err, count) => {
  console.log('Total population: ' + count);
}, 0);

Implement KeyCollector

var keyCollector = new metasync.KeyCollector(
  ['user', 'config', 'readme', 'timer'],
  function(data) {
    console.dir(Object.keys(data));
    console.log('Collector test done');
    end('KeyCollector');
  }
);

keyCollector.collect('user', { name: 'Marcus Aurelius' });

fs.readFile('HISTORY.md', function(err, data) {
  keyCollector.collect('history', data);
});

fs.readFile('README.md', function(err, data) {
  keyCollector.collect('readme', data);
});

setTimeout(function() {
  keyCollector.collect('timer', { date: new Date() });
}, ASYNC_TIMEOUT);

Implement metasync.ConcurrentQueue

var queue = new metasync.ConcurrentQueue(concurrency, timeout);
queue.add({item});
queue.on('empty', fn);
queue.on('process', fn(item, callback));
queue.pause();
queue.resume();
queue.stop();

Chaining implementation comparison

Here we have 5 branches:

• With promises in master
• Without promises but working just with async callbacks in chain-alternative
• Good for sync and async using process.nextTick to fire chain in chain-alternative-nexttick
• Using .fire() to fire chain in chain-alternative-fire
• Using functor: chain-alternative-functor

I'm afraid that promises will affect performance and they may be not so effectivу.
I think they also use something like nextTick to fire chain.
That's because I like manual .fire()
What do you think @aqrln ?

Use setImmediate in metasync.series

To prevent RangeError: Maximum call stack size exceeded and break recursion we need setImmediate in metasync.series

Add options

options: {
  expected: 4, // expected data pices for DataCollector
  timeout: 5000, // global timeout
  each: 1000, // timeout for each function
  error: { // TODO: invent better options structure
    finish: true // finish on error if true
  },
  on: {
    done: function(err, data) { // on finish
    },
    timeout: function(err, data) { // on timeout
    }
  }
}

Implement metasync.reduceRight()

Asynchronous implementation of Array.prototype.reduceRight()

Implement metasync.some()

Asynchronous implementation of Array.prototype.some()

Alternative chaining syntax for collectors

const dc1 = metasync
  .collect(3)
  .timeout(5000)
  .done((err, data) => {});

dc1(item);

const dc2 = metasync
  .collect(['key1', 'key2', 'key3'])
  .timeout(5000)
  .done((err, data) => {});

dc2(key, value);

WDYT, @aqrln about .timeout(number) and firing .catch on timeout ?

Impement asyncronous each in parallel

metasync.each(
  ['a', 'b', 'c'],
  function iterator(item) { callback(); },
  function done(data) { ... }
);

Delegate synchronous Array methods

• .concat()
• .includes()
• .reverse()
• .sort()
• .slice()
• .splice()
• .shift()
• .unshift()
• .push()
• .pop()

You can add more

Alternative chaining syntax for Queue

We have ConcurrentQueue in lib/queue.js but new chaining metasync.queue may implement more features from Queueing theory:

• Prioritized queueing
• Ordering: FIFO/LIFO
• Process throttling
• Balancing
• Timeouts
• Statistics and optimization

const cq = metasync.queue(3) // concurrency
  .wait(2000) // task may wait in queue not more then 2 sec
  .timeout(5000) // task processing timeout 5 sec
  .throttle(100, 1000) // throttle processing to 100 rps
  .process((item, cb) => cb(err, result)) // task processing function
  .success((item) => {}) // on success
  .failure((item) => {}) // on fail
  .drain(() => {}); // on drain

@aqrln @belochub @nechaido @DzyubSpirit maybe you need more in JSTP or can add something for GlobalStorage or Impress use cases?

Alternative chaining syntax for..each

As we mentioned in #17 it would be great to have better syntax, chaining:

metasync.for(items).each(fn).fetch(fn);

metasync.for(items).reduce(fn).fetch(fn);

metasync.for(items).filter(fn).map(fn).then(fn).reduce(fn).fetch(fn);

Storing error in collectors

Both DataCollector and KeyCollector initialize field errs as empty arrays.
https://github.com/metarhia/MetaSync/blob/master/lib/collectors.js#L38
https://github.com/metarhia/MetaSync/blob/master/lib/collectors.js#L85
Also they use length field like errs is array.
https://github.com/metarhia/MetaSync/blob/master/lib/collectors.js#L69
https://github.com/metarhia/MetaSync/blob/master/lib/collectors.js#L115
However adding errors is like errs is object.
https://github.com/metarhia/MetaSync/blob/master/lib/collectors.js#L62
https://github.com/metarhia/MetaSync/blob/master/lib/collectors.js#L108

I think a right way is use errs as object everywhere

Chaining optimization

Maybe it is better to have consistent function signature in chaining.
Now in following example fn1 and fn2 have different signature then fn3 and fn4:

api.metasync.for(array).filter(fn1).map(fn2).then(fn3).catch(fn4);

• Signature for fn1 and fn2 is: (element, callback) => callback(error, result)
• Signature for fn3 is: (data) => {}
• Signature for fn4 is: (error) => {}
  But we can combine .then() and .catch() into .done()

api.metasync.for(array).filter(fn1).map(fn2).done(fn3);

Signature for fn1 and fn2 will be: (element, callback) => callback(err, result)
But here is 2 alternative syntax for f3:

• (data, callback) => callback(error, data) where result may be instance of Error or data
• (error, data, callback) => callback(error, data)
  We need callback in .done() because chaining may contain multiple steps after .done()
  For example:

api.metasync.for(array).filter(fn1).map(fn2).done(fn3).reduce(fn4).done(fn5);

@aqrln what do you think?

Implement metasync.every()

Asynchronous implementation of Array.prototype.every()

Impement asyncronous series

metasync.series(
  ['a', 'b', 'c'],
  function iterator(item) { callback(); },
  function done(data) { ... }
);

metasync.map with empty array

metasync.map doesn't call done function when is called with empty array. I think it's good to call done immediately instead of not calling at all.

Refactor testing and add more tests

Impement asyncronous functional composition

metasync.composition(
  [f1,f2,f3,[[f4,f5,[f6,f7],f8]],f9], // functional composition syntax
  function done(data) { ... } // done
);

Refactor to ES6

Refactor and remove old Node.js support.
MetaSync will support just Node.js 6.x and 7.x

Prevent repeated done callback

Callback done must not be called more then once per async primitive: each, series, ...

Implement asynchronous filter

Function metasync.filter(items, iterator, done); is an asynchronous analogue for synchronous Array.filter

Dynamically build ArrayChain.prototype

['then', 'catch', 'map', 'filter', 'reduce', 'each', 'series', 'find'] on library load

Broken test

Tests for metasync.filter and metasync.find are broken due to missing err argument in callback.

Failing tests should exit loudly

Now tests silently failing, #77

Rename find to every and create Array compatible find

metasync.every - find all that satisfies passed function
metasync.find - find first that satisfies passed function

Do not use Array.forEach

In .map() and .filter() we use .forEach but we can't break on error, so better to iterate with loops.

Implement asynchronous find

Function metasync.find(items, iterator, done); is an asynchronous analogue for synchronous Array.find

Broken metasync.find

We have totally broken metasync.find, I tested it writing chaining test

Move metasync.cb to common.cb

This wrapper needed even inside common so to avoid recursive dependency is will be good to move it from metasync.cb to common.cb. Mark metasync.cb as deprecated and support it in [email protected]. We will remove support in [email protected].

Implement metasync.timeout

metasync.timeout(1000, asyncFunction, doneFunction)
FYI: @aqrln @belochub

Rewrite sequential putting next as callback of prev

Having .fetch() we need no .execute()

Now .fetch can start chain ececution

Avoid sort in metasync.filter

Now it use push and sort. It is better to allocate new Array(len) and remove filtered. But tests needed.

Implement metasync.firstOf

metasync.firstOf([f1...fN], (err, args...) => {
...
});

FYI: @aqrln @belochub

Handle errors

• callback(err, data)
• stop everything if function returns error (optional)