/*

 * 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 push promise handlers and their

 *          response body handlers and subscribers throw unexpected exceptions.

 * @library /lib/testlibrary http2/server

 * @build jdk.testlibrary.SimpleSSLContext HttpServerAdapters ThrowingPushPromises

 * @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 -Djdk.internal.httpclient.debug=true ThrowingPushPromises

 */

import jdk.internal.net.http.common.HttpHeadersImpl;

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.io.UncheckedIOException;

import java.net.URI;

import java.net.URISyntaxException;

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.BodyHandlers;

import java.net.http.HttpResponse.BodySubscriber;

import java.net.http.HttpResponse.PushPromiseHandler;

import java.nio.ByteBuffer;

import java.nio.charset.StandardCharsets;

import java.util.List;

import java.util.Map;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.CompletionException;

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.Function;

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.System.err;

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 ThrowingPushPromises implements HttpServerAdapters {

    SSLContext sslContext;

    HttpTestServer http2TestServer;   // HTTP/2 ( h2c )

    HttpTestServer https2TestServer;  // HTTP/2 ( h2  )

    String http2URI_fixed;

    String http2URI_chunk;

    String https2URI_fixed;

    String https2URI_chunk;

    static final int ITERATION_COUNT = 1;

    // 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 final AtomicLong serverCount = new AtomicLong();

    static final AtomicLong clientCount = new AtomicLong();

    static final long start = System.nanoTime();

    public static String now() {

        long now = System.nanoTime() - start;

        long secs = now / 1000_000_000;

        long mill = (now % 1000_000_000) / 1000_000;

        long nan = now % 1000_000;

        return String.format("[%d s, %d ms, %d ns] ", secs, mill, nan);

    }

    private volatile HttpClient sharedClient;

    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;

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

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

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

                    throw t;

                }

            });

        }

    }

    @AfterClass

    static final void printFailedTests() {

        out.println("\n=========================");

        try {

            out.printf("%n%sCreated %d servers and %d clients%n",

                    now(), serverCount.get(), clientCount.get());

            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(out);

                e.getValue().printStackTrace();

            });

            if (tasksFailed) {

                out.println("WARNING: Some tasks failed");

            }

        } finally {

            out.println("\n=========================\n");

        }

    }

    private String[] uris() {

        return new String[] {

                http2URI_fixed,

                http2URI_chunk,

                https2URI_fixed,

                https2URI_chunk,

        };

    }

    @DataProvider(name = "noThrows")

    public Object[][] noThrows() {

        String[] uris = uris();

        Object[][] result = new Object[uris.length * 2][];

        int i = 0;

        for (boolean sameClient : List.of(false, true)) {

            for (String uri: uris()) {

                result[i++] = new Object[] {uri, sameClient};

            }

        }

        assert i == uris.length * 2;

        return result;

    }

    @DataProvider(name = "variants")

    public Object[][] variants() {

        String[] uris = uris();

        Object[][] result = new Object[uris.length * 2 * 2][];

        int i = 0;

        for (Thrower thrower : List.of(

                new UncheckedIOExceptionThrower(),

                new UncheckedCustomExceptionThrower())) {

            for (boolean sameClient : List.of(false, true)) {

                for (String uri : uris()) {

                    result[i++] = new Object[]{uri, sameClient, thrower};

                }

            }

        }

        assert i == uris.length * 2 * 2;

        return result;

    }

    private HttpClient makeNewClient() {

        clientCount.incrementAndGet();

        return HttpClient.newBuilder()

                .executor(executor)

                .sslContext(sslContext)

                .build();

    }

    HttpClient newHttpClient(boolean share) {

        if (!share) return makeNewClient();

        HttpClient shared = sharedClient;

        if (shared != null) return shared;

        synchronized (this) {

            shared = sharedClient;

            if (shared == null) {

                shared = sharedClient = makeNewClient();

            }

            return shared;

        }

    }

    @Test(dataProvider = "noThrows")

    public void testNoThrows(String uri, boolean sameClient)

            throws Exception {

        HttpClient client = null;

        out.printf("%ntestNoThrows(%s, %b)%n", uri, sameClient);

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

            if (!sameClient || client == null)

                client = newHttpClient(sameClient);

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

                    .build();

            BodyHandler<Stream<String>> handler =

                    new ThrowingBodyHandler((w) -> {},

                                            BodyHandlers.ofLines());

            Map<HttpRequest, CompletableFuture<HttpResponse<Stream<String>>>> pushPromises =

                    new ConcurrentHashMap<>();

            PushPromiseHandler<Stream<String>> pushHandler = new PushPromiseHandler<>() {

                @Override

                public void applyPushPromise(HttpRequest initiatingRequest,

                                             HttpRequest pushPromiseRequest,

                                             Function<BodyHandler<Stream<String>>,

                                                     CompletableFuture<HttpResponse<Stream<String>>>>

                                                     acceptor) {

                    pushPromises.putIfAbsent(pushPromiseRequest, acceptor.apply(handler));

                }

            };

            HttpResponse<Stream<String>> response =

                    client.sendAsync(req, BodyHandlers.ofLines(), pushHandler).get();

            String body = response.body().collect(Collectors.joining("|"));

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

            for (HttpRequest promised : pushPromises.keySet()) {

                out.printf("%s Received promise: %s%n\tresponse: %s%n",

                        now(), promised, pushPromises.get(promised).get());

                String promisedBody = pushPromises.get(promised).get().body()

                        .collect(Collectors.joining("|"));

                assertEquals(promisedBody, promised.uri().toASCIIString());

            }

            assertEquals(3, pushPromises.size());

        }

    }

    @Test(dataProvider = "variants")

    public void testThrowingAsString(String uri,

                                     boolean sameClient,

                                     Thrower thrower)

            throws Exception

    {

        String test = format("testThrowingAsString(%s, %b, %s)",

                             uri, sameClient, thrower);

        testThrowing(test, uri, sameClient, BodyHandlers::ofString,

                this::checkAsString, thrower);

    }

    @Test(dataProvider = "variants")

    public void testThrowingAsLines(String uri,

                                    boolean sameClient,

                                    Thrower thrower)

            throws Exception

    {

        String test =  format("testThrowingAsLines(%s, %b, %s)",

                uri, sameClient, thrower);

        testThrowing(test, uri, sameClient, BodyHandlers::ofLines,

                this::checkAsLines, thrower);

    }

    @Test(dataProvider = "variants")

    public void testThrowingAsInputStream(String uri,

                                          boolean sameClient,

                                          Thrower thrower)

            throws Exception

    {

        String test = format("testThrowingAsInputStream(%s, %b, %s)",

                uri, sameClient, thrower);

        testThrowing(test, uri, sameClient, BodyHandlers::ofInputStream,

                this::checkAsInputStream,  thrower);

    }

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

                                    Supplier<BodyHandler<T>> handlers,

                                    Finisher finisher, Thrower thrower)

            throws Exception

    {

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

        try {

            testThrowing(uri, sameClient, handlers, finisher, thrower);

        } catch (Error | Exception x) {

            FAILURES.putIfAbsent(name, x);

            throw x;

        }

    }

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

                                    Supplier<BodyHandler<T>> handlers,

                                    Finisher finisher, Thrower thrower)

            throws Exception

    {

        HttpClient client = null;

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

            if (where == Where.ON_ERROR) continue;

            if (!sameClient || client == null)

                client = newHttpClient(sameClient);

            HttpRequest req = HttpRequest.

                    newBuilder(URI.create(uri))

                    .build();

            ConcurrentMap<HttpRequest, CompletableFuture<HttpResponse<T>>> promiseMap =

                    new ConcurrentHashMap<>();

            Supplier<BodyHandler<T>> throwing = () ->

                    new ThrowingBodyHandler(where.select(thrower), handlers.get());

            PushPromiseHandler<T> pushHandler = new ThrowingPromiseHandler<>(

                    where.select(thrower),

                    PushPromiseHandler.of((r) -> throwing.get(), promiseMap));

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

            HttpResponse<T> response = null;

            try {

                response = client.sendAsync(req, handlers.get(), pushHandler).join();

            } catch (Error | Exception x) {

                throw x;

            }

            if (response != null) {

                finisher.finish(where, req.uri(), response, thrower, promiseMap);

            }

        }

    }

    enum Where {

        BODY_HANDLER, ON_SUBSCRIBE, ON_NEXT, ON_COMPLETE, ON_ERROR, GET_BODY, BODY_CF,

        BEFORE_ACCEPTING, AFTER_ACCEPTING;

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

            return new Consumer<Where>() {

                @Override

                public void accept(Where where) {

                    if (Where.this == where) {

                        consumer.accept(where);

                    }

                }

            };

        }

    }

    interface Thrower extends Consumer<Where>, Predicate<Throwable> {

    }

    interface Finisher<T,U> {

        U finish(Where w, URI requestURI, HttpResponse<T> resp, Thrower thrower,

                 Map<HttpRequest, CompletableFuture<HttpResponse<T>>> promises);

    }

    final <T,U> U shouldHaveThrown(Where w, HttpResponse<T> resp, Thrower thrower) {

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

    }

    final List<String> checkAsString(Where w, URI reqURI,

                                    HttpResponse<String> resp,

                                    Thrower thrower,

                                    Map<HttpRequest, CompletableFuture<HttpResponse<String>>> promises) {

        Function<HttpResponse<String>, List<String>> extractor =

                (r) -> List.of(r.body());

        return check(w, reqURI, resp, thrower, promises, extractor);

    }

    final List<String> checkAsLines(Where w, URI reqURI,

                                    HttpResponse<Stream<String>> resp,

                                    Thrower thrower,

                                    Map<HttpRequest, CompletableFuture<HttpResponse<Stream<String>>>> promises) {

        Function<HttpResponse<Stream<String>>, List<String>> extractor =

                (r) -> r.body().collect(Collectors.toList());

        return check(w, reqURI, resp, thrower, promises, extractor);

    }

    final List<String> checkAsInputStream(Where w, URI reqURI,

                                          HttpResponse<InputStream> resp,

                                          Thrower thrower,

                                          Map<HttpRequest, CompletableFuture<HttpResponse<InputStream>>> promises)

    {

        Function<HttpResponse<InputStream>, List<String>> extractor = (r) -> {

            List<String> result;

            try (InputStream is = r.body()) {

                result = new BufferedReader(new InputStreamReader(is))

                        .lines().collect(Collectors.toList());

            } catch (Throwable t) {

                throw new CompletionException(t);

            }

            return result;

        };

        return check(w, reqURI, resp, thrower, promises, extractor);

    }

    private final <T> List<String> check(Where w, URI reqURI,

                                 HttpResponse<T> resp,

                                 Thrower thrower,

                                 Map<HttpRequest, CompletableFuture<HttpResponse<T>>> promises,

                                 Function<HttpResponse<T>, List<String>> extractor)

    {

        List<String> result = extractor.apply(resp);

        for (HttpRequest req : promises.keySet()) {

            switch (w) {

                case BEFORE_ACCEPTING:

                    throw new RuntimeException("No push promise should have been received" +

                            " for " + reqURI + " in " + w + ": got " + promises.keySet());

                default:

                    break;

            }

            HttpResponse<T> presp;

            try {

                presp = promises.get(req).join();

            } catch (Error | Exception x) {

                Throwable cause = findCause(x, thrower);

                if (cause != null) {

                    out.println(now() + "Got expected exception in "

                            + w + ": " + cause);

                    continue;

                }

                throw x;

            }

            switch (w) {

                case BEFORE_ACCEPTING:

                case AFTER_ACCEPTING:

                case BODY_HANDLER:

                case ON_SUBSCRIBE:

                case GET_BODY:

                case BODY_CF:

                    return shouldHaveThrown(w, presp, thrower);

                default:

                    break;

            }

            List<String> presult = null;

            try {

                presult = extractor.apply(presp);

            } catch (Error | Exception x) {

                Throwable cause = findCause(x, thrower);

                if (cause != null) {

                    out.println(now() + "Got expected exception for "

                            + req + " in " + w + ": " + cause);

                    continue;

                }

                throw x;

            }

            throw new RuntimeException("Expected exception not thrown for "

                    + req + " in " + w);

        }

        final int expectedCount;

        switch (w) {

            case BEFORE_ACCEPTING:

                expectedCount = 0;

                break;

            default:

                expectedCount = 3;

        }

        assertEquals(promises.size(), expectedCount,

                "bad promise count for " + reqURI + " with " + w);

        assertEquals(result, List.of(reqURI.toASCIIString()));

        return result;

    }

    private static Throwable findCause(Throwable x,

                                       Predicate<Throwable> filter) {

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

        return x;

    }

    static final class UncheckedCustomExceptionThrower implements Thrower {

        @Override

        public void accept(Where where) {

            out.println(now() + "Throwing in " + where);

            throw new UncheckedCustomException(where.name());

        }

        @Override

        public boolean test(Throwable throwable) {

            return UncheckedCustomException.class.isInstance(throwable);

        }

        @Override

        public String toString() {

            return "UncheckedCustomExceptionThrower";

        }

    }

    static final class UncheckedIOExceptionThrower implements Thrower {

        @Override

        public void accept(Where where) {

            out.println(now() + "Throwing in " + where);

            throw new UncheckedIOException(new CustomIOException(where.name()));

        }

        @Override

        public boolean test(Throwable throwable) {

            return UncheckedIOException.class.isInstance(throwable)

                    && CustomIOException.class.isInstance(throwable.getCause());

        }

        @Override

        public String toString() {

            return "UncheckedIOExceptionThrower";

        }

    }

    static final class UncheckedCustomException extends RuntimeException {

        UncheckedCustomException(String message) {

            super(message);

        }

        UncheckedCustomException(String message, Throwable cause) {

            super(message, cause);