toAsync

Author: mattpodwysocki

src/modular/observable/toasync.js

@@ -0,0 +1,35 @@
+'use strict';
+
+var AsyncSubject = require('../asyncsubject');
+var asObservable = require('./asobservable');
+var isScheduler = require('../scheduler').isScheduler;
+var tryCatch = require('../internal/trycatchutils').tryCatch;
+
+global._Rx || (global._Rx = {});
+if (!global._Rx.defaultScheduler) {
+  require('../scheduler/defaultscheduler');
+}
+
+function scheduleMethod(s, state) {
+  var result = tryCatch(state.func).apply(state.context, state.args);
+  if (result === global._Rx.errorObj) { return state.subject.onError(result.e); }
+  state.subject.onNext(result);
+  state.subject.onCompleted();
+}
+
+module.exports = function toAsync(func, context, scheduler) {
+  isScheduler(scheduler) || (scheduler = global._Rx.defaultScheduler);
+  return function asyncFn () {
+    var subject = new AsyncSubject(), len = arguments.length, args = new Array(len);
+    for (var i = 0; i < len; i++) { args[i] = arguments[i]; }
+    var state = {
+      subject: subject,
+      args: args,
+      func: func,
+      context: context
+    };
+
+    scheduler.schedule(state, scheduleMethod);
+    return asObservable(subject);
+  };
+};

src/modular/test/toasync.js

@@ -0,0 +1,136 @@
+'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({
+  toAsync: require('../observable/toasync')
+});
+
+test('Observable.toAsync context', function (t) {
+  var context = { value: 42 };
+
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.toAsync(function (x) {
+      return this.value + x;
+    }, context, scheduler)(42);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, 84),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.toAsync 0', function (t) {
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.toAsync(function () {
+      return 0;
+    }, null, scheduler)();
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, 0),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.toAsync 1', function (t) {
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.toAsync(function (x) {
+      return x;
+    }, null, scheduler)(1);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, 1),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.toAsync 2', function (t) {
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.toAsync(function (x, y) {
+        return x + y;
+    }, null, scheduler)(1, 2);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, 3),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.toAsync 3', function (t) {
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.toAsync(function (x, y, z) {
+      return x + y + z;
+    }, null, scheduler)(1, 2, 3);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, 6),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.toAsync 4', function (t) {
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.toAsync(function (a, b, c, d) {
+      return a + b + c + d;
+    }, null, scheduler)(1, 2, 3, 4);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(200, 10),
+    onCompleted(200)
+  ]);
+
+  t.end();
+});
+
+test('Observable.toAsync error', function (t) {
+  var error = new Error();
+
+  var scheduler = new TestScheduler();
+
+  var results = scheduler.startScheduler(function () {
+    return Observable.toAsync(function () {
+      throw error;
+    }, null, scheduler)();
+  });
+
+  reactiveAssert(t, results.messages, [
+    onError(200, error)
+  ]);
+
+  t.end();
+});