Handle Falcor style observables from unhandled path datasource.

lrowe · lrowe · commit 07ce5ff874fa · 2017-05-31T19:49:25.000-07:00
diff --git a/src/router/call.js b/src/router/call.js
@@ -9,6 +9,7 @@ var collapse = pathUtils.collapse;
 var Observable = require('../RouterRx.js').Observable;
 var MaxPathsExceededError = require('../errors/MaxPathsExceededError');
 var getPathsCount = require('./getPathsCount');
+var outputToObservable = require('../run/conversion/outputToObservable');
 var rxNewToRxNewAndOld = require('../run/conversion/rxNewToRxNewAndOld');
 
 /**
@@ -88,8 +89,9 @@ module.exports = function routerCall(callPath, args,
             // we will try the next dataSource in the line.
                 catch(function catchException(e) {
                     if (e instanceof CallNotFoundError && router._unhandled) {
-                        return router._unhandled.
-                            call(callPath, args, refPaths, thisPaths);
+                        return outputToObservable(
+                            router._unhandled.
+                            call(callPath, args, refPaths, thisPaths));
                     }
                     throw e;
                 });
diff --git a/src/router/get.js b/src/router/get.js
@@ -7,6 +7,7 @@ var Observable = require('../RouterRx.js').Observable;
 var mCGRI = require('./../run/mergeCacheAndGatherRefsAndInvalidations');
 var MaxPathsExceededError = require('../errors/MaxPathsExceededError');
 var getPathsCount = require('./getPathsCount');
+var outputToObservable = require('../run/conversion/outputToObservable');
 var rxNewToRxNewAndOld = require('../run/conversion/rxNewToRxNewAndOld');
 
 /**
@@ -56,8 +57,8 @@ module.exports = function routerGet(paths) {
                         // The 3rd argument is the beginning of the actions
                         // arguments, which for get is the same as the
                         // unhandledPaths.
-                        return router._unhandled.
-                            get(unhandledPaths).
+                        return outputToObservable(
+                            router._unhandled.get(unhandledPaths)).
 
                             // Merge the solution back into the overall message.
                             map(function(jsonGraphFragment) {
diff --git a/src/router/set.js b/src/router/set.js
@@ -15,6 +15,7 @@ var collapse = pathUtils.collapse;
 var mCGRI = require('./../run/mergeCacheAndGatherRefsAndInvalidations');
 var MaxPathsExceededError = require('../errors/MaxPathsExceededError');
 var getPathsCount = require('./getPathsCount');
+var outputToObservable = require('../run/conversion/outputToObservable');
 var rxNewToRxNewAndOld = require('../run/conversion/rxNewToRxNewAndOld');
 
 /**
@@ -141,8 +142,8 @@ module.exports = function routerSet(jsonGraph) {
                             return pV.path;
                         }));
 
-                    return router._unhandled.
-                        set(jsonGraphEnvelope).
+                    return outputToObservable(
+                        router._unhandled.set(jsonGraphEnvelope)).
 
                         // Merge the solution back into the overall message.
                         map(function(unhandledJsonGraphEnv) {
diff --git a/test/FalcorObservable.js b/test/FalcorObservable.js
@@ -0,0 +1,78 @@
+"use strict";
+
+function noop() {
+    // do nothing
+}
+
+var disposeNoop = { dispose: noop };
+
+function disposable(dispose) {
+    if (typeof dispose === "function") {
+        return { dispose: dispose };
+    }
+    if (typeof dispose === "undefined" || dispose === null) {
+        return disposeNoop;
+    }
+    return dispose;
+}
+
+function functionsObserver(onNext, onError, onCompleted) {
+    return {
+        onNext: typeof onNext === "function" ? onNext : noop,
+        onError: typeof onError === "function" ? onError : noop,
+        onCompleted: typeof onCompleted === "function" ? onCompleted : noop
+    };
+}
+
+function partialObserver(partial) {
+    return {
+        onNext: typeof partial.onNext === "function"
+            ? partial.onNext.bind(partial)
+            : noop,
+        onError: function onError(e) {
+            if (typeof partial.onError === "function") {
+                partial.onError(e);
+            }
+        },
+        onCompleted: function onCompleted() {
+            if (typeof partial.onCompleted === "function") {
+                partial.onCompleted();
+            }
+        }
+    };
+}
+
+function observer(observerOrOnNext, onError, onCompleted) {
+    return typeof observerOrOnNext === "object" && observerOrOnNext !== null
+        ? partialObserver(observerOrOnNext)
+        : functionsObserver(observerOrOnNext, onError, onCompleted);
+}
+
+function create(subscribe) {
+    return {
+        subscribe: function(observerOrOnNext, onError, onCompleted) {
+            return disposable(
+                subscribe(observer(observerOrOnNext, onError, onCompleted))
+            );
+        }
+    };
+}
+
+function empty() {
+    return create(function(obs) {
+        obs.onCompleted();
+    });
+}
+
+function of(value) {
+    return create(function(obs) {
+        obs.onNext(value);
+        obs.onCompleted();
+    });
+}
+
+module.exports = {
+    create: create,
+    empty: empty,
+    of: of
+};
diff --git a/test/unit/functional/unhandled.call.spec.js b/test/unit/functional/unhandled.call.spec.js
@@ -3,13 +3,13 @@ var noOp = function() {};
 var chai = require('chai');
 var expect = chai.expect;
 var sinon = require('sinon');
-var Observable = require('../../../src/RouterRx').Observable;
+var FalcorObservable = require('../../FalcorObservable');
 
 describe('#call', function() {
     it('should ensure a missing function gets chained.', function(done) {
         var router = new R([]);
         var onCall = sinon.spy(function() {
-            return Observable.of({
+            return FalcorObservable.of({
                 jsonGraph: {
                     videos: {
                         summary: 5
@@ -45,7 +45,7 @@ describe('#call', function() {
     it('should ensure a missing function gets chained and will not materialize properly.', function(done) {
         var router = new R([]);
         var onCall = sinon.spy(function() {
-            return Observable.of({
+            return FalcorObservable.of({
                 jsonGraph: { },
                 paths: [
                     ['videos', 'summary']
diff --git a/test/unit/functional/unhandled.get.spec.js b/test/unit/functional/unhandled.get.spec.js
@@ -4,14 +4,14 @@ var chai = require('chai');
 var expect = chai.expect;
 var sinon = require('sinon');
 var pathValueMerge = require('./../../../src/cache/pathValueMerge');
-var Observable = require('../../../src/RouterRx').Observable;
+var FalcorObservable = require('../../FalcorObservable');
 var $atom = require('./../../../src/support/types').$atom;
 
 describe('#get', function() {
     it('should return an empty Observable and just materialize values.', function(done) {
         var router = new R([]);
         var onUnhandledPaths = sinon.spy(function convert(paths) {
-            return Observable.empty();
+            return FalcorObservable.empty();
         });
         router.routeUnhandledPathsTo({get: onUnhandledPaths});
 
@@ -45,7 +45,7 @@ describe('#get', function() {
                 });
                 return jsonGraph;
             }, {jsonGraph: {}});
-            return Observable.of(returnValue);
+            return FalcorObservable.of(returnValue);
         });
         router.routeUnhandledPathsTo({get: onUnhandledPaths});
 
@@ -88,7 +88,7 @@ describe('#get', function() {
                 });
                 return jsonGraph;
             }, {jsonGraph: {}});
-            return Observable.of(returnValue);
+            return FalcorObservable.of(returnValue);
         });
         router.routeUnhandledPathsTo({get: onUnhandledPaths});
 
diff --git a/test/unit/functional/unhandled.set.spec.js b/test/unit/functional/unhandled.set.spec.js
@@ -4,15 +4,15 @@ var chai = require('chai');
 var expect = chai.expect;
 var sinon = require('sinon');
 var pathValueMerge = require('./../../../src/cache/pathValueMerge');
-var Observable = require('../../../src/RouterRx').Observable;
+var FalcorObservable = require('../../FalcorObservable');
 var $atom = require('./../../../src/support/types').$atom;
 var $ref = require('./../../../src/support/types').$ref;
 
 describe('#set', function() {
     it('should return an empty Observable and just materialize values.', function(done) {
         var router = new R([]);
         var onUnhandledPaths = sinon.spy(function convert(paths) {
-            return Observable.empty();
+            return FalcorObservable.empty();
         });
         router.routeUnhandledPathsTo({set: onUnhandledPaths});
 
@@ -63,7 +63,7 @@ describe('#set', function() {
                 return jsonGraph;
             }, {jsonGraph: {}});
 
-            return Observable.of(returnValue);
+            return FalcorObservable.of(returnValue);
         });
         router.routeUnhandledPathsTo({set: onUnhandledPaths});
 
@@ -122,7 +122,7 @@ describe('#set', function() {
                 });
                 return jsonGraph;
             }, {jsonGraph: {}});
-            return Observable.of(returnValue);
+            return FalcorObservable.of(returnValue);
         });
         router.routeUnhandledPathsTo({set: onUnhandledPaths});
 
@@ -202,7 +202,7 @@ describe('#set', function() {
                 path: unicorn,
                 value: 'missing'
             });
-            return Observable.of(next);
+            return FalcorObservable.of(next);
         });
         router.routeUnhandledPathsTo({set: onUnhandledPaths});
 
