package io.searchbox.client.http;

import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.HttpObject;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponse;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.JestResult;
import io.searchbox.client.JestResultHandler;
import io.searchbox.client.config.HttpClientConfig;
import io.searchbox.indices.Stats;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.node.Node;
import org.elasticsearch.test.ESIntegTestCase;
import org.junit.*;
import org.littleshoot.proxy.HttpFilters;
import org.littleshoot.proxy.HttpFiltersAdapter;
import org.littleshoot.proxy.HttpFiltersSourceAdapter;
import org.littleshoot.proxy.HttpProxyServer;
import org.littleshoot.proxy.impl.DefaultHttpProxyServer;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author cihat keser
 */
@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0)
public class JestHttpClientSystemWideProxyIntegrationTest extends ESIntegTestCase {

    private static final int PROXY_PORT = 8790;

    private AtomicInteger numProxyRequests = new AtomicInteger(0);
    private JestClientFactory factory = new JestClientFactory();
    private HttpProxyServer server = null;

    private static String nonProxyHostsDefault;
    private static String proxyHostDefault;
    private static String proxyPortDefault;
    private static String useSystemProxiesDefault;

    @BeforeClass
    public static void setupOnce() throws URISyntaxException {
        proxyHostDefault = System.getProperty("http.proxyHost");
        proxyPortDefault = System.getProperty("http.proxyPort");
        useSystemProxiesDefault = System.getProperty("java.net.useSystemProxies");

        System.setProperty("http.proxyHost", "localhost");
        System.setProperty("http.proxyPort", Integer.toString(PROXY_PORT));
        nonProxyHostsDefault = System.getProperty("http.nonProxyHosts");
        System.setProperty("http.nonProxyHosts", ""); // we want localhost to go through proxy
        System.setProperty("java.net.useSystemProxies", "true");
    }

    @Before
    public void setup() {
        server = DefaultHttpProxyServer
                .bootstrap()
                .withPort(PROXY_PORT)
                .withFiltersSource(new HttpFiltersSourceAdapter() {
                    public HttpFilters filterRequest(HttpRequest originalRequest, ChannelHandlerContext ctx) {
                        return new HttpFiltersAdapter(originalRequest) {
                            @Override
                            public void proxyToServerRequestSent() {
                                numProxyRequests.incrementAndGet();
                            }
                        };
                    }
                }).start();
    }

    @After
    public void destroy() {
        if (server != null) server.stop();
    }

    @AfterClass
    public static void destroyOnce() {
        rollbackSystemProperty("http.proxyHost", proxyHostDefault);
        rollbackSystemProperty("http.proxyPort", proxyPortDefault);
        rollbackSystemProperty("http.nonProxyHosts", nonProxyHostsDefault);
        rollbackSystemProperty("java.net.useSystemProxies", useSystemProxiesDefault);
    }

    private static void rollbackSystemProperty(String key, String value) {
        if (value == null) {
            System.clearProperty(key);
        } else {
            System.setProperty(key, value);
        }
    }

    @Test
    public void testConnectionThroughDefaultProxy() throws IOException, ExecutionException, InterruptedException {
        internalCluster().ensureAtLeastNumDataNodes(1);
        assertEquals("All nodes in cluster should have HTTP endpoint exposed", 1, cluster().httpAddresses().length);

        // test sync execution
        factory.setHttpClientConfig(new HttpClientConfig
                .Builder("http://localhost:" + cluster().httpAddresses()[0].getPort())
                .build());
        JestHttpClient jestClient = (JestHttpClient) factory.getObject();
        assertNotNull(jestClient);

        JestResult result = jestClient.execute(new Stats.Builder().build());
        assertTrue(result.getErrorMessage(), result.isSucceeded());
        assertEquals(1, numProxyRequests.intValue());
        jestClient.shutdownClient();

        // test async execution
        factory.setHttpClientConfig(new HttpClientConfig
                .Builder("http://localhost:" + cluster().httpAddresses()[0].getPort())
                .multiThreaded(true)
                .build());
        jestClient = (JestHttpClient) factory.getObject();
        assertNotNull(jestClient);

        final CountDownLatch actionExecuted = new CountDownLatch(1);
        jestClient.executeAsync(new Stats.Builder().build(), new JestResultHandler<JestResult>() {
            @Override
            public void completed(JestResult result) {
                actionExecuted.countDown();
            }

            @Override
            public void failed(Exception ex) {
                throw new RuntimeException(ex);
            }
        });

        boolean finishedAsync = actionExecuted.await(2, TimeUnit.SECONDS);
        if (!finishedAsync) {
            fail("Execution took too long to complete");
        }

        assertEquals(2, numProxyRequests.intValue());
        jestClient.shutdownClient();
    }

    @Override
    protected Settings nodeSettings(int nodeOrdinal) {
        return Settings.builder()
                .put(super.nodeSettings(nodeOrdinal))
                .build();
    }
}

@@ -118,111 +118,15 @@
 Http
-Object;%0Aimport io.netty.handler.codec.http.HttpRequest;%0Aimport io.netty.handler.codec.http.HttpResponse
+Request
 ;%0Aim
@@ -341,16 +341,71 @@
 .Stats;%0A
+import org.elasticsearch.common.network.NetworkModule;%0A
 import o
@@ -476,17 +476,22 @@
 rch.
-node.Node
+plugins.Plugin
 ;%0Aim
@@ -550,15 +550,175 @@
 org.
-junit.*
+elasticsearch.transport.Netty4Plugin;%0Aimport org.junit.After;%0Aimport org.junit.AfterClass;%0Aimport org.junit.Before;%0Aimport org.junit.BeforeClass;%0Aimport org.junit.Test
 ;%0Aim
@@ -1022,32 +1022,91 @@
 yntaxException;%0A
+import java.util.Collection;%0Aimport java.util.Collections;%0A
 import java.util
@@ -3230,32 +3230,33 @@
 start();%0A    %7D%0A%0A
+%0A
     @After%0A    p
@@ -5917,86 +5917,372 @@
 er()
-%0A                .put(super.nodeSettings(nodeOrdinal))%0A                .build(
+.put(super.nodeSettings(nodeOrdinal))%0A                .put(NetworkModule.HTTP_TYPE_KEY, Netty4Plugin.NETTY_HTTP_TRANSPORT_NAME)%0A                .put(NetworkModule.HTTP_ENABLED.getKey(), true)%0A                .build();%0A    %7D%0A%0A    @Override%0A    protected Collection%3CClass%3C? extends Plugin%3E%3E nodePlugins() %7B%0A        return Collections.singletonList(Netty4Plugin.class
 );%0A 

Fix JestHttpClientSystemWideProxyIntegrationTest