OperationToFuture must throw CancellationException on get() if cancelled.

The documentation for Future.get() requires the method to throw CancellationException if the Future was cancelled before the task completed. The Futures returned by OperationToFuture.toFuture() did not respect this contract. Now they do.

Author: rickbw

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;
     }