Отмена http-запроса в Java 11 HttpClient

Question

Я пытаюсь отменить http-запрос через новый Java 11 HttpClient.

Это мой тестовый код:

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

Я ожидаю, что задача запроса будет отменена сразу после вызова строки future.cancel(true);. И, следовательно, последняя напечатанная строка в консоли должна быть isCancelled: true

Но когда я запускаю этот код, я вижу что-то вроде этого:

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

Это означает, что задание запроса все еще выполняется после того, как я отменил его ... Так что это правильный способ отменить запрос?

UPD

Правильный способ отменить запрос - (как предложил Даниэль, + UPD2: избегать NPE при вызове метода cancel()):

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

Вы можете отменить HTTP-запрос, используя java.net.http.HttpClient API, отменив объект Flow.Subscription, который передается в ответ BodySubscriber.Должно быть относительно просто обернуть одну из предоставленных BodyHandler / BodySubscriber реализаций, чтобы получить доступ к объекту подписки.К сожалению, нет никакой связи между методом cancel CompletableFuture, возвращаемым клиентом, и методом cancel Flow.Subscription, переданным BodySubscriber.Правильный способ отменить запрос - это метод подписки cancel.

Отмена подписки будет работать как с синхронным (HttpClient::send), так и с асинхронным (HttpClient::sendAsync) методами.Однако это будет иметь разные последствия в зависимости от того, был ли запрос отправлен через HTTP / 1.1 или HTTP / 2.0 (с HTTP / 1.1 это приведет к закрытию соединения, с HTTP / 2.0 это приведет к сбросу потока).И, конечно, это может вообще не иметь никакого эффекта, если последний байт ответа уже доставлен в BodySubscriber.

Answer 2

Синхронный VS асинхронный

Запрос может быть отправлен либо синхронно, либо асинхронно. Синхронный API блокирует до тех пор, пока Http-ответ не станет доступен

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

Асинхронный API немедленно возвращается с CompletableFuture, который завершается с HttpResponse, когда он становится доступным. CompletableFuture был добавлен в Java 8 и поддерживает композитное асинхронное программирование.

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

Будущий объект

Будущее представляет собой результат асинхронных вычислений. Java Doc

Это означает, что это не синхронная функция, и что ваше предположение «Я ожидаю, что запрос будет отменен сразу после», будет верно только для синхронного метода.

Проверка отмены объекта Future

Существует полезный метод isCancelled(), если вы хотите проверить, отменена ли ваша задача.

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

sendAsync () возвращает объект CompletableFuture

Метод sendAsync() возвращает CompletableFuture . Обратите внимание, что CompletableFuture реализует интерфейс Future.

Вы можете сделать что-то вроде:

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

В техническом плане метод thenAccept добавляет Consumer, который вызывается, когда отклик становится доступным.

Почему метод отмены через CompeletableFuture не будет работать

Поскольку (в отличие от FutureTask) этот класс не имеет прямого контроля над вычислениями, которые приводят к его завершению, отмена рассматривается как еще одна форма исключительного завершения. Метод отмены имеет тот же эффект, что и completeExceptionally(new CancellationException()). Метод isCompletedExceptionally() может использоваться для определения того, завершен ли CompletableFuture каким-либо исключительным образом.

В случае исключительного завершения с CompletionException, методы get() и get(long, TimeUnit) выдают ExecutionException по той же причине, что и соответствующая CompletionException. Чтобы упростить использование в большинстве контекстов, этот класс также определяет методы join() и getNow (T), которые вместо этого напрямую выбрасывают CompletionException в этих случаях.

Другими словами

Метод cancel() не использует прерывания для отмены, и поэтому он не работает. Вы должны использовать completeExceptionally(new CancellationException())

Ссылка

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