Adds functions tream. Dont know why.

modules/stream.js

@@ -7,6 +7,7 @@ import BroadcastStream from './stream/broadcast-stream.js';
 import BufferStream    from './stream/buffer-stream.js';
 import BatchStream     from './stream/batch-stream.js';
 import CombineStream   from './stream/combine-stream.js';
+import FunctionStream  from './stream/function-stream.js';
 import MergeStream     from './stream/merge-stream.js';
 import PromiseStream   from './stream/promise-stream.js';
 //import ZipProducer     from './stream/zip-producer.js';
@@ -40,8 +41,12 @@ assign(Stream, {
         return source.pipe ? new Stream(source) :
             // Source is a promise
             source.then ? new PromiseStream(source) :
-            // Source is an array-like
-            typeof source.length === 'number' ? new BufferStream(source) :
+            // Source has length
+            typeof source.length === 'number' ?
+                // Source is a function
+                typeof source === 'function' ? new FunctionStream(source) :
+                // Source is an array-like
+                new BufferStream(source) :
             // Source cannot be made into a stream
             throwTypeError(source) ;
     },

modules/stream/function-stream.js

@@ -0,0 +1,30 @@
+
+import Stream, { stop, push } from './stream.js';
+
+const assign = Object.assign;
+const create = Object.create;
+
+
+/*
+FunctionStream(fn)
+A FunctionStream is a readable stream of values produced by `fn(push, stop)`,
+which is called when the stream is first consumed.
+*/
+
+export default function FunctionStream(fn) {
+    this.fn = fn;
+}
+
+FunctionStream.prototype = assign(create(Stream.prototype), {
+    pipe: function(output) {
+        // Connect stream to output
+        output.done(this);
+        this[0] = output;
+
+        // Call fn(push, stop)
+        this.fn((value) => this.push(value), (value) => this.stop(value));
+
+        // Return output stream
+        return output;
+    }
+});

modules/stream/stream.js

@@ -73,7 +73,7 @@ assign(Stream.prototype, Stopable.prototype, {
     **/
     pipe: function(output) {
         if (this[0]) {
-            throw new Error('Stream: Attempt to .pipe() a unicast stream multiple times. Create a multicast stream with stream.broadcast().');
+            throw new Error('Stream: Attempt to .pipe() a unicast stream multiple times. Create a multicast stream with .broadcast().');
         }
 
         this[0] = output;