exhaustMap.js
5.77 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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
"use strict";
var __extends = (this && this.__extends) || function (d, b) {
for (var p in b) if (b.hasOwnProperty(p)) d[p] = b[p];
function __() { this.constructor = d; }
d.prototype = b === null ? Object.create(b) : (__.prototype = b.prototype, new __());
};
var OuterSubscriber_1 = require('../OuterSubscriber');
var subscribeToResult_1 = require('../util/subscribeToResult');
/* tslint:disable: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>
*
* <img src="./img/exhaustMap.png" width="100%">
*
* 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 <caption>Run a finite timer for each click, only if there is no currently active timer</caption>
* var clicks = Rx.Observable.fromEvent(document, 'click');
* var result = clicks.exhaustMap((ev) => Rx.Observable.interval(1000));
* result.subscribe(x => console.log(x));
*
* @see {@link concatMap}
* @see {@link exhaust}
* @see {@link mergeMap}
* @see {@link switchMap}
*
* @param {function(value: T, ?index: number): Observable} project A function
* that, when applied to an item emitted by the source Observable, returns an
* Observable.
* @param {function(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number): any} [resultSelector]
* A function to produce the value on the output Observable based on the values
* and the indices of the source (outer) emission and the inner Observable
* emission. The arguments passed to this function are:
* - `outerValue`: the value that came from the source
* - `innerValue`: the value that came from the projected Observable
* - `outerIndex`: the "index" of the value that came from the source
* - `innerIndex`: the "index" of the value from the projected 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
*/
function exhaustMap(project, resultSelector) {
return this.lift(new SwitchFirstMapOperator(project, resultSelector));
}
exports.exhaustMap = exhaustMap;
var SwitchFirstMapOperator = (function () {
function SwitchFirstMapOperator(project, resultSelector) {
this.project = project;
this.resultSelector = resultSelector;
}
SwitchFirstMapOperator.prototype.call = function (subscriber, source) {
return source.subscribe(new SwitchFirstMapSubscriber(subscriber, this.project, this.resultSelector));
};
return SwitchFirstMapOperator;
}());
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
var SwitchFirstMapSubscriber = (function (_super) {
__extends(SwitchFirstMapSubscriber, _super);
function SwitchFirstMapSubscriber(destination, project, resultSelector) {
_super.call(this, destination);
this.project = project;
this.resultSelector = resultSelector;
this.hasSubscription = false;
this.hasCompleted = false;
this.index = 0;
}
SwitchFirstMapSubscriber.prototype._next = function (value) {
if (!this.hasSubscription) {
this.tryNext(value);
}
};
SwitchFirstMapSubscriber.prototype.tryNext = function (value) {
var index = this.index++;
var destination = this.destination;
try {
var result = this.project(value, index);
this.hasSubscription = true;
this.add(subscribeToResult_1.subscribeToResult(this, result, value, index));
}
catch (err) {
destination.error(err);
}
};
SwitchFirstMapSubscriber.prototype._complete = function () {
this.hasCompleted = true;
if (!this.hasSubscription) {
this.destination.complete();
}
};
SwitchFirstMapSubscriber.prototype.notifyNext = function (outerValue, innerValue, outerIndex, innerIndex, innerSub) {
var _a = this, resultSelector = _a.resultSelector, destination = _a.destination;
if (resultSelector) {
this.trySelectResult(outerValue, innerValue, outerIndex, innerIndex);
}
else {
destination.next(innerValue);
}
};
SwitchFirstMapSubscriber.prototype.trySelectResult = function (outerValue, innerValue, outerIndex, innerIndex) {
var _a = this, resultSelector = _a.resultSelector, destination = _a.destination;
try {
var result = resultSelector(outerValue, innerValue, outerIndex, innerIndex);
destination.next(result);
}
catch (err) {
destination.error(err);
}
};
SwitchFirstMapSubscriber.prototype.notifyError = function (err) {
this.destination.error(err);
};
SwitchFirstMapSubscriber.prototype.notifyComplete = function (innerSub) {
this.remove(innerSub);
this.hasSubscription = false;
if (this.hasCompleted) {
this.destination.complete();
}
};
return SwitchFirstMapSubscriber;
}(OuterSubscriber_1.OuterSubscriber));
//# sourceMappingURL=exhaustMap.js.map