innerSubscribe.ts
3.18 KB
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
/** @prettier */
import { Subscription } from './Subscription';
import { Subscriber } from './Subscriber';
import { Observable } from './Observable';
import { subscribeTo } from './util/subscribeTo';
interface SimpleOuterSubscriberLike<T> {
/**
* A handler for inner next notifications from the inner subscription
* @param innerValue the value nexted by the inner producer
*/
notifyNext(innerValue: T): void;
/**
* A handler for inner error notifications from the inner subscription
* @param err the error from the inner producer
*/
notifyError(err: any): void;
/**
* A handler for inner complete notifications from the inner subscription.
*/
notifyComplete(): void;
}
export class SimpleInnerSubscriber<T> extends Subscriber<T> {
constructor(private parent: SimpleOuterSubscriberLike<any>) {
super();
}
protected _next(value: T): void {
this.parent.notifyNext(value);
}
protected _error(error: any): void {
this.parent.notifyError(error);
this.unsubscribe();
}
protected _complete(): void {
this.parent.notifyComplete();
this.unsubscribe();
}
}
export class ComplexInnerSubscriber<T, R> extends Subscriber<R> {
constructor(private parent: ComplexOuterSubscriber<T, R>, public outerValue: T, public outerIndex: number) {
super();
}
protected _next(value: R): void {
this.parent.notifyNext(this.outerValue, value, this.outerIndex, this);
}
protected _error(error: any): void {
this.parent.notifyError(error);
this.unsubscribe();
}
protected _complete(): void {
this.parent.notifyComplete(this);
this.unsubscribe();
}
}
export class SimpleOuterSubscriber<T, R> extends Subscriber<T> implements SimpleOuterSubscriberLike<R> {
notifyNext(innerValue: R): void {
this.destination.next(innerValue);
}
notifyError(err: any): void {
this.destination.error(err);
}
notifyComplete(): void {
this.destination.complete();
}
}
/**
* DO NOT USE (formerly "OuterSubscriber")
* TODO: We want to refactor this and remove it. It is retaining values it shouldn't for long
* periods of time.
*/
export class ComplexOuterSubscriber<T, R> extends Subscriber<T> {
/**
* @param _outerValue Used by: bufferToggle, delayWhen, windowToggle
* @param innerValue Used by: subclass default, combineLatest, race, bufferToggle, windowToggle, withLatestFrom
* @param _outerIndex Used by: combineLatest, race, withLatestFrom
* @param _innerSub Used by: delayWhen
*/
notifyNext(_outerValue: T, innerValue: R, _outerIndex: number, _innerSub: ComplexInnerSubscriber<T, R>): void {
this.destination.next(innerValue);
}
notifyError(error: any): void {
this.destination.error(error);
}
/**
* @param _innerSub Used by: race, bufferToggle, delayWhen, windowToggle, windowWhen
*/
notifyComplete(_innerSub: ComplexInnerSubscriber<T, R>): void {
this.destination.complete();
}
}
export function innerSubscribe(result: any, innerSubscriber: Subscriber<any>): Subscription | undefined {
if (innerSubscriber.closed) {
return undefined;
}
if (result instanceof Observable) {
return result.subscribe(innerSubscriber);
}
return subscribeTo(result)(innerSubscriber) as Subscription;
}