Used readable/writable properties on a stream + updated tests.

src/index.js

@@ -37,7 +37,8 @@ const groupFunctions = (output, fn, index, fns) => {
     output.push(fn);
     return output;
   }
-  if (typeof fn != 'function') throw TypeError('Item #' + index + ' is not a proper stream, nor a function.');
+  if (typeof fn != 'function')
+    throw TypeError('Item #' + index + ' is not a proper stream, nor a function.');
   if (!output.length) output.push([]);
   const last = output[output.length - 1];
   if (Array.isArray(last)) {
@@ -118,7 +119,7 @@ const chain = (fns, options) => {
 
   if (!isWritableNodeStream(input)) {
     writeMethod = (_1, _2, callback) => callback(null);
-    finalMethod = callback => callback(null); // unavailable in Node 6
+    finalMethod = callback => callback(null);
     input.on('end', () => stream.end());
   }
 
@@ -131,7 +132,9 @@ const chain = (fns, options) => {
   }
 
   stream = new Duplex(
-    Object.assign({}, {writableObjectMode: true, readableObjectMode: true}, options, {
+    Object.assign({writableObjectMode: true, readableObjectMode: true}, options, {
+      readable: isReadableNodeStream(output),
+      writable: isWritableNodeStream(input),
       write: writeMethod,
       final: finalMethod,
       read: readMethod

tests/test-readWrite.mjs

@@ -3,7 +3,7 @@
 import test from 'tape-six';
 
 import {streamToArray} from './helpers.mjs';
-import chain from '../src/index.js';
+import chain, {dataSource} from '../src/index.js';
 import fromIterable from '../src/utils/readableFrom.js';
 
 test.asPromise('readWrite: readable', (t, resolve) => {
@@ -70,16 +70,17 @@ test.asPromise('readWrite: single writable', (t, resolve) => {
   });
 });
 
-test.asPromise('readWrite: pipeable', (t, resolve) => {
+test.asPromise('readWrite: pipeable', (t, resolve, reject) => {
   const output1 = [],
     output2 = [],
-    c = chain([fromIterable([1, 2, 3]), streamToArray(output1)]);
+    c = chain([dataSource([1, 2, 3]), streamToArray(output1)]);
 
   fromIterable([4, 5, 6]).pipe(c).pipe(streamToArray(output2));
 
   c.on('end', () => {
-    t.deepEqual(output1, [1, 2, 3]);
+    t.deepEqual(output1, [1, 2, 3, 1, 2, 3, 1, 2, 3]);
     t.deepEqual(output2, []);
     resolve();
   });
+  c.on('error', error => reject(error));
 });

tests/test-simple.mjs

@@ -110,6 +110,4 @@ test.asPromise('simple: factory', (t, resolve) => {
     t.deepEqual(output, [1, 1, 2, 4, 8, 4, 9, 27, 6]);
     resolve();
   });
-
-  c.end(0); // start the chain
 });