Filtering (lossy)

Why: user wants to mute notifications for some time. Notifications during enabled mute will be skipped

What: when paused we simply filter out events from Observable source, keeping server connection intact

How: probably, windowToggle operator would be the best fit for this. Using it we can indicate when to pass events from the source, and when to filter them out

source$.pipe(

windowToggle(

offs$,

() => ons$

),

// then flattern window Observables

flatMap(x => x)

)

Play with this example using windowToggle operator in a playground.

Unsubscribe / resubscribe (lossy)

Why: when user switches to another window — we can unsubscribe from notifications stream, therefore lowering network, process and server load. Once user switches back — we restore notifications stream connection

(here source is a cold timer, so upon resubscription it starts emitting from 0)

What: when paused we unsubscribe from Observable and resubscribe on resume

How:

takeUntil with repeatWhen

takeUntil will complete Observable when we pause

and repeatWhen will subscribe again when we resume

source$.pipe(

takeUntil(ons$),

repeatWhen(() => offs$)

)

Play with this example using takeUntil and repeatWhen in a playground.

switchMap with EMPTY or NEVER

when paused we switchMap to an empty Observable

when resumed we switchMap to source stream (this triggers subscription)

pause$.pipe(

switchMap(value =>

value

? EMPTY

: source$

)

)

Play with the example using switchMap operator in a playground.

An article describing this approach with switchMap.

or manual use of subscribe / unsubscribe

Old geek note: there used to be a pausable operator in rxjs 4 that implemented this kind of pausing. Official advice is to use switchMap to achieve this in 5+ versions

Sampling, throttling, debouncing and audition (lossy)

Why: when you don’t want to bother your user with frequent updates — there are several ways to dose amount of messages shown in a given time period

(sampling source every 10ms)

What: every N ms take the latest value emitted since previous sampling

(debounceTime vs throttleTime vs auditTime)

Note: in the marble diagram above I keep source marble colors, while updating their values to represent time when they were emitted. E.g. the second yellow marble from the source was emitted at 5ms, in debounceTime(10) it was emitted at 15ms, in auditTime(10) it was emitted at 10ms, and it was ignored in throttleTime(10) stream.

What: debouncing emits a value if after a given time no other values were emitted. Throttling does exactly opposite: when a value is emitted — ignore consequent emissions for a given time period. And audition acts similar to throttling, though it emits the latest value in a given period, not the first.

How: for all four we have out-of-the-box operators

sample

debounce , debounceTime

, throttle , throttleTime

, audit , auditTime

Play with sample operator in a playground.

Check out this comparison of debounceTime vs throttleTime vs auditTime.

Spacing (lossless)

Why: when push notifications come rapidly — we’d like to ensure that user has time to grasp them. So we make at least 1 sec pause after displaying a notification

What: making at least N-long gap between emissions

How: concatMap with a timer , that starts with source value and completes in N time

source$.pipe(

concatMap(value => timer(N).pipe(

ignoreElements(),

startWith(value)

))

)

Play with this example of spacing values on an observable using concatMap.

Buffering (lossless)

Why: imagine that our client has lost connection to the server, and is trying to send a bunch of messages. We can delay sending those messages until we get connected back

What: when paused we buffer values on the stream, emit buffer once resumed

How:

merge( windowToggle between resume and pause , bufferToggle between pause and resume )

merge(

source$.pipe( bufferToggle(off$, ()=>on$) ),

source$.pipe( windowToggle(on$, ()=>off$) )

).pipe(

// then flatten buffer arrays and window Observables

flatMap(x => x)

)

Run this example using windowToggle with bufferToggle to make pausable buffered Observable.

switchMap pause stream onto buffer when paused, onto raw source on resume

Old geek note: there used to be a pausableBuffered operator in rxjs 4 that implemented this kind of pausing

A weird bonus!

Delaying events on mouse hover with event spacing (lossless)

When mouse hovers a message — we want to pause a stream to give user time to read the message. Notifications stream will be paused until user drives mouse away. Also, all the events are spaced to give user at least 1 second to read the message

Play with this complex example using lots of stuff.

Stackblitz interactive app.

Notes and honorable mentions

Ability of Observable consumer to control producer’s emission rate, time or amount is often called back pressure. So pausing is just a subset of back pressure.

Obviously, we could not cover all back pressure techniques in this article, though here are some I’d like to mention: