/*

 * Copyright (c) 2018, Oracle and/or its affiliates. All rights reserved.

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.

 *

 * This code is distributed in the hope that it will be useful, but WITHOUT

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

 * version 2 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 */

/*

 * @test

 * @summary Tests what happens when response body handlers and subscribers

 *          throw unexpected exceptions.

 * @library /lib/testlibrary http2/server

 * @build jdk.testlibrary.SimpleSSLContext HttpServerAdapters ThrowingSubscribers

 * @modules java.base/sun.net.www.http

 *          java.net.http/jdk.internal.net.http.common

 *          java.net.http/jdk.internal.net.http.frame

 *          java.net.http/jdk.internal.net.http.hpack

 * @run testng/othervm ThrowingSubscribers

 */

import com.sun.net.httpserver.HttpServer;

import com.sun.net.httpserver.HttpsConfigurator;

import com.sun.net.httpserver.HttpsServer;

import jdk.testlibrary.SimpleSSLContext;

import org.testng.annotations.AfterTest;

import org.testng.annotations.AfterClass;

import org.testng.annotations.BeforeTest;

import org.testng.annotations.DataProvider;

import org.testng.annotations.Test;

import javax.net.ssl.SSLContext;

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStream;

import java.io.InputStreamReader;

import java.io.OutputStream;

import java.net.InetSocketAddress;

import java.net.URI;

import java.net.http.HttpClient;

import java.net.http.HttpHeaders;

import java.net.http.HttpRequest;

import java.net.http.HttpResponse;

import java.net.http.HttpResponse.BodyHandler;

import java.net.http.HttpResponse.BodySubscriber;

import java.nio.ByteBuffer;

import java.nio.charset.StandardCharsets;

import java.util.List;

import java.util.concurrent.CompletionStage;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.ConcurrentMap;

import java.util.concurrent.Executor;

import java.util.concurrent.Executors;

import java.util.concurrent.Flow;

import java.util.concurrent.atomic.AtomicLong;

import java.util.function.Consumer;

import java.util.function.Predicate;

import java.util.function.Supplier;

import java.util.stream.Collectors;

import java.util.stream.Stream;

import static java.lang.System.out;

import static java.lang.String.format;

import static java.nio.charset.StandardCharsets.UTF_8;

import static org.testng.Assert.assertEquals;

import static org.testng.Assert.assertTrue;

public class ThrowingSubscribers implements HttpServerAdapters {

    SSLContext sslContext;

    HttpTestServer httpTestServer;    // HTTP/1.1    [ 4 servers ]

    HttpTestServer httpsTestServer;   // HTTPS/1.1

    HttpTestServer http2TestServer;   // HTTP/2 ( h2c )

    HttpTestServer https2TestServer;  // HTTP/2 ( h2  )

    String httpURI_fixed;

    String httpURI_chunk;

    String httpsURI_fixed;

    String httpsURI_chunk;

    String http2URI_fixed;

    String http2URI_chunk;

    String https2URI_fixed;

    String https2URI_chunk;

    static final int ITERATION_COUNT = 2;

    // a shared executor helps reduce the amount of threads created by the test

    static final Executor executor = new TestExecutor(Executors.newCachedThreadPool());

    static final ConcurrentMap<String, Throwable> FAILURES = new ConcurrentHashMap<>();

    static volatile boolean tasksFailed;

    static class TestExecutor implements Executor {

        final AtomicLong tasks = new AtomicLong();

        Executor executor;

        TestExecutor(Executor executor) {

            this.executor = executor;

        }

        @Override

        public void execute(Runnable command) {

            long id = tasks.incrementAndGet();

            executor.execute(() -> {

                try {

                    command.run();

                } catch (Throwable t) {

                    tasksFailed = true;

                    System.out.printf("Task %s failed: %s%n", id, t);

                    System.err.printf("Task %s failed: %s%n", id, t);

                    FAILURES.putIfAbsent("Task " + id, t);

                    throw t;

                }

            });

        }

    }

    @AfterClass

    static final void printFailedTests() {

        if (FAILURES.isEmpty()) return;

        out.println("Failed tests: ");

        FAILURES.entrySet().forEach((e) -> {

                out.printf("\t%s: %s%n", e.getKey(), e.getValue());

                e.getValue().printStackTrace();

        });

        if (tasksFailed) {

            throw new RuntimeException("Some tasks failed");

        }

    }

    @DataProvider(name = "variants")

    public Object[][] variants() {

    }

    HttpClient newHttpClient() {

        return HttpClient.newBuilder()

                         .executor(executor)

                         .sslContext(sslContext)

                         .build();

    }

        HttpClient client = null;

        for (int i=0; i< ITERATION_COUNT; i++) {

            if (!sameClient || client == null)

                client = newHttpClient();

            HttpRequest req = HttpRequest.newBuilder(URI.create(uri))

                    .build();

            HttpResponse<String> response = client.send(req, handler);

            String body = response.body();

            assertEquals(URI.create(body).getPath(), URI.create(uri).getPath());

        }

    }

    @Test(dataProvider = "variants")

        testThrowing(test, uri, sameClient, BodyHandler::asString,

    }

    @Test(dataProvider = "variants")

        testThrowing(test, uri, sameClient, BodyHandler::asLines,

    }

    @Test(dataProvider = "variants")

        testThrowing(test, uri, sameClient, BodyHandler::asInputStream,

    }

    @Test(dataProvider = "variants")

        testThrowing(uri, sameClient, BodyHandler::asString,

    }

    @Test(dataProvider = "variants")

        testThrowing(test, uri, sameClient, BodyHandler::asLines,

    }

    @Test(dataProvider = "variants")

        testThrowing(test, uri, sameClient, BodyHandler::asInputStream,

    }

    private <T,U> void testThrowing(String name, String uri, boolean sameClient,

                                    Supplier<BodyHandler<T>> handlers,

        out.printf("%n%s%n", name);

        try {

        } catch (Error | Exception x) {

            FAILURES.putIfAbsent(name, x);

            throw x;

        }

    }

    private <T,U> void testThrowing(String uri, boolean sameClient,

        HttpClient client = null;

        for (Where where : Where.values()) {

            if (where == Where.ON_SUBSCRIBE) continue;

            if (where == Where.ON_ERROR) continue;

            if (where == Where.GET_BODY) continue; // doesn't work with HTTP/2

            if (!sameClient || client == null)

                client = newHttpClient();

                    .build();

            System.out.println("try throwing in " + where);

            HttpResponse<T> response = null;

            if (async) {

                try {

                    response = client.sendAsync(req, handler).join();

                } catch (Error | Exception x) {

                    if (cause == null) throw x;

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

                }

            } else {

                try {

                    response = client.send(req, handler);

                }

            }

            if (response != null) {

            }

        }

    }

    enum Where {BODY_HANDLER, ON_SUBSCRIBE, ON_NEXT, ON_COMPLETE, ON_ERROR, GET_BODY;

        public Consumer<Where> select(Consumer<Where> consumer) {

            return new Consumer<Where>() {

                @Override

                public void accept(Where where) {

                    if (Where.this == where) {

                        consumer.accept(where);

                    }

                }

            };

        }

    }

    }

        throw new RuntimeException("Expected exception not thrown in " + w);

    }

        switch(w) {

            default: break;

        }

        List<String> result = null;

        try {

            result = resp.body().collect(Collectors.toList());

        } catch (Error | Exception x) {

            if (cause != null) {

                return result;

            }

            throw x;

        }

        throw new RuntimeException("Expected exception not thrown in " + w);

    }

            throws IOException

    {

        switch(w) {

            default: break;

        }

        List<String> result = null;

        try (InputStreamReader r1 = new InputStreamReader(resp.body(), UTF_8);

             BufferedReader r = new BufferedReader(r1)) {

            try {

                result = r.lines().collect(Collectors.toList());

            } catch (Error | Exception x) {

                if (cause != null) {

                    out.println("Got expected exception in " + w + ": " + x);

                    return result;

                }

                throw x;

            }

        }

    }

        while (x != null && !filter.test(x)) x = x.getCause();

    }

        @Override

        public void accept(Where where) {

            throw new UncheckedCustomException(where.name());

        }

    }

    static class UncheckedCustomException extends RuntimeException {

        UncheckedCustomException(String message) {

            super(message);

        }

        UncheckedCustomException(String message, Throwable cause) {

            super(message, cause);

        }

    }

    static class ThrowingBodyHandler<T> implements BodyHandler<T> {

        final Consumer<Where> throwing;

        final BodyHandler<T> bodyHandler;

        ThrowingBodyHandler(Consumer<Where> throwing, BodyHandler<T> bodyHandler) {

            this.throwing = throwing;

            this.bodyHandler = bodyHandler;

        }

        @Override

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

            throwing.accept(Where.BODY_HANDLER);

            BodySubscriber<T> subscriber = bodyHandler.apply(statusCode, responseHeaders);

            return new ThrowingBodySubscriber(throwing, subscriber);

        }

    }

    static class ThrowingBodySubscriber<T> implements BodySubscriber<T> {

        private final BodySubscriber<T> subscriber;

        volatile boolean onSubscribeCalled;

        final Consumer<Where> throwing;

        ThrowingBodySubscriber(Consumer<Where> throwing, BodySubscriber<T> subscriber) {

            this.throwing = throwing;

            this.subscriber = subscriber;

        }

        @Override

        public void onSubscribe(Flow.Subscription subscription) {

            //out.println("onSubscribe ");

            onSubscribeCalled = true;

            throwing.accept(Where.ON_SUBSCRIBE);

            subscriber.onSubscribe(subscription);

        }

        @Override

        public void onNext(List<ByteBuffer> item) {

           // out.println("onNext " + item);