propagation of _.nil in async scenario

[jeromew]

Hello,

I had a 'freeze' bug in a complex pipeline that I managed to simplify to the following.

var _ = require('./lib/index.js');
_([1])
.consume(function(err, x, push, next) {
  if (err !== null) {
    push(err);
    next();
  }
  else if (x === _.nil) {
    console.log('first nil')
    setImmediate(function() {
      push(null, x);
    })
  }
  else {
    push(null, x);
    next();
  }
})
.consume(function(err, x, push, next) {
  if (err !== null) {
    push(err);
    next();
  }
  else if (x === _.nil) {
    console.log('second nil')
    push(null, x);
  }
  else {
    push(null, x);
    next();
  }
})
.resume();

I had the impression that this should log first nil and second nil (which it does in the sync case) but by forwarding the _.nil in an async fashion with setImmediate as a test, only first nil is logged.

Am I missing something ?

I noticed that if I call next() in the async case, then it starts working again so I concluded that there is either a bug or something that I don't understand about calling next() in the _.nil case

[vqvu]

If you call resume() again, you'll see the second nil. It's because the consume stream will pause if you don't synchronously call next. Thus, when you finally push the nil, you're pushing onto a paused stream, and it gets buffered instead of emitted downstream. Normally this is OK, since next will be called after the push if you need to do an async push (and next will resume the stream), but this isn't the case when pushing nil.

It's definitely a bug not to resume the stream after an async nil. It's not trivial to fix though. The reason for the pause is to guard against upstream producers that don't call next in between two push, but it's not the correct thing to do.

Here's another, related, bug with consume.

var s = _([1, 2])
.consume(function(err, x, push, next) {
  if (err !== null) {
    push(err);
    next();
  }
  else if (x === _.nil) {
    push(null, x);
  }
  else {
    if (x === 1) {
      setImmediate(function () {
        push(null, x);
        next();
      });
    } else {
      push(null, x);
      next();
    }
  }
})

s.toArray(_.log);
// => [2, 1]

I'll take ownership of this, since what I'm working on for #41 is related.

[vqvu]

Actually, what I mentioned is not a bug. It doesn't happen. I seemed to be working with the wrong branch.