Cannot read property '__ConsumeGenerator__' of null

[pavel]

Environment

Highland: 3.0.0-beta.2
Node: v6.3.1
OS: Linux

Example (CoffeeScript)

Stream = require "highland"

fn = ->
    Stream Promise.resolve([])
        .map (data) ->
            console.log "map", data
            Stream Promise.resolve([])
        .parallel 1
        .consume (err, data, push, next) ->
            push null, Stream.nil

Stream([
    Stream([["data"]])
    fn().collect()
])
    .parallel 2
    .tap (data) -> console.log "tap", data
    .collect()
    .each (data) -> console.log "each", data

Error

/home/user/highland-bug/node_modules/highland/lib/index.js:1280
        if (gen.__ConsumeGenerator__) {
               ^

TypeError: Cannot read property '__ConsumeGenerator__' of null
  at Stream._runGenerator (/home/user/highland-bug/node_modules/highland/lib/index.js:1280:16)
  at Immediate.<anonymous> (/home/user/highland-bug/node_modules/highland/lib/index.js:791:26)
  at runCallback (timers.js:570:20)
  at tryOnImmediate (timers.js:550:5)
  at processImmediate [as _immediateCallback] (timers.js:529:5)

---

Does only happen when Promise are in use.
Substituting Promise.resolve([]) to [[]] works without error.

[pavel]

Replacing .parallel 1 with .series() eliminates the error.

[vqvu]

I think this is an easy fix. The call to _runGenerator here needs to check if the generator still exists. It could have been destroyed by downstream.

I'm away from my computer for the new week or so though, so I won't be able to get around to this until then.

[pavel]

@vqvu You mean this check?

[vqvu]

Yeah.

[svozza]

I'm having trouble reproducing this, I translated it into ES5 here to create a failing test case but it passes for me. Is there something wrong with my code here?

var fn = function() {
        return _(Promise.resolve([]))
            .map(function (x) {
                return _(Promise.resolve([]));
            })
            .parallel(1)
            .consume(function (err, data, push, next) {
                push(null, _.nil);
            });
    };

    _([_([['data']]),
    fn().collect()])
        .parallel(2)
        .collect()
        .each(function(xs) {
            console.log(xs);
        });

[vqvu]

Here's a more explicit test case.

var _ = require("highland");

// Code path is this:
// - push(...) causes s.destroy() to be queued.
// - next2 is queued.
// - next1 is executed, which queues _runGenerator
// - destroy() runs and clears the _generator.
// - next2 runs and sets the _generator_requested flag.
// - _runGenerator runs and is allowed to be executed because _generator_requested
//   was set.
var s = _(function (push, next) {
    setTimeout(function () {
        push(new Error("error"));
        _.setImmediate(next); // next2
        next(); // next1
    }, 0);
});

s.pull(function (err, x) {
    // This pull is required to un-pause the stream.
    s.pull(function () {
        // Ignore.
    });

    _.setImmediate(function () {
        s.destroy();
    });
});

I thought about this more and I think the correct fix is to check for _explicitly_destroyed at the start of _next_fn and immediately return if it's true. I believe this also needs to be done in _writeOutgoing.

[svozza]

Ah yes, this causes failures for me now too. I'll raise a PR in the morning.

[vqvu]

@pavel, since we're not testing against the test case that you provided, can you verify that #521 fixes the issue for you?

[pavel]

@vqvu I've verified, and the fix does resolve the problem.

[vqvu]

#521 was merged, so I will close this.