Tests for async proxy functionality.

jetty-http2/http2-client/pom.xml

@@ -52,6 +52,12 @@
             <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-proxy</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.eclipse.jetty.http2</groupId>
             <artifactId>http2-server</artifactId>

jetty-http2/http2-client/src/test/java/org/eclipse/jetty/http2/client/ProxyTest.java

@@ -0,0 +1,195 @@
+//
+//  ========================================================================
+//  Copyright (c) 1995-2015 Mort Bay Consulting Pty. Ltd.
+//  ------------------------------------------------------------------------
+//  All rights reserved. This program and the accompanying materials
+//  are made available under the terms of the Eclipse Public License v1.0
+//  and Apache License v2.0 which accompanies this distribution.
+//
+//      The Eclipse Public License is available at
+//      http://www.eclipse.org/legal/epl-v10.html
+//
+//      The Apache License v2.0 is available at
+//      http://www.opensource.org/licenses/apache2.0.php
+//
+//  You may elect to redistribute this code under either of these licenses.
+//  ========================================================================
+//
+
+package org.eclipse.jetty.http2.client;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.http.HostPortHttpField;
+import org.eclipse.jetty.http.HttpFields;
+import org.eclipse.jetty.http.HttpScheme;
+import org.eclipse.jetty.http.HttpVersion;
+import org.eclipse.jetty.http.MetaData;
+import org.eclipse.jetty.http2.api.Session;
+import org.eclipse.jetty.http2.api.Stream;
+import org.eclipse.jetty.http2.frames.DataFrame;
+import org.eclipse.jetty.http2.frames.HeadersFrame;
+import org.eclipse.jetty.http2.server.HTTP2ServerConnectionFactory;
+import org.eclipse.jetty.proxy.AsyncProxyServlet;
+import org.eclipse.jetty.server.HttpConfiguration;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.eclipse.jetty.toolchain.test.TestTracker;
+import org.eclipse.jetty.util.Callback;
+import org.eclipse.jetty.util.FuturePromise;
+import org.eclipse.jetty.util.Promise;
+import org.eclipse.jetty.util.thread.QueuedThreadPool;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class ProxyTest
+{
+    @Rule
+    public final TestTracker tracker = new TestTracker();
+    private HTTP2Client client;
+    private Server proxy;
+    private ServerConnector proxyConnector;
+    private Server server;
+    private ServerConnector serverConnector;
+
+    private void startServer(HttpServlet servlet) throws Exception
+    {
+        QueuedThreadPool serverPool = new QueuedThreadPool();
+        serverPool.setName("server");
+        server = new Server(serverPool);
+        serverConnector = new ServerConnector(server);
+        server.addConnector(serverConnector);
+
+        ServletContextHandler appCtx = new ServletContextHandler(server, "/", true, false);
+        ServletHolder appServletHolder = new ServletHolder(servlet);
+        appCtx.addServlet(appServletHolder, "/*");
+
+        server.start();
+    }
+
+    private void startProxy(HttpServlet proxyServlet, Map<String, String> initParams) throws Exception
+    {
+        QueuedThreadPool proxyPool = new QueuedThreadPool();
+        proxyPool.setName("proxy");
+        proxy = new Server(proxyPool);
+
+        HttpConfiguration configuration = new HttpConfiguration();
+        configuration.setSendDateHeader(false);
+        configuration.setSendServerVersion(false);
+        String value = initParams.get("outputBufferSize");
+        if (value != null)
+            configuration.setOutputBufferSize(Integer.valueOf(value));
+        proxyConnector = new ServerConnector(proxy, new HTTP2ServerConnectionFactory(configuration));
+        proxy.addConnector(proxyConnector);
+
+        ServletContextHandler proxyContext = new ServletContextHandler(proxy, "/", true, false);
+        ServletHolder proxyServletHolder = new ServletHolder(proxyServlet);
+        proxyServletHolder.setInitParameters(initParams);
+        proxyContext.addServlet(proxyServletHolder, "/*");
+
+        proxy.start();
+    }
+
+    private void startClient() throws Exception
+    {
+        QueuedThreadPool clientExecutor = new QueuedThreadPool();
+        clientExecutor.setName("client");
+        client = new HTTP2Client();
+        client.setExecutor(clientExecutor);
+        client.start();
+    }
+
+    private Session newClient(Session.Listener listener) throws Exception
+    {
+        String host = "localhost";
+        int port = proxyConnector.getLocalPort();
+        InetSocketAddress address = new InetSocketAddress(host, port);
+        FuturePromise<Session> promise = new FuturePromise<>();
+        client.connect(address, listener, promise);
+        return promise.get(5, TimeUnit.SECONDS);
+    }
+
+    private MetaData.Request newRequest(String method, String path, HttpFields fields)
+    {
+        String host = "localhost";
+        int port = proxyConnector.getLocalPort();
+        String authority = host + ":" + port;
+        return new MetaData.Request(method, HttpScheme.HTTP, new HostPortHttpField(authority), path, HttpVersion.HTTP_2, fields);
+    }
+
+    @After
+    public void dispose() throws Exception
+    {
+        client.stop();
+        proxy.stop();
+        server.stop();
+    }
+
+    @Test
+    public void testServerBigDownloadSlowClient() throws Exception
+    {
+        final CountDownLatch serverLatch = new CountDownLatch(1);
+        final byte[] content = new byte[1024 * 1024];
+        startServer(new HttpServlet()
+        {
+            @Override
+            protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
+            {
+                response.getOutputStream().write(content);
+                serverLatch.countDown();
+            }
+        });
+        Map<String, String> params = new HashMap<>();
+        params.put("proxyTo", "http://localhost:" + serverConnector.getLocalPort());
+        startProxy(new AsyncProxyServlet.Transparent()
+        {
+            @Override
+            protected void sendProxyRequest(HttpServletRequest clientRequest, HttpServletResponse proxyResponse, Request proxyRequest)
+            {
+                proxyRequest.version(HttpVersion.HTTP_1_1);
+                super.sendProxyRequest(clientRequest, proxyResponse, proxyRequest);
+            }
+        }, params);
+        startClient();
+
+        final CountDownLatch clientLatch = new CountDownLatch(1);
+        Session session = newClient(new Session.Listener.Adapter());
+        MetaData.Request metaData = newRequest("GET", "/", new HttpFields());
+        HeadersFrame frame = new HeadersFrame(1, metaData, null, true);
+        session.newStream(frame, new Promise.Adapter<Stream>(), new Stream.Listener.Adapter()
+        {
+            @Override
+            public void onData(Stream stream, DataFrame frame, Callback callback)
+            {
+                try
+                {
+                    TimeUnit.MILLISECONDS.sleep(1);
+                    callback.succeeded();
+                    if (frame.isEndStream())
+                        clientLatch.countDown();
+                }
+                catch (InterruptedException x)
+                {
+                    callback.failed(x);
+                }
+            }
+        });
+
+        Assert.assertTrue(serverLatch.await(15, TimeUnit.SECONDS));
+        Assert.assertTrue(clientLatch.await(15, TimeUnit.SECONDS));
+    }
+}