How do i handle sub sequential URL calls in RxJS that pass data to each other?

Question

I have a call to a service, that potentially returns to me a queue URL, if the server is to busy to handle the response.

I'm writing a service in angular, to handle these types of calls, and i'm struggling to figure out what kind of RXJS 6+ operator to handle this for me.

This means that i want to make a fallback, that if the response returns to me a queuing URL, i will subscribe to this call, and retry it until i get an answer.

The wait time can be up to 30 seconds (< insert frustrations here>).

From what i can tell from the rxjs documentation page, i need to use the concatMap operator, and somehow retry the call until i get the correct response? Maybe with some delay operators to restrict the amount of calls?

I found this snippet from https://www.learnrxjs.io/.

Thanks in advance!

Answer 1

This is a recursive call structure, so you need to write a recursive observable. You haven't provided exact response structures so I can't give exact code but at a high level it will look like this:

getQueuedResponse<T>(url) {
  return this.http.get<T>(url).pipe( // fetch the first URL
    switchMap(res => 
      (res.queueUrl) // if queued (your actual queue indicator may be different)
        ? this.getQueuedResponse<T>(res.queueUrl) //then recurse (your actual next url may be different or it may be the original url again)
        : of(res))); // else break (what you actually return here may be different)
}

you can add in a delay if you'd like with a simple timer:

getQueuedResponse<T>(url) {
  return this.http.get<T>(url).pipe( // fetch the first URL
    switchMap(res => 
      (res.queueUrl) // if queued, recurse after 5 seconds
        ? timer(5000).pipe(switchMap(t => this.getQueuedResponse<T>(res.queueUrl))
        : of(res))); // else break
}

Alternatively, if your needs are a little different and you can just call the same URL over and over, you can view this is a polling problem:

pollForResponse<T>(url) {
  return timer(0, 5000).pipe( // start right away then emit every 5 seconds
    switchMap(i => this.http.get<T>(url)), // request the URL
    takeWhile(r => !!r.queued), // keep taking while it's queued
    last() // only emit the last response
  );
}