sink the saveStream on nextTick to start flowing and avoid highWaterM…

…ark - add tests

lib/dest/index.js

@@ -3,6 +3,7 @@
 var through2 = require('through2');
 var sourcemaps = require('gulp-sourcemaps');
 var duplexify = require('duplexify');
+var sink = require('../sink');
 var prepareWrite = require('../prepareWrite');
 var writeContents = require('./writeContents');
 
@@ -21,7 +22,12 @@ function dest(outFolder, opt) {
   }
 
   var saveStream = through2.obj(saveFile);
+
   if (!opt.sourcemaps) {
+    // Sink the save stream to start flowing
+    // Do this on nextTick, it will flow at slowest speed of piped streams
+    process.nextTick(sink(saveStream));
+
     return saveStream;
   }
 
@@ -34,6 +40,10 @@ function dest(outFolder, opt) {
   var outputStream = duplexify.obj(mapStream, saveStream);
   mapStream.pipe(saveStream);
 
+  // Sink the output stream to start flowing
+  // Do this on nextTick, it will flow at slowest speed of piped streams
+  process.nextTick(sink(outputStream));
+
   return outputStream;
 }
 

lib/sink.js

@@ -0,0 +1,16 @@
+'use strict';
+
+var Writable = require('readable-stream/writable');
+
+function sink(stream) {
+  var sinkStream = new Writable({
+    objectMode: true,
+    write: function(file, enc, cb) { cb(); }
+  });
+
+  return function() {
+    stream.pipe(sinkStream);
+  };
+}
+
+module.exports = sink;

package.json

@@ -19,6 +19,7 @@
     "merge-stream": "^1.0.0",
     "mkdirp": "^0.5.0",
     "object-assign": "^4.0.0",
+    "readable-stream": "^2.0.4",
     "strip-bom": "^2.0.0",
     "strip-bom-stream": "^1.0.0",
     "through2": "^2.0.0",
@@ -27,6 +28,7 @@
   },
   "devDependencies": {
     "buffer-equal": "^0.0.1",
+    "del": "^2.2.0",
     "istanbul": "^0.3.0",
     "istanbul-coveralls": "^1.0.1",
     "jshint": "^2.4.1",

test/dest.js

@@ -8,7 +8,8 @@ var vfs = require('../');
 
 var path = require('path');
 var fs = require('graceful-fs');
-var rimraf = require('rimraf');
+var del = require('del');
+var Writeable = require('readable-stream/writable');
 
 var bufEqual = require('buffer-equal');
 var through = require('through2');
@@ -17,11 +18,12 @@ var File = require('vinyl');
 var should = require('should');
 require('mocha');
 
-var wipeOut = function(cb) {
-  rimraf(path.join(__dirname, './out-fixtures/'), cb);
+var wipeOut = function() {
   spies.setError('false');
   statSpy.reset();
   chmodSpy.reset();
+  del.sync(path.join(__dirname, './fixtures/highwatermark'));
+  del.sync(path.join(__dirname, './out-fixtures/'));
 };
 
 var dataWrap = function(fn) {
@@ -1256,4 +1258,72 @@ describe('dest stream', function() {
     stream.write(file);
     stream.end();
   });
+
+  it('does not get clogged by highWaterMark', function(done) {
+    fs.mkdirSync(path.join(__dirname, './fixtures/highwatermark'));
+    for (var idx = 0; idx < 17; idx++) {
+      fs.writeFileSync(path.join(__dirname, './fixtures/highwatermark/', 'file' + idx + '.txt'));
+    }
+
+    var srcPath = path.join(__dirname, './fixtures/highwatermark/*.txt');
+    var srcStream = vfs.src(srcPath);
+    var destStream = vfs.dest('./out-fixtures/', {cwd: __dirname});
+
+    var fileCount = 0;
+    var countFiles = through.obj(function(file, enc, cb) {
+      fileCount++;
+
+      cb(null, file);
+    });
+
+    destStream.once('finish', function() {
+      fileCount.should.equal(17);
+      done();
+    });
+
+    srcStream.pipe(countFiles).pipe(destStream);
+  });
+
+  it('allows backpressure when piped to another, slower stream', function(done) {
+    this.timeout(8000);
+
+    fs.mkdirSync(path.join(__dirname, './fixtures/highwatermark'));
+    for (var idx = 0; idx < 24; idx++) {
+      fs.writeFileSync(path.join(__dirname, './fixtures/highwatermark/', 'file' + idx + '.txt'));
+    }
+
+    var srcPath = path.join(__dirname, './fixtures/highwatermark/*.txt');
+    var srcStream = vfs.src(srcPath);
+    var destStream = vfs.dest('./out-fixtures/', {cwd: __dirname});
+
+    var fileCount = 0;
+    var countFiles = through.obj(function(file, enc, cb) {
+      fileCount++;
+
+      cb(null, file);
+    });
+
+    var slowFileCount = 0;
+    var slowCountFiles = new Writeable({
+      objectMode: true,
+      write: function(file, enc, cb){
+        slowFileCount++;
+
+        setTimeout(function() {
+          cb(null, file);
+        }, 250);
+      }
+    });
+
+    slowCountFiles.once('finish', function() {
+      fileCount.should.equal(24);
+      slowFileCount.should.equal(24);
+      done();
+    });
+
+    srcStream
+      .pipe(countFiles)
+      .pipe(destStream)
+      .pipe(slowCountFiles);
+  });
 });