Added asGen().

tests/test_gen.js

@@ -0,0 +1,204 @@
+'use strict';
+
+const unit = require('heya-unit');
+
+const Chain = require('../index');
+const {streamToArray, delay} = require('./helpers');
+
+const {fromIterable} = require('../utils/FromIterable');
+const gen = require('../utils/gen');
+const asGen = require('../utils/asGen');
+
+const {none, final, many} = Chain;
+
+unit.add(module, [
+  function test_gen(t) {
+    const async = t.startAsync('test_gen');
+
+    const output = [],
+      chain = new Chain([fromIterable([1, 2, 3]), gen(x => x * x, x => 2 * x + 1), streamToArray(output)]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [3, 9, 19])'));
+      async.done();
+    });
+  },
+  function test_genFinal(t) {
+    const async = t.startAsync('test_genFinal');
+
+    const output = [],
+      chain = new Chain([
+        fromIterable([1, 2, 3]),
+        gen(x => x * x, x => final(x), x => 2 * x + 1),
+        streamToArray(output)
+      ]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [1, 4, 9])'));
+      async.done();
+    });
+  },
+  function test_compNothing(t) {
+    const async = t.startAsync('test_compNothing');
+
+    const output = [],
+      chain = new Chain([fromIterable([1, 2, 3]), gen(x => x * x, () => none, x => 2 * x + 1), streamToArray(output)]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [])'));
+      async.done();
+    });
+  },
+  function test_genEmpty(t) {
+    const async = t.startAsync('test_genEmpty');
+
+    const output = [],
+      chain = new Chain([fromIterable([1, 2, 3]), x => x * x, gen(), streamToArray(output)]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [1, 4, 9])'));
+      async.done();
+    });
+  },
+  function test_genAsync(t) {
+    const async = t.startAsync('test_genAsync');
+
+    const output = [],
+      chain = new Chain([fromIterable([1, 2, 3]), gen(delay(x => x * x), x => 2 * x + 1), streamToArray(output)]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [3, 9, 19])'));
+      async.done();
+    });
+  },
+  function test_genGenerator(t) {
+    const async = t.startAsync('test_genGenerator');
+
+    const output = [],
+      chain = new Chain([
+        fromIterable([1, 2, 3]),
+        gen(
+          x => x * x,
+          function*(x) {
+            yield x;
+            yield x + 1;
+            yield x + 2;
+          },
+          x => 2 * x + 1
+        ),
+        streamToArray(output)
+      ]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [3, 5, 7, 9, 11, 13, 19, 21, 23])'));
+      async.done();
+    });
+  },
+  function test_genMany(t) {
+    const async = t.startAsync('test_genMany');
+
+    const output = [],
+      chain = new Chain([
+        fromIterable([1, 2, 3]),
+        gen(x => x * x, x => many([x, x + 1, x + 2]), x => 2 * x + 1),
+        streamToArray(output)
+      ]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [3, 5, 7, 9, 11, 13, 19, 21, 23])'));
+      async.done();
+    });
+  },
+  function test_genCombined(t) {
+    const async = t.startAsync('test_genCombined');
+
+    const output = [],
+      chain = new Chain([
+        fromIterable([1, 2]),
+        gen(
+          delay(x => -x),
+          x => many([x, x * 10]),
+          function*(x) {
+            yield x;
+            yield x - 1;
+          },
+          x => -x
+        ),
+        streamToArray(output)
+      ]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [1, 2, 10, 11, 2, 3, 20, 21])'));
+      async.done();
+    });
+  },
+  function test_genCombinedFinal(t) {
+    const async = t.startAsync('test_genCombinedFinal');
+
+    const output = [],
+      chain = new Chain([
+        fromIterable([1, 2]),
+        gen(
+          delay(x => -x),
+          x => many([x, x * 10]),
+          function*(x) {
+            yield x;
+            yield final(x - 1);
+          },
+          x => -x
+        ),
+        streamToArray(output)
+      ]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [1, -2, 10, -11, 2, -3, 20, -21])'));
+      async.done();
+    });
+  },
+  function test_genAsGen(t) {
+    const async = t.startAsync('test_genAsGen');
+
+    const output = [],
+      chain = new Chain([
+        fromIterable([1, 2]),
+        asGen(
+          delay(x => -x),
+          x => many([x, x * 10]),
+          function*(x) {
+            yield x;
+            yield final(x - 1);
+          },
+          x => -x
+        ),
+        streamToArray(output)
+      ]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [1, -2, 10, -11, 2, -3, 20, -21])'));
+      async.done();
+    });
+  },
+  function test_genAsAsyncGen(t) {
+    const async = t.startAsync('test_genAsAsyncGen');
+
+    const output = [],
+      chain = new Chain([
+        fromIterable([1, 2]),
+        asGen(
+          delay(x => -x),
+          x => many([x, x * 10]),
+          async function*(x) {
+            yield delay(x => x)(x);
+            yield delay(x => final(x - 1))(x);
+          },
+          x => -x
+        ),
+        streamToArray(output)
+      ]);
+
+    chain.on('end', () => {
+      eval(t.TEST('t.unify(output, [1, -2, 10, -11, 2, -3, 20, -21])'));
+      async.done();
+    });
+  }
+]);

tests/tests.js

@@ -1,5 +1,7 @@
 'use strict';
 
+const nodeVersion = /^v?(\d+)\./.exec(process.version);
+
 const unit = require('heya-unit');
 
 require('./test_FromIterable');
@@ -10,6 +12,7 @@ require('./test_errors');
 
 require('./test_transducers');
 require('./test_comp');
+nodeVersion && +nodeVersion[1] >= 10 && require('./test_gen');
 
 require('./test_take');
 require('./test_skip');

utils/asGen.js

@@ -0,0 +1,69 @@
+'use strict';
+
+const {none, Final, Many} = require('../defs');
+
+const next = async function*(value, fns, index) {
+  for (let i = index; i <= fns.length; ++i) {
+    if (value && typeof value.then == 'function') {
+      // thenable
+      value = await value;
+    }
+    if (value === none) break;
+    if (value instanceof Final) {
+      if (value !== none) yield value;
+      break;
+    }
+    if (value instanceof Many) {
+      const values = value.values;
+      if (i == fns.length) {
+        for (let j = 0; j < values.length; ++j) {
+          yield values[j];
+        }
+      } else {
+        for (let j = 0; j < values.length; ++j) {
+          yield* next(values[j], fns, i);
+        }
+      }
+      break;
+    }
+    if (value && typeof value.next == 'function') {
+      // generator
+      for (;;) {
+        let data = value.next();
+        if (data && typeof data.then == 'function') {
+          data = await data;
+        }
+        if (data.done) break;
+        if (i == fns.length) {
+          yield data.value;
+        } else {
+          yield* next(data.value, fns, i);
+        }
+      }
+      break;
+    }
+    if (i == fns.length) {
+      yield value;
+      break;
+    }
+    value = fns[i](value);
+  }
+};
+
+const nop = async function*() {};
+
+const asGen = (...fns) => {
+  fns = fns.filter(fn => fn);
+  if (!fns.length) return nop;
+  return async function*(value) {
+    yield* next(value, fns, 0);
+  };
+};
+
+asGen.next = next;
+
+asGen.none = none;
+asGen.Final = Final;
+asGen.Many = Many;
+
+module.exports = asGen;

utils/gen.js

@@ -0,0 +1,24 @@
+'use strict';
+
+const {Transform} = require('stream');
+const {next} = require('./asGen');
+const {sanitize} = require('../index');
+
+const gen = (...fns) => {
+  fns = fns.filter(fn => fn);
+  return fns.length
+    ? new Transform({
+        writableObjectMode: true,
+        readableObjectMode: true,
+        transform(chunk, encoding, callback) {
+          (async () => {
+            for await (let value of next(chunk, fns, 0)) {
+              sanitize(value, this);
+            }
+          })().then(() => callback(null), error => callback(error));
+        }
+      })
+    : null;
+};
+
+module.exports = gen;