Skip to content

Commit

Permalink
feat(retryWhen): notifier now supports any ObservableInput (#7105)
Browse files Browse the repository at this point in the history
* feat(retryWhen): retryWhen's notifier supports ObservableInput

* test(retryWhen): added tests for ObservableInput notifier

* chore(retryWhen): updated docs grammar

* Delete index.d.ts

* Delete index.d.ts

Co-authored-by: Ben Lesh <ben@benlesh.com>
  • Loading branch information
jeremymwells and benlesh authored Dec 15, 2022
1 parent b18c2eb commit 794f806
Show file tree
Hide file tree
Showing 2 changed files with 73 additions and 8 deletions.
66 changes: 64 additions & 2 deletions spec-dtslint/operators/retryWhen-spec.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { of } from 'rxjs';
import { retryWhen } from 'rxjs/operators';
import { asInteropObservable } from '../../spec/helpers/interop-helper';

it('should infer correctly', () => {
const o = of(1, 2, 3).pipe(retryWhen(errors => errors)); // $ExpectType Observable<number>
Expand All @@ -13,10 +14,71 @@ it('should enforce types', () => {
const o = of(1, 2, 3).pipe(retryWhen()); // $ExpectError
});

it('should accept interop observable notifier', () => {
of(1, 2, 3).pipe(retryWhen(() => asInteropObservable(of(true)))); // $ExpectType Observable<number>
});

it('should accept promise notifier', () => {
of(1, 2, 3).pipe(retryWhen(() => Promise.resolve(true))); // $ExpectType Observable<number>
});

it('should async iterable notifier', () => {
const asyncRange = {
from: 1,
to: 2,
[Symbol.asyncIterator]() {
return {
current: this.from,
last: this.to,
async next() {
await Promise.resolve();
const done = (this.current > this.last);
return {
done,
value: done ? this.current++ : undefined
};
}
};
}
};
of(1, 2, 3).pipe(retryWhen(() => asyncRange)); // $ExpectType Observable<number>
});

it('should accept iterable notifier', () => {
const syncRange = {
from: 1,
to: 2,
[Symbol.iterator]() {
return {
current: this.from,
last: this.to,
next() {
const done = (this.current > this.last);
return {
done,
value: done ? this.current++ : undefined
};
}
};
}
};
of(1, 2, 3).pipe(retryWhen(() => syncRange)); // $ExpectType Observable<number>
});

it('should accept readable stream notifier', () => {
const readableStream = new ReadableStream<string>({
pull(controller) {
controller.enqueue('x');
controller.close();
},
});
of(1, 2, 3).pipe(retryWhen(() => readableStream)); // $ExpectType Observable<number>
});

it('should enforce types of the notifier', () => {
const o = of(1, 2, 3).pipe(retryWhen(() => 8)); // $ExpectError
of(1, 2, 3).pipe(retryWhen(() => 8)); // $ExpectError
});

it('should be deprecated', () => {
const o = of(1, 2, 3).pipe(retryWhen(() => of(true))); // $ExpectDeprecation
of(1, 2, 3).pipe(retryWhen(() => of(true))); // $ExpectDeprecation
});
15 changes: 9 additions & 6 deletions src/internal/operators/retryWhen.ts
Original file line number Diff line number Diff line change
@@ -1,14 +1,15 @@
import { Observable } from '../Observable';
import { innerFrom } from '../observable/innerFrom';
import { Subject } from '../Subject';
import { Subscription } from '../Subscription';

import { MonoTypeOperatorFunction } from '../types';
import { MonoTypeOperatorFunction, ObservableInput } from '../types';
import { operate } from '../util/lift';
import { createOperatorSubscriber } from './OperatorSubscriber';

/**
* Returns an Observable that mirrors the source Observable with the exception of an `error`. If the source Observable
* calls `error`, this method will emit the Throwable that caused the error to the Observable returned from `notifier`.
* calls `error`, this method will emit the Throwable that caused the error to the `ObservableInput` returned from `notifier`.
* If that Observable calls `complete` or `error` then this method will call `complete` or `error` on the child
* subscription. Otherwise this method will resubscribe to the source Observable.
*
Expand Down Expand Up @@ -55,13 +56,15 @@ import { createOperatorSubscriber } from './OperatorSubscriber';
*
* @see {@link retry}
*
* @param {function(errors: Observable): Observable} notifier - Receives an Observable of notifications with which a
* @param notifier Function that receives an Observable of notifications with which a
* user can `complete` or `error`, aborting the retry.
* @return A function that returns an Observable that mirrors the source
* @return A function that returns an `ObservableInput` that mirrors the source
* Observable with the exception of an `error`.
* @deprecated Will be removed in v9 or v10, use {@link retry}'s `delay` option instead.
* Will be removed in v9 or v10. Use {@link retry}'s {@link RetryConfig#delay delay} option instead.
* Instead of `retryWhen(() => notify$)`, use: `retry({ delay: () => notify$ })`.
*/
export function retryWhen<T>(notifier: (errors: Observable<any>) => Observable<any>): MonoTypeOperatorFunction<T> {
export function retryWhen<T>(notifier: (errors: Observable<any>) => ObservableInput<any>): MonoTypeOperatorFunction<T> {
return operate((source, subscriber) => {
let innerSub: Subscription | null;
let syncResub = false;
Expand All @@ -72,7 +75,7 @@ export function retryWhen<T>(notifier: (errors: Observable<any>) => Observable<a
createOperatorSubscriber(subscriber, undefined, undefined, (err) => {
if (!errors$) {
errors$ = new Subject();
notifier(errors$).subscribe(
innerFrom(notifier(errors$)).subscribe(
createOperatorSubscriber(subscriber, () =>
// If we have an innerSub, this was an asynchronous call, kick off the retry.
// Otherwise, if we don't have an innerSub yet, that's because the inner subscription
Expand Down

0 comments on commit 794f806

Please sign in to comment.