Filtering RxJS stream after emission of other observable until timer runs out

user_14388607 0
Viewed 356

I want to achieve the following behavior in RxJS but could not find a way using the available operators:

  • Stream A: Generated by a continuous stream of events (e.g. browser scroll)
  • Stream B: Generated by another arbitrary event (e.g. some kind of user input)
  • When B emits a value, I want to pause the processing of A, until a specified amount of time has passed. All values emitted by A in this timeframe are thrown away.
  • When B emits another value during this interval, the interval is reset.
  • After the interval has passed, the emitted values of A are no longer filtered.
// Example usage.
streamA$
  .pipe(
    unknownOperator(streamB$, 800),
    tap(val => doSomething(val))
  )
// Output: E.g. [event1, event2, <skips processing because streamB$ emitted>, event10, ...]

// Operator API.
const unknownOperator = (pauseProcessingWhenEmits: Observable<any>, pauseIntervalInMs: number) => ...

I thought that throttle could be used for this use case, however it will not let any emission through, until B has emitted for the first time (which might be never!).

streamA$
  .pipe(
    // If B does not emit, this never lets any emission of A pass through!
    throttle(() => streamB$.pipe(delay(800)), {leading: false}),
    tap(val => doSomething(val))
  )

An easy hack would be to e.g. subscribe manually to B, store the timestamp when a value was emitted in the Angular component and then filter until the specified time has passed:
(obviously goes against the side-effect avoidance of a reactive framework)

streamB$
  .pipe(
    tap(() => this.timestamp = Date.now())
  ).subscribe()

streamA$
  .pipe(
    filter(() => Date.now() - this.timestamp > 800),
    tap(val => doSomething(val))
  )

I wanted to check with the experts here if somebody knows an operator (combination) that does this without introducing side-effects, before I build my own custom operator :)

rxjs angular javascript
1 Answers

I think this would be an approach:

bModified$ = b$.pipe(
  switchMap(
    () => of(null).pipe(
      delay(ms),
      switchMapTo(subject),
      ignoreElements(),
      startWith(null).
    )
  )
)

a$.pipe(
  multicast(
    new Subject(),
    subject => merge(
      subject.pipe(
        takeUntil(bModified$)
      ),
      NEVER,
    )
  ),
  refCount(),
)

It may seem that this is not a problem whose solution would necessarily involve multicasting, but in the above approach I used a sort of local multicasting.

It's not that expected multicasting behavior because if you subscribe to a$ multiple times(let's say N times), the source will be reached N times, so the multicasting does not occur at that level.

So, let's examine each relevant part:

multicast(
  new Subject(),
  subject => merge(
    subject.pipe(
      takeUntil(bModified$)
    ),
    NEVER,
  )
),

The first argument will indicate the type of Subject to be used in order to achieve that local multicasting. The second argument is a function, more accurately called a selector. Its single argument is the argument specified before(the Subject instance). This selector function will be called every time a$ is being subscribed to.

As we can see from the source code:

selector(subject).subscribe(subscriber).add(source.subscribe(subject));

the source is subscribed, with source.subscribe(subject). What's achieved through selector(subject).subscribe(subscriber) is a new subscriber that will be part of the Subject's observers list(it's always the same Subject instance), because merge internally subscribes to the provided observables.

We used merge(..., NEVER) because, if the subscriber that subscribed to the selector completes, then, next time the a$ stream becomes active again, the source would have to be resubscribed. By appending NEVER, the observable resulted form calling select(subject) will never complete, because, in order for merge to complete, all of its observables have to complete.

subscribe(subscriber).add(source.subscribe(subject)) creates a connection between subscribed and the Subject, such that when subscriber completes, the Subject instance will have its unsubscribe method called.

So, let's assume we have subscribed to a$: a$.pipe(...).subscribe(mySubscriber). The Subject instance in use will have one subscriber and if a$ emits something, mySubscriber will receive it(through the subject).

Now let's cover the case when bModified$ emits

bModified$ = b$.pipe(
  switchMap(
    () => of(null).pipe(
      delay(ms),
      switchMapTo(subject),
      ignoreElements(),
      startWith(null).
    )
  )
)

First of all, we're using switchMap because one requirement is that when b$ emits, the timer should reset. But, the way I see this problem, 2 things have to happen when b$ emits:

  • start a timer (1)
  • pause a$'s emissions (2)

(1) is achieved by using takeUntil in the Subject's subscribers. By using startWith, b$ will emit right away, so that a$'s emissions are ignored. In the switchMap's inner observable we're using delay(ms) to specify how long the timer should take. After it elapses, with the help of switchMapTo(subject), the Subject will now get a new subscriber, meaning that a$'s emissions will be received by mySubscriber(without having to resubscribe to the source). Lastly, ignoreElements is used because otherwise when a$ emits, it would mean that b$ also emit, which will cause a$ to be stopped again. What comes after switchMapTo(subject) are a$'s notifications.

Basically, we're able to achieve the pausable behavior this way: when the Subject instance as one subscriber(it will have at most one in this solution), it is not paused. When it has none, it means it is paused.

EDIT: alternatively, you could have a look at the pause operator from rxjs-etc.

user_96326210
Related