1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { Observable } from '../Observable';
import { Subject } from '../Subject';
import { Subscription } from '../Subscription';
import { OuterSubscriber } from '../OuterSubscriber';
import { InnerSubscriber } from '../InnerSubscriber';
import { subscribeToResult } from '../util/subscribeToResult';
import { MonoTypeOperatorFunction, TeardownLogic } from '../types';
/**
* Returns an Observable that mirrors the source Observable with the exception of a `complete`. If the source
* Observable calls `complete`, this method will emit to the Observable 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.
*
* 
*
* @param {function(notifications: Observable): Observable} notifier - Receives an Observable of notifications with
* which a user can `complete` or `error`, aborting the repetition.
* @return {Observable} The source Observable modified with repeat logic.
* @method repeatWhen
* @owner Observable
*/
export function repeatWhen<T>(notifier: (notifications: Observable<any>) => Observable<any>): MonoTypeOperatorFunction<T> {
return (source: Observable<T>) => source.lift(new RepeatWhenOperator(notifier));
}
class RepeatWhenOperator<T> implements Operator<T, T> {
constructor(protected notifier: (notifications: Observable<any>) => Observable<any>) {
}
call(subscriber: Subscriber<T>, source: any): TeardownLogic {
return source.subscribe(new RepeatWhenSubscriber(subscriber, this.notifier, source));
}
}
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class RepeatWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
private notifications: Subject<any>;
private retries: Observable<any>;
private retriesSubscription: Subscription;
private sourceIsBeingSubscribedTo: boolean = true;
constructor(destination: Subscriber<R>,
private notifier: (notifications: Observable<any>) => Observable<any>,
private source: Observable<T>) {
super(destination);
}
notifyNext(outerValue: T, innerValue: R,
outerIndex: number, innerIndex: number,
innerSub: InnerSubscriber<T, R>): void {
this.sourceIsBeingSubscribedTo = true;
this.source.subscribe(this);
}
notifyComplete(innerSub: InnerSubscriber<T, R>): void {
if (this.sourceIsBeingSubscribedTo === false) {
return super.complete();
}
}
complete() {
this.sourceIsBeingSubscribedTo = false;
if (!this.isStopped) {
if (!this.retries) {
this.subscribeToRetries();
}
if (!this.retriesSubscription || this.retriesSubscription.closed) {
return super.complete();
}
this._unsubscribeAndRecycle();
this.notifications.next();
}
}
/** @deprecated This is an internal implementation detail, do not use. */
_unsubscribe() {
const { notifications, retriesSubscription } = this;
if (notifications) {
notifications.unsubscribe();
this.notifications = null;
}
if (retriesSubscription) {
retriesSubscription.unsubscribe();
this.retriesSubscription = null;
}
this.retries = null;
}
/** @deprecated This is an internal implementation detail, do not use. */
_unsubscribeAndRecycle(): Subscriber<T> {
const { _unsubscribe } = this;
this._unsubscribe = null;
super._unsubscribeAndRecycle();
this._unsubscribe = _unsubscribe;
return this;
}
private subscribeToRetries() {
this.notifications = new Subject();
let retries;
try {
const { notifier } = this;
retries = notifier(this.notifications);
} catch (e) {
return super.complete();
}
this.retries = retries;
this.retriesSubscription = subscribeToResult(this, retries);
}
}
|
