Fix ActionStateSubscriber onError to throw error forward and not be s…

…ilent
techery · Mar 11, 2016 · 8ae60d9 · 8ae60d9
1 parent 3a1caf5
commit 8ae60d9
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/janet/src/main/java/io/techery/janet/ActionPipe.java b/janet/src/main/java/io/techery/janet/ActionPipe.java
@@ -70,7 +70,7 @@ public Observable<ActionState<A>> observe() {
      * @see Observable#replay(int)
      */
     public Observable<ActionState<A>> observeWithReplay() {
-        return cachedPipeline;
+        return cachedPipeline.asObservable();
     }
 
     /**
@@ -89,7 +89,7 @@ public Observable<A> observeSuccess() {
      * @see ActionPipe#observeSuccess()
      */
     public Observable<A> observeSuccessWithReplay() {
-        return cachedSuccessPipeline;
+        return cachedSuccessPipeline.asObservable();
     }
 
     /**

diff --git a/janet/src/main/java/io/techery/janet/helper/ActionStateSubscriber.java b/janet/src/main/java/io/techery/janet/helper/ActionStateSubscriber.java
@@ -2,6 +2,7 @@
 
 import io.techery.janet.ActionState;
 import rx.Subscriber;
+import rx.exceptions.OnErrorNotImplementedException;
 import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Action2;
@@ -69,5 +70,7 @@ public ActionStateSubscriber<A> afterEach(Action1<ActionState<A>> afterEach) {
 
     @Override public void onCompleted() { }
 
-    @Override public void onError(Throwable e) { }
+    @Override public void onError(Throwable e) {
+        throw new OnErrorNotImplementedException(e);
+    }
 }