Ping API continuous with RXJS, timeout cancels?

user_5657968 0
Viewed 183

I try to ping a URL/API continuous with rxjs.

My first try was:

timer(3000, 2000)
   .pipe(mergeMap(() => this._http.get(environment.pingUrl)))
   .subscribe(res => console.log(res), err => console.log(err))

But when the URL is not reachable i never get a response. So I decide to set a timeout:

timer(3000, 2000)
   .pipe(mergeMap(() => this._http.get(environment.pingUrl).pipe(timeout(5000)))
   .subscribe(res => console.log(res), err => console.log(err))

Now I get the first timeout/response when the api is not reachable, but then it seems the timeout trigger an unsubscribe for the timer and there is no further response. I have no idea to prevent this unsubscribe.

observable rxjs angular
2 Answers

From what I understand, you would like to ping an API continuously and keep going when this API is not reachable.

The problem you have is, when the api fail, the observable is completed, and won't trigger again.

You will need to use retry or retryWhen

and simply do

timer(2000, 3000)
  .pipe(
    mergeMap(() => this._http.get(environment.pingUrl)),
    timeout(5000),
    retryWhen(() => timer(3000))
  )
  .subscribe(
    (res) => console.log(res),
    (err) => console.log(err)
  );

ex : https://stackblitz.com/edit/rxjs-playground-test-mgppcj

Be sure to unsubscribe for this observable at some point tho.

user_31905320

About Errors

In RxJS, Observables have three types of emissions.

  1. next: A value in the steam, there can be 0 to arbitrarily many of these
  2. error: A terminal emission. This instance of the observable has an error and is closed. It can never emit again.
  3. complete: A terminal emission. This instance of the observable is done and closed. It will never emit again.

You'll notice that an instance of an observable can never emit more than one error or complete emission, but any observable can be run (or be restarted/ retried) any number of times. Every time you subscribe or some operator subscribes for you, you create a new instance of an observable.

Creating a stream that emits errors, but still continues afterwards

Since observables can never emit after an error, the only way to emit errors while continuing is to catch the error emission and turn it into a next emission.

An Example:

In this example, any time the source observable errors, I tell catchError to resubscribe to the source.

  • I emit regular emissions inside an object with a {value: } property
  • I emit errors inside an object with an {error: } property.

I deal with these caught errors by just printing them to the console, but you'll likely want to do something nicer than that :)


timer(3000, 2000).pipe(
  mergeMap(_ => this._http.get(environment.pingUrl)),
  map(value => ({value})),
  timeout({each: 5000}),
  catchError((error, src) => {
    // Only catch/handle TimeoutError
    if(error instanceof TimeoutError){
      return src.pipe(startWith({error}));
    } else {
      return throwError(() => error);
    }
  })
).subscribe({
  next: emitted => {
    if("error" in emitted){
      // Found an error! In this case, we only handled
      // TimeoutErrors, so that'll appear here as a next
      // emission. The observable is still running.
      console.log("Caught an error: ", emitted.error);
    }else{
      // got a result!
      console.log(emitted.value);
    }
  },
  // All other errors appear here and are error emissions, 
  // so they're terminal. This observable is closed
  error: err => console.log("Uncaught error: ", err),
  // The complete emission is also terminal. This 
  // observable is now closed.
  complete: () => console.log("Complete")
});
user_135009860
Related