Added dataSource() to wrap iterable objects.

src/index.js

@@ -153,6 +153,15 @@ const chain = (fns, options) => {
   return stream;
 };
 
+const dataSource = fn => {
+  if (typeof fn == 'function') return fn;
+  if (fn) {
+    if (typeof fn[Symbol.asyncIterator] == 'function') return fn[Symbol.asyncIterator].bind(fn);
+    if (typeof fn[Symbol.iterator] == 'function') return fn[Symbol.iterator].bind(fn);
+  }
+  throw new TypeError('The argument should be a function or an iterable object.');
+};
+
 module.exports = chain;
 
 // to keep ESM happy
@@ -175,3 +184,5 @@ module.exports.final = chain.final = defs.final;
 module.exports.chain = chain.chain = chain; // for compatibility with 2.x
 module.exports.gen = chain.gen = gen;
 module.exports.asStream = chain.asStream = asStream;
+
+module.exports.dataSource = dataSource;

tests/test-dataSource.mjs

@@ -0,0 +1,98 @@
+'use strict';
+
+import test from 'tape-six';
+
+import {streamToArray, delay} from './helpers.mjs';
+import chain, {dataSource} from '../src/index.js';
+
+test.asPromise('dataSource: smoke test', (t, resolve) => {
+  const output = [],
+    c = chain([dataSource([1, 2, 3]), streamToArray(output)]);
+
+  c.on('end', () => {
+    t.deepEqual(output, [1, 2, 3]);
+    resolve();
+  });
+
+  c.end(1); // start the chain
+});
+
+test.asPromise('dataSource: function', (t, resolve) => {
+  const output = [],
+    c = chain([dataSource(() => 0), streamToArray(output)]);
+
+  c.on('end', () => {
+    t.deepEqual(output, [0]);
+    resolve();
+  });
+
+  c.end(1); // start the chain
+});
+
+test.asPromise('dataSource: async function', (t, resolve) => {
+  const output = [],
+    c = chain([dataSource(delay(() => 0)), streamToArray(output)]);
+
+  c.on('end', () => {
+    t.deepEqual(output, [0]);
+    resolve();
+  });
+
+  c.end(1); // start the chain
+});
+
+test.asPromise('dataSource: generator', (t, resolve) => {
+  const output = [],
+    c = chain([
+      dataSource(function* () {
+        yield 0;
+        yield 1;
+      }),
+      streamToArray(output)
+    ]);
+
+  c.on('end', () => {
+    t.deepEqual(output, [0, 1]);
+    resolve();
+  });
+
+  c.end(1); // start the chain
+});
+
+test.asPromise('dataSource: async generator', (t, resolve) => {
+  const output = [],
+    c = chain([
+      dataSource(async function* () {
+        yield delay(() => 0)();
+        yield delay(() => 1)();
+      }),
+      streamToArray(output)
+    ]);
+
+  c.on('end', () => {
+    t.deepEqual(output, [0, 1]);
+    resolve();
+  });
+
+  c.end(1); // start the chain
+});
+
+test.asPromise('dataSource: nextable', (t, resolve) => {
+  const output = [],
+    c = chain([
+      dataSource(
+        (function* () {
+          yield 0;
+          yield 1;
+        })()
+      ),
+      streamToArray(output)
+    ]);
+
+  c.on('end', () => {
+    t.deepEqual(output, [0, 1]);
+    resolve();
+  });
+
+  c.end(1); // start the chain
+});