import jdk.incubator.http.HttpHeaders;

import jdk.jshell.spi.ExecutionControl;

import javax.net.ServerSocketFactory;

import javax.net.ssl.SSLContext;

import javax.net.ssl.SSLException;

import javax.net.ssl.SSLServerSocketFactory;

import java.io.IOException;

import java.net.SocketException;

import java.net.URI;

import java.nio.ByteBuffer;

import java.util.List;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.CompletionException;

import java.util.concurrent.CompletionStage;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.Flow;

import java.util.concurrent.atomic.AtomicBoolean;

import java.util.concurrent.atomic.AtomicInteger;

import jdk.incubator.http.HttpResponse.BodyHandler;

import jdk.incubator.http.HttpResponse.BodySubscriber;

import static java.lang.String.format;

 * @run main/othervm  CancelledResponse

 * @run main/othervm  CancelledResponse SSL

 * Similar test to SplitResponse except that the client will cancel the response

 * before receiving it fully.

public class CancelledResponse {

        "Lorem ipsum dolor sit amet consectetur adipiscing elit,",

        "sed do eiusmod tempor quis nostrud exercitation ullamco laboris nisi ut",

        "aliquip ex ea commodo consequat."

    CancelledResponse(boolean useSSL) throws IOException {

        CancelledResponse sp = new CancelledResponse(useSSL);

    static class CancelException extends IOException {

    }

        try {

            for (int i = 0; i < responses.length; i++) {

                HttpResponse<String> r = null;

                CompletableFuture<HttpResponse<String>> cf1;

                CancelException expected = null;

                AtomicBoolean cancelled = new AtomicBoolean();

                Thread t = sendSplitResponse(response(body, serverKeepalive), server, cancelled);

                try {

                    if (async) {

                        out.println("send async: " + request);

                        cf1 = client.sendAsync(request, asString(body, cancelled));

                        r = cf1.get();

                    } else { // sync

                        out.println("send sync: " + request);

                        r = client.send(request, asString(body, cancelled));

                    }

                } catch (CancelException c1) {

                    System.out.println("Got expected exception: " + c1);

                    expected = c1;

                } catch (IOException | ExecutionException | CompletionException c2) {

                    Throwable c = c2;

                    while (c != null && !(c instanceof CancelException)) {

                        c = c.getCause();

                    }

                    if (c instanceof CancelException) {

                        System.out.println("Got expected exception: " + c);

                        expected = (CancelException)c;

                    } else throw c2;

                if (r != null) {

                    if (r.statusCode() != 200)

                        throw new RuntimeException("Failed");

                    String rxbody = r.body();

                    out.println("received " + rxbody);

                    if (!rxbody.equals(body))

                        throw new RuntimeException(format("Expected:%s, got:%s", body, rxbody));

                }

                if (expected == null) {

                    throw new RuntimeException("Expected exception not raised for "

                            + i + " cancelled=" + cancelled.get());

                }

    static class CancellingSubscriber implements BodySubscriber<String> {

        private final String expected;

        private final CompletableFuture<String> result;

        private Flow.Subscription subscription;

        final AtomicInteger index = new AtomicInteger();

        final AtomicBoolean cancelled;

        CancellingSubscriber(String expected, AtomicBoolean cancelled) {

            this.cancelled = cancelled;

            this.expected = expected;

            result = new CompletableFuture<>();

        }

        @Override

        public CompletionStage<String> getBody() {

            return result;

        }

        @Override

        public void onSubscribe(Flow.Subscription subscription) {

            this.subscription = subscription;

            subscription.request(1);

        }

        @Override

        public void onNext(List<ByteBuffer> item) {

            //if (result.isDone())

            for (ByteBuffer b : item) {

                while (b.hasRemaining() && !result.isDone()) {

                    int i = index.getAndIncrement();

                    char at = expected.charAt(i);

                    byte[] data = new byte[b.remaining()];

                    b.get(data); // we know that the server writes 1 char

                    String s = new String(data);

                    char c = s.charAt(0);

                    if (c != at) {

                        Throwable x = new IllegalStateException("char at "

                                + i + " is '" + c + "' expected '"

                                + at + "' for \"" + expected +"\"");

                        out.println("unexpected char received, cancelling");

                        subscription.cancel();

                        result.completeExceptionally(x);

                        return;

                    }

                }

            }

            if (index.get() > 0 && !result.isDone()) {

                // we should complete the result here, but let's

                // see if we get something back...

                out.println("Cancelling subscription after reading " + index.get());

                cancelled.set(true);

                subscription.cancel();

                result.completeExceptionally(new CancelException());

                return;

            }

            if (!result.isDone()) {

                out.println("requesting 1 more");

                subscription.request(1);

            }

        }

        @Override

        public void onError(Throwable throwable) {

            result.completeExceptionally(throwable);

        }

        @Override

        public void onComplete() {

            int len = index.get();

            if (len == expected.length()) {

                result.complete(expected);

            } else {

                Throwable x = new IllegalStateException("received only "

                        + len + " chars, expected " + expected.length()

                        + " for \"" + expected +"\"");

                result.completeExceptionally(x);

            }

        }

    }

    static class CancellingHandler implements BodyHandler<String> {

        final String expected;

        final AtomicBoolean cancelled;

        CancellingHandler(String expected, AtomicBoolean cancelled) {

            this.expected = expected;

            this.cancelled = cancelled;

        }

        @Override

        public BodySubscriber<String> apply(int statusCode, HttpHeaders responseHeaders) {

            assert !cancelled.get();

            return new CancellingSubscriber(expected, cancelled);

        }

    }

    BodyHandler<String> asString(String expected, AtomicBoolean cancelled) {

        return new CancellingHandler(expected, cancelled);

    }

    Thread sendSplitResponse(String s, MockServer server, AtomicBoolean cancelled) {

            int sent = 0;

                    sent++;

            } catch (SSLException | SocketException x) {

                if (x instanceof SocketException

                        || x.getCause() instanceof SocketException) {

                    // if SSL then we might get a "Broken Pipe", otherwise

                    // a "Socket closed".

                    boolean expected = cancelled.get();

                    if (sent > 0 && expected) {

                        System.out.println("Connection closed by peer as expected: " + x);

                        return;

                    } else {

                        System.out.println("Unexpected exception (sent="

                            + sent + ", cancelled=" + expected + "): " + x);

                    }

                }

                throw new RuntimeException(x);