New Abstractions: Effect, QueueingEffect, SwitchingEffect, BlockingEffect

[deanrad]

RxJS - despite many internal improvements over a dozen years - hasn’t grown much in its level of abstraction, and this keeps it hard to learn, and makes devs inclined to reimplement it. RxJS still deals in pipes, Subjects and obscurely-named operators which no amount of documentation seems to sort out for developers, and I think it can provide a shorter on-ramp than it does.

This PR aims to introduce an abstraction - Effect, and its concurrency-controlled kin like QueueingEffect - that makes the desirable and common use-cases of concurrency control and cancelation easier, with fewer imports, more mnemonic names, and more direct cancelation. (It could be extended to a version that keeps track of state such as StatefulQueueingEffect - but that will be proposed separately).

How does it work? Imagine a function playBellSound that plays a bell ringing sound. We want to use a concatMap style of behavior to reduce concurrency to 1 by enqueueing rings one after another. Under this proposal, this would be:

import { QueueingEffect } from ‘rxjs/effects’

const ringIt = new QueueingEffect<void>(playBellSound);
ringIt.next() // ring 
ringIt.next() // enqueue the next ring

Without this proposal, the status quo is:

import { Subject } from ‘rxjs’;
import { concatMap } from ‘rxjs/operators’;
const ringIt = new Subject<void>();
const rings = ringIt.pipe(
  concatMap(playBellSound)
).subscribe()

ringIt.next() // ring 
ringIt.next() // enqueue the next ring

Only one of these: a) highlights the concurrency mode b) names it plainly c) is free of noise words like pipe and subscribe.

Let’s introduce cancelation, and the contrast will only increase. Under the new proposal:

import { QueueingEffect } from ‘rxjs/effects’

const ringIt = new QueueingEffect<void>(playBellSound);
ringIt.next() // ring 
ringIt.cancelCurrent(); // cancel the current ring (advancing to the next in the queue)

While the status quo would be:

const ringer = new Subject<void>();
const cancels = new Subject<void>();
ringer
  .pipe(
    concatMap(() => {
      return defer(playBellWebAudio).pipe(
        takeUntil(cancels)
      );
    })
  )
  .subscribe();

ringer.next() // ring 
cancels.next(); // cancel the current ring (advancing to the next in the queue)

The number of imports has grown, as well as the noisiness of the code. This shows there is an abstraction just begging to be factored out - why not provide it inside RxJS so that people can have a simpler initial interface?

Lastly, if a user may want to cancel the bell, they may want to cancel the entire queue. This could be simply:

ringIt.cancelCurrentAndQueued(); 

While the raw RxJS of canceling the queue probably involves a new Subject, a switchMap, and other code contortions. (See this StackOverflow thread for community-proposed versions.)

In this proposal, each of the *map operators would have a corresponding effect:

operator	Subject
mergeMap	Effect
concatMap	QueueingEffect
switchMap	SwitchingEffect
exhaustMap	BlockingEffect

They would share the same interface, where the single argument would be a function returning an ObservableInput of a Promise, Observable, etc.. and the return value would be an object with a next() method, which would forward its arguments through to the underlying effect.

Sample (but not minimal) implementation code is provided in the @rxfx/service library RxFx Github - my own ‘contrib’ library. A Service, in RxFx, concurrency controls an effect as in this proposal, but also accepts a reducer to keep track of a state Observable by aggregating over lifecycle events of the effect via scan.

In short - I don’t think the answer to a shorter learning curve for RxJS is only in having better documentation, or having more blog posts claiming to demystify Observables - I think it’s having abstractions that are more self-explanatory.

Let me know if this is something we can consider for an enhancement to RxJS or if there’s another repo you think this would be more suited to. Thanks.

PS Regarding Error handling, it's worth discussing whether an error should finalize the effect (as if it were a Subject), or if the effect should try and be more generous - rescuing errors by default and making them available through some property like an .errors Observable.