feat(operators): combineAll operator

[EarthyOrange]

Description:
An operator to combine values from source observables. The values emitted by early observers are saved until the last observer emits, following which all the previously saved values are emitted.

first$: ---a---b---c--
second$: ---------------d--
result$: [ad, bd, cd]

[EarthyOrange]

@benlesh @cartant Can you please take a look and let me know your thoughts on the changes? Please consider it a draft change. I want a green light before I add more to this PR. Thanks in advance.

[kwonoj]

Same as prior feature implementation requests (#5939 (comment), #5089, and others)

Core team strongly suggest to create new features via userland custom operators instead of including it in core as we generally trying to reduce core api surface, also with pipe it is straightforward to create a new operator as separate library. If a new features should require core lib change we'd like to ask to have some details what part of core lib change is required & usecases it should be included in core.

[EarthyOrange]

@kwonoj I have been implementing operators in my library and would have preferred to do the same for this operator as well. But looking at the implementation of the combineAll operator, it looked like I needed to have access to some internals like OperatorSubscriber and executeSchedule methods. If there is a replacement of these internals, that is already present in the exposed APIs, I will be happy to use those and close this PR.

From what I further understand from your comment is that the direction of this library development now is towards making it a framework which can be used to implement the operators in user libraries or perhaps even make the current operators available in a separate library which would rely on this core library. Is that right?

[benlesh]

It seems like this PR is reimplementing combineLatestAll? I guess I'm not sure I understand this PR. Is there an issue that this is solving?

[benlesh]

Based off of the name, I'm not sure what the intent of this class is. Why is it called "Tetris"?

[EarthyOrange]

@benlesh

Apologies for the poor documentation. I wasn't sure if this change would be accepted so I just over-wrote a deprecated operator for now, to demonstrate the change. It should, of-course, be renamed to something more appropriate and the code broken down and moved to a different folder (Happy to do it).

Setup: On subscription to 2 observables (which have the replay(buffer: 5) and share operators) the first one emits 4 values (a, b, c, d) and the second one emits 1 (e).
Expectation: Since both the hot observables have values ready to be forwarded, the expectation was that combineLatest would give (ae, be, ce, de).
Actual: The received value is (de).

The operator is similar to combineLatestAll, but differs from it by storing all the values that it receives from the time of subscription in the Tetris object.
Consider the example in the description. The combineLatestAll operator would have just emitted the combination cd. But this operator stored the values emitted by the first observable until the second one emitted its first, following which it sequentially emitted all the values from the first observable paired with the one value of the second.

The purpose of the Tetris object is:

• to maintain one array of emitted values per observable (column value in constructor)
• to keep track of the last emitted values (lastEmittedIndices)
• to let us know if there is a new available value (hasNext)
• return an array of most recently emitted values, in case of a new emission (getNext)

I chose the name Tetris because I played a lot of it when I was a kid. When I saw how the values were "falling" and "disappearing" in the columns, it reminded me of it. It can/should be renamed to something more appropriate that betrays its intent a bit more clearly.

[benlesh]

I see. So what you're trying to solve here is the condition where the first values of the combined observables all show up synchronously, and you might miss a few values from the "earlier" observables while waiting for the subsequent values to arrive?

I'm not sure if we can land this in RxJS core, as it's so similar to combineLatest, and probably confusing. I've seen requests for things like this before, but usually it's for people wanting this:

a$:   ---1----2-----4-----5-----
b$;   ------------3----------6----
out$: ---A----B---C--D-----E--F--- 

A: [1, null]
B: [2, null]
C: [2, 3]
D: [4, 3]
E: [5, 3]
F: [5, 6]

And I think that operator already exists in @cartant's rxjs-etc library.

[EarthyOrange]

a$:   ---1----2-----------4-----5-----
b$;   ------------3----------------6----
out$: ------------(AB)----C-----D--E--- 

A: [1, 3]
B: [2, 3]
C: [4, 3]
D: [5, 3]
E: [5, 6]

[EarthyOrange]

@benlesh The example posted in your comment doesn't reflect the way this operator works. Can you please recheck the expected output I have posted and confirm if this operator indeed exists in rxjs-etc?