Rewrite to promisify and bring back firstOf realization

Author: Yehor Khilchenko

lib/array.js

@@ -2,6 +2,7 @@
 
 const common = require('@metarhia/common');
 const { asyncIter } = require('./async-iterator.js');
+const { promisify } = require('util');
 
 // Asynchronous map (iterate parallel)
 //   items - <Array>, incoming
@@ -20,15 +21,7 @@ const map = (items, fn, done) => {
     return;
   }
   const isArray = Array.isArray(items);
-  const data = asyncIter(items).map(
-    item =>
-      new Promise((resolve, reject) => {
-        fn(item, (err, res) => {
-          if (err) reject(err);
-          else resolve(res);
-        });
-      })
-  );
+  const data = asyncIter(items).map(item => promisify(fn)(item));
   const promise = isArray ? data.toArray() : data.collectTo(items.constructor);
   promise.then(res => done(null, res), done);
 };
@@ -122,15 +115,7 @@ const filter = (items, fn, done) => {
     return;
   }
   const isArray = Array.isArray(items);
-  const data = asyncIter(items).filter(
-    item =>
-      new Promise((resolve, reject) => {
-        fn(item, (err, res) => {
-          if (err) reject(err);
-          else resolve(res);
-        });
-      })
-  );
+  const data = asyncIter(items).filter(item => promisify(fn)(item));
   const promise = isArray ? data.toArray() : data.collectTo(items.constructor);
   promise.then(res => done(null, res), err => done(err));
 };
@@ -155,22 +140,10 @@ const filter = (items, fn, done) => {
 const reduce = (items, fn, done, initial) => {
   done = done || common.emptyness;
   asyncIter(items)
-    .reduce(
-      (prev, cur) =>
-        new Promise((resolve, reject) => {
-          fn(prev, cur, (err, res) => {
-            if (err) reject(err);
-            else resolve(res);
-          });
-        }),
-      initial
-    )
+    .reduce((prev, cur) => promisify(fn)(prev, cur), initial)
     .then(r => done(null, r), done);
 };
 
-const REDUCE_RIGHT_EMPTY_ARR =
-  'Metasync: reduceRight of empty array with no initial value';
-
 // Asynchronous reduceRight
 //   items - <Array>, incoming
 //   fn - <Function>, to be executed for each value in array
@@ -190,41 +163,9 @@ const REDUCE_RIGHT_EMPTY_ARR =
 //       argument in first iteration
 const reduceRight = (items, fn, done, initial) => {
   done = done || common.emptyness;
-  const len = items.length || items.size;
-  items = [...items];
-  const hasInitial = typeof initial !== 'undefined';
-
-  if (!len && !hasInitial) {
-    done(new TypeError(REDUCE_RIGHT_EMPTY_ARR), initial);
-    return;
-  }
-
-  let previous = hasInitial ? initial : items[len - 1];
-  if ((!len && hasInitial) || (len === 1 && !hasInitial)) {
-    done(null, previous);
-    return;
-  }
-
-  let count = hasInitial ? len - 1 : len - 2;
-  let current = items[count];
-  const last = 0;
-
-  const next = (err, data) => {
-    if (err) {
-      done(err);
-      return;
-    }
-    if (count === last) {
-      done(null, data);
-      return;
-    }
-    count--;
-    previous = data;
-    current = items[count];
-    fn(previous, current, next, count, items);
-  };
-
-  fn(previous, current, next, count, items);
+  asyncIter(items)
+    .reduceRight((prev, cur) => promisify(fn)(prev, cur), initial)
+    .then(r => done(null, r), done);
 };
 
 // Asynchronous each (iterate in parallel)
@@ -239,17 +180,8 @@ const reduceRight = (items, fn, done, initial) => {
 const each = (items, fn, done) => {
   done = done || common.emptyness;
   asyncIter(items)
-    .parallel(
-      item =>
-        new Promise((resolve, reject) => {
-          fn(item, (err, res) => {
-            if (err) reject(err);
-            else resolve(res);
-          });
-        })
-    )
-    .then(r => done(null, r))
-    .catch(e => done(e));
+    .parallel(item => promisify(fn)(item))
+    .then(r => done(null, r), done);
 };
 
 // Asynchronous series
@@ -264,15 +196,7 @@ const each = (items, fn, done) => {
 const series = (items, fn, done) => {
   done = done || common.emptyness;
   asyncIter(items)
-    .each(
-      item =>
-        new Promise((resolve, reject) => {
-          fn(item, (err, res) => {
-            if (err) reject(err);
-            else resolve(res);
-          });
-        })
-    )
+    .each(item => promisify(fn)(item))
     .then(r => done(null, r), done);
 };
 
@@ -289,15 +213,7 @@ const series = (items, fn, done) => {
 const find = (items, fn, done) => {
   done = done || common.emptyness;
   asyncIter(items)
-    .find(
-      item =>
-        new Promise((resolve, reject) => {
-          fn(item, (err, res) => {
-            if (err) reject(err);
-            else resolve(res);
-          });
-        })
-    )
+    .find(item => promisify(fn)(item))
     .then(r => done(null, r))
     .catch(e => done(e));
 };
@@ -315,15 +231,7 @@ const find = (items, fn, done) => {
 const every = (items, fn, done) => {
   done = done || common.emptyness;
   asyncIter(items)
-    .every(
-      item =>
-        new Promise((resolve, reject) => {
-          fn(item, (err, res) => {
-            if (err) reject(err);
-            else resolve(res);
-          });
-        })
-    )
+    .every(item => promisify(fn)(item))
     .then(r => done(null, r))
     .catch(e => done(e));
 };
@@ -341,15 +249,7 @@ const every = (items, fn, done) => {
 const some = (items, fn, done) => {
   done = done || common.emptyness;
   asyncIter(items)
-    .some(
-      item =>
-        new Promise((resolve, reject) => {
-          fn(item, (err, res) => {
-            if (err) reject(err);
-            else resolve(res);
-          });
-        })
-    )
+    .some(item => promisify(fn)(item))
     .then(r => done(null, r))
     .catch(e => done(e));
 };

lib/control.js

@@ -1,13 +1,19 @@
 'use strict';
 
 const common = require('@metarhia/common');
+const { each } = require('./array.js');
 
 // Executes all asynchronous functions and pass first result to callback
 //   fns - <Function[]>, callback-last / err-first
 //   callback - <Function>, on done, err-first
 const firstOf = (fns, callback) => {
   const done = common.once(callback);
-  fns.forEach(fn => fn((...args) => done(...args)));
+  each(fns, (f, iterCb) =>
+    f((...args) => {
+      done(...args);
+      iterCb(...args);
+    })
+  );
 };
 
 // Parallel execution