Combining two streams in RxJS and terminating the merged stream when a particular input is triggered

Question

Combining two streams in RxJS and terminating the merged stream when a particular input is triggered

I am currently developing an Angular application and working on implementing a system where an NGRX effect will make requests to a service.

This service will essentially perform two tasks:

Firstly, it will check the local cache (sqlite) for the requested data,
Then, it will contact the API to retrieve the most recent data

My goal is to define an observable response from this service that will either emit the API's response alone (in case of a cache miss) or emit the local data followed by the API data. I have illustrated these two scenarios below:

Scenario 1 (API request is first/cache miss)

API   ---1
CACHE ---------2
MERGE ---1

Scenario 2 (Cache hit)

API   ---------1
CACHE ---2
MERGE ---2-----1

I am considering creating a Subject, triggering both calls, and completing the Subject when the API responds (whether successfully or with an error).

Pseudo code:

const subject = new Subject();

this.cache.getData().then(res => subject.next(res)).catch(err => subject.error(err));
this.api.getData().then(res => subject.next(res)).catch(err => subject.error(err)).finally(() => subject.complete());

return subject.asObservable();

However, I am curious if there is a more elegant solution to achieve what I need here. Any suggestions would be greatly appreciated.

angular typescript rxjs observable ngrx

Answer 1

Answer №1

If my understanding is correct, you are looking to make two calls - A) API, B) Cache. If A returns first, emit that result without waiting for the response from B. If B returns first, emit that result and then also emit the response from A when it arrives.

Basically, you always want a response from A. You only want a response from B if it is received before A.

To achieve this, you can define two source observables and use the merge operator.

private apiData$ = defer(() => this.api.getData()).pipe(share());

private cachedData$ = defer(() => this.cache.getData()).pipe(
  filter(existing => !!existing),
  takeUntil(apiData$)
);

data$ = merge(this.cachedData$, this.apiData$);

In this solution, we utilize defer to convert Promises into Observables upon subscription.

apiData$ uses share() to prevent duplicate API calls as we will have two subscriptions.

cachedData$ applies a filter to exclude emissions when data does not exist in the cache.

The takeUntil operator ensures that the cachedData$ observable completes once api$ emits a value, preventing further emissions from cachedData$ after that point.

Lastly, we use merge to combine both sources into a single observable, so that data$ emits whenever either source emits.

You can view a demo on StackBlitz.

Answer 2

If my understanding is correct, you are looking to make two calls - A) API, B) Cache. If A returns first, emit that result without waiting for the response from B. If B returns first, emit that result and then also emit the response from A when it arrives.

Basically, you always want a response from A. You only want a response from B if it is received before A.

To achieve this, you can define two source observables and use the merge operator.

private apiData$ = defer(() => this.api.getData()).pipe(share());

private cachedData$ = defer(() => this.cache.getData()).pipe(
  filter(existing => !!existing),
  takeUntil(apiData$)
);

data$ = merge(this.cachedData$, this.apiData$);

In this solution, we utilize defer to convert Promises into Observables upon subscription.

apiData$ uses share() to prevent duplicate API calls as we will have two subscriptions.

cachedData$ applies a filter to exclude emissions when data does not exist in the cache.

The takeUntil operator ensures that the cachedData$ observable completes once api$ emits a value, preventing further emissions from cachedData$ after that point.

Lastly, we use merge to combine both sources into a single observable, so that data$ emits whenever either source emits.

You can view a demo on StackBlitz.

Combining two streams in RxJS and terminating the merged stream when a particular input is triggered

Answer №1

Similar questions

When updating the data in a datatables.net table within Angular 7, the previous data from the initial table load is retained

Proper positioning of try/catch block in scenarios involving delayed async/await operations

What are the steps for personalizing themes in the Monaco editor?

Cypress - AG-Grid table: Typing command causing focus loss in Cell

Struggling with Primeng's KeyFilter functionality?

Issue with bi-directional data binding in Angular's matInput component

HammerJS swipe functionality does not seem to be functioning properly on elements that have the overflow CSS

When utilizing express-handlebars to render, the error message "req.next() is not a valid function

Using TypeScript to Implement Content Security Policy Nonce

Implementing experimental decorators and type reconciliation in TypeScript - A step-by-step guide

When working with Angular and NGRX, I often wonder if opening my app in multiple tabs within the same browser will result in all tabs pointing to a single NGRX instance

What is the best way to monitor updates made to a function that utilizes firestore's onSnapShot method?

Instance property value driven class property type guard

Is this Firebase regulation accurate and suitable for PUT and GET requests in an Angular and Firebase environment?

Tips for bundling and inlining .json files within an Angular npm package

AmplifyJS is throwing an error: TypeError - It seems like the property 'state' is undefined and cannot be read

Ways to apply the strategy pattern in Vue component implementation

Having trouble retrieving the selected dropdown value in Angular 5 using <p-dropdown> from PrimeNG due to issues with the ngOnInit function. Need some assistance with

Can someone provide guidance on effectively implementing this JavaScript (TypeScript) Tree Recursion function?

What is the most appropriate form to use, and what factors should be considered in determining