adding pipe interface, readable-stream module

gulpfile.js

@@ -8,7 +8,7 @@ var fs = Promise.promisifyAll(require('fs'));
 var excluded = {
   sqlite3: ['sqlite3'],
   mysql: ['mysql'],
-  pg: ['pg', 'pg.js', 'pg-though-stream'],
+  pg: ['pg', 'pg.js', 'pg-query-stream'],
   websql: ['sqlite3']
 };
 
@@ -22,7 +22,7 @@ var bases = {
 var all = ['mysql', 'pg', 'sqlite3', 'websql'];
 
 var externals = ['lodash', 'bluebird'];
-var alwaysExcluded = ['generic-pool-redux', 'stream', './lib/migrate/index.js'];
+var alwaysExcluded = ['generic-pool-redux', 'node-pg-stream', 'readable-stream', './lib/migrate/index.js'];
 
 gulp.task('build', function() {
   var targets = argv.t || 'all';

lib/runner.js

@@ -48,7 +48,11 @@ Runner.prototype.stream = Promise.method(function(options, handler) {
       options = {};
     }
   }
-  return Promise.bind(this)
+
+  // Lazy-load the "PassThrough" dependency.
+  PassThrough = PassThrough || require('readable-stream').PassThrough;
+  var stream  = new PassThrough({objectMode: true});
+  var promise = Promise.bind(this)
     .then(this.ensureConnection)
     .then(function(connection) {
       this.connection = connection;
@@ -60,23 +64,24 @@ Runner.prototype.stream = Promise.method(function(options, handler) {
       }
       return sql;
     }).then(function(sql) {
-
-      // Lazy-load the "PassThrough" dependency.
-      PassThrough = PassThrough || require('stream').PassThrough;
-      var stream  = new PassThrough({objectMode: true});
-
-      // If a function is passed to handle the stream, send the stream
-      // there and return the promise, otherwise just return the stream
-      // and the promise will take care of itsself.
-      if (_.isFunction(handler)) {
-	handler(stream);
-	return this._stream(sql, stream, options);
-      }
-      this._stream(sql, stream, options);
-      return stream;
+      return this._stream(sql, stream, options);
     }).finally(this.cleanupConnection);
+
+  // If a function is passed to handle the stream, send the stream
+  // there and return the promise, otherwise just return the stream
+  // and the promise will take care of itsself.
+  if (_.isFunction(handler)) {
+    handler(stream);
+    return promise;
+  }
+  return stream;
 });
 
+// Allow you to pipe the stream to a writable stream.
+Runner.prototype.pipe = function(writable) {
+  return this.stream().pipe(writable);
+};
+
 // "Runs" a query, returning a promise. All queries specified by the builder are guaranteed
 // to run in sequence, and on the same connection, especially helpful when schema building
 // and dealing with foreign key constraints, etc.

package.json

@@ -33,7 +33,8 @@
     "inherits": "~2.0.1",
     "lodash": "~2.4.0",
     "mkdirp": "~0.3.5",
-    "optimist": "~0.6.0"
+    "optimist": "~0.6.0",
+    "readable-stream": "1.0.27-1"
   },
   "scripts": {
     "test": "jshint knex.js lib/* && mocha -b --check-leaks -R spec test/index.js",