Merge pull request #96 from ThunderCatsJS/feature/action-map-observable

[Add] Action map now accepts observables

circle.yml

@@ -4,4 +4,4 @@ general:
       - gh-pages
 machine:
   node:
-    version: 4.1.1
+    version: 4.2.1

docs/API.md

@@ -21,7 +21,14 @@ methods of these factories instances. These are taken as the specifications of t
 For every key on spec,
 there will be a corresponding method with that name. If the keys value on spec
 is a function, it is used as an initial mapping function. If value is null, the
-indentity function, `((x) => x)` is used.
+indentity function, `((x) => x)`, is used.
+
+A mapping function has the following signature
+
+`(value : any) : any|Observable<any>`
+
+If a mapping function returns an observable, that observable is subscribed to
+and those values it returns are passed to that actions observables.
 
 ### ActionsFactory(instanceProperties) : actions
 

src/Actions.js

@@ -1,4 +1,4 @@
-import Rx from 'rx';
+import { Observable, Subject, Disposable, helpers } from 'rx';
 import stampit from 'stampit';
 import debugFactory from 'debug';
 
@@ -32,7 +32,7 @@ export function getActionDef(ctx) {
     .map(name => ({ name: name, map: ctx[name] }))
     .map(def => {
       if (typeof def.map !== 'function') {
-        def.map = Rx.helpers.identity;
+        def.map = helpers.identity;
       }
       return def;
     });
@@ -41,33 +41,42 @@ export function getActionDef(ctx) {
 
 export function create(shouldBind, { name, map }) {
   let observers = [];
-  let actionStart = new Rx.Subject();
+  let actionStart = new Subject();
   let maybeBound = shouldBind ?
     map.bind(this) :
     map;
 
   function action(value) {
-    let err = null;
-    try {
-      value = maybeBound(value);
-    } catch (e) {
-      err = e;
-    }
-
-    actionStart.onNext(value);
-    observers.forEach((observer) => {
-      if (err) {
-        return observer.onError(err);
-      }
-      observer.onNext(value);
-    });
+    Observable.just(value)
+      .map(maybeBound)
+      .flatMap(value => {
+        if (Observable.isObservable(value)) {
+          return value;
+        }
+        return Observable.just(value);
+      })
+      // notify of action start
+      .do(value => actionStart.onNext(value))
+      // notify action observers
+      .doOnNext(
+        value => observers.forEach(observer => observer.onNext(value))
+      )
+      // notify action observers of error
+      .doOnError(
+        value => observers.forEach(observer => observer.onError(value))
+      )
+      // prevent error from calling on error
+      .catch(e => Observable.just(e))
+      .subscribe(
+        () => debug('action % onNext', name)
+      );
 
     return value;
   }
 
   action.displayName = name;
   action.observers = observers;
-  assign(action, Rx.Observable.prototype);
+  assign(action, Observable.prototype);
 
   action.hasObservers = function hasObservers() {
     return observers.length > 0 ||
@@ -81,12 +90,12 @@ export function create(shouldBind, { name, map }) {
 
   action._subscribe = function subscribeToAction(observer) {
     observers.push(observer);
-    return new Rx.Disposable(() => {
+    return new Disposable(() => {
       observers.splice(observers.indexOf(observer), 1);
     });
   };
 
-  Rx.Observable.call(action);
+  Observable.call(action);
 
   debug('action %s created', action.displayName);
   return action;

test/action.js

@@ -32,14 +32,16 @@ describe('Actions', function() {
         },
         errorMap() {
           throw new Error('test');
-        }
-      })
-        .refs({ displayName: 'catActions' });
+        },
+        returnObservable(obs) {
+          return obs;
+        },
+
+        refs: { displayName: 'catActions' }
+      });
+
       catActions = CatActions();
     });
-    it('should add displayName as property, not observable', () => {
-      catActions.displayName.should.equal('catActions');
-    });
 
     it('should be extend-able', function() {
       catActions.getInBox.should.exist;
@@ -134,9 +136,19 @@ describe('Actions', function() {
       }
     );
 
+    it('should accept observables from map function', done => {
+      catActions.returnObservable.subscribe(val => {
+        val.should.equal(3);
+        done();
+      });
+      catActions.returnObservable(Rx.Observable.just(3));
+    });
+
     it('should call onError when map throws', function(done) {
       catActions.errorMap.subscribe(
-        () => {},
+        () => {
+          throw new Error('should not call on next');
+        },
         err => {
           expect(err).to.be.an.instanceOf(Error);
           done();
@@ -145,6 +157,25 @@ describe('Actions', function() {
       );
       catActions.errorMap();
     });
+
+    it(
+      'should accept observables from map function and call onError',
+      done => {
+        const err = new Error('Happy Cat Day!');
+        catActions.returnObservable.subscribe(
+          () => {
+            throw new Error('should not call onNext!');
+          },
+          _err => {
+            _err.should.equal(err);
+            done();
+          },
+          done
+        );
+
+        catActions.returnObservable(Rx.Observable.throw(err));
+      }
+    );
   });
 
   describe('spec', function() {

test/store.js

@@ -542,15 +542,22 @@ describe('Store', function() {
           replace: {},
           optimistic: defer2
         });
+        store.history.size.should.equal(2);
         defer2.onError('boo');
+        store.history.size.should.equal(1);
       });
 
-      it('should remove that entry on promise resolve', function(done) {
+      it('should remove that entry on promise resolve', function() {
+        const delay = Q.defer();
+        setTimeout(delay.resolve, 100);
+        store.history.size.should.equal(1);
         defer.resolve();
-        defer.promise.then(function() {
-          store.history.size.should.equal(0);
-          done();
-        });
+        return delay
+          .promise
+          .then(() => defer.promise)
+          .then(() => {
+            store.history.size.should.equal(0);
+          });
       });
     });
 
@@ -621,20 +628,20 @@ describe('Store', function() {
             .init(({ instance }) => instance.register(catActions));
           store = CatStore();
           store.subscribe(spy);
-
-          catActions.onNext({
-            transform: function(arr) {
-              return arr.concat('foo');
-            },
-            optimistic: deferred1.promise
-          });
         });
 
         it(
           'should notify observers with the transformed value ' +
           'after the first operation has been applied',
           function() {
+            catActions.onNext({
+              transform: function(arr) {
+                return arr.concat('foo');
+              },
+              optimistic: deferred1.promise
+            });
             spy.should.have.been.calledWith(['foo']);
+            spy.should.have.been.callCount(2);
           }
         );
 
@@ -649,31 +656,42 @@ describe('Store', function() {
               optimistic: deferred2.promise
             });
             spy.should.have.been.calledWith(['foo', 'bar']);
+            spy.should.have.been.callCount(3);
           }
         );
 
         it(
           'should notify observers with result of applying the ' +
           'second operation on the old value after the first ' +
           'operation has failed',
-          function(done) {
+          function() {
+            const delay = Q.defer();
+            setTimeout(delay.resolve, 100);
             deferred1.reject();
-            deferred1.promise.catch(function() {
-              spy.should.have.been.calledWith(['bar']);
-              done();
-            });
+            return delay
+              .promise
+              .then(() => deferred1.promise)
+              .catch(() => {
+                spy.should.have.been.callCount(4);
+                spy.should.have.been.calledWith(['bar']);
+              });
           }
         );
 
         it(
           'should notify observers with the initial value after ' +
           'the second operation has failed',
-          function(done) {
+          function() {
+            const delay = Q.defer();
+            setTimeout(delay.resolve, 100);
             deferred2.reject();
-            deferred2.promise.catch(function() {
-              spy.should.have.been.calledWith([]);
-              done();
-            });
+            return delay
+              .promise
+              .then(() => deferred2.promise)
+              .catch(function() {
+                spy.should.have.been.callCount(5);
+                spy.should.have.been.calledWith([]);
+              });
           }
         );
       });