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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
import { Operator } from '../Operator';
import { Observable } from '../Observable';
import { Subscriber } from '../Subscriber';
import { Subscription } from '../Subscription';
import { OuterSubscriber } from '../OuterSubscriber';
import { InnerSubscriber } from '../InnerSubscriber';
import { subscribeToResult } from '../util/subscribeToResult';
import { ObservableInput, OperatorFunction } from '../types';
import { map } from './map';
import { from } from '../observable/from';
/* tslint:disable:max-line-length */
export function exhaustMap<T, R>(project: (value: T, index: number) => ObservableInput<R>): OperatorFunction<T, R>;
/** @deprecated resultSelector is no longer supported. Use inner map instead. */
export function exhaustMap<T, R>(project: (value: T, index: number) => ObservableInput<R>, resultSelector: undefined): OperatorFunction<T, R>;
/** @deprecated resultSelector is no longer supported. Use inner map instead. */
export function exhaustMap<T, I, R>(project: (value: T, index: number) => ObservableInput<I>, resultSelector: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R): OperatorFunction<T, R>;
/* tslint:enable:max-line-length */
/**
* Projects each source value to an Observable which is merged in the output
* Observable only if the previous projected Observable has completed.
*
* <span class="informal">Maps each value to an Observable, then flattens all of
* these inner Observables using {@link exhaust}.</span>
*
* 
*
* Returns an Observable that emits items based on applying a function that you
* supply to each item emitted by the source Observable, where that function
* returns an (so-called "inner") Observable. When it projects a source value to
* an Observable, the output Observable begins emitting the items emitted by
* that projected Observable. However, `exhaustMap` ignores every new projected
* Observable if the previous projected Observable has not yet completed. Once
* that one completes, it will accept and flatten the next projected Observable
* and repeat this process.
*
* ## Example
* Run a finite timer for each click, only if there is no currently active timer
* ```javascript
* const clicks = fromEvent(document, 'click');
* const result = clicks.pipe(
* exhaustMap((ev) => interval(1000).pipe(take(5))),
* );
* result.subscribe(x => console.log(x));
* ```
*
* @see {@link concatMap}
* @see {@link exhaust}
* @see {@link mergeMap}
* @see {@link switchMap}
*
* @param {function(value: T, ?index: number): ObservableInput} project A function
* that, when applied to an item emitted by the source Observable, returns an
* Observable.
* @return {Observable} An Observable containing projected Observables
* of each item of the source, ignoring projected Observables that start before
* their preceding Observable has completed.
* @method exhaustMap
* @owner Observable
*/
export function exhaustMap<T, I, R>(
project: (value: T, index: number) => ObservableInput<I>,
resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R,
): OperatorFunction<T, I|R> {
if (resultSelector) {
// DEPRECATED PATH
return (source: Observable<T>) => source.pipe(
exhaustMap((a, i) => from(project(a, i)).pipe(
map((b, ii) => resultSelector(a, b, i, ii)),
)),
);
}
return (source: Observable<T>) =>
source.lift(new ExhauseMapOperator(project));
}
class ExhauseMapOperator<T, R> implements Operator<T, R> {
constructor(private project: (value: T, index: number) => ObservableInput<R>) {
}
call(subscriber: Subscriber<R>, source: any): any {
return source.subscribe(new ExhaustMapSubscriber(subscriber, this.project));
}
}
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class ExhaustMapSubscriber<T, R> extends OuterSubscriber<T, R> {
private hasSubscription = false;
private hasCompleted = false;
private index = 0;
constructor(destination: Subscriber<R>,
private project: (value: T, index: number) => ObservableInput<R>) {
super(destination);
}
protected _next(value: T): void {
if (!this.hasSubscription) {
this.tryNext(value);
}
}
private tryNext(value: T): void {
let result: ObservableInput<R>;
const index = this.index++;
try {
result = this.project(value, index);
} catch (err) {
this.destination.error(err);
return;
}
this.hasSubscription = true;
this._innerSub(result, value, index);
}
private _innerSub(result: ObservableInput<R>, value: T, index: number): void {
const innerSubscriber = new InnerSubscriber(this, undefined, undefined);
this.add(innerSubscriber);
subscribeToResult<T, R>(this, result, value, index, innerSubscriber);
}
protected _complete(): void {
this.hasCompleted = true;
if (!this.hasSubscription) {
this.destination.complete();
}
}
notifyNext(outerValue: T, innerValue: R,
outerIndex: number, innerIndex: number,
innerSub: InnerSubscriber<T, R>): void {
this.destination.next(innerValue);
}
notifyError(err: any): void {
this.destination.error(err);
}
notifyComplete(innerSub: Subscription): void {
this.remove(innerSub);
this.hasSubscription = false;
if (this.hasCompleted) {
this.destination.complete();
}
}
}
|
