Skip to content

Commit 794f806

Browse files
jeremymwellsbenlesh
andauthoredDec 15, 2022
feat(retryWhen): notifier now supports any ObservableInput (#7105)
* 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>
1 parent b18c2eb commit 794f806

File tree

2 files changed

+73
-8
lines changed

2 files changed

+73
-8
lines changed
 

‎spec-dtslint/operators/retryWhen-spec.ts

+64-2
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { of } from 'rxjs';
22
import { retryWhen } from 'rxjs/operators';
3+
import { asInteropObservable } from '../../spec/helpers/interop-helper';
34

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

17+
it('should accept interop observable notifier', () => {
18+
of(1, 2, 3).pipe(retryWhen(() => asInteropObservable(of(true)))); // $ExpectType Observable<number>
19+
});
20+
21+
it('should accept promise notifier', () => {
22+
of(1, 2, 3).pipe(retryWhen(() => Promise.resolve(true))); // $ExpectType Observable<number>
23+
});
24+
25+
it('should async iterable notifier', () => {
26+
const asyncRange = {
27+
from: 1,
28+
to: 2,
29+
[Symbol.asyncIterator]() {
30+
return {
31+
current: this.from,
32+
last: this.to,
33+
async next() {
34+
await Promise.resolve();
35+
const done = (this.current > this.last);
36+
return {
37+
done,
38+
value: done ? this.current++ : undefined
39+
};
40+
}
41+
};
42+
}
43+
};
44+
of(1, 2, 3).pipe(retryWhen(() => asyncRange)); // $ExpectType Observable<number>
45+
});
46+
47+
it('should accept iterable notifier', () => {
48+
const syncRange = {
49+
from: 1,
50+
to: 2,
51+
[Symbol.iterator]() {
52+
return {
53+
current: this.from,
54+
last: this.to,
55+
next() {
56+
const done = (this.current > this.last);
57+
return {
58+
done,
59+
value: done ? this.current++ : undefined
60+
};
61+
}
62+
};
63+
}
64+
};
65+
of(1, 2, 3).pipe(retryWhen(() => syncRange)); // $ExpectType Observable<number>
66+
});
67+
68+
it('should accept readable stream notifier', () => {
69+
const readableStream = new ReadableStream<string>({
70+
pull(controller) {
71+
controller.enqueue('x');
72+
controller.close();
73+
},
74+
});
75+
of(1, 2, 3).pipe(retryWhen(() => readableStream)); // $ExpectType Observable<number>
76+
});
77+
1678
it('should enforce types of the notifier', () => {
17-
const o = of(1, 2, 3).pipe(retryWhen(() => 8)); // $ExpectError
79+
of(1, 2, 3).pipe(retryWhen(() => 8)); // $ExpectError
1880
});
1981

2082
it('should be deprecated', () => {
21-
const o = of(1, 2, 3).pipe(retryWhen(() => of(true))); // $ExpectDeprecation
83+
of(1, 2, 3).pipe(retryWhen(() => of(true))); // $ExpectDeprecation
2284
});

‎src/internal/operators/retryWhen.ts

+9-6
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,15 @@
11
import { Observable } from '../Observable';
2+
import { innerFrom } from '../observable/innerFrom';
23
import { Subject } from '../Subject';
34
import { Subscription } from '../Subscription';
45

5-
import { MonoTypeOperatorFunction } from '../types';
6+
import { MonoTypeOperatorFunction, ObservableInput } from '../types';
67
import { operate } from '../util/lift';
78
import { createOperatorSubscriber } from './OperatorSubscriber';
89

910
/**
1011
* Returns an Observable that mirrors the source Observable with the exception of an `error`. If the source Observable
11-
* calls `error`, this method will emit the Throwable that caused the error to the Observable returned from `notifier`.
12+
* calls `error`, this method will emit the Throwable that caused the error to the `ObservableInput` returned from `notifier`.
1213
* If that Observable calls `complete` or `error` then this method will call `complete` or `error` on the child
1314
* subscription. Otherwise this method will resubscribe to the source Observable.
1415
*
@@ -55,13 +56,15 @@ import { createOperatorSubscriber } from './OperatorSubscriber';
5556
*
5657
* @see {@link retry}
5758
*
58-
* @param {function(errors: Observable): Observable} notifier - Receives an Observable of notifications with which a
59+
* @param notifier Function that receives an Observable of notifications with which a
5960
* user can `complete` or `error`, aborting the retry.
60-
* @return A function that returns an Observable that mirrors the source
61+
* @return A function that returns an `ObservableInput` that mirrors the source
6162
* Observable with the exception of an `error`.
6263
* @deprecated Will be removed in v9 or v10, use {@link retry}'s `delay` option instead.
64+
* Will be removed in v9 or v10. Use {@link retry}'s {@link RetryConfig#delay delay} option instead.
65+
* Instead of `retryWhen(() => notify$)`, use: `retry({ delay: () => notify$ })`.
6366
*/
64-
export function retryWhen<T>(notifier: (errors: Observable<any>) => Observable<any>): MonoTypeOperatorFunction<T> {
67+
export function retryWhen<T>(notifier: (errors: Observable<any>) => ObservableInput<any>): MonoTypeOperatorFunction<T> {
6568
return operate((source, subscriber) => {
6669
let innerSub: Subscription | null;
6770
let syncResub = false;
@@ -72,7 +75,7 @@ export function retryWhen<T>(notifier: (errors: Observable<any>) => Observable<a
7275
createOperatorSubscriber(subscriber, undefined, undefined, (err) => {
7376
if (!errors$) {
7477
errors$ = new Subject();
75-
notifier(errors$).subscribe(
78+
innerFrom(notifier(errors$)).subscribe(
7679
createOperatorSubscriber(subscriber, () =>
7780
// If we have an innerSub, this was an asynchronous call, kick off the retry.
7881
// Otherwise, if we don't have an innerSub yet, that's because the inner subscription

0 commit comments

Comments
 (0)
Please sign in to comment.