Parallel Ajax requests with Rxjs
I am currently moving from promises to observables. I am using Redux-Observable for my application app. Basically, I'm looking for a better operator that will include mutliple, concurrent ajax calls, and return responses when all observables have completed execution successfully. Here is a code snippet from my application.
let epicPostAd = (action$, store, {ajax}) =>
action$.ofType(POST_AD)
.debounceTime(1000)
.mergeMap(({ payload }) =>
ajax(generateAjaxRequestSetting(POSTS_URL, 'post', payload,CT_JSON))
.map(response => postAdSuccessful(response))
.catch(e => Observable.of(postAdUnsuccessful(e.xhr.response)))
.takeUntil(action$.ofType(LOCATION_CHANGE))
)
This is a simple ajax request that sends messages and dispatches POST_AD_SUCCESSFUL
when the response is still 201 sends POST_AD_UNSUCCESSFUL
on error. But the problem is I want to make a follow-up observable ajax stream when there is a response. For example,
.map(response => /* start a stream of ajax observables then process the response */)
I would appreciate it if you could show me the best way to achieve this.
source to share
It looks like you are looking for an operator forkJoin
.
It will subscribe to whatever Observable data you pass to it, and after all completes , it will output the last value from each one inside the array.
It's not entirely clear where in your Epic you wanted to do this, so I just made a generic example:
const somethingEpic = (action$, store, { ajax }) =>
action$.ofType(SOMETHING)
.mergeMap(() =>
Observable.forkJoin(
ajax('/first'),
ajax('/second'),
ajax('/third')
)
.do(results => {
// the results is an array, containing each
const [first, second, third] = results;
console.log(first, second, third);
})
.map(results => ({
type: 'SOME_RESULTS',
results
}))
);
Technically, it supports a final argument resultSelector
that you can use instead of using the operator map
after it, but I tend not to use it because I found it less clear with minor performance benefits in the overall shortening of served styles. But it's still good to know. May be useful for "normalizing data" rather than "turning it into action".
const somethingEpic = (action$, store, { ajax }) =>
action$.ofType(SOMETHING)
.mergeMap(() =>
Observable.forkJoin(
ajax('/first'),
ajax('/second'),
ajax('/third'),
results => ({
type: 'SOME_RESULTS',
results
})
)
);
ALSO if you are asking yourself, "Which operator am I using?" you should try the master operator located in the documentation: http://reactivex.io/rxjs/
Scroll down to the part that reads:
Do you need to find an operator for your problem? Start by choosing an option from the list below:
- I have one existing Observable and ...
- I have some Observables to be bundled together as one Observable and ...
- I don't have Observables yet and ...
Hint: Open DevTools to experiment with RxJS.
Although it is correctly suggested in this case forkJoin
, but when you click on it, it is not yet documented: sadface: But a google search will present many different sites explaining what it does and how to use it (in RxJS and other Rx implementations on others languages). Like this helpful website
source to share
Here is the answer to my question. Although JayPelps answered, I realized my question was not that clear. Using Jay's recommendation. I came up with the following:
let epicPostAd = (action$, store, {ajax, observable}) =>
action$.ofType(POST_AD)
.debounceTime(1000)
.mergeMap(({ payload }) =>
ajax(generateAjaxRequestSetting(POSTS_URL, 'post', payload, CT_JSON))
.mergeMap(response =>
observable.forkJoin(
ajax(''),
ajax('')
)
.do(res => {
const [first, second, third] = results;
console.log(first, second, third);
})
.map(res => postAdSuccessful(res))
)
.catch(e => observable.of(postAdUnsuccessful(e.xhr.response)))
.takeUntil(action$.ofType(LOCATION_CHANGE))
)
So how does it work. I make a post request and right after the ajax request completes, I .mergeMap
reply to the ovservables ajax stream with .forkJoin()
. Then process the results
source to share