Skip to content

Commit 497bb0d

Browse files
kwonojbenlesh
kwonoj
authored and
benlesh
committed
feat(bindNodeCallback): add Observable.bindNodeCallback
- add bindNodeCallback corresponding to fromNodeCallback in RxJS4 closes #736
1 parent a5927ba commit 497bb0d

File tree

7 files changed

+421
-0
lines changed

7 files changed

+421
-0
lines changed

‎MIGRATION.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,8 @@ enabling "composite" subscription behavior.
5959
|`flatMapFirst`|`exhaustMap`|
6060
|`flatMapLatest`|`switchMap`|
6161
|`flatMapWithMaxConcurrent`|`mergeMap` or `flatMap`(alias)|
62+
|`fromCallback`|`bindCallback`|
63+
|`fromNodeCallback`|`bindNodeCallback`|
6264
|`publishValue`|`publishBehavior`|
6365
|`replay`|`publishReplay`|
6466
|`select`|`map`|

‎spec/observables/bindNodeCallback-spec.js‎

Lines changed: 292 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,292 @@
1+
/* globals describe, it, expect, rxTestScheduler */
2+
var Rx = require('../../dist/cjs/Rx');
3+
var Observable = Rx.Observable;
4+
5+
describe('Observable.bindNodeCallback', function () {
6+
describe('when not scheduled', function () {
7+
it('should emit one value from a callback', function () {
8+
function callback(datum, cb) {
9+
cb(null, datum);
10+
}
11+
var boundCallback = Observable.bindNodeCallback(callback);
12+
var results = [];
13+
14+
boundCallback(42)
15+
.subscribe(function (x) {
16+
results.push(x);
17+
}, null, function () {
18+
results.push('done');
19+
});
20+
21+
expect(results).toEqual([42, 'done']);
22+
});
23+
24+
it('should emit one value chosen by a selector', function () {
25+
function callback(datum, cb) {
26+
cb(null, datum);
27+
}
28+
var boundCallback = Observable.bindNodeCallback(callback, function (datum) { return datum; });
29+
var results = [];
30+
31+
boundCallback(42)
32+
.subscribe(function (x) {
33+
results.push(x);
34+
}, null, function () {
35+
results.push('done');
36+
});
37+
38+
expect(results).toEqual([42, 'done']);
39+
});
40+
41+
it('should raise error from callback', function () {
42+
var error = new Error();
43+
44+
function callback(cb) {
45+
cb(error);
46+
}
47+
48+
var boundCallback = Observable.bindNodeCallback(callback);
49+
var results = [];
50+
51+
boundCallback()
52+
.subscribe(function () {
53+
throw 'should not next';
54+
}, function (err) {
55+
results.push(err);
56+
}, function () {
57+
throw 'should not complete';
58+
});
59+
60+
expect(results).toEqual([error]);
61+
});
62+
63+
it('should emit an error when the selector throws', function () {
64+
function callback(cb) {
65+
cb(null, 42);
66+
}
67+
var boundCallback = Observable.bindNodeCallback(callback, function (err) { throw new Error('Yikes!'); });
68+
var results = [];
69+
70+
boundCallback()
71+
.subscribe(function () {
72+
throw 'should not next';
73+
}, function (err) {
74+
results.push(err);
75+
}, function () {
76+
throw 'should not complete';
77+
});
78+
79+
expect(results).toEqual([new Error('Yikes!')]);
80+
});
81+
82+
it('should not emit, throw or complete if immediately unsubscribed', function (done) {
83+
var nextSpy = jasmine.createSpy('next');
84+
var throwSpy = jasmine.createSpy('throw');
85+
var completeSpy = jasmine.createSpy('complete');
86+
var timeout;
87+
function callback(datum, cb) {
88+
// Need to cb async in order for the unsub to trigger
89+
timeout = setTimeout(function () {
90+
cb(null, datum);
91+
});
92+
}
93+
var subscription = Observable.bindNodeCallback(callback)(42)
94+
.subscribe(nextSpy, throwSpy, completeSpy);
95+
subscription.unsubscribe();
96+
97+
setTimeout(function () {
98+
expect(nextSpy).not.toHaveBeenCalled();
99+
expect(throwSpy).not.toHaveBeenCalled();
100+
expect(completeSpy).not.toHaveBeenCalled();
101+
102+
clearTimeout(timeout);
103+
done();
104+
});
105+
});
106+
});
107+
108+
describe('when scheduled', function () {
109+
it('should emit one value from a callback', function () {
110+
function callback(datum, cb) {
111+
cb(null, datum);
112+
}
113+
var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
114+
var results = [];
115+
116+
boundCallback(42)
117+
.subscribe(function (x) {
118+
results.push(x);
119+
}, null, function () {
120+
results.push('done');
121+
});
122+
123+
rxTestScheduler.flush();
124+
125+
expect(results).toEqual([42, 'done']);
126+
});
127+
128+
it('should error if callback throws', function () {
129+
function callback(datum, cb) {
130+
throw new Error('haha no callback for you');
131+
}
132+
var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
133+
var results = [];
134+
135+
boundCallback(42)
136+
.subscribe(function (x) {
137+
throw 'should not next';
138+
}, function (err) {
139+
results.push(err);
140+
}, function () {
141+
throw 'should not complete';
142+
});
143+
144+
rxTestScheduler.flush();
145+
146+
expect(results).toEqual([new Error('haha no callback for you')]);
147+
});
148+
149+
it('should raise error from callback', function () {
150+
var error = new Error();
151+
152+
function callback(cb) {
153+
cb(error);
154+
}
155+
156+
var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
157+
var results = [];
158+
159+
boundCallback()
160+
.subscribe(function () {
161+
throw 'should not next';
162+
}, function (err) {
163+
results.push(err);
164+
}, function () {
165+
throw 'should not complete';
166+
});
167+
168+
rxTestScheduler.flush();
169+
170+
expect(results).toEqual([error]);
171+
});
172+
173+
it('should error if selector throws', function () {
174+
function callback(datum, cb) {
175+
cb(null, datum);
176+
}
177+
function selector() {
178+
throw new Error('what? a selector? I don\'t think so');
179+
}
180+
var boundCallback = Observable.bindNodeCallback(callback, selector, rxTestScheduler);
181+
var results = [];
182+
183+
boundCallback(42)
184+
.subscribe(function (x) {
185+
throw 'should not next';
186+
}, function (err) {
187+
results.push(err);
188+
}, function () {
189+
throw 'should not complete';
190+
});
191+
192+
rxTestScheduler.flush();
193+
194+
expect(results).toEqual([new Error('what? a selector? I don\'t think so')]);
195+
});
196+
197+
it('should use a selector', function () {
198+
function callback(datum, cb) {
199+
cb(null, datum);
200+
}
201+
function selector(x) {
202+
return x + '!!!';
203+
}
204+
var boundCallback = Observable.bindNodeCallback(callback, selector, rxTestScheduler);
205+
var results = [];
206+
207+
boundCallback(42)
208+
.subscribe(function (x) {
209+
results.push(x);
210+
}, null, function () {
211+
results.push('done');
212+
});
213+
214+
rxTestScheduler.flush();
215+
216+
expect(results).toEqual(['42!!!', 'done']);
217+
});
218+
});
219+
220+
it('should pass multiple inner arguments as an array', function () {
221+
function callback(datum, cb) {
222+
cb(null, datum, 1, 2, 3);
223+
}
224+
var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
225+
var results = [];
226+
227+
boundCallback(42)
228+
.subscribe(function (x) {
229+
results.push(x);
230+
}, null, function () {
231+
results.push('done');
232+
});
233+
234+
rxTestScheduler.flush();
235+
236+
expect(results).toEqual([[42, 1, 2, 3], 'done']);
237+
});
238+
239+
it('should pass multiple inner arguments to the selector if there is one', function () {
240+
function callback(datum, cb) {
241+
cb(null, datum, 1, 2, 3);
242+
}
243+
function selector(a, b, c, d) {
244+
expect([a, b, c, d]).toEqual([42, 1, 2, 3]);
245+
return a + b + c + d;
246+
}
247+
var boundCallback = Observable.bindNodeCallback(callback, selector, rxTestScheduler);
248+
var results = [];
249+
250+
boundCallback(42)
251+
.subscribe(function (x) {
252+
results.push(x);
253+
}, null, function () {
254+
results.push('done');
255+
});
256+
257+
rxTestScheduler.flush();
258+
259+
expect(results).toEqual([48, 'done']);
260+
});
261+
262+
it('should cache value for next subscription and not call callbackFunc again', function () {
263+
var calls = 0;
264+
function callback(datum, cb) {
265+
calls++;
266+
cb(null, datum);
267+
}
268+
var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
269+
var results1 = [];
270+
var results2 = [];
271+
272+
var source = boundCallback(42);
273+
274+
source.subscribe(function (x) {
275+
results1.push(x);
276+
}, null, function () {
277+
results1.push('done');
278+
});
279+
280+
source.subscribe(function (x) {
281+
results2.push(x);
282+
}, null, function () {
283+
results2.push('done');
284+
});
285+
286+
rxTestScheduler.flush();
287+
288+
expect(calls).toBe(1);
289+
expect(results1).toEqual([42, 'done']);
290+
expect(results2).toEqual([42, 'done']);
291+
});
292+
});

‎src/Observable.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import {concat as concatStatic} from './operator/concat-static';
1717
import {merge as mergeStatic} from './operator/merge-static';
1818
import {zip as zipStatic} from './operator/zip-static';
1919
import {BoundCallbackObservable} from './observable/bindCallback';
20+
import {BoundNodeCallbackObservable} from './observable/bindNodeCallback';
2021
import {DeferObservable} from './observable/defer';
2122
import {EmptyObservable} from './observable/empty';
2223
import {ForkJoinObservable} from './observable/forkJoin';
@@ -165,6 +166,7 @@ export class Observable<T> implements CoreOperators<T> {
165166
// static method stubs
166167
static ajax: AjaxCreationMethod;
167168
static bindCallback: typeof BoundCallbackObservable.create;
169+
static bindNodeCallback: typeof BoundNodeCallbackObservable.create;
168170
static combineLatest: typeof combineLatestStatic;
169171
static concat: typeof concatStatic;
170172
static defer: typeof DeferObservable.create;

‎src/Rx.KitchenSink.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import './add/operator/concat-static';
2828
import './add/operator/merge-static';
2929
import './add/operator/race-static';
3030
import './add/observable/bindCallback';
31+
import './add/observable/bindNodeCallback';
3132
import './add/observable/defer';
3233
import './add/observable/empty';
3334
import './add/observable/forkJoin';

‎src/Rx.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import './add/operator/concat-static';
1313
import './add/operator/merge-static';
1414
import './add/operator/race-static';
1515
import './add/observable/bindCallback';
16+
import './add/observable/bindNodeCallback';
1617
import './add/observable/defer';
1718
import './add/observable/empty';
1819
import './add/observable/forkJoin';

‎src/add/observable/bindNodeCallback.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
import {Observable} from '../../Observable';
2+
import {BoundNodeCallbackObservable} from '../../observable/bindNodeCallback';
3+
Observable.bindNodeCallback = BoundNodeCallbackObservable.create;
4+
5+
export var _void: void;

0 commit comments

Comments
 (0)