Adding in support for Node streams to queue-flow.

You can simple wrap a readable stream with `q(readable)` and you can push results into a writable stream with `.pipe(writable)`

lib/queue-flow.js

@@ -37,6 +37,13 @@ function qFunc(nameOrArray, QType) {
         return this.namedQueues[nameOrArray];
     } else if(nameOrArray instanceof Array) {
         return new QType(nameOrArray, q);
+    } else if(nameOrArray instanceof Object &&
+        typeof(nameOrArray.pipe) === 'function' &&
+        typeof(nameOrArray.on) === 'function') {
+        var newQ = new QType(undefined, q);
+        nameOrArray.on('data', newQ.push.bind(newQ));
+        nameOrArray.on('end', newQ.close.bind(newQ));
+        return newQ;
     } else {
         return new QType(undefined, q);
     }
@@ -709,7 +716,21 @@ Q.prototype.promise = function promise(callback, error) {
 Q.prototype.promiseAsync = Q.prototype.promise;
 Q.prototype.promiseSync = function() { throw "Synchronous Promises are Nonsensical!"; };
 
-// ``drain`` is a simple callback that empties the attached queue and throws away the 
+// ``pipe`` pushes the queue results into the provided writeable object (and returns it so
+// if it is also readable, you can continue to ``pipe`` it just as you'd expect). Throws
+// an error if it can't find the ``write`` and ``end`` methods.
+Q.prototype.pipe = function pipe(writable) {
+    if(typeof(writable.write) !== 'function' || typeof(writable.end) !== 'function') throw new Error('Not a valid writeable object!');
+    this.setHandler(function(value, next) {
+        if(!next) return writable.end();
+        next(function() {
+            writable.write(value);
+        });
+    });
+    return writable;
+};
+
+// ``drain`` is a simple callback that empties the attached queue and throws away the
 // results. This is useful for long-running queues to eliminate references to effectively
 // "dead" data without using the ``reduce`` hack to do so. No chaining is possible
 // after this call (for obvious reasons).

test/test.js

@@ -956,6 +956,31 @@ exports.pushToClosedQueue = function(test) {
     test.done();
 };
 
+exports.readStreams = function(test) {
+    bootstrap(test);
+    test.expect(1);
+    q(fs.createReadStream('./test/test.js', { encoding: 'utf8' }))
+        .toArray(function(arr) {
+            test.ok(arr.length > 0, 'got the array');
+            test.done();
+        });
+};
+
+exports.pipe = function(test) {
+    bootstrap(test);
+    test.expect(2);
+    var fakeWritable = {
+        write: function write(data) {
+            test.ok(!!data, 'received data');
+        },
+        end: function end() {
+            test.ok(true, 'stream was closed');
+            test.done();
+        }
+    };
+    q(['foo']).pipe(fakeWritable);
+};
+
 exports.complexity = function(test) {
     bootstrap(test);
     test.expect(1);