feat(bindNodeCallback): add Observable.bindNodeCallback

- add bindNodeCallback corresponding to fromNodeCallback in RxJS4

closes #736

MIGRATION.md

@@ -59,6 +59,8 @@ enabling "composite" subscription behavior.
 |`flatMapFirst`|`exhaustMap`|
 |`flatMapLatest`|`switchMap`|
 |`flatMapWithMaxConcurrent`|`mergeMap` or `flatMap`(alias)|
+|`fromCallback`|`bindCallback`|
+|`fromNodeCallback`|`bindNodeCallback`|
 |`publishValue`|`publishBehavior`|
 |`replay`|`publishReplay`|
 |`select`|`map`|

spec/observables/bindNodeCallback-spec.js

@@ -0,0 +1,292 @@
+/* globals describe, it, expect, rxTestScheduler */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.bindNodeCallback', function () {
+ describe('when not scheduled', function () {
+ it('should emit one value from a callback', function () {
+ function callback(datum, cb) {
+ cb(null, datum);
+ }
+ var boundCallback = Observable.bindNodeCallback(callback);
+ var results = [];
+
+ boundCallback(42)
+ .subscribe(function (x) {
+ results.push(x);
+ }, null, function () {
+ results.push('done');
+ });
+
+ expect(results).toEqual([42, 'done']);
+ });
+
+ it('should emit one value chosen by a selector', function () {
+ function callback(datum, cb) {
+ cb(null, datum);
+ }
+ var boundCallback = Observable.bindNodeCallback(callback, function (datum) { return datum; });
+ var results = [];
+
+ boundCallback(42)
+ .subscribe(function (x) {
+ results.push(x);
+ }, null, function () {
+ results.push('done');
+ });
+
+ expect(results).toEqual([42, 'done']);
+ });
+
+ it('should raise error from callback', function () {
+ var error = new Error();
+
+ function callback(cb) {
+ cb(error);
+ }
+
+ var boundCallback = Observable.bindNodeCallback(callback);
+ var results = [];
+
+ boundCallback()
+ .subscribe(function () {
+ throw 'should not next';
+ }, function (err) {
+ results.push(err);
+ }, function () {
+ throw 'should not complete';
+ });
+
+ expect(results).toEqual([error]);
+ });
+
+ it('should emit an error when the selector throws', function () {
+ function callback(cb) {
+ cb(null, 42);
+ }
+ var boundCallback = Observable.bindNodeCallback(callback, function (err) { throw new Error('Yikes!'); });
+ var results = [];
+
+ boundCallback()
+ .subscribe(function () {
+ throw 'should not next';
+ }, function (err) {
+ results.push(err);
+ }, function () {
+ throw 'should not complete';
+ });
+
+ expect(results).toEqual([new Error('Yikes!')]);
+ });
+
+ it('should not emit, throw or complete if immediately unsubscribed', function (done) {
+ var nextSpy = jasmine.createSpy('next');
+ var throwSpy = jasmine.createSpy('throw');
+ var completeSpy = jasmine.createSpy('complete');
+ var timeout;
+ function callback(datum, cb) {
+ // Need to cb async in order for the unsub to trigger
+ timeout = setTimeout(function () {
+ cb(null, datum);
+ });
+ }
+ var subscription = Observable.bindNodeCallback(callback)(42)
+ .subscribe(nextSpy, throwSpy, completeSpy);
+ subscription.unsubscribe();
+
+ setTimeout(function () {
+ expect(nextSpy).not.toHaveBeenCalled();
+ expect(throwSpy).not.toHaveBeenCalled();
+ expect(completeSpy).not.toHaveBeenCalled();
+
+ clearTimeout(timeout);
+ done();
+ });
+ });
+ });
+
+ describe('when scheduled', function () {
+ it('should emit one value from a callback', function () {
+ function callback(datum, cb) {
+ cb(null, datum);
+ }
+ var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
+ var results = [];
+
+ boundCallback(42)
+ .subscribe(function (x) {
+ results.push(x);
+ }, null, function () {
+ results.push('done');
+ });
+
+ rxTestScheduler.flush();
+
+ expect(results).toEqual([42, 'done']);
+ });
+
+ it('should error if callback throws', function () {
+ function callback(datum, cb) {
+ throw new Error('haha no callback for you');
+ }
+ var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
+ var results = [];
+
+ boundCallback(42)
+ .subscribe(function (x) {
+ throw 'should not next';
+ }, function (err) {
+ results.push(err);
+ }, function () {
+ throw 'should not complete';
+ });
+
+ rxTestScheduler.flush();
+
+ expect(results).toEqual([new Error('haha no callback for you')]);
+ });
+
+ it('should raise error from callback', function () {
+ var error = new Error();
+
+ function callback(cb) {
+ cb(error);
+ }
+
+ var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
+ var results = [];
+
+ boundCallback()
+ .subscribe(function () {
+ throw 'should not next';
+ }, function (err) {
+ results.push(err);
+ }, function () {
+ throw 'should not complete';
+ });
+
+ rxTestScheduler.flush();
+
+ expect(results).toEqual([error]);
+ });
+
+ it('should error if selector throws', function () {
+ function callback(datum, cb) {
+ cb(null, datum);
+ }
+ function selector() {
+ throw new Error('what? a selector? I don\'t think so');
+ }
+ var boundCallback = Observable.bindNodeCallback(callback, selector, rxTestScheduler);
+ var results = [];
+
+ boundCallback(42)
+ .subscribe(function (x) {
+ throw 'should not next';
+ }, function (err) {
+ results.push(err);
+ }, function () {
+ throw 'should not complete';
+ });
+
+ rxTestScheduler.flush();
+
+ expect(results).toEqual([new Error('what? a selector? I don\'t think so')]);
+ });
+
+ it('should use a selector', function () {
+ function callback(datum, cb) {
+ cb(null, datum);
+ }
+ function selector(x) {
+ return x + '!!!';
+ }
+ var boundCallback = Observable.bindNodeCallback(callback, selector, rxTestScheduler);
+ var results = [];
+
+ boundCallback(42)
+ .subscribe(function (x) {
+ results.push(x);
+ }, null, function () {
+ results.push('done');
+ });
+
+ rxTestScheduler.flush();
+
+ expect(results).toEqual(['42!!!', 'done']);
+ });
+ });
+
+ it('should pass multiple inner arguments as an array', function () {
+ function callback(datum, cb) {
+ cb(null, datum, 1, 2, 3);
+ }
+ var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
+ var results = [];
+
+ boundCallback(42)
+ .subscribe(function (x) {
+ results.push(x);
+ }, null, function () {
+ results.push('done');
+ });
+
+ rxTestScheduler.flush();
+
+ expect(results).toEqual([[42, 1, 2, 3], 'done']);
+ });
+
+ it('should pass multiple inner arguments to the selector if there is one', function () {
+ function callback(datum, cb) {
+ cb(null, datum, 1, 2, 3);
+ }
+ function selector(a, b, c, d) {
+ expect([a, b, c, d]).toEqual([42, 1, 2, 3]);
+ return a + b + c + d;
+ }
+ var boundCallback = Observable.bindNodeCallback(callback, selector, rxTestScheduler);
+ var results = [];
+
+ boundCallback(42)
+ .subscribe(function (x) {
+ results.push(x);
+ }, null, function () {
+ results.push('done');
+ });
+
+ rxTestScheduler.flush();
+
+ expect(results).toEqual([48, 'done']);
+ });
+
+ it('should cache value for next subscription and not call callbackFunc again', function () {
+ var calls = 0;
+ function callback(datum, cb) {
+ calls++;
+ cb(null, datum);
+ }
+ var boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
+ var results1 = [];
+ var results2 = [];
+
+ var source = boundCallback(42);
+
+ source.subscribe(function (x) {
+ results1.push(x);
+ }, null, function () {
+ results1.push('done');
+ });
+
+ source.subscribe(function (x) {
+ results2.push(x);
+ }, null, function () {
+ results2.push('done');
+ });
+
+ rxTestScheduler.flush();
+
+ expect(calls).toBe(1);
+ expect(results1).toEqual([42, 'done']);
+ expect(results2).toEqual([42, 'done']);
+ });
+});

src/Observable.ts

@@ -17,6 +17,7 @@ import {concat as concatStatic} from './operator/concat-static';
 import {merge as mergeStatic} from './operator/merge-static';
 import {zip as zipStatic} from './operator/zip-static';
 import {BoundCallbackObservable} from './observable/bindCallback';
+import {BoundNodeCallbackObservable} from './observable/bindNodeCallback';
 import {DeferObservable} from './observable/defer';
 import {EmptyObservable} from './observable/empty';
 import {ForkJoinObservable} from './observable/forkJoin';
@@ -165,6 +166,7 @@ export class Observable<T> implements CoreOperators<T> {
  // static method stubs
  static ajax: AjaxCreationMethod;
  static bindCallback: typeof BoundCallbackObservable.create;
+ static bindNodeCallback: typeof BoundNodeCallbackObservable.create;
  static combineLatest: typeof combineLatestStatic;
  static concat: typeof concatStatic;
  static defer: typeof DeferObservable.create;

src/Rx.KitchenSink.ts

@@ -28,6 +28,7 @@ import './add/operator/concat-static';
 import './add/operator/merge-static';
 import './add/operator/race-static';
 import './add/observable/bindCallback';
+import './add/observable/bindNodeCallback';
 import './add/observable/defer';
 import './add/observable/empty';
 import './add/observable/forkJoin';

src/Rx.ts

@@ -13,6 +13,7 @@ import './add/operator/concat-static';
 import './add/operator/merge-static';
 import './add/operator/race-static';
 import './add/observable/bindCallback';
+import './add/observable/bindNodeCallback';
 import './add/observable/defer';
 import './add/observable/empty';
 import './add/observable/forkJoin';

src/add/observable/bindNodeCallback.ts

@@ -0,0 +1,5 @@
+import {Observable} from '../../Observable';
+import {BoundNodeCallbackObservable} from '../../observable/bindNodeCallback';
+Observable.bindNodeCallback = BoundNodeCallbackObservable.create;
+
+export var _void: void;