consume push mutates _consumers list.

[jgrund]

Consider the following code:

s = highland();
s1 = s.map(id).map(id).each(highland.log);
s2 = s.fork().map(id).each(highland.log);
s.end();
function id (xs) { return xs; }

This code throws the following error:

Uncaught TypeError: Cannot read property 'write' of undefined

What is happening is the consumer is being removed at the same time we are iterating a list of consumers to write to. We should either slice (copy) the list of consumers before iterating, or not mutate the consumers list upon getting a nil in consume.

[jgrund]

Any thoughts on fix? Happy to do a PR.

[vqvu]

It's a pretty easy fix. We don't even need to splice before iterating, just save the reference. I already have a PR.

[vqvu]

Side note, support for forking a consumed stream is going away in 3.0.0. That is, you can either fork a stream or you can consume it. Not both. So you're better off just sticking a fork at the start of s1 if you can.

This bug only happens when you do a fork of a stream that has already been consumed.

[jgrund]

Beat me to it 😄

[jgrund]

Left a note in that PR: you are relying on the same reference in memory when you do that assignment. It could introduce other issues down the line if someone mutates that reference within the loop.

[jgrund]

@vqvu Regarding your earlier comment, I should be fine to do forks only. I end up using highland streams with forking / destroying forks in a dynamic manner. Seems like I can always just fork the original stream since it will just push a new stream on to the _consumers.

Why not just have consume use a fork internally? Then you'd have one method of adding consumers.

[vqvu]

Why not just have consume use a fork internally?

We want to force people to explicitly choose a back pressure strategy (fork or observe) when they multiplex the result of a stream. consume needs to be able to throw an exception when users make a mistake and try to consume from the same stream multiple times, so it can't just use fork internally.