Merge pull request #959 from rickbw/toFuture-cancellation

OperationToFuture must throw CancellationException on get() if cancelled

rxjava-core/src/main/java/rx/operators/OperationToFuture.java

@@ -15,6 +15,7 @@
  */
 package rx.operators;
 
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -120,6 +121,9 @@ public T get(long timeout, TimeUnit unit) throws InterruptedException, Execution
             private T getValue() throws ExecutionException {
                 if (error.get() != null) {
                     throw new ExecutionException("Observable onError", error.get());
+                } else if (cancelled) {
+                    // Contract of Future.get() requires us to throw this:
+                    throw new CancellationException("Subscription unsubscribed");
                 } else {
                     return value.get();
                 }

rxjava-core/src/test/java/rx/operators/OperationToFutureTest.java

@@ -15,18 +15,23 @@
  */
 package rx.operators;
 
-import static org.junit.Assert.*;
-import static rx.operators.OperationToFuture.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static rx.operators.OperationToFuture.toFuture;
 
 import java.util.List;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
 import rx.Observable;
 import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.subscriptions.Subscriptions;
 
@@ -77,6 +82,34 @@ public Subscription onSubscribe(Observer<? super String> observer) {
         }
     }
 
+    @Test(expected=CancellationException.class)
+    public void testGetAfterCancel() throws Exception {
+        Observable<String> obs = Observable.create(new OperationNeverComplete<String>());
+        Future<String> f = toFuture(obs);
+        boolean cancelled = f.cancel(true);
+        assertTrue(cancelled);  // because OperationNeverComplete never does
+        f.get();                // Future.get() docs require this to throw
+    }
+
+    @Test(expected=CancellationException.class)
+    public void testGetWithTimeoutAfterCancel() throws Exception {
+        Observable<String> obs = Observable.create(new OperationNeverComplete<String>());
+        Future<String> f = toFuture(obs);
+        boolean cancelled = f.cancel(true);
+        assertTrue(cancelled);  // because OperationNeverComplete never does
+        f.get(Long.MAX_VALUE, TimeUnit.NANOSECONDS);    // Future.get() docs require this to throw
+    }
+
+    /**
+     * Emits no observations. Used to simulate a long-running asynchronous operation.
+     */
+    private static class OperationNeverComplete<T> implements Observable.OnSubscribe<T> {
+        @Override
+        public void call(Subscriber<? super T> unused) {
+            // do nothing
+        }
+    }
+
     private static class TestException extends RuntimeException {
         private static final long serialVersionUID = 1L;
     }