Operator "consume" internals

[vitaly-t]

Just to show what operator consume really does underneath - compare the following custom operator examples...

If we were to take the code example from operator consume, and make it a custom operator, it would be like this:

function toReadable<T>(): Operation<T, Readable> {
    return source => consume(i => Readable.from(i))(source);
}

To do the same, but without consume, and using concurrencyFork instead:

function toReadable<T>(): Operation<T, Readable> {
    return source => concurrencyFork({
        onSync(i: Iterable<T>) {
            return [Readable.from(i)];
        },
        onAsync(i: AsyncIterable<T>) {
            return toAsync([Readable.from(i)]) as any;
        }
    })(source) as any;
}

And without using any helpers, just direct implementation:

function toReadable<T>(): Operation<T, Readable> {
    return source => {
        const sync = typeof source[Symbol.iterator] === 'function';
        const r = [Readable.from(source)];
        return sync ? r : toAsync(r);
    };
}

All of the above would produce the same result, though the latter would not be able to handle errors thrown during pipeline construction, and it is what should be generally considered an unsafe approach.

To understand why operator consume represents a good pattern, consider an example from this discussion:

function streamCursor<T>(externalResource): Operation<T, Readable> {
    return source => consume(i => {
        const r = Readable.from(i);
        r.once('close', () => {
            // release externalResource here
        });
        return r;
    })(source);
}

Operator consume makes it easy and safe to link up external objects like that, inside custom operators, and manage their lifecycle. And when used directly, it is the only way to consume the entire source when it is an iterator (not iterable).