Eager ConcatMap

[akarnokd]

Related discussion in #3017.

[davidmoten]

Thanks a lot for contributing this one @akarnokd. I've had one look through and no problems jump out at me but I will review further. This will be useful.

[davidmoten]

For readability I'd suggest longer names for variables like d, a, r, e, p.

[akarnokd]

Thanks for the review @davidmoten . Usually I use one letter variables because I find it easier to parse the code and instead of long variable names, I use newlines to separate logical blocks. This way, I don't have to type that many letters and I don't have to wait while Eclipse returns with the content assist (which is blocking by the way and may take hundreds of milliseconds, even on my i7 + SSD).

I took the time and renamed variables as you asked for and added some more unit tests that check the code paths. That being said, I don't plan to do such renames in my contributions very often in the future and encourage anybody to post their PRs with their proposed cleanups/renames.

[stealthcode]

Should we name these eagerConcat instead of eagerConcatMap? These overloads do not take a FuncN so they aren't really mapping.

[davidmoten]

Ditto that @stealthcode and for discoverability call it concatEager?

[akarnokd]

Thanks for the feedback. I've updated the method names to concatMapEager and concatEager, added the @Experimental tags, some javadoc and a missing test for a 2-parameter concatEager overload.

[akarnokd]

Updated with a capacity fix for the SpscLinkedArrayQueue.

[stevegury]

I reviewed in details in the code and I believe it is correct.

But I wonder if subscribing to the sources in a "unbounded mode" is the right thing to do.
At the expense of complexifying the code, I believe it could be possible to split the request(n) between the sources. Maybe requesting 1 to all the m sources expect the first one, which we request n - m + 1 ... and so on.

The devil is in the details, and I believe this would significantly complexify the code, but I would like to know what you are thinking about this?

[akarnokd]

@stevegury This operator consumes source observables in order and doesn't make sense to split any request from downstream. If the downstream requests n and the first is requested n / count, that would hang the sequence because only the first is allowed to emit and it won't emit enough to trigger a new request.

It is possible to use a bounded buffer per source so while they are not consumed, they don't grow indefinitely. However, it means that each of them can produce only RxRingBuffer.SIZE elements before stopping and thus the operator wouldn't be eager anymore.

[stevegury]

@akarnokd sorry I wasn't very clear but what I proposed was roughly what you described (replacing RxRingBuffer.SIZE by 1).

My main concern here is the use of unbounded buffer, which, I think, could be avoided.

[akarnokd]

@davidmoten What do you think about the bounded buffering and the fact that such sources would act more like a delayed source?

[davidmoten]

I'd be happy to see bounded buffering (RxRingBuffer.SIZE sounds a better default size than 1 to me and would be consistent with the request eagerness of other operators like merge for instance). Could we add an overload to allow the specification of buffer size as well?

If this is significant rework I'm content to see this unbounded version documented with its buffering characteristics merged and we can defer the bounding work to another PR.

[akarnokd]

Done.

[davidmoten]

Great, thanks @akarnokd ! Lightning quick as always.

[stevegury]

LGTM 👍

[abersnaze]

The three of you seem happy with it so I'm merging it.