Pipeline not propagating Promise casted error properly

[cphoover]

it seems like when a stream is passed through a pipeline the error is not being propagated properly and the error handler is never called.

For instance while this works (as it should)

'use strict';

var hi = require('highland'),
    Bluebird = require('bluebird');

function prependBlah(s) {
    return s.map(function (x) {
        return 'blah ' + x.toString();
    });
}

function appendAsdf(s) {
    return s.map(function (x) {
        return x.toString() + 'Asdf';
    });
}

function appendNewline(s) {
    return s.map(function (x) {
        return x.toString() + '\n';
    });
}

hi(function (push, next) {
    push(null, hi(Bluebird.resolve(1)));
    push(null, hi(Bluebird.resolve(2)));
    push(null, hi(Bluebird.reject(3)));
    setTimeout(next, 1000);
})
    .merge()
    .errors(function (err) {
        console.log('err', err);
    })
    .through(hi.pipeline.apply(null, [
        prependBlah,
        appendAsdf,
        appendNewline
    ]))
    .pipe(process.stdout);

the below snippet of code causes the following error to be thrown: Unhandled rejection Error: 3

'use strict';

var hi = require('highland'),
    Bluebird = require('bluebird');

function prependBlah(s) {
    return s.map(function (x) {
        return 'blah ' + x.toString();
    });
}

function appendAsdf(s) {
    return s.map(function (x) {
        return x.toString() + 'Asdf';
    });
}

function appendNewline(s) {
    return s.map(function (x) {
        return x.toString() + '\n';
    });
}

hi(function (push, next) {
    push(null, hi(Bluebird.resolve(1)));
    push(null, hi(Bluebird.resolve(2)));
    push(null, hi(Bluebird.reject(3)));
    setTimeout(next, 1000);
})
    .merge()
    .through(hi.pipeline.apply(null, [
        prependBlah,
        appendAsdf,
        appendNewline
    ]))
    .errors(function (err) {
        console.log('err', err);
    })
    .pipe(process.stdout);

Seems like the error should be handled in both instances, and thus the promise chain should handle the rejection. Maybe I am misunderstanding how pipeline and highland stream error handling is supposed to function.

[vqvu]

The two examples you gave should behave the same way, so you're not misunderstanding.

But I can't reproduce the bug. What version of highland are you using? I'm using 2.5.1 and both of your examples print

blah 1Asdf
blah 2Asdf
err 3

every second.

[cphoover]

I'm also using 2.5.1 strange....

[vqvu]

Nevermind, I see it. I thought I correctly copied your second example, but apparently I didn't.

I ran it against master though, and it seems to work. My guess is it was fixed by #372. That PR was meant to fix pipeline behavior as it applies to node streams, but it did do a general rewrite of pipeline.

Can you verify that the code in master works for you?

Aside: Looks like 2.5.1 was released in May, which means it's probably time to do another release. I'll probably do this after #393 is merged.

[vqvu]

Btw, using pipeline with through in this way is something of an anti-pattern. You can just as easily call through three times and pass it your handlers each time. pipeline is more meant to be used for interop with Node streams. You can even use seq if you don't want to type "through" all the time.

This is functionally equivalent to the version you had with pipeline.

    .through(hi.seq(
        prependBlah,
        appendAsdf,
        appendNewline
    ))

[vqvu]

Closing because the issue seems to be resolved in the latest version.

@daywiss, I got a notification of your comment, but since you've deleted it, I'll assume that you've worked out the issue that you had.