Mutable-reductions friendly version of reduce (something similar to transduce / RxJava's collect)

[asaf-romano]

When it comes to non-primitive reductions (e.g. reducing an observable into an object or an array), Observable.reduce only works well with immutable reductions. Without a library like ImmutableJS, this is fine only as long as the amount of incoming items is relatively small. For high-volume observables, mutable reductions are a better option. While RxJS technically supports mutable reductions (nothing blocks object reuse in reduce), doing so can lead to terrible bugs when an Observable is subscribed multiple times. Consider the following:

Observable.range(1, 10).reduce((acc, curr) => { acc[curr % 2] = (acc[curr % 2] || 0) + 1; return acc; },  {})
                  .repeat(4)
                  .toArray()
                  .toPromise();

/* => 
 [ { '0': 20, '1': 20 },
  { '0': 20, '1': 20 },
  { '0': 20, '1': 20 },
  { '0': 20, '1': 20 } ]
*/

This is a common issue with mutable reductions that has come up in many libraries, and is very often addressed by providing another reduction operator that - in one way or the other - takes an initial value supplier rather than the initial value itself. This also applies to other Rx implementations, including RxJava, which implements such an operator as "collect" (there it simply takes an initial value supplier) and RxJS 4, which implemented the more powerful/generic/complex transducers protocol.

[asaf-romano]

Just noticed this practice is recommended here:
https://github.com/ReactiveX/rxjs/blob/d58feb7ce4e7c8b8c145f70af4021392e062a947/MIGRATION.md

for toMap and toSet. As you can see above, this operations are unsafe. I would suggest changing those to "No longer implemented".

[asaf-romano]

Working implementation:

export default curry(
    (initialValueSupplier, reducer, source) =>
    {
        return Observable.concat(
            source.pipe(
                scan((accum, curr, idx) => {
                    if (idx == 0)
                    {
                        return reducer(initialValueSupplier(), curr);
                    }
                    return reducer(accum, curr);
                }, null),
                skip(1),
                takeLast(1)
            ),
            Observable.defer(() => Observable.of(initialValueSupplier()))
        ).first();
    });

The additional complexity is necessary in order to avoid calling initialValueSupplier twice without breaking the no-items case (while not delaying the subscription to the source observable).

[BioPhoton]

Hi @asaf-romano :)

I noticed this discussion relates to RxJS major version 4. As the current major version is 6 I guess this issue can be closed.

If yes please do so, if no let me know how I can help further

[0xTomDaniel]

I am trying to use a Set as the accumulator value of the Scan operator. A mutable accumulator would be an amazing option!

[cartant]

@tomdaniel0x01 You can solve the problem using defer in the composition of the observable. There's more information on defer and per-subscription state here.

Closing this as defer can be used to solve the problem and any new reduce-like operator should be introduced as a user-land operator before being considered for inclusion in the core.