RXJS retryWhen reset waiting interval

Question

I want to trigger retrywhen() with increasing time interval,

   socketResponse.retryWhen(attempts => {
    return attempts.zip(Observable.range(1, 4)).mergeMap(([error, i]) => {
        console.log(`[socket] Wait ${i} seconds, then retry!`);
        if (i === 4) {
            console.log(`[socket] maxReconnectAttempts ${i} reached!`);
        }
        return Observable.timer(i * 1000);
    });
});

above code works fine. current implementation output :

on connection error (1st time)

---

• [socket] Wait 1 seconds, then retry! // wait for 1 second
• [socket] Wait 2 seconds, then retry! // wait for 2 seconds

on connection successful

---

// successful connection.

on connection error (2nd time)

---

• [socket] Wait 3 seconds, then retry! // wait for 3 second
• [socket] Wait 4 seconds, then retry! // wait for 4 seconds

Now I want to reset waiting time when socket connection is successful.

desired output :

on connection error (1st time)

---

• [socket] Wait 1 seconds, then retry! // wait for 1 second
• [socket] Wait 2 seconds, then retry! // wait for 2 seconds

on connection successful

---

// successful connection.

on connection error (2nd time)

---

• [socket] Wait 1 seconds, then retry! // wait for 1 second

• [socket] Wait 2 seconds, then retry! // wait for 2 seconds

but I don't know how to reset retrywhen() time interval.

Answer 1

I also had the same problem. My goal was to wrap a socket connection with observables. I wanted the network observable to never complete (unless specifically asked to) and infinitely retry connecting in case of any error.

Here's how I accomplished this (with rxjs version 6+). This my retryWhen operator, retrying infinitely while adding a scaling wait duration scalingDuration topped at maxDuration.

import { Observable, timer } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

export const infiniteRetryStrategy = (opts: { scalingDuration: number, maxDuration: number }, attempt?: () => number) => (attempts: Observable<{}>) => {
    return attempts.pipe(
        mergeMap((error, i) => {
            // use attempt if we have it, otherwise use the number of errors received
            const retryAttempt = attempt ? attempt() : i + 1;
            const waitDuration = Math.min(opts.maxDuration, retryAttempt * opts.scalingDuration);

            console.error(`Got error: ${error}. Retrying attempt ${retryAttempt} in ${waitDuration}ms`);
            return timer(waitDuration);
        })
    );
};

Usage is pretty straight forward. If you provide the attempt function, you can manage the retry number externally and you can therefore reset it on a next tick.

import { retryWhen, tap } from 'rxjs/operators';

class Connection {
    private reconnectionAttempt = 0;

    public connect() {
        // replace myConnectionObservable$ by your own observale
        myConnectionObservable$.pipe(
            retryWhen(
                infiniteRetryStrategy({
                    scalingDuration: 1000,
                    maxDuration: 10000
                }, () => ++this.reconnectionAttempt)
            ),
            tap(() => this.reconnectionAttempt = 0)
        ).subscribe(
            (cmd) => console.log(cmd),
            (err) => console.error(err)
        );
    }
}

This isn't perfect as you need to keep state outside of the stream but it's the only way I managed to do it. Any cleaner solution is welcomed.