What is the best way to link together Angular observables?

Question

What is the best way to link together Angular observables?

In order for my component to make API requests, it needs to check if certain app preferences are set. Currently, I have implemented a method where the component's data is refreshed every 2 minutes using a timer:

ngOnInit(): void {
    this.subscription = timer(0, 120 * 1000).subscribe(() => {
        this.shopService.getPreferencesAsObservable().subscribe(preferences => {
            if(preferences) {
                this.getInitialPendingSlotsOrders();
                this.getInitialPendingNoSlotsOrders();
            }
        });
    });
}

getInitialPendingSlotsOrders(){
    this.apiService.fetchShopOrders("status=PENDING&only_slots=true").subscribe(orders=> {
        /* do stuff with orders */
        /* may need to retrieve additional data */
        if(orders.next_page){
            this.getNextSlotsSetOfPendingOrders();
        }
    });
}

getInitialPendingNoSlotsOrders(){
    this.apiService.fetchShopOrders("status=PENDING").subscribe(orders=> {
        /* do stuff with orders */
        /* may need to retrieve additional data */
        if(orders.next_page){
            this.getNextNoSlotsSetOfPendingOrders();
        }
    });
}

getNextSlotsSetOfPendingOrders() { 
    this.apiService.fetchNextSetOfOrders(this.nextSlotsUrl).subscribe(nextSetOfOrders => {
        /* do stuff with next set of orders */
    })
}

getNextNoSlotsSetOfPendingOrders() { 
    this.apiService.fetchNextSetOfOrders(this.nextNoSlotsUrl).subscribe(nextSetOfOrders => {
        /* do stuff with next set of orders */
    })
}

Although I believed this approach would be effective, there seems to be an issue with extra API calls being made in some cases. I suspect this could be due to chaining observables. How can I improve and optimize this setup?

Appreciate your assistance.

angular typescript rxjs

Answer 1

Answer №1

Having multiple nested subscriptions can result in open subscriptions that may not be closed. It is recommended to utilize RxJS operators to limit it to a single subscription.

If you need to trigger two requests simultaneously, consider using the RxJS forkJoin function.

For more information, check out this link.

To summarize:

switchMap: Maps from one observable to another
filter: Continues the operator chain based on a condition
forkJoin: Combines and triggers multiple observables in parallel

Here's an example to try:

ngOnInit(): void {
  this.subscription = timer(0, 120 * 1000).pipe(
    switchMap(() => this.shopService.getPreferencesAsObservable()),
    filter(preferences => !!preferences),
    switchMap(() => 
      forkJoin({
        slots: getInitialPendingOrders(true),
        noSlots: getInitialPendingOrders(false)
      })
    )
  ).subscribe({
    next: ({ slots, noSlots }) => {
      // handle orders from `slots` and `noSlots`
    },
    error: (error: any) => {
      // handle error
    }
  });
}

getInitialPendingOrders(slots: boolean): Observable<any> {
  return this.apiService.fetchShopOrders("status=PENDING" + (slots ? "&only_slots=true" : ''));
}

Update

It is advisable to return the observable and subscribe only when the response is needed. You can use switchMap for each argument of forkJoin and return an observable conditionally. Use RxJS constant EMPTY to emit results from forkJoin when nothing needs to be returned. Note that forkJoin will emit only when all source observables complete.

ngOnInit(): void {
  this.subscription = timer(0, 120 * 1000).pipe(
    switchMap(() => this.shopService.getPreferencesAsObservable()),
    filter(preferences => !!preferences),
    switchMap(() => 
      forkJoin({
        slots: getInitialPendingOrders(true).pipe(
          switchMap((orders: any) => {
            // perform operations with orders
            return orders.next_page ? this.getNextSlotsSetOfPendingOrders() : EMPTY;
          })
        ),
        noSlots: getInitialPendingOrders(false).pipe(
          switchMap((orders: any) => {
            // perform operations with orders
            return orders.next_page ? this.getNextNoSlotsSetOfPendingOrders() : EMPTY;
          })
        )
      })
    )
  ).subscribe({
    next: ({ slots, noSlots }) => {
      // handle next set of orders from `slots` and `noSlots`
    },
    error: (error: any) => {
      // handle error
    }
  });
}

getInitialPendingOrders(slots: boolean): Observable<any> {
  return this.apiService.fetchShopOrders("status=PENDING" + (!!slots ? "&only_slots=true" : ''));
}

getNextSlotsSetOfPendingOrders(): Observable<any> { 
  return this.apiService.fetchNextSetOfOrders(this.nextSlotsUrl);
}

getNextNoSlotsSetOfPendingOrders(): Observable<any> { 
  return this.apiService.fetchNextSetOfOrders(this.nextNoSlotsUrl);
}

Answer 2

Having multiple nested subscriptions can result in open subscriptions that may not be closed. It is recommended to utilize RxJS operators to limit it to a single subscription.

If you need to trigger two requests simultaneously, consider using the RxJS forkJoin function.

For more information, check out this link.

To summarize:

switchMap: Maps from one observable to another
filter: Continues the operator chain based on a condition
forkJoin: Combines and triggers multiple observables in parallel

Here's an example to try:

ngOnInit(): void {
  this.subscription = timer(0, 120 * 1000).pipe(
    switchMap(() => this.shopService.getPreferencesAsObservable()),
    filter(preferences => !!preferences),
    switchMap(() => 
      forkJoin({
        slots: getInitialPendingOrders(true),
        noSlots: getInitialPendingOrders(false)
      })
    )
  ).subscribe({
    next: ({ slots, noSlots }) => {
      // handle orders from `slots` and `noSlots`
    },
    error: (error: any) => {
      // handle error
    }
  });
}

getInitialPendingOrders(slots: boolean): Observable<any> {
  return this.apiService.fetchShopOrders("status=PENDING" + (slots ? "&only_slots=true" : ''));
}

Update

It is advisable to return the observable and subscribe only when the response is needed. You can use switchMap for each argument of forkJoin and return an observable conditionally. Use RxJS constant EMPTY to emit results from forkJoin when nothing needs to be returned. Note that forkJoin will emit only when all source observables complete.

ngOnInit(): void {
  this.subscription = timer(0, 120 * 1000).pipe(
    switchMap(() => this.shopService.getPreferencesAsObservable()),
    filter(preferences => !!preferences),
    switchMap(() => 
      forkJoin({
        slots: getInitialPendingOrders(true).pipe(
          switchMap((orders: any) => {
            // perform operations with orders
            return orders.next_page ? this.getNextSlotsSetOfPendingOrders() : EMPTY;
          })
        ),
        noSlots: getInitialPendingOrders(false).pipe(
          switchMap((orders: any) => {
            // perform operations with orders
            return orders.next_page ? this.getNextNoSlotsSetOfPendingOrders() : EMPTY;
          })
        )
      })
    )
  ).subscribe({
    next: ({ slots, noSlots }) => {
      // handle next set of orders from `slots` and `noSlots`
    },
    error: (error: any) => {
      // handle error
    }
  });
}

getInitialPendingOrders(slots: boolean): Observable<any> {
  return this.apiService.fetchShopOrders("status=PENDING" + (!!slots ? "&only_slots=true" : ''));
}

getNextSlotsSetOfPendingOrders(): Observable<any> { 
  return this.apiService.fetchNextSetOfOrders(this.nextSlotsUrl);
}

getNextNoSlotsSetOfPendingOrders(): Observable<any> { 
  return this.apiService.fetchNextSetOfOrders(this.nextNoSlotsUrl);
}

Answer 3

Answer №2

It's generally advised not to use the subscribe function inside another subscribe (although there are exceptions).

I suggest taking a look at Is it good way to call subscribe inside subscribe? and familiarizing yourself with various operators such as forkJoin, mergeMap, etc., that RxJs offers for combining observables in different ways based on your specific needs.

Answer 4

It's generally advised not to use the subscribe function inside another subscribe (although there are exceptions).

I suggest taking a look at Is it good way to call subscribe inside subscribe? and familiarizing yourself with various operators such as forkJoin, mergeMap, etc., that RxJs offers for combining observables in different ways based on your specific needs.

Answer 5

Answer №3

There are numerous ways to connect observables, making this a versatile question. For a comprehensive list of combination operators, check out the Combination operators on Learn RxJs.

In my experience with rxjs, I've found it beneficial to break down complex processes into const variables or readonly properties, and then chain them together using a separate observable as needed. This approach helps in maintaining clarity and structure as a project expands. Additionally, it eliminates redundant method calls by executing streams only upon subscription.

In the scenario provided below, after each timer interval, the stream transitions to the getPreferencesAsObservable() method, which likely emits once. Upon emission, the filter() is applied based on valid preferences. The results from both observables retrieving shop orders are combined using forkJoin().

I faced a choice between using concatMap and switchMap. The former ensured that no emissions were skipped even if subsequent emissions differed. On the other hand, with switchMap, any incomplete previous emission would be terminated before the next one. The decision ultimately depends on the desired behavior for the application.

readonly noSlotsProcess$ = 
  this.apiService.fetchShopOrders("status=PENDING").pipe(
    tap((x) => /* do stuff */)
  );
readonly slotsProcess$ = 
  this.apiService.fetchShopOrders("status=PENDING&only_slots=true").pipe(
    tap((x) => /* do stuff */)
  );
readonly refresh$ = timer(0, 120 * 1000).pipe(
  concatMap(() => this.shopService.getPreferencesAsObservable()),
  filter((preferences => !!preferences),
  switchMap(() => { forkJoin( noSlots: this.noSlotsProcess$, slots: this.slotsProcess$ }))
);
ngOnInit(): void {
    this.subscription = this.refresh$.subscribe();
}

Answer 6

There are numerous ways to connect observables, making this a versatile question. For a comprehensive list of combination operators, check out the Combination operators on Learn RxJs.

In my experience with rxjs, I've found it beneficial to break down complex processes into const variables or readonly properties, and then chain them together using a separate observable as needed. This approach helps in maintaining clarity and structure as a project expands. Additionally, it eliminates redundant method calls by executing streams only upon subscription.

In the scenario provided below, after each timer interval, the stream transitions to the getPreferencesAsObservable() method, which likely emits once. Upon emission, the filter() is applied based on valid preferences. The results from both observables retrieving shop orders are combined using forkJoin().

I faced a choice between using concatMap and switchMap. The former ensured that no emissions were skipped even if subsequent emissions differed. On the other hand, with switchMap, any incomplete previous emission would be terminated before the next one. The decision ultimately depends on the desired behavior for the application.

readonly noSlotsProcess$ = 
  this.apiService.fetchShopOrders("status=PENDING").pipe(
    tap((x) => /* do stuff */)
  );
readonly slotsProcess$ = 
  this.apiService.fetchShopOrders("status=PENDING&only_slots=true").pipe(
    tap((x) => /* do stuff */)
  );
readonly refresh$ = timer(0, 120 * 1000).pipe(
  concatMap(() => this.shopService.getPreferencesAsObservable()),
  filter((preferences => !!preferences),
  switchMap(() => { forkJoin( noSlots: this.noSlotsProcess$, slots: this.slotsProcess$ }))
);
ngOnInit(): void {
    this.subscription = this.refresh$.subscribe();
}

What is the best way to link together Angular observables?

Answer №1

Update

Answer №2

Answer №3

Similar questions

The selected value of the PrimeNG p-checkbox cannot be determined using a function when binding to [ngModel]

Angular - Implementing a debounce feature for event handling

Error message in Angular 2: "__generator is not recognized"

Executing a PHP function within a Laravel controller using Ajax

Changing icons within an ngFor loop in Angular 2

Merging RXJS observable outputs into a single result

The JSON object cannot be assigned to the IntrinsicAttributes and customType data types

The ngFor directive in Angular2 consistently collapses the identical <tr> element

Discovering specific values for an ID using API calls in Angular (Implementing CRUD Operations in Angular with API Integration)

Issue: Module '@nrwl/workspace/src/utilities/perf-logging' not found

Customizing the design of the datepicker in Angular and then passing the formatted date to a

angular change digits to Persian script

Deciphering a mysterious message in Typescript while defining a function for one of my tasks

The NX Monorepo housing a variety of applications with unique design themes all utilizing a single, comprehensive UI component

Customizing the appearance of mat-form-field when it is not in focus and has text in the mat-input field

The role of callback functions in TypeScript

Adding a second interface to a Prop in Typescript React: a step-by-step guide

Icon appearing outside the button instead of inside

Automatically generated error notifications for Express-Validator

Issue with TagCloud functionality when deployed on Vercel platform