How to create a conditonal polling operator in Rxjs

Question

I need to create a polling operator using Rxjs, that does not use a time source (e.g. timer or interval). This will be used to poll a third party api that is wildly inconsistent with the amount of time it takes to return a value. I only want to poll again if the value returned is empty.

What I've tried so far is

this.someService.getStoredValue().pipe(
  repeatWhen(newPoll => newPoll.pipe( // delay each resubmission by two seconds
    delay(2000)
  )),
  pluck('Data), // only interested in the Data property of the returned object
  filter(isNotNull => isNotNull !== ''), 
  timeout(20000), // if gone 20 seconds without a successfull polling attempt, throw error
).subscribe({
 next: value => {
    // Only if polling is succesfull
   },
 error: err => {
      // I want to end up here if the api can not be reached, or the timeout occurs.
   },
 complete: () => console.log('completed')
})

I expect the output of the operator to try polling, for instance 10 times. If the Data property of the returned object has a non-empty value, unsubscribe from the polling service. If 10 attempts has been made, unsubscribe and emit an error.

But what happens is that the subscription goes on for infinity if the polling is never successful.

Perhaps you want this answer. It uses timer but I'm not sure why you're asking not to include this. — Andrew Allen
I would definitely recommend rxjs-take-while-inclusive for this: npmjs.com/package/rxjs-take-while-inclusive — jburtondev
interval(1000).pipe( takeWhileInclusive(v => v < 5), ).subscribe(console.log); — jburtondev
@AndrewAllen Hi, will definitely check it out. The reason why I do not want to rely on a timer, is in the case the 3rd party api takes more than one second to reply. This would result in my polling function to poll again, when the 3rd party is still formulating a valid reply. — Anders
@Anders in that case, link is not too relevant - it's more about timing when to retry on failures. I would use mbojko answer (perhaps with linked retryWhen(generalRetryStrategy()) to add delay on failures) — Andrew Allen

mbojko mbojko · Accepted Answer · 2019-10-16T08:34:28

It seems like a retry case, with a catch that an empty answer should be treated as an error. Something like:

this.someService.getStoredValue().pipe(
    pluck('Data'),
    flatMap(value => !!value ? of(value) : throwError("Empty!")),
    retry(10),
)

How to create a conditonal polling operator in Rxjs

2 Answers