RxJS Operator to delay handling subsequent events

Question

I'm looking for an RxJS operator (or a combination of operators) which will allow me to achieve this:

Every event fired by the observable should be handled by the order it was triggered (no skipping, throttling, debouncing any of the values).
I should be abled to define a set interval between the handling of 2 subsequent events.
In case no previous events are currently handled, the next event should be handled immediately (no delay).

I created the following example which I hope will make this clearer:

import { Component, OnInit } from "@angular/core";
import { Subject } from "rxjs";
import { delay } from "rxjs/operators";

@Component({
  selector: "app-root",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent implements OnInit {
  counter = 1;
  displayedValue = 1;

  valueEmitter = new Subject();
  valueEmitted$ = this.valueEmitter.asObservable();

  ngOnInit() {
    // I want to allow each incremented value to be displayed for at least 2 seconds
    this.valueEmitted$.pipe(delay(2000)).subscribe((value) => {
      this.displayedValue = value;
    });
  }

  onClick() {
    const nextCounter = ++this.counter;
    this.counter = nextCounter;
    this.valueEmitter.next(nextCounter);
  }
}

In this example, I would like each value incremented in the counter to be shown as the displayedValue for at least 2 seconds.

So the first time the user clicks onClick the displayed value should immediately change from 1 to 2 but in case the user keeps on rapidly incrementing the counter the displyedValue should follow along in a delayed manner allowing each of the previously incremented numbers to be displayed for 2 seconds before changing, slowly catching up with the current counter value.

Can this be achieved with RxJs operators?

UPDATE

For the moment I came up with the following solution:

import { Component, OnInit } from "@angular/core";
import { Subject } from "rxjs";

@Component({
  selector: "app-root",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent implements OnInit {
  counter = 1;
  displayedValue = 1;
  currentDisplayedValueStartTime = 0;
  readonly DISPLAY_DURATION = 2000;
  valuesToDisplay = [];
  displayedValueSwitcherInterval: number | null = null;

  valueEmitter = new Subject();
  valueEmitted$ = this.valueEmitter.asObservable();

  ngOnInit() {
    this.valueEmitted$.pipe().subscribe((value) => {
      this.handleValueDisplay(value);
    });
  }

  onClick() {
    const nextCounter = ++this.counter;
    this.counter = nextCounter;
    this.valueEmitter.next(nextCounter);
  }

  handleValueDisplay(value) {
    this.valuesToDisplay.push(value);

    if (!this.displayedValueSwitcherInterval) {
      this.displayedValue = this.valuesToDisplay.shift();

      this.displayedValueSwitcherInterval = window.setInterval(() => {
        const nextDiplayedValue = this.valuesToDisplay.shift();
        if (nextDiplayedValue) {
          this.displayedValue = nextDiplayedValue;
        } else {
          clearInterval(this.displayedValueSwitcherInterval);
          this.displayedValueSwitcherInterval = null;
        }
      }, this.DISPLAY_DURATION);
    }
  }
}

It's not what I was looking for since it relies on the component state and on setInterval in order to manage the notification queue. I was hoping to find something cleaner and more declarative which relies on RxJs operators to handle my notifications stream, and their interval queue - but at least I got the feature to work exactly as I wanted. Still open to suggestions on how to make this better.

user_9632621 · Answer

I think this could be a way to solve it:

this.valueEmitted$
  .pipe(
    concatMap(
      (v) => concat(
        of(v),
        timer(2000).pipe(ignoreElements())
      )
    )
  )
  .subscribe((value) => {
    this.displayedValue = value;
  });

Because timer also emits some values(0, 1, 2 and so forth), we use ignoreElements since we're not interested in those values and all we want from timer is the complete notification which will indicate the moment when the next inner observable can be subscribed. Note that the beforementioned inner observable is created by invoking the concatMap's callback function with v as the argument.

Working demo.

Every event fired by the observable should be handled by the order it was triggered (no skipping, throttling, debouncing any of the values).

This is ensured with the help of concatMap.

I should be able to define a set interval between the handling of 2 subsequent events.

We can achieve that with:

concat(
  // First, we send the value.
  of(v),
  // Then we set up a timer so that in case a new notification is emitted, it will have to
  // wait for the timer to end.
  timer(2000).pipe(ignoreElements())
)

In case no previous events are currently handled, the next event should be handled immediately (no delay).

If there are no active inner observables created by concatMap, then the value v will be sent immediately to the consumer.

This would also work:

this.valueEmitted$
  .pipe(
    concatMap(
      (v) => timer(2000).pipe(
        ignoreElements(), startWith(v)
      )
    )
  )
  .subscribe((value) => {
    this.displayedValue = value;
  });

user_7493840 · Answer

So your comment made it a bit more clear.

Obviously, you need some state. Because you can't display everything at once, user's screen is not infinite. It can't handle more than.. 3.. 4.. 5 notification at once.

scan is the way you handle state within the rxjs emitting each event.

Your scan may hold two lists: one for those notification which are currently rendered and the other one for pending notifications. Lists do preserve order as well as scan does, so ordering won't be violated.

Then you need to be able to act differently based on your events. The simplest thing I can imagine is tuple (event_type_as_sting_or_number, event_specific_params). Having that you will be able to correctly compute your next state (which will be emitted by the scan) and then handle it gracefully (like rerender notifications).

Seems you need three events:

New event has came from the outer space.
Some notification has expired and must be taken off the screen.
User acked some notification and it does not make any sens to keep it.

So the entire pipeline is: merge all your events into a single stream, having them "marked" by event_type; scan on that stream with [], [] as your initial state (no pending, nor active notifications yet); if else if else based on your event_type within the scan to compute next state correctly; take an outcome and render all active notifications.

From the comments:

I still don't see how I manage the intervals in this solution. I mean, how do I move a notification from the pending to the active array, only after a set interval in which the currently active notification has been displayed.

Basically, two options. Since you're not strictly bound to a given limit for displaying some notification you can simply have global interval that emits, let's say, every your_limit / 2 time units. Each notification you have may be extended with simple timestamp at the moment you move it to the list of active notifications. Next time your interval fires, you loop over that list and filter out all the notifications with time delta >= your_limit. That's a bit inaccurate, but can be easily configured and adjusted to be transparent for your users. I vote for this solutions for its simplicity.

The other solution is to fire timer per each notification once placed on the active list. Once timer done, you send an event (notification_must_die, { notification_id: ... }) and your scan handles that; it must be prepared that such notification_id does not exist any more (it could be cancelled right before for example). It's complicated, but much more accurate.

RxJS Operator to delay handling subsequent events

2 Answers