Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

396606 make spdy proxy capable of receiving SPDY and talk HTTP to the upstream server

This commit is contained in:
Thomas Becker 2013-01-31 13:06:36 +01:00
parent cd30ac104d
commit 95698ca248
10 changed files with 574 additions and 180 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
jetty-spdy/spdy-core/src/main/java/org/eclipse/jetty/spdy/StandardSession.java
View File

@ -281,7 +281,6 @@ public class StandardSession implements ISession, Parser.Listener, Dumpable
private void goAway(SessionStatus sessionStatus, long timeout, TimeUnit unit, Callback callback) private void goAway(SessionStatus sessionStatus, long timeout, TimeUnit unit, Callback callback)
{ {
new Exception().printStackTrace();
if (goAwaySent.compareAndSet(false, true)) if (goAwaySent.compareAndSet(false, true))
{ {
if (!goAwayReceived.get()) if (!goAwayReceived.get())

89
jetty-spdy/spdy-core/src/main/java/org/eclipse/jetty/spdy/api/Stream.java
View File

@ -27,34 +27,28 @@ import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.Promise; import org.eclipse.jetty.util.Promise;
/** /**
* <p>A {@link Stream} represents a bidirectional exchange of data on top of a {@link Session}.</p> * <p>A {@link Stream} represents a bidirectional exchange of data on top of a {@link Session}.</p> <p>Differently from
* <p>Differently from socket streams, where the input and output streams are permanently associated * socket streams, where the input and output streams are permanently associated with the socket (and hence with the
* with the socket (and hence with the connection that the socket represents), there can be multiple * connection that the socket represents), there can be multiple SPDY streams for a SPDY session.</p> <p>SPDY streams
* SPDY streams for a SPDY session.</p> * may terminate without this implying that the SPDY session is terminated.</p> <p>If SPDY is used to transport the HTTP
* <p>SPDY streams may terminate without this implying that the SPDY session is terminated.</p> * protocol, then a SPDY stream maps to a HTTP request/response cycle, and after the request/response cycle is
* <p>If SPDY is used to transport the HTTP protocol, then a SPDY stream maps to a HTTP request/response * completed, the stream is closed, and other streams may be opened. Differently from HTTP, though, multiple SPDY
* cycle, and after the request/response cycle is completed, the stream is closed, and other streams * streams may be opened concurrently on the same SPDY session.</p> <p>Like {@link Session}, {@link Stream} is the
* may be opened. Differently from HTTP, though, multiple SPDY streams may be opened concurrently * active part and by calling its API applications can generate events on the stream; conversely, {@link
* on the same SPDY session.</p> * StreamFrameListener} is the passive part, and its callbacks are invoked when events happen on the stream.</p> <p>A
* <p>Like {@link Session}, {@link Stream} is the active part and by calling its API applications * {@link Stream} can send multiple data frames one after the other but implementations use a flow control mechanism
* can generate events on the stream; conversely, {@link StreamFrameListener} is the passive part, and its * that only sends the data frames if the other end has signalled that it can accept the frame.</p> <p>Data frames
* callbacks are invoked when events happen on the stream.</p> * should be sent sequentially only when the previous frame has been completely sent. The reason for this requirement is
* <p>A {@link Stream} can send multiple data frames one after the other but implementations use a * to avoid potentially confusing code such as:</p>
* flow control mechanism that only sends the data frames if the other end has signalled that it can
* accept the frame.</p>
* <p>Data frames should be sent sequentially only when the previous frame has been completely sent.
* The reason for this requirement is to avoid potentially confusing code such as:</p>
* <pre> * <pre>
* // WRONG CODE, DO NOT USE IT * // WRONG CODE, DO NOT USE IT
* final Stream stream = ...; * final Stream stream = ...;
* stream.data(StringDataInfo("chunk1", false), 5, TimeUnit.SECONDS, new Handler&lt;Void&gt;() { ... }); * stream.data(StringDataInfo("chunk1", false), 5, TimeUnit.SECONDS, new Handler&lt;Void&gt;() { ... });
* stream.data(StringDataInfo("chunk2", true), 1, TimeUnit.SECONDS, new Handler&lt;Void&gt;() { ... }); * stream.data(StringDataInfo("chunk2", true), 1, TimeUnit.SECONDS, new Handler&lt;Void&gt;() { ... });
* </pre> * </pre>
* <p>where the second call to {@link #data(DataInfo, Callback)} has a timeout smaller * <p>where the second call to {@link #data(DataInfo, Callback)} has a timeout smaller than the previous call.</p>
* than the previous call.</p> * <p>The behavior of such style of invocations is unspecified (it may even throw an exception - similar to {@link
* <p>The behavior of such style of invocations is unspecified (it may even throw an exception - similar * WritePendingException}).</p> <p>The correct sending of data frames is the following:</p>
* to {@link WritePendingException}).</p>
* <p>The correct sending of data frames is the following:</p>
* <pre> * <pre>
* final Stream stream = ...; * final Stream stream = ...;
* ... * ...
@ -92,10 +86,8 @@ public interface Stream
public Session getSession(); public Session getSession();
/** /**
* <p>Initiate a unidirectional spdy pushstream associated to this stream asynchronously<p> * <p>Initiate a unidirectional spdy pushstream associated to this stream asynchronously<p> <p>Callers may use the
* <p>Callers may use the returned future to get the pushstream once it got created</p> * returned future to get the pushstream once it got created</p>
*
*
* *
* @param pushInfo the metadata to send on stream creation * @param pushInfo the metadata to send on stream creation
* @return a future containing the stream once it got established * @return a future containing the stream once it got established
@ -104,10 +96,8 @@ public interface Stream
public Stream push(PushInfo pushInfo) throws InterruptedException, ExecutionException, TimeoutException; public Stream push(PushInfo pushInfo) throws InterruptedException, ExecutionException, TimeoutException;
/** /**
* <p>Initiate a unidirectional spdy pushstream associated to this stream asynchronously<p> * <p>Initiate a unidirectional spdy pushstream associated to this stream asynchronously<p> <p>Callers may pass a
* <p>Callers may pass a non-null completion callback to be notified of when the * non-null completion callback to be notified of when the pushstream has been established.</p>
* pushstream has been established.</p>
*
* *
* @param pushInfo the metadata to send on stream creation * @param pushInfo the metadata to send on stream creation
* @param callback the completion callback that gets notified once the pushstream is established * @param callback the completion callback that gets notified once the pushstream is established
@ -116,9 +106,8 @@ public interface Stream
public void push(PushInfo pushInfo, Promise<Stream> callback); public void push(PushInfo pushInfo, Promise<Stream> callback);
/** /**
* <p>Sends asynchronously a SYN_REPLY frame in response to a SYN_STREAM frame.</p> * <p>Sends asynchronously a SYN_REPLY frame in response to a SYN_STREAM frame.</p> <p>Callers may use the returned
* <p>Callers may use the returned future to wait for the reply to be actually sent.</p> * future to wait for the reply to be actually sent.</p>
*
* *
* @param replyInfo the metadata to send * @param replyInfo the metadata to send
* @return a future to wait for the reply to be sent * @return a future to wait for the reply to be sent
@ -128,9 +117,8 @@ public interface Stream
public void reply(ReplyInfo replyInfo) throws InterruptedException, ExecutionException, TimeoutException; public void reply(ReplyInfo replyInfo) throws InterruptedException, ExecutionException, TimeoutException;
/** /**
* <p>Sends asynchronously a SYN_REPLY frame in response to a SYN_STREAM frame.</p> * <p>Sends asynchronously a SYN_REPLY frame in response to a SYN_STREAM frame.</p> <p>Callers may pass a non-null
* <p>Callers may pass a non-null completion callback to be notified of when the * completion callback to be notified of when the reply has been actually sent.</p>
* reply has been actually sent.</p>
* *
* @param replyInfo the metadata to send * @param replyInfo the metadata to send
* @param callback the completion callback that gets notified of reply sent * @param callback the completion callback that gets notified of reply sent
@ -139,10 +127,8 @@ public interface Stream
public void reply(ReplyInfo replyInfo, Callback callback); public void reply(ReplyInfo replyInfo, Callback callback);
/** /**
* <p>Sends asynchronously a DATA frame on this stream.</p> * <p>Sends asynchronously a DATA frame on this stream.</p> <p>DATA frames should always be sent after a SYN_REPLY
* <p>DATA frames should always be sent after a SYN_REPLY frame.</p> * frame.</p> <p>Callers may use the returned future to wait for the data to be actually sent.</p>
* <p>Callers may use the returned future to wait for the data to be actually sent.</p>
*
* *
* @param dataInfo the metadata to send * @param dataInfo the metadata to send
* @return a future to wait for the data to be sent * @return a future to wait for the data to be sent
@ -152,10 +138,9 @@ public interface Stream
public void data(DataInfo dataInfo) throws InterruptedException, ExecutionException, TimeoutException; public void data(DataInfo dataInfo) throws InterruptedException, ExecutionException, TimeoutException;
/** /**
* <p>Sends asynchronously a DATA frame on this stream.</p> * <p>Sends asynchronously a DATA frame on this stream.</p> <p>DATA frames should always be sent after a SYN_REPLY
* <p>DATA frames should always be sent after a SYN_REPLY frame.</p> * frame.</p> <p>Callers may pass a non-null completion callback to be notified of when the data has been actually
* <p>Callers may pass a non-null completion callback to be notified of when the * sent.</p>
* data has been actually sent.</p>
* *
* @param dataInfo the metadata to send * @param dataInfo the metadata to send
* @param callback the completion callback that gets notified of data sent * @param callback the completion callback that gets notified of data sent
@ -164,10 +149,8 @@ public interface Stream
public void data(DataInfo dataInfo, Callback callback); public void data(DataInfo dataInfo, Callback callback);
/** /**
* <p>Sends asynchronously a HEADER frame on this stream.</p> * <p>Sends asynchronously a HEADER frame on this stream.</p> <p>HEADERS frames should always be sent after a
* <p>HEADERS frames should always be sent after a SYN_REPLY frame.</p> * SYN_REPLY frame.</p> <p>Callers may use the returned future to wait for the headers to be actually sent.</p>
* <p>Callers may use the returned future to wait for the headers to be actually sent.</p>
*
* *
* @param headersInfo the metadata to send * @param headersInfo the metadata to send
* @return a future to wait for the headers to be sent * @return a future to wait for the headers to be sent
@ -177,10 +160,9 @@ public interface Stream
public void headers(HeadersInfo headersInfo) throws InterruptedException, ExecutionException, TimeoutException; public void headers(HeadersInfo headersInfo) throws InterruptedException, ExecutionException, TimeoutException;
/** /**
* <p>Sends asynchronously a HEADER frame on this stream.</p> * <p>Sends asynchronously a HEADER frame on this stream.</p> <p>HEADERS frames should always be sent after a
* <p>HEADERS frames should always be sent after a SYN_REPLY frame.</p> * SYN_REPLY frame.</p> <p>Callers may pass a non-null completion callback to be notified of when the headers have
* <p>Callers may pass a non-null completion callback to be notified of when the * been actually sent.</p>
* headers have been actually sent.</p>
* *
* @param headersInfo the metadata to send * @param headersInfo the metadata to send
* @param callback the completion callback that gets notified of headers sent * @param callback the completion callback that gets notified of headers sent
@ -212,7 +194,8 @@ public interface Stream
/** /**
* @param key the attribute key * @param key the attribute key
* @return an arbitrary object associated with the given key to this stream * @return an arbitrary object associated with the given key to this stream or null if no object can be found for
* the given key.
* @see #setAttribute(String, Object) * @see #setAttribute(String, Object)
*/ */
public Object getAttribute(String key); public Object getAttribute(String key);

113
jetty-spdy/spdy-http-server/src/main/java/org/eclipse/jetty/spdy/server/proxy/HTTPProxyEngine.java
View File

@ -43,6 +43,7 @@ import org.eclipse.jetty.spdy.server.http.HTTPSPDYHeader;
import org.eclipse.jetty.util.BufferUtil; import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback; import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.Fields; import org.eclipse.jetty.util.Fields;
import org.eclipse.jetty.util.HttpCookieStore;
import org.eclipse.jetty.util.log.Log; import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger; import org.eclipse.jetty.util.log.Logger;
@ -50,45 +51,25 @@ import org.eclipse.jetty.util.log.Logger;
* <p>{@link HTTPProxyEngine} implements a SPDY to HTTP proxy, that is, converts SPDY events received by clients into * <p>{@link HTTPProxyEngine} implements a SPDY to HTTP proxy, that is, converts SPDY events received by clients into
* HTTP events for the servers.</p> * HTTP events for the servers.</p>
*/ */
public class HTTPProxyEngine extends ProxyEngine implements StreamFrameListener public class HTTPProxyEngine extends ProxyEngine
{ {
private static final Logger LOG = Log.getLogger(HTTPProxyEngine.class); private static final Logger LOG = Log.getLogger(HTTPProxyEngine.class);
private static final String CLIENT_REQUEST_ATTRIBUTE = "org.eclipse.jetty.spdy.server.http.proxy.request";
private static final Callback LOGGING_CALLBACK = new LoggingCallback(); private static final Callback LOGGING_CALLBACK = new LoggingCallback();
private final HttpClient httpClient = new HttpClient(); private final HttpClient httpClient;
private volatile boolean committed;
public HTTPProxyEngine() public HTTPProxyEngine(HttpClient httpClient)
{ {
try this.httpClient = httpClient;
{ configureHttpClient(httpClient);
httpClient.start();
}
catch (Exception e)
{
LOG.warn("Exception while starting HttpClient: ", e);
}
} }
public long getConnectTimeout() private void configureHttpClient(HttpClient httpClient)
{ {
return httpClient.getConnectTimeout(); // Redirects must be proxied as is, not followed
} httpClient.setFollowRedirects(false);
// Must not store cookies, otherwise cookies of different clients will mix
public void setConnectTimeout(long connectTimeout) httpClient.setCookieStore(new HttpCookieStore.Empty());
{
httpClient.setConnectTimeout(connectTimeout);
}
public long getIdleTimeout()
{
return httpClient.getIdleTimeout();
}
public void setIdleTimeout(long idleTimeout)
{
httpClient.setIdleTimeout(idleTimeout);
} }
public StreamFrameListener proxy(final Stream clientStream, SynInfo clientSynInfo, ProxyEngineSelector.ProxyServerInfo proxyServerInfo) public StreamFrameListener proxy(final Stream clientStream, SynInfo clientSynInfo, ProxyEngineSelector.ProxyServerInfo proxyServerInfo)
@ -105,27 +86,55 @@ public class HTTPProxyEngine extends ProxyEngine implements StreamFrameListener
String host = proxyServerInfo.getHost(); String host = proxyServerInfo.getHost();
int port = proxyServerInfo.getAddress().getPort(); int port = proxyServerInfo.getAddress().getPort();
LOG.debug("Sending HTTP request to: {}", host + ":" + port); LOG.debug("Sending HTTP request to: {}", host + ":" + port);
Request request = httpClient.newRequest(host, port) final Request request = httpClient.newRequest(host, port)
.path(path) .path(path)
.method(HttpMethod.fromString(method)); .method(HttpMethod.fromString(method));
addNonSpdyHeadersToRequest(version, headers, request); addNonSpdyHeadersToRequest(version, headers, request);
if (!clientSynInfo.isClose()) if (!clientSynInfo.isClose())
{ {
clientStream.setAttribute(CLIENT_REQUEST_ATTRIBUTE, request);
request.content(new DeferredContentProvider()); request.content(new DeferredContentProvider());
} }
sendRequest(clientStream, request); sendRequest(clientStream, request);
return this; return new StreamFrameListener.Adapter()
{
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
// We proxy to HTTP so we do not receive replies
throw new UnsupportedOperationException("Not Yet Implemented");
}
@Override
public void onHeaders(Stream stream, HeadersInfo headersInfo)
{
throw new UnsupportedOperationException("Not Yet Implemented");
}
@Override
public void onData(Stream clientStream, final DataInfo clientDataInfo)
{
LOG.debug("received clientDataInfo: {} for stream: {}", clientDataInfo, clientStream);
DeferredContentProvider contentProvider = (DeferredContentProvider)request.getContent();
contentProvider.offer(clientDataInfo.asByteBuffer(true));
if (clientDataInfo.isClose())
contentProvider.close();
}
};
} }
private void sendRequest(final Stream clientStream, Request request) private void sendRequest(final Stream clientStream, Request request)
{ {
request.send(new Response.Listener.Empty() request.send(new Response.Listener.Empty()
{ {
private volatile boolean committed;
@Override @Override
public void onHeaders(final Response response) public void onHeaders(final Response response)
{ {
@ -140,9 +149,14 @@ public class HTTPProxyEngine extends ProxyEngine implements StreamFrameListener
LOG.debug("failed: ", x); LOG.debug("failed: ", x);
response.abort(x); response.abort(x);
} }
});
@Override
public void succeeded()
{
committed = true; committed = true;
} }
});
}
@Override @Override
public void onContent(final Response response, ByteBuffer content) public void onContent(final Response response, ByteBuffer content)
@ -185,7 +199,7 @@ public class HTTPProxyEngine extends ProxyEngine implements StreamFrameListener
@Override @Override
public void onFailure(Response response, Throwable failure) public void onFailure(Response response, Throwable failure)
{ {
LOG.debug("onFailure called: {}", failure); LOG.debug("onFailure called: ", failure);
if (committed) if (committed)
{ {
LOG.debug("clientStream already committed. Resetting stream."); LOG.debug("clientStream already committed. Resetting stream.");
@ -221,6 +235,9 @@ public class HTTPProxyEngine extends ProxyEngine implements StreamFrameListener
Fields responseHeaders = new Fields(); Fields responseHeaders = new Fields();
for (HttpField header : response.getHeaders()) for (HttpField header : response.getHeaders())
responseHeaders.add(header.getName(), header.getValue()); responseHeaders.add(header.getName(), header.getValue());
if (response.getStatus() > 0)
responseHeaders.add(HTTPSPDYHeader.STATUS.name(clientStream.getSession().getVersion()),
String.valueOf(response.getStatus()));
addResponseProxyHeaders(clientStream, responseHeaders); addResponseProxyHeaders(clientStream, responseHeaders);
return responseHeaders; return responseHeaders;
} }
@ -232,32 +249,6 @@ public class HTTPProxyEngine extends ProxyEngine implements StreamFrameListener
request.header(header.name(), header.value()); request.header(header.name(), header.value());
} }
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
// We proxy to HTTP so we do not receive replies
throw new UnsupportedOperationException("Not Yet Implemented");
}
@Override
public void onHeaders(Stream stream, HeadersInfo headersInfo)
{
throw new UnsupportedOperationException("Not Yet Implemented");
}
@Override
public void onData(Stream clientStream, final DataInfo clientDataInfo)
{
LOG.debug("received clientDataInfo: {} for stream: {}", clientDataInfo, clientStream);
Request request = (Request)clientStream.getAttribute(CLIENT_REQUEST_ATTRIBUTE);
DeferredContentProvider contentProvider = (DeferredContentProvider)request.getContent();
contentProvider.offer(clientDataInfo.asByteBuffer(true));
if (clientDataInfo.isClose())
contentProvider.close();
}
static class LoggingCallback extends Callback.Adapter static class LoggingCallback extends Callback.Adapter
{ {
@Override @Override

4
jetty-spdy/spdy-http-server/src/test/java/org/eclipse/jetty/spdy/server/http/SPDYTestUtils.java
View File

@ -24,14 +24,14 @@ import org.eclipse.jetty.util.ssl.SslContextFactory;
public class SPDYTestUtils public class SPDYTestUtils
{ {
public static Fields createHeaders(int port, short version, String httpMethod, String path) public static Fields createHeaders(String host, int port, short version, String httpMethod, String path)
{ {
Fields headers = new Fields(); Fields headers = new Fields();
headers.put(HTTPSPDYHeader.METHOD.name(version), httpMethod); headers.put(HTTPSPDYHeader.METHOD.name(version), httpMethod);
headers.put(HTTPSPDYHeader.URI.name(version), path); headers.put(HTTPSPDYHeader.URI.name(version), path);
headers.put(HTTPSPDYHeader.VERSION.name(version), "HTTP/1.1"); headers.put(HTTPSPDYHeader.VERSION.name(version), "HTTP/1.1");
headers.put(HTTPSPDYHeader.SCHEME.name(version), "http"); headers.put(HTTPSPDYHeader.SCHEME.name(version), "http");
headers.put(HTTPSPDYHeader.HOST.name(version), "localhost:" + port); headers.put(HTTPSPDYHeader.HOST.name(version), host + ":" + port);
return headers; return headers;
} }

50
jetty-spdy/spdy-http-server/src/test/java/org/eclipse/jetty/spdy/server/http/ServerHTTPSPDYTest.java
View File

@ -80,7 +80,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", path); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", path);
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{ {
@ -119,7 +119,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", uri); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", uri);
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{ {
@ -155,7 +155,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "HEAD", path); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "HEAD", path);
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{ {
@ -200,7 +200,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "POST", path); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "POST", path);
headers.put("content-type", "application/x-www-form-urlencoded"); headers.put("content-type", "application/x-www-form-urlencoded");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0),
@ -242,7 +242,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "POST", path); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "POST", path);
headers.put("content-type", "application/x-www-form-urlencoded"); headers.put("content-type", "application/x-www-form-urlencoded");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0),
@ -287,7 +287,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "POST", path); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "POST", path);
headers.put("content-type", "application/x-www-form-urlencoded"); headers.put("content-type", "application/x-www-form-urlencoded");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter() Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter()
@ -329,7 +329,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -375,7 +375,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -426,7 +426,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(2); final CountDownLatch dataLatch = new CountDownLatch(2);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -481,7 +481,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -534,7 +534,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -587,7 +587,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -645,7 +645,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -698,7 +698,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{ {
@ -735,7 +735,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -780,7 +780,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch latch = new CountDownLatch(1); final CountDownLatch latch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -834,7 +834,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(2); final CountDownLatch dataLatch = new CountDownLatch(2);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -916,7 +916,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() session.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
@ -972,7 +972,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "GET", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "GET", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
final AtomicInteger contentLength = new AtomicInteger(); final AtomicInteger contentLength = new AtomicInteger();
@ -1036,7 +1036,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "POST", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "POST", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter() Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter()
{ {
@ -1079,7 +1079,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "POST", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "POST", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, true, (byte)0), new StreamFrameListener.Adapter() session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, true, (byte)0), new StreamFrameListener.Adapter()
{ {
@ -1124,7 +1124,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "POST", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "POST", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter() Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter()
{ {
@ -1192,7 +1192,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "POST", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "POST", "/foo");
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter() Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter()
{ {
@ -1260,7 +1260,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "POST", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "POST", "/foo");
final CountDownLatch responseLatch = new CountDownLatch(2); final CountDownLatch responseLatch = new CountDownLatch(2);
Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter() Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter()
{ {
@ -1301,7 +1301,7 @@ public class ServerHTTPSPDYTest extends AbstractHTTPSPDYTest
} }
}), null); }), null);
Fields headers = SPDYTestUtils.createHeaders(connector.getPort(), version, "POST", "/foo"); Fields headers = SPDYTestUtils.createHeaders("localhost", connector.getPort(), version, "POST", "/foo");
final CountDownLatch responseLatch = new CountDownLatch(1); final CountDownLatch responseLatch = new CountDownLatch(1);
Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter() Stream stream = session.syn(new SynInfo(5, TimeUnit.SECONDS, headers, false, (byte)0), new StreamFrameListener.Adapter()
{ {

39
jetty-spdy/spdy-http-server/src/test/java/org/eclipse/jetty/spdy/server/proxy/ProxyHTTPToSPDYTest.java
View File

@ -256,8 +256,43 @@ public class ProxyHTTPToSPDYTest
} }
@Test @Test
public void testHEADRequest() throws Exception{ public void testHEADRequest() throws Exception
// fail("Not yet implemented"); //TODO: {
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Assert.assertTrue(synInfo.isClose());
Fields requestHeaders = synInfo.getHeaders();
Assert.assertNotNull(requestHeaders.get("via"));
Fields responseHeaders = new Fields();
responseHeaders.put(HTTPSPDYHeader.VERSION.name(version), "HTTP/1.1");
responseHeaders.put(HTTPSPDYHeader.STATUS.name(version), "200 OK");
ReplyInfo replyInfo = new ReplyInfo(responseHeaders, true);
stream.reply(replyInfo, new Callback.Adapter());
return null;
}
}));
Socket client = new Socket();
client.connect(proxyAddress);
OutputStream output = client.getOutputStream();
String request = "" +
"HEAD / HTTP/1.1\r\n" +
"Host: localhost:" + proxyAddress.getPort() + "\r\n" +
"\r\n";
output.write(request.getBytes("UTF-8"));
output.flush();
InputStream input = client.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(input, "UTF-8"));
String line = reader.readLine();
Assert.assertTrue(line.contains(" 200 "));
client.close();
} }
@Test @Test

301
jetty-spdy/spdy-http-server/src/test/java/org/eclipse/jetty/spdy/server/proxy/ProxySPDYToHTTPLoadTest.java Normal file
View File

@ -0,0 +1,301 @@
//
// ========================================================================
// Copyright (c) 1995-2013 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.spdy.server.proxy;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.server.handler.DefaultHandler;
import org.eclipse.jetty.spdy.api.DataInfo;
import org.eclipse.jetty.spdy.api.ReplyInfo;
import org.eclipse.jetty.spdy.api.SPDY;
import org.eclipse.jetty.spdy.api.Session;
import org.eclipse.jetty.spdy.api.Stream;
import org.eclipse.jetty.spdy.api.StreamFrameListener;
import org.eclipse.jetty.spdy.api.StringDataInfo;
import org.eclipse.jetty.spdy.api.SynInfo;
import org.eclipse.jetty.spdy.client.SPDYClient;
import org.eclipse.jetty.spdy.server.http.SPDYTestUtils;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.Fields;
import org.eclipse.jetty.util.IO;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import static junit.framework.Assert.fail;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.Assert.assertThat;
@RunWith(value = Parameterized.class)
public class ProxySPDYToHTTPLoadTest
{
@Rule
public final TestWatcher testName = new TestWatcher()
{
@Override
public void starting(Description description)
{
super.starting(description);
System.err.printf("Running %s.%s()%n",
description.getClassName(),
description.getMethodName());
}
};
private final short version;
@Parameterized.Parameters
public static Collection<Short[]> parameters()
{
return Arrays.asList(new Short[]{SPDY.V2}, new Short[]{SPDY.V3});
}
private SPDYClient.Factory factory;
private Server server;
private Server proxy;
private ServerConnector proxyConnector;
private SslContextFactory sslContextFactory = SPDYTestUtils.newSslContextFactory();
public ProxySPDYToHTTPLoadTest(short version)
{
this.version = version;
}
protected InetSocketAddress startServer(Handler handler) throws Exception
{
server = new Server();
ServerConnector connector = new ServerConnector(server);
server.setHandler(handler);
server.addConnector(connector);
server.start();
return new InetSocketAddress("localhost", connector.getLocalPort());
}
protected InetSocketAddress startProxy(InetSocketAddress server1, InetSocketAddress server2,
long proxyConnectorTimeout, long proxyEngineTimeout) throws Exception
{
proxy = new Server();
ProxyEngineSelector proxyEngineSelector = new ProxyEngineSelector();
HttpClient httpClient = new HttpClient();
httpClient.start();
httpClient.setIdleTimeout(proxyEngineTimeout);
HTTPProxyEngine httpProxyEngine = new HTTPProxyEngine(httpClient);
proxyEngineSelector.putProxyEngine("http/1.1", httpProxyEngine);
proxyEngineSelector.putProxyServerInfo("localhost", new ProxyEngineSelector.ProxyServerInfo("http/1.1",
server1.getHostName(), server1.getPort()));
// server2 will be available at two different ProxyServerInfos with different hosts
proxyEngineSelector.putProxyServerInfo("127.0.0.1", new ProxyEngineSelector.ProxyServerInfo("http/1.1",
server2.getHostName(), server2.getPort()));
proxyEngineSelector.putProxyServerInfo("127.0.0.2", new ProxyEngineSelector.ProxyServerInfo("http/1.1",
server2.getHostName(), server2.getPort()));
proxyConnector = new HTTPSPDYProxyServerConnector(proxy, sslContextFactory, proxyEngineSelector);
proxyConnector.setPort(0);
proxyConnector.setIdleTimeout(proxyConnectorTimeout);
proxy.addConnector(proxyConnector);
proxy.start();
return new InetSocketAddress("localhost", proxyConnector.getLocalPort());
}
@Before
public void init() throws Exception
{
factory = new SPDYClient.Factory(sslContextFactory);
factory.start();
}
@After
public void destroy() throws Exception
{
if (server != null)
{
server.stop();
server.join();
}
if (proxy != null)
{
proxy.stop();
proxy.join();
}
factory.stop();
}
@Test
public void testSimpleLoadTest() throws Exception
{
String server1String = "server1";
String server2String = "server2";
InetSocketAddress server1 = startServer(new TestServerHandler(server1String, null));
InetSocketAddress server2 = startServer(new TestServerHandler(server2String, null));
final InetSocketAddress proxyAddress = startProxy(server1, server2, 30000, 30000);
final int requestsPerClient = 50;
ExecutorService executorService = Executors.newFixedThreadPool(3);
Runnable client1 = createClientRunnable(proxyAddress, requestsPerClient, server1String, "localhost");
Runnable client2 = createClientRunnable(proxyAddress, requestsPerClient, server2String, "127.0.0.1");
Runnable client3 = createClientRunnable(proxyAddress, requestsPerClient, server2String, "127.0.0.2");
List<Future> futures = new ArrayList<>();
futures.add(executorService.submit(client1));
futures.add(executorService.submit(client2));
futures.add(executorService.submit(client3));
for (Future future : futures)
{
future.get(60, TimeUnit.SECONDS);
}
}
private Runnable createClientRunnable(final InetSocketAddress proxyAddress, final int requestsPerClient,
final String serverIdentificationString, final String serverHost)
{
Runnable client = new Runnable()
{
@Override
public void run()
{
try
{
Session client = factory.newSPDYClient(version).connect(proxyAddress, null).get(5, TimeUnit.SECONDS);
for (int i = 0; i < requestsPerClient; i++)
{
sendSingleClientRequest(proxyAddress, client, serverIdentificationString, serverHost);
}
}
catch (InterruptedException | ExecutionException | TimeoutException | IOException e)
{
fail();
e.printStackTrace();
}
}
};
return client;
}
private void sendSingleClientRequest(InetSocketAddress proxyAddress, Session client, final String serverIdentificationString, String serverHost) throws ExecutionException, InterruptedException, TimeoutException
{
final String data = UUID.randomUUID().toString();
final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1);
Fields headers = SPDYTestUtils.createHeaders(serverHost, proxyAddress.getPort(), version, "POST", "/");
Stream stream = client.syn(new SynInfo(headers, false), new StreamFrameListener.Adapter()
{
private final ByteArrayOutputStream result = new ByteArrayOutputStream();
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
Fields headers = replyInfo.getHeaders();
assertThat("response comes from the given server", headers.get(serverIdentificationString),
is(notNullValue()));
replyLatch.countDown();
}
@Override
public void onData(Stream stream, DataInfo dataInfo)
{
result.write(dataInfo.asBytes(true), 0, dataInfo.length());
if (dataInfo.isClose())
{
assertThat("received data matches send data", data, is(result.toString()));
dataLatch.countDown();
}
}
});
stream.data(new StringDataInfo(data, true), new Callback.Adapter());
assertThat("reply has been received", replyLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat("data has been received", dataLatch.await(5, TimeUnit.SECONDS), is(true));
}
private class TestServerHandler extends DefaultHandler
{
private final String responseHeader;
private final byte[] responseData;
private TestServerHandler(String responseHeader, byte[] responseData)
{
this.responseHeader = responseHeader;
this.responseData = responseData;
}
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request,
HttpServletResponse response) throws IOException, ServletException
{
assertThat("Via Header is set", baseRequest.getHeader("X-Forwarded-For"), is(notNullValue()));
assertThat("X-Forwarded-For Header is set", baseRequest.getHeader("X-Forwarded-For"),
is(notNullValue()));
assertThat("X-Forwarded-Host Header is set", baseRequest.getHeader("X-Forwarded-Host"),
is(notNullValue()));
assertThat("X-Forwarded-Proto Header is set", baseRequest.getHeader("X-Forwarded-Proto"),
is(notNullValue()));
assertThat("X-Forwarded-Server Header is set", baseRequest.getHeader("X-Forwarded-Server"),
is(notNullValue()));
baseRequest.setHandled(true);
IO.copy(request.getInputStream(), response.getOutputStream());
if (responseHeader != null)
response.addHeader(responseHeader, "bar");
if (responseData != null)
response.getOutputStream().write(responseData);
}
}
}

143
jetty-spdy/spdy-http-server/src/test/java/org/eclipse/jetty/spdy/server/proxy/ProxySPDYToHTTPTest.java
View File

@ -22,12 +22,15 @@ import java.io.BufferedReader;
import java.io.ByteArrayOutputStream; import java.io.ByteArrayOutputStream;
import java.io.IOException; import java.io.IOException;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import javax.servlet.ServletException; import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse; import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.server.Handler; import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Request; import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.Server;
@ -38,6 +41,7 @@ import org.eclipse.jetty.spdy.api.GoAwayReceivedInfo;
import org.eclipse.jetty.spdy.api.PingInfo; import org.eclipse.jetty.spdy.api.PingInfo;
import org.eclipse.jetty.spdy.api.PingResultInfo; import org.eclipse.jetty.spdy.api.PingResultInfo;
import org.eclipse.jetty.spdy.api.ReplyInfo; import org.eclipse.jetty.spdy.api.ReplyInfo;
import org.eclipse.jetty.spdy.api.RstInfo;
import org.eclipse.jetty.spdy.api.SPDY; import org.eclipse.jetty.spdy.api.SPDY;
import org.eclipse.jetty.spdy.api.Session; import org.eclipse.jetty.spdy.api.Session;
import org.eclipse.jetty.spdy.api.SessionFrameListener; import org.eclipse.jetty.spdy.api.SessionFrameListener;
@ -45,6 +49,7 @@ import org.eclipse.jetty.spdy.api.Stream;
import org.eclipse.jetty.spdy.api.StreamFrameListener; import org.eclipse.jetty.spdy.api.StreamFrameListener;
import org.eclipse.jetty.spdy.api.StringDataInfo; import org.eclipse.jetty.spdy.api.StringDataInfo;
import org.eclipse.jetty.spdy.api.SynInfo; import org.eclipse.jetty.spdy.api.SynInfo;
import org.eclipse.jetty.spdy.api.server.ServerSessionFrameListener;
import org.eclipse.jetty.spdy.client.SPDYClient; import org.eclipse.jetty.spdy.client.SPDYClient;
import org.eclipse.jetty.spdy.server.http.HTTPSPDYHeader; import org.eclipse.jetty.spdy.server.http.HTTPSPDYHeader;
import org.eclipse.jetty.spdy.server.http.SPDYTestUtils; import org.eclipse.jetty.spdy.server.http.SPDYTestUtils;
@ -59,15 +64,14 @@ import org.junit.Test;
import org.junit.rules.TestWatcher; import org.junit.rules.TestWatcher;
import org.junit.runner.Description; import org.junit.runner.Description;
import org.junit.runner.RunWith; import org.junit.runner.RunWith;
import org.junit.runners.JUnit4; import org.junit.runners.Parameterized;
import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue; import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue; import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertThat; import static org.junit.Assert.assertThat;
//@RunWith(value = Parameterized.class) @RunWith(value = Parameterized.class)
@RunWith(JUnit4.class)
public class ProxySPDYToHTTPTest public class ProxySPDYToHTTPTest
{ {
@Rule @Rule
@ -83,13 +87,14 @@ public class ProxySPDYToHTTPTest
description.getMethodName()); description.getMethodName());
} }
}; };
private final short version = SPDY.V3;
// @Parameterized.Parameters private final short version;
// public static Collection<Short[]> parameters()
// { @Parameterized.Parameters
// return Arrays.asList(new Short[]{SPDY.V2}, new Short[]{SPDY.V3}); public static Collection<Short[]> parameters()
// } {
return Arrays.asList(new Short[]{SPDY.V2}, new Short[]{SPDY.V3});
}
private SPDYClient.Factory factory; private SPDYClient.Factory factory;
private Server server; private Server server;
@ -97,10 +102,10 @@ public class ProxySPDYToHTTPTest
private ServerConnector proxyConnector; private ServerConnector proxyConnector;
private SslContextFactory sslContextFactory = SPDYTestUtils.newSslContextFactory(); private SslContextFactory sslContextFactory = SPDYTestUtils.newSslContextFactory();
// public ProxySPDYToHTTPTest(short version) public ProxySPDYToHTTPTest(short version)
// { {
// this.version = version; this.version = version;
// } }
protected InetSocketAddress startServer(Handler handler) throws Exception protected InetSocketAddress startServer(Handler handler) throws Exception
{ {
@ -116,8 +121,10 @@ public class ProxySPDYToHTTPTest
{ {
proxy = new Server(); proxy = new Server();
ProxyEngineSelector proxyEngineSelector = new ProxyEngineSelector(); ProxyEngineSelector proxyEngineSelector = new ProxyEngineSelector();
HTTPProxyEngine httpProxyEngine = new HTTPProxyEngine(); HttpClient httpClient = new HttpClient();
httpProxyEngine.setIdleTimeout(proxyEngineTimeout); httpClient.start();
httpClient.setIdleTimeout(proxyEngineTimeout);
HTTPProxyEngine httpProxyEngine = new HTTPProxyEngine(httpClient);
proxyEngineSelector.putProxyEngine("http/1.1", httpProxyEngine); proxyEngineSelector.putProxyEngine("http/1.1", httpProxyEngine);
proxyEngineSelector.putProxyServerInfo("localhost", new ProxyEngineSelector.ProxyServerInfo("http/1.1", address.getHostName(), address.getPort())); proxyEngineSelector.putProxyServerInfo("localhost", new ProxyEngineSelector.ProxyServerInfo("http/1.1", address.getHostName(), address.getPort()));
proxyConnector = new HTTPSPDYProxyServerConnector(proxy, sslContextFactory, proxyEngineSelector); proxyConnector = new HTTPSPDYProxyServerConnector(proxy, sslContextFactory, proxyEngineSelector);
@ -162,7 +169,7 @@ public class ProxySPDYToHTTPTest
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
Fields headers = SPDYTestUtils.createHeaders(proxyAddress.getPort(), version, "GET", "/"); Fields headers = SPDYTestUtils.createHeaders("localhost", proxyAddress.getPort(), version, "GET", "/");
headers.put(header, "bar"); headers.put(header, "bar");
headers.put("connection", "close"); headers.put("connection", "close");
@ -195,7 +202,7 @@ public class ProxySPDYToHTTPTest
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
Fields headers = SPDYTestUtils.createHeaders(proxyAddress.getPort(), version, "GET", "/"); Fields headers = SPDYTestUtils.createHeaders("localhost", proxyAddress.getPort(), version, "GET", "/");
headers.put(header, "bar"); headers.put(header, "bar");
headers.put("connection", "close"); headers.put("connection", "close");
@ -227,6 +234,92 @@ public class ProxySPDYToHTTPTest
assertThat("data has been received", dataLatch.await(5, TimeUnit.SECONDS), is(true)); assertThat("data has been received", dataLatch.await(5, TimeUnit.SECONDS), is(true));
} }
@Test
public void testHttpServerCommitsResponseTwice() throws Exception
{
final long timeout = 1000;
InetSocketAddress proxyAddress = startProxy(startServer(new DefaultHandler()
{
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
response.addHeader("some response", "header");
response.flushBuffer();
try
{
Thread.sleep(timeout * 2);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
}), 30000, timeout);
final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch resetLatch = new CountDownLatch(1);
Session client = factory.newSPDYClient(version).connect(proxyAddress, new ServerSessionFrameListener.Adapter()
{
@Override
public void onRst(Session session, RstInfo rstInfo)
{
resetLatch.countDown();
}
}).get(5, TimeUnit.SECONDS);
Fields headers = SPDYTestUtils.createHeaders("localhost", proxyAddress.getPort(), version, "GET", "/");
headers.put("connection", "close");
client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
replyLatch.countDown();
}
});
assertThat("reply has been received", replyLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat("stream is reset", resetLatch.await(5, TimeUnit.SECONDS), is(true));
}
@Test
public void testHttpServerSendsRedirect() throws Exception
{
InetSocketAddress proxyAddress = startProxy(startServer(new DefaultHandler()
{
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
response.setStatus(HttpServletResponse.SC_FOUND);
response.setHeader("Location", "http://doesnot.exist");
}
}), 30000, 30000);
final CountDownLatch replyLatch = new CountDownLatch(1);
Session client = factory.newSPDYClient(version).connect(proxyAddress, null).get(5, TimeUnit.SECONDS);
Fields headers = SPDYTestUtils.createHeaders("localhost", proxyAddress.getPort(), version, "GET", "/");
client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
assertThat("Status code is 302", replyInfo.getHeaders().get(HTTPSPDYHeader.STATUS.name(version)).value(),
is("302"));
assertThat("Location header has been received", replyInfo.getHeaders().get("Location"), is(notNullValue()));
replyLatch.countDown();
}
});
assertThat("reply has been received", replyLatch.await(5, TimeUnit.SECONDS), is(true));
}
@Test @Test
public void testSYNWithRequestContentThenREPLYAndDATA() throws Exception public void testSYNWithRequestContentThenREPLYAndDATA() throws Exception
{ {
@ -240,7 +333,7 @@ public class ProxySPDYToHTTPTest
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
Fields headers = SPDYTestUtils.createHeaders(proxyAddress.getPort(), version, "POST", "/"); Fields headers = SPDYTestUtils.createHeaders("localhost", proxyAddress.getPort(), version, "POST", "/");
headers.put(header, "bar"); headers.put(header, "bar");
headers.put("connection", "close"); headers.put("connection", "close");
@ -288,7 +381,7 @@ public class ProxySPDYToHTTPTest
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1);
Fields headers = SPDYTestUtils.createHeaders(proxyAddress.getPort(), version, "POST", "/"); Fields headers = SPDYTestUtils.createHeaders("localhost", proxyAddress.getPort(), version, "POST", "/");
headers.put(header, "bar"); headers.put(header, "bar");
headers.put("connection", "close"); headers.put("connection", "close");
@ -310,16 +403,13 @@ public class ProxySPDYToHTTPTest
result.write(dataInfo.asBytes(true), 0, dataInfo.length()); result.write(dataInfo.asBytes(true), 0, dataInfo.length());
if (dataInfo.isClose()) if (dataInfo.isClose())
{ {
System.out.println("client received DATA: " + result);
assertThat("received data matches send data", result.toString(), is(data + data2)); assertThat("received data matches send data", result.toString(), is(data + data2));
dataLatch.countDown(); dataLatch.countDown();
} }
} }
}); });
System.out.println("DATA1 sent!!!!!!!!");
stream.data(new StringDataInfo(data, false), new Callback.Adapter()); stream.data(new StringDataInfo(data, false), new Callback.Adapter());
System.out.println("DATA2 sent!!!!!!!!");
stream.data(new StringDataInfo(data2, true), new Callback.Adapter()); stream.data(new StringDataInfo(data2, true), new Callback.Adapter());
assertThat("reply has been received", replyLatch.await(5, TimeUnit.SECONDS), is(true)); assertThat("reply has been received", replyLatch.await(5, TimeUnit.SECONDS), is(true));
@ -344,7 +434,7 @@ public class ProxySPDYToHTTPTest
} }
}).get(5, TimeUnit.SECONDS); }).get(5, TimeUnit.SECONDS);
Fields headers = SPDYTestUtils.createHeaders(proxyAddress.getPort(), version, "POST", "/"); Fields headers = SPDYTestUtils.createHeaders("localhost", proxyAddress.getPort(), version, "POST", "/");
client.syn(new SynInfo(headers, false), null); client.syn(new SynInfo(headers, false), null);
assertThat("goAway has been received by proxy", goAwayLatch.await(2 * timeout, TimeUnit.MILLISECONDS), assertThat("goAway has been received by proxy", goAwayLatch.await(2 * timeout, TimeUnit.MILLISECONDS),
is(true)); is(true));
@ -376,7 +466,7 @@ public class ProxySPDYToHTTPTest
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
Fields headers = SPDYTestUtils.createHeaders(proxyAddress.getPort(), version, "POST", "/"); Fields headers = SPDYTestUtils.createHeaders("localhost", proxyAddress.getPort(), version, "POST", "/");
headers.put(header, "bar"); headers.put(header, "bar");
headers.put("connection", "close"); headers.put("connection", "close");
@ -393,8 +483,6 @@ public class ProxySPDYToHTTPTest
}); });
assertThat("reply has been received", replyLatch.await(5, TimeUnit.SECONDS), is(true)); assertThat("reply has been received", replyLatch.await(5, TimeUnit.SECONDS), is(true));
// client.goAway(new GoAwayInfo(5, TimeUnit.SECONDS));
} }
@Test @Test
@ -419,8 +507,6 @@ public class ProxySPDYToHTTPTest
client.ping(new PingInfo(5, TimeUnit.SECONDS)); client.ping(new PingInfo(5, TimeUnit.SECONDS));
Assert.assertTrue(pingLatch.await(5, TimeUnit.SECONDS)); Assert.assertTrue(pingLatch.await(5, TimeUnit.SECONDS));
// client.goAway(new GoAwayInfo(5, TimeUnit.SECONDS));
} }
private class TestServerHandler extends DefaultHandler private class TestServerHandler extends DefaultHandler
@ -438,7 +524,6 @@ public class ProxySPDYToHTTPTest
public void handle(String target, Request baseRequest, HttpServletRequest request, public void handle(String target, Request baseRequest, HttpServletRequest request,
HttpServletResponse response) throws IOException, ServletException HttpServletResponse response) throws IOException, ServletException
{ {
System.out.println("HANDLER CALLED!!!");
assertThat("Via Header is set", baseRequest.getHeader("X-Forwarded-For"), is(notNullValue())); assertThat("Via Header is set", baseRequest.getHeader("X-Forwarded-For"), is(notNullValue()));
assertThat("X-Forwarded-For Header is set", baseRequest.getHeader("X-Forwarded-For"), assertThat("X-Forwarded-For Header is set", baseRequest.getHeader("X-Forwarded-For"),
is(notNullValue())); is(notNullValue()));

2
jetty-spdy/spdy-http-server/src/test/java/org/eclipse/jetty/spdy/server/proxy/ProxySPDYToSPDYTest.java
View File

@ -169,7 +169,7 @@ public class ProxySPDYToSPDYTest
Session client = factory.newSPDYClient(version).connect(proxyAddress, null).get(5, TimeUnit.SECONDS); Session client = factory.newSPDYClient(version).connect(proxyAddress, null).get(5, TimeUnit.SECONDS);
final CountDownLatch replyLatch = new CountDownLatch(1); final CountDownLatch replyLatch = new CountDownLatch(1);
Fields headers = SPDYTestUtils.createHeaders(proxyAddress.getPort(), version, "GET", "/"); Fields headers = SPDYTestUtils.createHeaders("localhost", proxyAddress.getPort(), version, "GET", "/");
headers.put(header, "bar"); headers.put(header, "bar");
client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter() client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{ {

4
jetty-spdy/spdy-http-server/src/test/resources/jetty-logging.properties
View File

@ -2,6 +2,6 @@ org.eclipse.jetty.util.log.class=org.eclipse.jetty.util.log.StdErrLog
#org.eclipse.jetty.spdy.LEVEL=DEBUG #org.eclipse.jetty.spdy.LEVEL=DEBUG
#org.eclipse.jetty.server.LEVEL=DEBUG #org.eclipse.jetty.server.LEVEL=DEBUG
#org.eclipse.jetty.io.ssl.LEVEL=DEBUG #org.eclipse.jetty.io.ssl.LEVEL=DEBUG
org.eclipse.jetty.spdy.LEVEL=DEBUG #org.eclipse.jetty.spdy.LEVEL=DEBUG
org.eclipse.jetty.client.LEVEL=DEBUG #org.eclipse.jetty.client.LEVEL=DEBUG
#org.mortbay.LEVEL=DEBUG #org.mortbay.LEVEL=DEBUG