I have spent quite some time researching this (for learning purposes) and will try to explain the workings of this code as thoroughly as possible.
First, here’s the original code, annotated:
import { pipe, range, timer, zip } from 'rxjs';
import { ajax } from 'rxjs/ajax';
import { retryWhen, map, mergeMap } from 'rxjs/operators';
function backoff(maxTries, ms) { // (1)
return pipe( // (2)
retryWhen(attempts => range(1, maxTries) // (3)
.pipe(
zip(attempts, (i) => i), // (4)
map(i => i * i), // (5)
mergeMap(i => timer(i * ms)) // (6)
)
)
); // (7)
}
ajax('/api/endpoint')
.pipe(backoff(3, 250))
.subscribe(data => handleData(data));
function handleData(data) {
// ...
}
- Easy enough, we’re creating custom
backoff
operator out of retryWhen
operator. We’ll be able to apply this later within pipe
function.
- In this context,
pipe
method returns a custom operator.
Our custom operator is going to be a modified retryWhen
operator. It takes a function argument. This function is going to be called once — specifically, when this retryWhen
is first encountered/invoked. By the way, retryWhen
gets into play only when the source observable produces an error. It then prevents error from propagating further and resubscribes to the source. If the source produces a non-error result (whether on first subscription or on a retry), retryWhen
is passed over and is not involved.
A few words on attempts
. It’s an observable. It is not the source observable. It is created specifically for retryWhen
. It has one use and one use only: whenever subscription (or re-subscription) to the source observable results in an error, attempts
fires a next
. We are given attempts
and are free to use it in order to react in some way to each failed subscription attempt to the source observable.
So that’s what we are going to do.
First we create range(1, maxTries)
, an observable that has an integer for every retry we are willing to perform. range
is ready to fire all it’s numbers right then and there, but we have to hold its horses: we only need a new number when another retry happens. So, that’s why we...
... zip it with the attempts
. Meaning, marry each emitted value of attempts
with a single value of range
.
Remember, function we’re currently in is going to be called only once, and at that time, attempts
will have only fired next
once — for the initial failed subscription. So, at this point, our two zipped observables have produced just one value.
Btw, what are the values of the two observables zipped into one? This function decides that: (i) => i
. For clarity it can be written (itemFromRange, itemFromAttempts) => itemFromRange
. Second argument is not used, so it’s dropped, and first is renamed into i
.
What happens here, is we simply disregard the values fired by attempts
, we are only interested in the fact that they are fired. And whenever that happens we pull the next value from range
observable...
...and square it. This is for the exponential part of the exponential backoff.
So, now whenever (re-)subscription to source fails, we have an ever increasing integer on our hands (1, 4, 9, 16...). How do we transform that integer into a time delay until next re-subscription?
Remember, this function we are currently inside of, it must return an observable, using attempts
as input. This resulting observable is only built once. retryWhen
then subscribes to that resulting observable and: retries subscribing to source observable whenever resulting observable fires next
; calls complete
or error
on source observable whenever resulting observable fires those corresponding events.
Long story short, we need to make retryWhen
wait a bit. delay
operator could maybe be used, but setting up exponential growth of the delay would likely be pain. Instead, mergeMap
operator comes into play.
mergeMap
is a shortcut for two operators combined: map
and mergeAll
. map
simply converts every increasing integer (1, 4, 9, 16...) into a timer
observable which fires next
after passed number of milliseconds. mergeAll
forces retryWhen
to actually subscribe to timer
. If that last bit didn’t happen, our resulting observable would just fire next
immediately with timer
observable instance as value.
At this point, we’ve built our custom observable which will be used by retryWhen
to decide when exactly to attempt to re-subscribe to source observable.
As it stands I see two problems with this implementation:
As soon as our resulting observable fires its last next
(causing the last attempt to resubscribe), it also immediately fires complete
. Unless the source observable returns result very quickly (assuming that the very last retry will be the one that succeeds), that result is going to be ignored.
This is because as soon as retryWhen
hears complete
from our observable, it calls complete
on source, which may still be in the process of making AJAX request.
If all retries were unsuccessful, source actually calls complete
instead of more logical error
.
To solve both these issues, I think that our resulting observable should fire error
at the very end, after giving the last retry some reasonable time to attempt to do its job.
Here’s my implementation of said fix, which also takes into account deprecation of zip
operator in latest rxjs v6
:
import { delay, dematerialize, map, materialize, retryWhen, switchMap } from "rxjs/operators";
import { concat, pipe, range, throwError, timer, zip } from "rxjs";
function backoffImproved(maxTries, ms) {
return pipe(
retryWhen(attempts => {
const observableForRetries =
zip(range(1, maxTries), attempts)
.pipe(
map(([elemFromRange, elemFromAttempts]) => elemFromRange),
map(i => i * i),
switchMap(i => timer(i * ms))
);
const observableForFailure =
throwError(new Error('Could not complete AJAX request'))
.pipe(
materialize(),
delay(1000),
dematerialize()
);
return concat(observableForRetries, observableForFailure);
})
);
}
I tested this code and it seems to work properly in all cases. I can’t be bothered to explain it in detail right now; I doubt anyone will even read the wall of text above.
Anyway, big thanks to @BenjaminGruenbaum and @cartant for setting me onto right path for wrapping my head around all this.
retryWhen
retries on a criteria -attempts
is the error stream.zip
grabs the next index from the range.map
multiplies` it by itself (that's the exponential part since x * x is just x ** 2), mapMerge waits for the timer to finish before the next attempt continues. Piping it to the ajax will tell it to retry when the ajax errors and then back off exponentially (when the error stream emits an error - the zip continues, the map runs and the mergeMap waits for the timer before continuing). – Benjamin Gruenbaum Oct 26 '18 at 19:33fetch
and implement it without Rx which would be ~5 LoC of regular JavaScript and a for loop with a try/catch :D – Benjamin Gruenbaum Oct 26 '18 at 19:34zip
is working here ... why doesn't it create a new Observable fromzip
each time it resubscribes? – Explosion Pills Oct 26 '18 at 20:42retryWhen
operator only subscribes once to the observable that's composed within it. It feeds any errors received from the source to the observable composed within it - theattempts
parameter to the arrow function is an observable of errors. Essentially, therange
observable 'seeds' thezip
with the number of retries that should be attempted. – cartant Oct 26 '18 at 23:10zip
operator grabs therange
observable (1, 2, 3) and merges it with the inputattempts
? But why do this? I removedzip
from thepipe
and the function seems to have run the same way. So what was the point of zipping? Can’t we just ignoreattempts
? – CBlew Oct 27 '18 at 10:44