Rxjs retry operator with dynamic delay between attempts

retry-with-delay.rxjs.ts

import { Observable, MonoTypeOperatorFunction, of } from "rxjs";
import { retryWhen, delay, take, catchError, concatMap } from "rxjs/operators";

let globalCounter = 0;

const emitter$ = new Observable((subscriber) => {
  if (globalCounter >= 5) {
    subscriber.next("Heey, finally i'am work!");
    subscriber.complete();
  }
  globalCounter += 1;
  subscriber.error(" Ooops! Too early :(");
});

type RetryDelayStrategy = number | number[] | ((arg0: number) => number);

function retryWithDelay<T>(
  maxRetryCount: number = 10,
  retryDelay: RetryDelayStrategy = 1000
): MonoTypeOperatorFunction<T> {
  let retryCount = 0;
  let nextDelay: number;

  const recalculateDelay = () => {
    if (retryDelay instanceof Function) {
      nextDelay = retryDelay(retryCount);
      return;
    }
    if (retryDelay instanceof Array) {
      const delayIndex =
        retryCount >= retryDelay.length
          ? Math.trunc(retryCount / retryDelay.length)
          : retryCount;
      nextDelay = retryDelay[delayIndex];
    }
    nextDelay = retryDelay as number;
  };
  recalculateDelay();
  return function <T>(source: Observable<T>): Observable<T> {
    return source.pipe(
      retryWhen((errors) => {
        return errors.pipe(
          concatMap(e => {
            retryCount = retryCount + 1;
            console.debug(
              "[line 91][retry-policy.operator.ts] 🚀 retryCount: ",
              retryCount
            );
            recalculateDelay();
            return of(e).pipe(delay(nextDelay))
          }),
          take(maxRetryCount)
        );
      })
    );
  };
}

emitter$
  .pipe(
    catchError(e => {
      console.log(e)
      throw e;
    }),
    retryWithDelay(8, (v) => v+1000),
  )
  .subscribe((r) => {
    console.log(r);
  });