BoundCallbackObservable.js
7.26 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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
"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 Observable_1 = require('../Observable');
var tryCatch_1 = require('../util/tryCatch');
var errorObject_1 = require('../util/errorObject');
var AsyncSubject_1 = require('../AsyncSubject');
/**
* We need this JSDoc comment for affecting ESDoc.
* @extends {Ignored}
* @hide true
*/
var BoundCallbackObservable = (function (_super) {
__extends(BoundCallbackObservable, _super);
function BoundCallbackObservable(callbackFunc, selector, args, scheduler) {
_super.call(this);
this.callbackFunc = callbackFunc;
this.selector = selector;
this.args = args;
this.scheduler = scheduler;
}
/* tslint:enable:max-line-length */
/**
* Converts a callback API to a function that returns an Observable.
*
* <span class="informal">Give it a function `f` of type `f(x, callback)` and
* it will return a function `g` that when called as `g(x)` will output an
* Observable.</span>
*
* `bindCallback` is not an operator because its input and output are not
* Observables. The input is a function `func` with some parameters, but the
* last parameter must be a callback function that `func` calls when it is
* done. The output of `bindCallback` is a function that takes the same
* parameters as `func`, except the last one (the callback). When the output
* function is called with arguments, it will return an Observable where the
* results will be delivered to.
*
* @example <caption>Convert jQuery's getJSON to an Observable API</caption>
* // Suppose we have jQuery.getJSON('/my/url', callback)
* var getJSONAsObservable = Rx.Observable.bindCallback(jQuery.getJSON);
* var result = getJSONAsObservable('/my/url');
* result.subscribe(x => console.log(x), e => console.error(e));
*
* @see {@link bindNodeCallback}
* @see {@link from}
* @see {@link fromPromise}
*
* @param {function} func Function with a callback as the last parameter.
* @param {function} [selector] A function which takes the arguments from the
* callback and maps those a value to emit on the output Observable.
* @param {Scheduler} [scheduler] The scheduler on which to schedule the
* callbacks.
* @return {function(...params: *): Observable} A function which returns the
* Observable that delivers the same values the callback would deliver.
* @static true
* @name bindCallback
* @owner Observable
*/
BoundCallbackObservable.create = function (func, selector, scheduler) {
if (selector === void 0) { selector = undefined; }
return function () {
var args = [];
for (var _i = 0; _i < arguments.length; _i++) {
args[_i - 0] = arguments[_i];
}
return new BoundCallbackObservable(func, selector, args, scheduler);
};
};
BoundCallbackObservable.prototype._subscribe = function (subscriber) {
var callbackFunc = this.callbackFunc;
var args = this.args;
var scheduler = this.scheduler;
var subject = this.subject;
if (!scheduler) {
if (!subject) {
subject = this.subject = new AsyncSubject_1.AsyncSubject();
var handler = function handlerFn() {
var innerArgs = [];
for (var _i = 0; _i < arguments.length; _i++) {
innerArgs[_i - 0] = arguments[_i];
}
var source = handlerFn.source;
var selector = source.selector, subject = source.subject;
if (selector) {
var result_1 = tryCatch_1.tryCatch(selector).apply(this, innerArgs);
if (result_1 === errorObject_1.errorObject) {
subject.error(errorObject_1.errorObject.e);
}
else {
subject.next(result_1);
subject.complete();
}
}
else {
subject.next(innerArgs.length === 1 ? innerArgs[0] : innerArgs);
subject.complete();
}
};
// use named function instance to avoid closure.
handler.source = this;
var result = tryCatch_1.tryCatch(callbackFunc).apply(this, args.concat(handler));
if (result === errorObject_1.errorObject) {
subject.error(errorObject_1.errorObject.e);
}
}
return subject.subscribe(subscriber);
}
else {
return scheduler.schedule(BoundCallbackObservable.dispatch, 0, { source: this, subscriber: subscriber });
}
};
BoundCallbackObservable.dispatch = function (state) {
var self = this;
var source = state.source, subscriber = state.subscriber;
var callbackFunc = source.callbackFunc, args = source.args, scheduler = source.scheduler;
var subject = source.subject;
if (!subject) {
subject = source.subject = new AsyncSubject_1.AsyncSubject();
var handler = function handlerFn() {
var innerArgs = [];
for (var _i = 0; _i < arguments.length; _i++) {
innerArgs[_i - 0] = arguments[_i];
}
var source = handlerFn.source;
var selector = source.selector, subject = source.subject;
if (selector) {
var result_2 = tryCatch_1.tryCatch(selector).apply(this, innerArgs);
if (result_2 === errorObject_1.errorObject) {
self.add(scheduler.schedule(dispatchError, 0, { err: errorObject_1.errorObject.e, subject: subject }));
}
else {
self.add(scheduler.schedule(dispatchNext, 0, { value: result_2, subject: subject }));
}
}
else {
var value = innerArgs.length === 1 ? innerArgs[0] : innerArgs;
self.add(scheduler.schedule(dispatchNext, 0, { value: value, subject: subject }));
}
};
// use named function to pass values in without closure
handler.source = source;
var result = tryCatch_1.tryCatch(callbackFunc).apply(this, args.concat(handler));
if (result === errorObject_1.errorObject) {
subject.error(errorObject_1.errorObject.e);
}
}
self.add(subject.subscribe(subscriber));
};
return BoundCallbackObservable;
}(Observable_1.Observable));
exports.BoundCallbackObservable = BoundCallbackObservable;
function dispatchNext(arg) {
var value = arg.value, subject = arg.subject;
subject.next(value);
subject.complete();
}
function dispatchError(arg) {
var err = arg.err, subject = arg.subject;
subject.error(err);
}
//# sourceMappingURL=BoundCallbackObservable.js.map