RxJS Subject conditional next() update, based on current-new value

user_13994898 0
Viewed 311

I have an angular application that stores some data in a Subject

dataSubject = new Subject<SomeDataType>();

I have two different ways that the data gets updated from the server which look like

connection1.on('messageReceived', (newData: SomeDataType) => this.dataService.dataSubject.next(newData))

connection2.on('messageReceived', (newData: SomeDataType) => this.dataService.dataSubject.next(newData))

The problem is that the newData that arrives last, is not necessarily the latest. Each data object contains a timestamp that I want to use.

Is there a way to make the Subject only accept values that have newData.timestamp > currentData.timestamp? Something like a conditional next()

rxjs angular
1 Answers

sounds like a case for the scan operator:

latestData$ = this.dataSubject.pipe(
  // scan will give you the last emitted value and the current value coming through
  // so you can compare them and select which to emit
  scan((last, current) => {
    if (last.timestamp > current.timestamp) {
      return last
    }
    return current;
  }),
  // optional operator to prevent emitting the same value twice 
  distinctUntilChanged()
);

latestData$ will only emit values when the timestamp is greater than the previous emitted value with this set up.

user_48553060
Related