Restructured so asFun() is a stream-independent.

defs.js

@@ -0,0 +1,19 @@
+'use strict';
+
+const none = {};
+
+function Final(value) {
+  this.value = value;
+}
+const final = value => new Final(value);
+
+function Many(values) {
+  this.values = values;
+}
+const many = values => new Many(values);
+
+module.exports.none = none;
+module.exports.Final = Final;
+module.exports.final = final;
+module.exports.Many = Many;
+module.exports.many = many;

index.js

@@ -1,14 +1,7 @@
 'use strict';
 
 const {Readable, Writable, Duplex, Transform} = require('stream');
-
-const none = {};
-function Final(value) {
-  this.value = value;
-}
-function Many(values) {
-  this.values = values;
-}
+const {none, Final, Many, final, many} = require('./defs');
 
 const runAsyncGenerator = async (gen, stream) => {
   for (;;) {
@@ -145,12 +138,6 @@ class Chain extends Duplex {
   static make(fns, options) {
     return new Chain(fns, options);
   }
-  static final(value) {
-    return new Chain.Final(value);
-  }
-  static many(values) {
-    return new Chain.Many(values);
-  }
   static sanitize(result, stream) {
     if (result instanceof Chain.Final) {
       result = result.value;
@@ -174,7 +161,9 @@ class Chain extends Duplex {
 
 Chain.none = none;
 Chain.Final = Final;
+Chain.final = final;
 Chain.Many = Many;
+Chain.many = many;
 
 Chain.chain = Chain.make;
 Chain.make.Constructor = Chain;

tests/test_comp.js

@@ -7,6 +7,7 @@ const {streamToArray, delay} = require('./helpers');
 
 const {fromIterable} = require('../utils/FromIterable');
 const comp = require('../utils/comp');
+const asFun = require('../utils/asFun');
 
 const {none, final, many} = Chain;
 
@@ -160,7 +161,7 @@ unit.add(module, [
     const output = [],
       chain = new Chain([
         fromIterable([1, 2]),
-        comp.asFun(
+        asFun(
           delay(x => -x),
           x => many([x, x * 10]),
           function*(x) {

utils/asFun.js

@@ -0,0 +1,75 @@
+'use strict';
+
+const {none, Final, Many} = require('../defs');
+
+const next = async (value, fns, index, push) => {
+  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) {
+      value !== none && push(value.value);
+      break;
+    }
+    if (value instanceof Many) {
+      const values = value.values;
+      if (i == fns.length) {
+        values.forEach(val => push(val));
+      } else {
+        for (let j = 0; j < values.length; ++j) {
+          await next(values[j], fns, i, push);
+        }
+      }
+      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) {
+          push(data.value);
+        } else {
+          await next(data.value, fns, i, push);
+        }
+      }
+      break;
+    }
+    if (i == fns.length) {
+      push(value);
+      break;
+    }
+    value = fns[i](value);
+  }
+};
+
+const nop = () => {};
+
+const asFun = (...fns) => {
+  fns = fns.filter(fn => fn);
+  if (!fns.length) return nop;
+  return async value => {
+    const results = [];
+    await next(value, fns, 0, value => results.push(value));
+    switch (results.length) {
+      case 0:
+        return none;
+      case 1:
+        return results[0];
+    }
+    return new Many(results);
+  };
+};
+
+asFun.next = next;
+
+asFun.none = none;
+asFun.Final = Final;
+asFun.Many = Many;
+
+module.exports = asFun;

utils/comp.js

@@ -1,55 +1,8 @@
 'use strict';
 
 const {Transform} = require('stream');
-
-const {none, Final, Many, sanitize} = require('../index');
-
-const next = async (value, fns, index, push) => {
-  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) {
-      value !== none && push(value.value);
-      break;
-    }
-    if (value instanceof Many) {
-      if (i == fns.length) {
-        value.values.forEach(val => push(val));
-        break;
-      }
-      const values = value.values;
-      for (let j = 0; j < values.length; ++j) {
-        await next(values[j], fns, i, push);
-      }
-      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) {
-          push(data.value);
-        } else {
-          await next(data.value, fns, i, push);
-        }
-      }
-      break;
-    }
-    const fn = fns[i];
-    if (!fn) {
-      push(value);
-      break;
-    }
-    value = fn(value);
-  }
-};
+const {next} = require('./asFun');
+const {sanitize} = require('../index');
 
 const comp = (...fns) => {
   fns = fns.filter(fn => fn);
@@ -64,22 +17,4 @@ const comp = (...fns) => {
     : null;
 };
 
-const nop = () => {};
-
-comp.asFun = (...fns) => {
-  fns = fns.filter(fn => fn);
-  if (!fns.length) return nop;
-  return async value => {
-    const results = [];
-    await next(value, fns, 0, value => results.push(value));
-    switch (results.length) {
-      case 0:
-        return none;
-      case 1:
-        return results[0];
-    }
-    return new Many(results);
-  };
-};
-
 module.exports = comp;