Function: retry()

retry<T>(factory, maxRetries?, delay?): Stream<T>

Defined in: streams/retry.ts:20

Creates a stream that subscribes to a source factory and retries on error.

This operator is useful for handling streams that may fail due to temporary issues, such as network problems. It will attempt to resubscribe to the source stream up to maxRetries times, with an optional delay between each attempt. If the stream completes successfully on any attempt, it will emit all of its values and then complete. If all retry attempts fail, the final error is propagated.

Type Parameters

T

T = any

The type of values emitted by the stream.

Parameters

factory

() => Stream<T> | Promise<T>

A function that returns a new stream instance for each subscription attempt.

maxRetries?

number = 3

The maximum number of times to retry the stream. A value of 0 means no retries.

delay?

number = 1000

The time in milliseconds to wait before each retry attempt.

Returns

Stream<T>

A new stream that applies the retry logic.

Examples

From retry.spec.ts:4

let attempt = 0;
const factory = jasmine.createSpy('factory').and.callFake(() => {
  attempt++;
  return createStream<number>('testStream', async function* () {
    if (attempt === 1) {
      yield 1;
      yield 2;
      throw new Error('Test Error');
    } else {
      yield 3;
      yield 4;
    }
  });
});
const result: number[] = [];
const stream$ = retry(factory, 3, 1000);
for await (const value of eachValueFrom(stream$)) {
  result.push(value);
}
expect(result).toEqual([3, 4]);
expect(factory).toHaveBeenCalledTimes(2);