Cancellation of http request in Java 11 HttpClient

Question

I'm trying to cancel http request via new Java 11 HttpClient.

This is my test code:

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

public class App {

    public static void main(String... args) throws InterruptedException {
        HttpClient client = HttpClient.newBuilder().build();

        URI uri = URI.create("http://releases.ubuntu.com/18.04.2/ubuntu-18.04.2-desktop-amd64.iso");
        HttpRequest request = HttpRequest.newBuilder().uri(uri).GET().build();

        var bodyHandler = HttpResponse.BodyHandlers.ofByteArrayConsumer(b -> System.out.println("#"));
        var future = client.sendAsync(request, bodyHandler);
        Thread.sleep(1000);

        future.cancel(true);
        System.out.println("\r\n----------CANCEL!!!------------");
        System.out.println("\r\nisCancelled: " + future.isCancelled());
        Thread.sleep(250);
    }
}

I expect, that request task will be cancelled right after future.cancel(true); line invoked. And, therefore, last printed line in console should be isCancelled: true

But, when I run this code, I see something like this:

####################################################################################################
----------CANCEL!!!------------
####
isCancelled: true
#######################################################################################################################################################

This means, that request task still running after I cancel it... So, that is the right way to cancel request?

UPD

Right way to cancel request is (As daniel suggested, + UPD2: avoiding NPE on cancel() method invoke):

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandler;
import java.net.http.HttpResponse.BodySubscriber;
import java.net.http.HttpResponse.ResponseInfo;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Flow.Subscription;

public class App {

    private static class SubscriberWrapper implements BodySubscriber<Void> {
        private final CountDownLatch latch;
        private final BodySubscriber<Void> subscriber;
        private Subscription subscription;

        private SubscriberWrapper(BodySubscriber<Void> subscriber, CountDownLatch latch) {
            this.subscriber = subscriber;
            this.latch = latch;
        }

        @Override
        public CompletionStage<Void> getBody() {
            return subscriber.getBody();
        }

        @Override
        public void onSubscribe(Subscription subscription) {
            subscriber.onSubscribe(subscription);
            this.subscription = subscription;
            latch.countDown();
        }

        @Override
        public void onNext(List<ByteBuffer> item) {
            subscriber.onNext(item);
        }

        @Override
        public void onError(Throwable throwable) {
            subscriber.onError(throwable);
        }

        @Override
        public void onComplete() {
            subscriber.onComplete();
        }

        public void cancel() {
            subscription.cancel();
            System.out.println("\r\n----------CANCEL!!!------------");
        }
    }

    private static class BodyHandlerWrapper implements BodyHandler<Void> {
        private final CountDownLatch latch = new CountDownLatch(1);
        private final BodyHandler<Void> handler;
        private SubscriberWrapper subscriberWrapper;

        private BodyHandlerWrapper(BodyHandler<Void> handler) {
            this.handler = handler;
        }

        @Override
        public BodySubscriber<Void> apply(ResponseInfo responseInfo) {
            subscriberWrapper = new SubscriberWrapper(handler.apply(responseInfo), latch);
            return subscriberWrapper;
        }

        public void cancel() {
            CompletableFuture.runAsync(() -> {
                try {
                    latch.await();
                    subscriberWrapper.cancel();
                } catch (InterruptedException e) {}
            });
        }
    }

    public static void main(String... args) throws InterruptedException, ExecutionException {
        HttpClient client = HttpClient.newBuilder().build();

        URI uri = URI.create("http://releases.ubuntu.com/18.04.2/ubuntu-18.04.2-desktop-amd64.iso");
        HttpRequest request = HttpRequest.newBuilder().uri(uri).GET().build();

        var handler = HttpResponse.BodyHandlers.ofByteArrayConsumer(b -> System.out.print("#"));
        BodyHandlerWrapper handlerWrapper = new BodyHandlerWrapper(handler);

        client.sendAsync(request, handlerWrapper).thenAccept(b -> System.out.println(b.statusCode()));
        Thread.sleep(1000);
        handlerWrapper.cancel();

        System.out.println("\r\n------Invoke cancel...---------");
        Thread.sleep(2500);
    }
}

Answer 1

You can cancel an HTTP request using the java.net.http.HttpClient API by cancelling the Flow.Subscription object that is passed to the response's BodySubscriber. It should be relatively easy to trivially wrap one of the provided BodyHandler/BodySubscriber implementations in order to get hold to the subscription object. There is unfortunately no relationship between the cancel method of the CompletableFuture returned by the client, and the cancel method of the Flow.Subscription passed to the BodySubscriber. The correct way to cancel a request is through the cancel method of the subscription.

Cancelling the subscription will work both with the synchronous (HttpClient::send) and asynchronous (HttpClient::sendAsync) methods. It will have different effects however depending on whether the request was sent through HTTP/1.1 or HTTP/2.0 (with HTTP/1.1 it will cause the connection to be closed, with HTTP/2.0 it will cause the stream to be reset). And of course it might have no effect at all if the last byte of the response was already delivered to the BodySubscriber.

Update: Since Java 16 it is possible to cancel a request by interrupting the thread that called HttpClient::send or by invoking cancel(true) on the CompletableFuture returned by HttpClient::sendAsync. This has been implemented by JDK-8245462

Answer 2

At least for synchronous requests you could just interrupt the thread that is calling httpClient.send(..)

The http client then aborts the request as fast a possible and throws an InterruptedException itself.

Answer 3

Synchronous VS asynchronous

The request can be sent either synchronously or asynchronously. The synchronous API blocks until the Http Response is available

HttpResponse<String> response =
      client.send(request, BodyHandlers.ofString());
System.out.println(response.statusCode());
System.out.println(response.body());

The asynchronous API returns immediately with a CompletableFuture that completes with the HttpResponse when it becomes available. CompletableFuture was added in Java 8 and supports composable asynchronous programming.

client.sendAsync(request, BodyHandlers.ofString())
      .thenApply(response -> { System.out.println(response.statusCode());
                               return response; } )
      .thenApply(HttpResponse::body)
      .thenAccept(System.out::println);

Future object

A Future represents the result of an asynchronous computation. Java Doc

Meaning that it's not a synchronous function and that your assumption "I expect, that request task will be cancelled right after" would be true only for synchronous method.

Check cancellation of Future object

There is a useful isCancelled() method if you want to check if your task is cancelled.

if(future.isCancelled()) {
  // Future object is cancelled, do smth
} else {
  // Future object is still running, do smth
}

sendAsync() returns a CompletableFuture object

The method sendAsync() returns a CompletableFuture. Note that a CompletableFuture implements the interface of Future.

You can do something like:

client.sendAsync(request, BodyHandlers.ofString())
          .thenAccept(response -> {
       // do action when completed;
});

In technical term, the thenAccept method adds a Consumer to be called when a response has become available.

Why cancel method over CompeletableFuture won't work

Since (unlike FutureTask) this class has no direct control over the computation that causes it to be completed, cancellation is treated as just another form of exceptional completion. Method cancel has the same effect as completeExceptionally(new CancellationException()). Method isCompletedExceptionally() can be used to determine if a CompletableFuture completed in any exceptional fashion.

In case of exceptional completion with a CompletionException, methods get() and get(long, TimeUnit) throw an ExecutionException with the same cause as held in the corresponding CompletionException. To simplify usage in most contexts, this class also defines methods join() and getNow(T) that instead throw the CompletionException directly in these cases.

In other words

The cancel() method do not employ the interrupts to do the cancellation and this is why it's not working. You should use completeExceptionally(new CancellationException())

Reference

• https://openjdk.java.net/groups/net/httpclient/intro.html
• https://docs.oracle.com/javase/8/docs/api/index.html?java/util/concurrent/Future.html