bindcallback

Author: mattpodwysocki

src/modular/index.js

@@ -9,48 +9,152 @@ Observer.addToObject({
 var Observable = require('./observable');
 
 Observable.addToObject({
+  bindCallback: require('./observable/bindcallback'),
+  bindNodeCallback: require('./observable/bindnodecallback'),
+  case: require('./observable/case'),
   catch: require('./observable/catch'),
+  combineLatest: require('./observable/combinelatest'),
   concat: require('./observable/concat'),
   create: require('./observable/create'),
+  defer: require('./observable/defer'),
   empty: require('./observable/empty'),
+  forkJoin: require('./observable/forkjoin'),
   from: require('./observable/from'),
   fromArray: require('./observable/fromarray'),
-  fromCallback: require('./observable/fromcallback'),
   fromEvent: require('./observable/fromevent'),
   fromEventPattern: require('./observable/fromeventpattern'),
-  fromNodeCallback: require('./observable/fromnodecallback'),
   fromPromise: require('./observable/frompromise'),
+  generate: require('./observable/generate'),
+  generateAbsolute: require('./observable/generateabsolute'),
+  generateRelative: require('./observable/generaterelative'),
   interval: require('./observable/interval'),
   just: require('./observable/just'),
   merge: require('./observable/merge'),
   never: require('./observable/never'),
   of: require('./observable/of'),
+  ofScheduled: require('./observable/ofscheduled'),
+  onErrorResumeNext: require('./observable/onerrorresumenext'),
+  pairs: require('./observable/pairs'),
   range: require('./observable/range'),
-  throw: require('./observable/throw')
+  repeat: require('./observable/repeatvalue'),
+  sequenceEqual: require('./observable/sequenceequal'),
+  start: require('./observable/start'),
+  startAsync: require('./observable/startasync'),
+  throw: require('./observable/throw'),
+  timer: require('./observable/timer'),
+  toAsync: require('./observable/toasync'),
+  when: require('./observable/when'),
+  zip: require('./observable/zip')
 });
 
 Observable.addToPrototype({
+  and: require('./observable/and'),
+  asObservable: require('./observable/asobservable'),
+  average: require('./observable/average'),
+  buffer: require('./observable/buffer'),
+  bufferCount: require('./observable/buffercount'),
+  bufferTime: require('./observable/buffertime'),
+  bufferTimeOrCount: require('./observable/buffertimeorcount'),
   catch: require('./observable/catch'),
+  catchHandler: require('./observable/catchhandler'),
   combineLatest: require('./observable/combinelatest'),
   concat: require('./observable/concat'),
   concatAll: require('./observable/concatall'),
+  count: require('./observable/count'),
   debounce: require('./observable/debounce'),
+  defaultIfEmpty: require('./observable/defaultifempty'),
+  delay: require('./observable/delay'),
+  delaySubscription: require('./observable/delaySubscription'),
+  dematerialize: require('./observable/dematerialize'),
+  distinct: require('./observable/distinct'),
   distinctUntilChanged: require('./observable/distinctuntilchanged'),
+  every: require('./observable/every'),
   filter: require('./observable/filter'),
   finally: require('./observable/finally'),
+  find: require('./observable/find'),
+  findIndex: require('./observable/findindex'),
+  first: require('./observable/first'),
   flatMap: require('./observable/flatMap'),
   flatMapLatest: require('./observable/flatmaplatest'),
+  forkJoin: require('./observable/forkjoin'),
+  groupJoin: require('./observable/groupjoin'),
+  ignoreElements: require('./observable/ignoreelements'),
+  includes: require('./observable/includes'),
+  indexOf: require('./observable/indexof'),
+  isEmpty: require('./observable/isempty'),
+  join: require('./observable/join'),
+  last: require('./observable/last'),
+  lastIndexOf: require('./observable/lastindexof'),
   map: require('./observable/map'),
+  materialize: require('./observable/materialize'),
+  max: require('./observable/max'),
+  maxBy: require('./observable/maxby'),
   merge: require('./observable/merge'),
   mergeAll: require('./observable/mergeall'),
+  mergeConcat: require('./observable/mergeconcat'),
+  min: require('./observable/min'),
+  minBy: require('./observable/minby'),
+  multicast: require('./observable/multicast'),
+  observeOn: require('./observable/observeon'),
+  onErrorResumeNext: require('./observable/onerrorresumenext'),
+  pairwise: require('./observable/pairwise'),
+  partition: require('./observable/partition'),
+  pluck: require('./observable/pluck'),
+  publish: require('./observable/publish'),
+  publishLast: require('./observable/publishlast'),
+  publishValue: require('./observable/publishvalue'),
+  reduce: require('./observable/reduce'),
+  repeat: require('./observable/repeat'),
+  repeatWhen: require('./observable/repeatwhen'),
+  replay: require('./observable/replay'),
+  retry: require('./observable/retry'),
+  retryWhen: require('./observable/retrywhen'),
+  sample: require('./observable/sample'),
   scan: require('./observable/scan'),
+  sequenceEqual: require('./observable/sequenceequal'),
+  share: require('./observable/share'),
+  shareReplay: require('./observable/sharereplay'),
+  shareValue: require('./observable/sharevalue'),
   skip: require('./observable/skip'),
+  skipLast: require('./observable/skiplast'),
+  skipLastTime: require('./observable/skiplastwithtime'),
   skipUntil: require('./observable/skipuntil'),
+  skipUntilTime: require('./observable/skipuntilwithtime'),
+  skipWhile: require('./observable/skipwhile'),
+  slice: require('./observable/slice'),
+  subscribeOn: require('./observable/subscribeon'),
+  some: require('./observable/some'),
+  sum: require('./observable/sum'),
   switch: require('./observable/switch'),
-  take: require('./obserable/take'),
+  switchFirst: require('./observable/switchfirst'),
+  take: require('./observable/take'),
+  takeLastBuffer: require('./observable/takelastbuffer'),
+  takeLastBufferTime: require('./observable/takelastbufferwithtime'),
+  lastLastTime: require('./observable/takelastwithtime'),
   takeUntil: require('./observable/takeuntil'),
+  takeUntilTime: require('./observable/takeuntilwithtime'),
+  takeWhile: require('./observable/takewhile'),
   tap: require('./observable/tap'),
-  zip: require('./observable/zip')
+  thenDo: require('./observable/thendo'),
+  throttle: require('./observable/throttle'),
+  timeInterval: require('./observable/timeinterval'),
+  timestamp: require('./observable/timestamp'),
+  toArray: require('./observable/toarray'),
+  toMap: require('./observable/tomap'),
+  toSet: require('./observable/toset'),
+  transduce: require('./observable/transduce'),
+  window: require('./observable/window'),
+  windowCount: require('./observable/windowcount'),
+  windowTime: require('./observable/windowtime'),
+  windowTimeOrCount: require('./observable/windowtimeorcount'),
+  withLatestFrom: require('./observable/withlatestfrom'),
+  zip: require('./observable/zip'),
+  zipIterable: require('./observable/zipiterable'),
+});
+
+var Subject = require('./subject');
+Subject.addToObject({
+  create: require('./subject/create')
 });
 
 var Rx = {
@@ -69,7 +173,7 @@ var Rx = {
   AsyncSubject: require('./asyncsubject'),
   BehaviorSubject: require('./behaviorsubject'),
   ReplaySubject: require('./replaysubject'),
-  Subject: require('./subject')
+  Subject: Subject
 };
 
 module.exports = Rx;

src/modular/observable/fromcallback.js src/modular/observable/bindcallback.jssrc/modular/observable/fromcallback.js renamed to src/modular/observable/bindcallback.js

@@ -1,9 +1,9 @@
 'use strict';
 
 var AsyncSubject = require('../asyncsubject');
+var asObservable  = require('./asobservable');
 var isFunction = require('../helpers/isfunction');
-var tryCatchUtils = require('../internal/trycatchutils');
-var tryCatch = tryCatchUtils.tryCatch;
+var tryCatch = require('../internal/trycatchutils').tryCatch;
 
 function createCbHandler(o, ctx, selector) {
   return function handler () {
@@ -30,9 +30,10 @@ function createCbObservable(fn, ctx, selector, args) {
   var o = new AsyncSubject();
 
   args.push(createCbHandler(o, ctx, selector));
-  fn.apply(ctx, args);
+  var res = tryCatch(fn).apply(ctx, args);
+  if (res === global._Rx.errorObj) { o.onError(res.e); }
 
-  return o.asObservable();
+  return asObservable(o);
 }
 
 /**
@@ -43,7 +44,7 @@ function createCbObservable(fn, ctx, selector, args) {
  * @param {Function} [selector] A selector which takes the arguments from the callback to produce a single item to yield on next.
  * @returns {Function} A function, when executed with the required parameters minus the callback, produces an Observable sequence with a single value of the arguments to the callback as an array.
  */
-module.exports = function fromCallback (fn, ctx, selector) {
+module.exports = function bindCallback (fn, ctx, selector) {
   return function () {
     typeof ctx === 'undefined' && (ctx = this);
 

src/modular/observable/bindnodecallback.js

@@ -0,0 +1,56 @@
+'use strict';
+
+var AsyncSubject = require('../asyncsubject');
+var asObservable  = require('./asobservable');
+var isFunction = require('../helpers/isfunction');
+var tryCatch = require('../internal/trycatchutils').tryCatch;
+
+function createNodeHandler(o, ctx, selector) {
+  return function handler () {
+    var err = arguments[0];
+    if (err) { return o.onError(err); }
+
+    var len = arguments.length, results = [];
+    for(var i = 1; i < len; i++) { results[i - 1] = arguments[i]; }
+
+    if (isFunction(selector)) {
+      results = tryCatch(selector).apply(ctx, results);
+      if (results === global._Rx.errorObj) { return o.onError(results.e); }
+      o.onNext(results);
+    } else {
+      if (results.length <= 1) {
+        o.onNext(results[0]);
+      } else {
+        o.onNext(results);
+      }
+    }
+
+    o.onCompleted();
+  };
+}
+
+function createNodeObservable(fn, ctx, selector, args) {
+  var o = new AsyncSubject();
+
+  args.push(createNodeHandler(o, ctx, selector));
+  var res = tryCatch(fn).apply(ctx, args);
+  if (res === global._Rx.errorObj) { o.onError(res.e); }
+
+  return asObservable(o);
+}
+
+/**
+ * Converts a Node.js callback style function to an observable sequence.  This must be in function (err, ...) format.
+ * @param {Function} fn The function to call
+ * @param {Mixed} [ctx] The context for the func parameter to be executed.  If not specified, defaults to undefined.
+ * @param {Function} [selector] A selector which takes the arguments from the callback minus the error to produce a single item to yield on next.
+ * @returns {Function} An async function which when applied, returns an observable sequence with the callback arguments as an array.
+ */
+module.exports = function bindNodeCallback (fn, ctx, selector) {
+  return function () {
+    typeof ctx === 'undefined' && (ctx = this);
+    var len = arguments.length, args = new Array(len);
+    for(var i = 0; i < len; i++) { args[i] = arguments[i]; }
+    return createNodeObservable(fn, ctx, selector, args);
+  };
+};

src/modular/test/bindcallback.js

@@ -0,0 +1,139 @@
+'use strict';
+
+var test = require('tape');
+var Observable = require('../observable');
+var TestScheduler = require('../testing/testscheduler');
+var reactiveAssert = require('../testing/reactiveassert');
+var ReactiveTest = require('../testing/reactivetest');
+var onNext = ReactiveTest.onNext,
+  onError = ReactiveTest.onError,
+  onCompleted = ReactiveTest.onCompleted;
+
+Observable.addToObject({
+  bindCallback: require('../observable/bindcallback')
+});
+
+test('Observable.bindCallback', function (t) {
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.bindCallback(function (cb) { cb(true); })();
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, true),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.bindCallback throws', function (t) {
+  var error = new Error();
+
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.bindCallback(function () { throw error; })();
+  });
+
+  reactiveAssert(t, results.messages, [
+    onError(200, error)
+  ]);
+
+  t.end();
+});
+
+test('Observable.bindCallback single argument', function (t) {
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.bindCallback(function (file, cb) { cb(file); })('file.txt');
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, 'file.txt'),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.bindCallback selector', function (t) {
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.bindCallback(
+      function (f, s, t, cb) { cb(1,2,3); },
+      null,
+      function (f) { return f; })(1,2,3);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, 1),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.bindCallback selector throws', function (t) {
+  var error = new Error();
+
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.bindCallback(
+      function (f, s, t, cb) { cb(1,2,3); },
+      null,
+      function () { throw error; })(1,2,3);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onError(200, error)
+  ]);
+
+  t.end();
+});
+
+test('Observable.bindCallback ctx', function (t) {
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.bindCallback(
+      function (cb) { t.equal(this, 42); cb(null); },
+      42)();
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, null),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.bindCallback resubscribe', function (t) {
+  var count = 0;
+
+  var res = Observable.bindCallback( function(cb) { cb(++count); })();
+
+  res.subscribe(function (x) {
+    t.equal(x, 1);
+  }, function () {
+    t.fail();
+  }, function () {
+    t.ok(true);
+  });
+
+  res.subscribe(function (x) {
+    t.equal(x, 1);
+  }, function () {
+    t.fail();
+  }, function () {
+    t.ok(true);
+  });
+
+  t.equal(1, count);
+  t.end();
+});