Outer groupBy doesn't obey backpressure

[abersnaze]

Someone here brought by some code that was throwing a MissingBackpressureException. Aaron and I narrowed it down to this snippet.

merge(range(0, 500).groupBy(i -> i % (RxRingBuffer.SIZE + 2)).observeOn(computation())).toBlocking().last()

it comes down observeOn requesting 128 Observables from groupBy and groupBy hard coded to request 1024 from range.

We fixed his problem changing the observeOn(computation()) to flatMap(grp -> grp.subscribeOn(computation()))2

[akarnokd]

It fails with 2.x too. I'll have a look at this within 2.x.

[akarnokd]

I've looked into this with 2.x. The problem is that a request(N) may result in a new group beyond the requested amount by the main subscriber (i.e., the one that receives GroupedObservables). Since consuming groups requires value replenishment from main, one can't be sure what the next value will be or where it will go (into the same group, into another group or into a completely new group).

The solution is to buffer the GroupedObservables for the main subscriber and hand them out based on requests while letting the groups be consumed. Since we can't know the number of groups, this buffering has to be unbounded, similar to how each group has to be unbounded (due to the same reason as before: asking for replenishment for a group may result in a value for another group).

I don't fully understand 1.x GroupBy so I don't know how to fix it with minimal changes. The best I can do is to port back the 2.x GroupBy, however, that requires SpscLinkedArrayQueue from #3169.