Added toCallback method

[0chroma]

Wrote this after reading through the discussion here: #484 #488

Main point of concern is that this is throwing an error, I've seen other functions call this.emit('error', ...) instead but wasn't sure if it was appropriate here. Since it's programmer error, and isn't really meant to be "caught", I think it should be okay, but open to feedback. There's a doc on Joyent's site that goes more into detail: https://www.joyent.com/developers/node/design/errors (specifically the section titled "(Not) handling programmer errors")

[vqvu]

Main point of concern is that this is throwing an error

It should definitely not throw new Error, I think. The way the rest of the library is written, if you have a programmer error that is checked synchronously, then it is thrown, but programmer errors that are only detected asynchronously are passed along whatever callback mechanism is available. For example, sequence does a push(new Error) even though the error here is a programmer error. In contrast, parallel throws Error when checking its arguments.

In general, I think the philosophy is that the user should be able to suppress all errors emitted by the library.

Also, I don't think the current implementation quite handles all of the error cases. Specifically, if you have

push(null, 1);
push(error);

or

push(null, 1);
push(null, 2);
push(error);

the callback will receive cb(error) instead of cb(new Error("stream emitted multiple values")).

The better way is probably to use a consume operator and detect everything directly. Something like this perhaps

var value;
var hasValue = false; // In case an emitted value === null or === undefined.

stream.consume(function (err, x, push, next) {
    if (err) {
        push(null, _.nil); // Signal that the stream no longer cares about future values.
        if (hasValue) {
            // emit error.
        }
        else {
            cb(err);
        }
    }
    else if (x === _.nil) {
        // Either cb() or cb(null, value);
    }
    else {
        if (hasValue) {
            push(null, _.nil); // Signal that the stream no longer cares about future values.
            // emit error
        }
        else {
            value = x;
            hasValue = true;
        }
    }
});

[0chroma]

I edited the tests quickly and found that this was the result for each case:

push(null, 1);
push(error);
>> cb(error) gets called

push(null, 1);
push(null, 2);
push(error);
>> throws multiple values error

The second case will work since we're only pulling 2 values at most, and since stopOnError is just passing values through until it reaches an error in the stream, it never gets up to it. You make a good point with the first case though, it is strange behavior that I didn't think about initially (was trying to get a rough solution down at first). I'll be sure to add test cases for those situations. Using consume() directly probably is the more efficient way of doing things anyway, so I'll rewrite it using your suggestion as well.

Also yeah, that's a good point about being able to suppress errors, the only reason I shied away from doing that was because it would be hard to differentiate between errors coming from the stream vs errors coming from toCallback, but maybe that's okay since it shouldn't happen during runtime unless the function is being used incorrectly.

[vqvu]

The second case will work since we're only pulling 2 values at most, and since stopOnError is just passing values through until it reaches an error in the stream, it never gets up to it.

Ah, you're right.

the only reason I shied away from doing that was because it would be hard to differentiate between errors coming from the stream vs errors coming from toCallback

Yeah, I understand your concern. But there's plenty of other ways for programmer errors to get to the errors of the callback, so I wouldn't worry too much much about separating them. For example,

// This is a programmer error. You can't flat map to a number.
_([1]).flatMap(x => x)
    .toCallback((err, x) => {
        // err should say "Expected Stream: got Number"
    });

[vqvu]

Your new changes look good. Can you add a call to test.expect(n) at the top of every test? Where n is the number of assertions you make in the test. It helps to detect when some assertions aren't run at all (and thus implicitly passes). It'll also make it so that you don't have to do the numCalled workaround in the last test.

[0chroma]

Ah okay, just added the test.expect() calls.

[vqvu]

Great! Thanks for the PR.

[0chroma]

Oh no problem, thanks for helping me get this implemented!

[vqvu]

This is released in version 2.8.1 if you want to use it.