Enable asynchronous cancellation of AsyncHttpSingle. (AsyncHttpClient#1250)

AsyncHttpSingle is currently not forwarding an unsubscription to AsyncHttpClient, it just aborts request processing when an AsyncHandler callback method is invoked. To actually eagerly cancel a request on unsubscription, use the request future to actually forward the cancellation to AsyncHttpClient.

Author: twz123

extras/rxjava/src/main/java/org/asynchttpclient/extras/rxjava/UnsubscribedException.java

@@ -12,16 +12,18 @@
  */
 package org.asynchttpclient.extras.rxjava;
 
+import java.util.concurrent.CancellationException;
+
 /**
  * Indicates that an {@code Observer} unsubscribed during the processing of a HTTP request.
  */
 @SuppressWarnings("serial")
-public class UnsubscribedException extends RuntimeException {
+public class UnsubscribedException extends CancellationException {
 
     public UnsubscribedException() {
     }
 
     public UnsubscribedException(final Throwable cause) {
-        super(cause);
+        initCause(cause);
     }
 }

extras/rxjava/src/main/java/org/asynchttpclient/extras/rxjava/single/AsyncHttpSingle.java

@@ -20,10 +20,13 @@
 import org.asynchttpclient.Response;
 import org.asynchttpclient.handler.ProgressAsyncHandler;
 
+import java.util.concurrent.Future;
+
 import rx.Single;
 import rx.SingleSubscriber;
-import rx.functions.Action1;
 import rx.functions.Func0;
+import rx.functions.Func1;
+import rx.subscriptions.Subscriptions;
 
 /**
  * Wraps HTTP requests into RxJava {@code Single} instances.
@@ -54,14 +57,17 @@ public static Single<Response> create(BoundRequestBuilder builder) {
      *
      * @param requestTemplate called to start the HTTP request with an
      *            {@code AysncHandler} that builds the HTTP response and
-     *            propagates results to the returned {@code Single}
+     *            propagates results to the returned {@code Single}. The
+     *            {@code Future} that is returned by {@code requestTemplate}
+     *            will be used to cancel the request when the {@code Single} is
+     *            unsubscribed.
      *
      * @return a {@code Single} that executes new requests on subscription by
      *         calling {@code requestTemplate} and that emits the response
      *
      * @throws NullPointerException if {@code requestTemplate} is {@code null}
      */
-    public static Single<Response> create(Action1<? super AsyncHandler<?>> requestTemplate) {
+    public static Single<Response> create(Func1<? super AsyncHandler<?>, ? extends Future<?>> requestTemplate) {
         return create(requestTemplate, AsyncCompletionHandlerBase::new);
     }
 
@@ -92,7 +98,10 @@ public static <T> Single<T> create(BoundRequestBuilder builder, Func0<? extends
      *
      * @param requestTemplate called to start the HTTP request with an
      *            {@code AysncHandler} that builds the HTTP response and
-     *            propagates results to the returned {@code Single}
+     *            propagates results to the returned {@code Single}.  The
+     *            {@code Future} that is returned by {@code requestTemplate}
+     *            will be used to cancel the request when the {@code Single} is
+     *            unsubscribed.
      * @param handlerSupplier supplies the desired {@code AsyncHandler}
      *            instances that are used to produce results
      *
@@ -104,13 +113,17 @@ public static <T> Single<T> create(BoundRequestBuilder builder, Func0<? extends
      * @throws NullPointerException if at least one of the parameters is
      *             {@code null}
      */
-    public static <T> Single<T> create(Action1<? super AsyncHandler<?>> requestTemplate,
+    public static <T> Single<T> create(Func1<? super AsyncHandler<?>, ? extends Future<?>> requestTemplate,
             Func0<? extends AsyncHandler<? extends T>> handlerSupplier) {
 
         requireNonNull(requestTemplate);
         requireNonNull(handlerSupplier);
 
-        return Single.create(subscriber -> requestTemplate.call(createBridge(subscriber, handlerSupplier.call())));
+        return Single.create(subscriber -> {
+            final AsyncHandler<?> bridge = createBridge(subscriber, handlerSupplier.call());
+            final Future<?> responseFuture = requestTemplate.call(bridge);
+            subscriber.add(Subscriptions.from(responseFuture));
+        });
     }
 
     static <T> AsyncHandler<?> createBridge(SingleSubscriber<? super T> subscriber, AsyncHandler<? extends T> handler) {

extras/rxjava/src/test/java/org/asynchttpclient/extras/rxjava/single/AsyncHttpSingleTest.java

@@ -18,13 +18,15 @@
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.isA;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.verifyZeroInteractions;
 import static org.mockito.Mockito.when;
 import static org.testng.Assert.assertEquals;
 
@@ -34,13 +36,16 @@
 import org.asynchttpclient.BoundRequestBuilder;
 import org.asynchttpclient.HttpResponseStatus;
 import org.asynchttpclient.Response;
+import org.asynchttpclient.extras.rxjava.UnsubscribedException;
 import org.asynchttpclient.handler.ProgressAsyncHandler;
 import org.mockito.InOrder;
 import org.testng.annotations.Test;
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Single;
 import rx.exceptions.CompositeException;
@@ -82,6 +87,8 @@ public void testSuccessfulCompletion() throws Exception {
             } catch (final Throwable t) {
                 bridge.onThrowable(t);
             }
+
+            return mock(Future.class);
         } , () -> handler);
 
         final TestSubscriber<Object> subscriber = new TestSubscriber<>();
@@ -132,6 +139,8 @@ public void testSuccessfulCompletionWithProgress() throws Exception {
             } catch (final Throwable t) {
                 bridge.onThrowable(t);
             }
+
+            return mock(Future.class);
         } , () -> handler);
 
         final TestSubscriber<Object> subscriber = new TestSubscriber<>();
@@ -186,6 +195,8 @@ public void testErrorPropagation() throws Exception {
             } catch (final Throwable t) {
                 bridge.onThrowable(t);
             }
+
+            return mock(Future.class);
         } , () -> handler);
 
         final TestSubscriber<Object> subscriber = new TestSubscriber<>();
@@ -209,6 +220,7 @@ public void testErrorInOnCompletedPropagation() throws Exception {
         final Single<?> underTest = AsyncHttpSingle.create(bridge -> {
             try {
                 bridge.onCompleted();
+                return mock(Future.class);
             } catch (final Throwable t) {
                 throw new AssertionError(t);
             }
@@ -237,6 +249,7 @@ public void testErrorInOnThrowablePropagation() throws Exception {
         final Single<?> underTest = AsyncHttpSingle.create(bridge -> {
             try {
                 bridge.onThrowable(processingException);
+                return mock(Future.class);
             } catch (final Throwable t) {
                 throw new AssertionError(t);
             }
@@ -281,4 +294,23 @@ public State onStatusReceived(HttpResponseStatus status) {
         subscriber.assertValue(null);
     }
 
+    @Test(groups = "standalone")
+    public void testUnsubscribe() throws Exception {
+        final AsyncHandler<Object> handler = mock(AsyncHandler.class);
+        final Future<?> future = mock(Future.class);
+        final AtomicReference<AsyncHandler<?>> bridgeRef = new AtomicReference<>();
+
+        final Single<?> underTest = AsyncHttpSingle.create(bridge -> {
+            bridgeRef.set(bridge);
+            return future;
+        } , () -> handler);
+
+        underTest.subscribe().unsubscribe();
+        verify(future).cancel(true);
+        verifyZeroInteractions(handler);
+
+        assertThat(bridgeRef.get().onStatusReceived(null), is(AsyncHandler.State.ABORT));
+        verify(handler).onThrowable(isA(UnsubscribedException.class));
+        verifyNoMoreInteractions(handler);
+    }
 }