Merge pull request #8 from stealthcode/1.x

Fixing issue with the DebugSubscriber and DebugSubscription swallowing exceptions
ReactiveX · Oct 21, 2015 · 832ca86 · 832ca86
2 parents 30a0218 + a6df92f
commit 832ca86
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 4 deletions.
diff --git a/src/main/java/rx/operators/DebugSubscriber.java b/src/main/java/rx/operators/DebugSubscriber.java
@@ -13,9 +13,9 @@
 package rx.operators;
 
 import rx.Observable.Operator;
-import rx.Observer;
 import rx.Producer;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
 import rx.plugins.DebugNotification;
 import rx.plugins.DebugNotificationListener;
 
@@ -43,6 +43,7 @@ public void onStart() {
             listener.complete(context);
         } catch (Throwable e) {
             listener.error(context, e);
+            throw Exceptions.propagate(e);
         }
     }
 
@@ -55,6 +56,7 @@ public void onCompleted() {
             listener.complete(context);
         } catch (Throwable e) {
             listener.error(context, e);
+            throw Exceptions.propagate(e);
         }
     }
 
@@ -67,6 +69,7 @@ public void onError(Throwable e) {
             listener.complete(context);
         } catch (Throwable e2) {
             listener.error(context, e2);
+            throw Exceptions.propagate(e);
         }
     }
 
@@ -81,6 +84,7 @@ public void onNext(T t) {
             listener.complete(context);
         } catch (Throwable e) {
             listener.error(context, e);
+            throw Exceptions.propagate(e);
         }
     }
 
@@ -97,6 +101,7 @@ public void request(long n) {
                     listener.complete(context);
                 } catch (Throwable e) {
                     listener.error(context, e);
+                    throw Exceptions.propagate(e);
                 }
             }
         });

diff --git a/src/main/java/rx/operators/DebugSubscription.java b/src/main/java/rx/operators/DebugSubscription.java
@@ -13,6 +13,7 @@
 package rx.operators;
 
 import rx.Subscription;
+import rx.exceptions.Exceptions;
 import rx.plugins.DebugNotification;
 import rx.plugins.DebugNotificationListener;
 
@@ -34,6 +35,7 @@ public void unsubscribe() {
             listener.complete(context);
         } catch (Throwable e) {
             listener.error(context, e);
+            throw Exceptions.propagate(e);
         }
     }
 

diff --git a/src/main/java/rx/plugins/DebugHook.java b/src/main/java/rx/plugins/DebugHook.java
@@ -17,6 +17,7 @@
 import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.Subscription;
+import rx.exceptions.Exceptions;
 import rx.operators.DebugSubscriber;
 
 /**
@@ -56,6 +57,7 @@ public void call(Subscriber<? super T> o) {
                     listener.complete(context);
                 } catch (Throwable e) {
                     listener.error(context, e);
+                    throw Exceptions.propagate(e);
                 }
             }
         };

diff --git a/src/test/java/rx/debug/DebugHookTest.java b/src/test/java/rx/debug/DebugHookTest.java
@@ -23,7 +23,6 @@
 import org.junit.Test;
 
 import rx.Observable;
-import rx.Subscriber;
 import rx.functions.Func1;
 import rx.observers.Subscribers;
 import rx.plugins.DebugHook;
@@ -36,8 +35,6 @@
 
 import java.util.Arrays;
 import java.util.Iterator;
-import java.util.Map.Entry;
-import java.util.Queue;
 import java.util.SortedSet;
 
 public class DebugHookTest {