Modify OnSubscribeToObservableFuture to return early if unsubscribe without emitting anything

mattrjacobs · mattrjacobs · commit e285a7b4cfcf · 2014-05-30T10:01:20.000-07:00
diff --git a/rxjava-core/src/main/java/rx/operators/OnSubscribeToObservableFuture.java b/rxjava-core/src/main/java/rx/operators/OnSubscribeToObservableFuture.java
@@ -62,6 +62,10 @@ public void call() {
                 }
             }));
             try {
+                //don't block or propagate CancellationException if already unsubscribed
+                if (subscriber.isUnsubscribed()) {
+                    return;
+                }
                 T value = (unit == null) ? (T) that.get() : (T) that.get(time, unit);
                 subscriber.onNext(value);
                 subscriber.onCompleted();
@@ -71,6 +75,10 @@ public void call() {
                 // since it's already subscribed.
                 // If the Future is canceled in other place, CancellationException will be still
                 // passed to the final Subscriber.
+                if (subscriber.isUnsubscribed()) {
+                    //refuse to emit onError if already unsubscribed
+                    return;
+                }
                 subscriber.onError(e);
             }
         }
