Consider a BehaviorSubject that is holding an array of objects. This subject is supposed to be subscribed in multiple places at once i.e multiple subscriptions to same subject. It receives data from backend via polling. Since polling same api can result in same data, I'm comparing the new response from backend with existing data of Subject using BehaviorSubject.value as shown below

interface Something {

}
class SomeService implements OnInit, OnDestroy {

    private readonly subject = new BehaviorSubject(undefined);
    private pollingSubscription: Subscription;

    constructor(private readonly httpClient: HttpClient) {

    }

    ngOnInit() {
        this.initPolling();
    }

    private initPolling(): void {
        this.pollingSubscription = timer(1000).pipe(
            concatMap(() => this.getDataFromBackend()),
            tap((newResponse) => {
                this.broadcastData(newResponse);
            })
        )
    }

    private getDataFromBackend(): Observable<Array<Something>> {
        return this.httpClient.get(url, options).pipe(
            map((response) => {
                if (response?.length > 0) {
                    return response;
                }
                return undefined;
            }),
            catchError(() => of(false))
        )
    }

    /**
     * Function to push data to subject only if data has changed
     */
    private broadcastData(newData: Array<Something>): void {
        if (!Lodash.isEqual(newData, this.subject.value)) {
            this.subject.next(newData);
        }
    }

    /**
     * Function to get observable of data.
     * Observable is returned so that calling function/component/service can chain observables
     * and create better workflows
     */
    private getDataObs(): Observable<Something> {
        return this.subject.asObservable();
    }

    ngOnDestroy() {
        if (this.subject) {
            this.subject.unsubscribe();
            this.subject = undefined;
        }
        if (this.pollingSubscription) {
            this.pollingSubscription.unsubscribe();
            this.pollingSubscription = undefined;
        }
    }
}

My teammates saw Ben Leah's comment on this thread How to get value from Subject

Based on @BenLesh's answer on this thread, my team is highly discouraging the use of .value or .getValue() for data comparison. If I have a service whose subject is initialised when class is loaded and is unsubscribed when class is destroyed (ngOnDestroy), is there any problem in using .value to compare data before calling .next()?

FYI, I try not to use distinctUntilChanged because if I have 1 subject subscribed to multiple subscriptions, it will trigger comparison multiple times (I.e. 1 time for each subscription).

CodePudding user response：

You can use pairwise() operator and compare the two values

documentation: https://rxjs.dev/api/operators/pairwise

concatMap(() => this.getDataFromBackend()),
pairwise(),
switchMap(([oldResponse, newResponse]) => {
    // ...logic
    if (!Lodash.isEqual(oldResponse, newResponse)) {
        this.subject.next(newResponse);
    }
})

demo: https://stackblitz.com/edit/jmumnz-amtlq4?file=index.ts

CodePudding user response：

The crux of Ben Lesh's answer is that using .value is a sign that you're not using RxJS to it's best ability.

If you're using getValue() you're doing something imperative in declarative paradigm.

To a lesser extent, that's true with Subjects in general. They're typically used for either of two purposes. Multicasting, or bridging between imperative and declarative code.

All you need here is the multi-casting component. In most cases, you can use a operator (they use subjects under the hood) to do that for you.

A lot of your song and dance here is to implement distinctUntilChanged declaratively. In so doing, you have created a version that is both much slower (shouldn't matter here) and much harder to maintain (should matter here).

Here is how I might refactor your code (using shareReplay & distinctUntilChanged) to be a bit more in line with dogmatic RxJS.

interface Something {
  length: number
}

class SomeService implements OnInit, OnDestroy {

  /* Errors are "false", Data without a length is "undefined", and 
    everything else is "something". I wouldn't reccomend this,
    but as an example, sure.
  */
  private dataOb$: Observable<(Something | Boolean | undefined)[]>
  private pollingSubscription: Subscription;

  constructor(private readonly httpClient: HttpClient) {

  }

  ngOnInit() {
    this.dataOb$ = timer(0,1000).pipe(
      concatMap(() => this.getDataFromBackend()),
      distinctUntilChanged(Lodash.isEqual),
      shareReplay(1) // multicasting
    )

    // This service is effectively "using" itself. This means
    // the polling continues even if nobody else is listening.
    this.pollingSubscription = this.getDataObs().subscribe()
  }

  private getDataFromBackend(): Observable<Array<Something | Boolean | undefined>> {
    // This is a bizzaar function, but I assume it's just as an example
    return this.httpClient.get(url, options).pipe(
      map((response: Something) => {
        if (response?.length > 0) {
          return response;
        }
        return undefined;
      }),
      catchError(() => of(false))
    )
  }

  private getDataObs(): Observable<(Something | Boolean | undefined)[]> {
    return this.dataOb$
  }

  ngOnDestroy() {
    this.pollingSubscription.unsubscribe();
  }
}