Breaks asynchronous down-stream streams

[wbyoung]

This simple code creates an asynchronous transform (that doesn't do anything other than wait a bit). It sets up a simple gulp config where you pipe through plumber and then to that asynchronous transform.

I think that it has something to do with gulp-plumber, but the transform never emits any data for anything downstream.

npm install gulp through gulp-plumber

// index.js
var gulp = require('gulp');
var plumber = require('gulp-plumber');
var through = require('through');

var transform = through(function(file) {
  this.pause();
  setTimeout(function() {
    this.queue(file);
    this.resume();
  }.bind(this), 100);
});

gulp.src('index.js')
  .pipe(plumber())
  // .pipe(through())
  .pipe(transform)
  .pipe(gulp.dest('out'));

The commented line in here is a workaround. If you simply create a synchronous stream following plumber, then everything works as expected. Plumber also seems to still notice errors properly in the asynchronous setup if you add this.emit('error', new Error('an example error')) regardless of whether or not that commented line is included.

[floatdrop]

Gulp guideline recommends to use through2 package to create pass-through streams:

// npm i gulp gulp-debug gulp-plumber through2

var gulp = require('gulp');
var plumber = require('gulp-plumber');
var through2 = require('through2');
var debug = require('gulp-debug');

var transform = through2.obj(function(file, enc, callback) {
  setTimeout(function() {
    this.push(file);
    callback();
  }.bind(this), 100);
});

gulp.src('index.js')
  .pipe(plumber())
  .pipe(transform)
  .pipe(debug())
  .pipe(gulp.dest('out'));
});

[wbyoung]

@floatdrop I still think it'd be nice for this to be compatible with other sources of streams. As I understand it, there's nothing wrong with using through.

[floatdrop]

@wbyoung I got same behaviour without gulp-plumber:

var gulp = require('gulp');
var through = require('through');
var debug = require('gulp-debug');

var transform = through(function(file) {
  this.pause();
  setTimeout(function() {
    this.queue(file);
    this.resume();
  }.bind(this), 100);
});

gulp.task('default', function () {
    gulp.src('index.js')
      .pipe(transform)
      .pipe(debug())
      .pipe(gulp.dest('out'));
});

Maybe this issue belongs to through package or we both misunderstand usage of this.pause/this.resume functions.

[wbyoung]

That's the way it's tested. Perhaps @dominictarr could shed some light on what may be wrong here.

[floatdrop]

@wbyoung I think it's because end event from gulp.src which can kill through unintentionally. Check this code:

var gulp = require('gulp');
var through = require('through');
var debug = require('gulp-debug');

var transform = through(function(file) {
  this.pause();
  setTimeout(function() {
    console.log('Pushing data');
    this.queue(file);
    this.resume();
  }.bind(this), 300);
}, function () {
    console.log('no end for you!');
});

gulp.task('default', function () {
    gulp.src('index.js')
      .pipe(debug())
      .pipe(transform)
      .on('data', function (data) {
        console.log(data);
      });
});

[wbyoung]

@floatdrop, I opened gulpjs/gulp#546 to see if we can get an answer. Thanks for looking into this, by the way.

[dominictarr]

hmm, i think @floatdrop is correct about end. You don't have an end handler, and so the default end is just this.queue(null). Then the timeout hits, and you try to queue something after the end marker. that is wrong so, through won't emit that. Through should prehaps make a better error message though. If you had already buffered something then it would work, but as far as through can tell, it's done.

[wbyoung]

@dominictarr then in this case would this be a more correct use of through:

var gulp = require('gulp');
var plumber = require('gulp-plumber');
var through = require('through');

var transform = through(function(file) {
  this.pause();
  setTimeout(function() {
    file.contents = new Buffer('hello world\n');
    this.queue(file);
    this.queue(null);
    this.resume();
  }.bind(this), 100);
}, function() {
  // end happens after async operation
});

gulp.src('index.js')
  .pipe(plumber())
  // .pipe(through())
  .pipe(transform)
  .pipe(gulp.dest('out'));

[wbyoung]

Oh, and @dominictarr if you have time, can you explain the difference here between why this needs to be set up this way and why the async test in through works? Thanks so much!

[dominictarr]

the difference is that in the through test, the source is well behaved, but in this case with gulp, the source is emitting end, even when the downstream is paused... maybe this is a streams2 thing. streams2 are complicated, so I'm not sure.

[wbyoung]

@dominictarr thanks!