How can I increase a <number> value of a ReplaySubject?-CodePudding

I'm using a BehaviourSubject but i'm finding it difficult to increase or decrease that number

I'm either creating an infinite loop using:

this.observer.pointsCollected$.subscribe(val => this.observer.pointsCollected$.next(val   10));

Or i'm only getting a single update using:

this.observer.pointsCollected$.pipe(take(1)).subscribe(val => this.observer.pointsCollected$.next(val   10));

I'm creating the observable using:

public pointsCollected$ = new ReplaySubject<number>();

I think I have solved this using a BehaviourSubject instead and using BehaviourSubject.getValue()

Please let me know if this is a good approach or not

CodePudding user response：

It's important to clearly define the desired behavior. A ReplaySubject (in the observable sense) doesn't have a value, it's merely an object that emits values to its subscribers.

To say "increase a value of a ReplaySubject" isn't quite precise enough.

Maybe you want to "emit the previously emitted value increased by a specific amount"

This sounds like a perfect use case for the scan operator as it will emit values based on the currently received value and the prior emitted value.

To set the up we can have a Subject that emits the "increment" to be added and derive an observable that emits the calculated value:

increment$ = new BehaviorSubject<number>(DEFAULT_NUMBER);
  
number$ = increment$.pipe(
  scan((previous, increment) => previous   increment)
);

We can now subscribe to number$ and initially receive the default value. Any time increment$.next() is called, number$ will emit the sum of the previous emission and the new increment.

In Angular, you could wrap this into a service and provide a public method to increase the number:

export class NumberService {

  private increment$ = new BehaviorSubject<number>(DEFAULT_NUMBER);
  
  public number$ = this.increment$.pipe(
    scan((previous, increment) => previous   increment)
  );

  public increaseNumber(amount: number) {
    this.increment$.next(amount);
  }

}

Here's a working StackBlitz demo.

CodePudding user response：

console.clear();
import { ReplaySubject } from 'rxjs';

const sub = new ReplaySubject(3);
let inc:any=0;
sub.next(1);
sub.next(9);
sub.subscribe((res) => {
  inc = inc   res;
});
console.log(inc);

Others think to use looping && Recursive Functions.

CodePudding user response：

The first approach is wrong because as you said, that's an infinite loop.

The second one I think its the best way "to increase the value of a ReplaySubject": it's getting just 1 single value (the current), and then incrementing it just once.

If you want to have a "signal" that triggers that, you could just wrap it around a Subject:

const performIncrement$ = new Subject<number>();

performIncrement$.pipe(
  withLatestFrom(pointsCollected$),
  map(([increment, v]) => v   increment)
).subscribe(newValue => pointsCollected$.next(newValue))

// Call performIncrement$ every time you want to increment
performIncrement$.next(10)

But in my opinion it's better to declare observables reactively, avoid ReplaySubjects/BehaviourSubjects as much as posible - it's a bit challenging first because instead of thinking the imperative way of "when this happens I need to increase pointsCollected$ by 10" you need to think the other way around "the value of pointsCollected$ is how many times this has happened multiplied by 10". This example would become something like:

const performIncrement$ = new Subject<number>();

const pointsCollected$ = performIncrement$.pipe(
  scan((acc, increment) => acc   increment, 0),
  startWith(0)
);

If you have multiple subscribers and want to share the same value, you can use a shareReplay(1) to keep the latest value - Similar to a ReplaySubject but without having to .next on it.

And ideally you wouldn't have performIncrement$ as a Subject either, it would be another observable that emits whenever that should happen based on your logic.