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

Jetty9 - Ported the HTTP SPDY Proxy code.

This commit is contained in:
Simone Bordet 2012-08-29 16:00:43 +02:00
parent 1265c27607
commit 8920cd16b6
15 changed files with 1731 additions and 1660 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
jetty-server/src/main/java/org/eclipse/jetty/server/HttpConnection.java
View File

@ -78,11 +78,16 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
_generator = new HttpGenerator(); // TODO: consider moving the generator to the transport, where it belongs
_generator.setSendServerVersion(getServer().getSendServerVersion());
_channel = new HttpChannelOverHttp(connector, config, endPoint, this, new Input());
_parser = new HttpParser(_channel,config.getRequestHeaderSize());
_parser = new HttpParser(newRequestHandler(),config.getRequestHeaderSize());
LOG.debug("New HTTP Connection {}", this);
}
protected HttpParser.RequestHandler<ByteBuffer> newRequestHandler()
{
return _channel;
}
public Server getServer()
{
return _connector.getServer();

20
jetty-spdy/spdy-jetty-http-webapp/pom.xml
View File

@ -29,16 +29,14 @@
</execution>
</executions>
</plugin>
<!--
<plugin>
<groupId>org.mortbay.jetty</groupId>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-maven-plugin</artifactId>
<version>${project.version}</version>
<configuration>
<stopPort>8888</stopPort>
<stopKey>quit</stopKey>
<jvmArgs>
-Dlog4j.configuration=file://${basedir}/src/main/resources/log4j.properties
-Xbootclasspath/p:${settings.localRepository}/org/mortbay/jetty/npn/npn-boot/${npn.version}/npn-boot-${npn.version}.jar
</jvmArgs>
<jettyXml>${basedir}/src/main/config/etc/jetty-spdy.xml</jettyXml>
@ -50,32 +48,24 @@
<artifactId>spdy-jetty-http</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>${slf4j-version}</version>
</dependency>
</dependencies>
</plugin>
-->
</plugins>
</build>
<!--
<profiles>
<profile>
<id>proxy</id>
<build>
<plugins>
<plugin>
<groupId>org.mortbay.jetty</groupId>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-maven-plugin</artifactId>
<version>${project.version}</version>
<configuration>
<stopPort>8888</stopPort>
<stopKey>quit</stopKey>
<jvmArgs>
-Dlog4j.configuration=file://${basedir}/src/main/resources/log4j.properties
-Xbootclasspath/p:${settings.localRepository}/org/mortbay/jetty/npn/npn-boot/${npn.version}/npn-boot-${npn.version}.jar
-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=5005
</jvmArgs>
@ -88,17 +78,11 @@
<artifactId>spdy-jetty-http</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>${slf4j-version}</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
</profile>
</profiles>
-->
</project>

26
jetty-spdy/spdy-jetty-http-webapp/src/main/config/etc/jetty-spdy-proxy.xml
View File

@ -21,9 +21,12 @@
<Call name="addConnector">
<Arg>
<New class="org.eclipse.jetty.spdy.http.HTTPSPDYServerConnector">
<Arg>
<Ref id="Server"/>
</Arg>
<Set name="Port">9090</Set>
<Set name="defaultAsyncConnectionFactory">
<Call name="getAsyncConnectionFactory">
<Set name="defaultConnectionFactory">
<Call name="getConnectionFactory">
<Arg>spdy/2</Arg>
</Call>
</Set>
@ -79,7 +82,12 @@
<Call name="addConnector">
<Arg>
<New class="org.eclipse.jetty.spdy.proxy.HTTPSPDYProxyConnector">
<Arg><Ref id="proxyEngineSelector" /></Arg>
<Arg>
<Ref id="Server"/>
</Arg>
<Arg>
<Ref id="proxyEngineSelector"/>
</Arg>
<Set name="Port">8080</Set>
</New>
</Arg>
@ -87,12 +95,18 @@
<Call name="addConnector">
<Arg>
<New class="org.eclipse.jetty.spdy.proxy.HTTPSPDYProxyConnector">
<Arg><Ref id="proxyEngineSelector" /></Arg>
<Arg><Ref id="sslContextFactory" /></Arg>
<Arg>
<Ref id="Server"/>
</Arg>
<Arg>
<Ref id="sslContextFactory"/>
</Arg>
<Arg>
<Ref id="proxyEngineSelector"/>
</Arg>
<Set name="Port">8443</Set>
</New>
</Arg>
</Call>
</Configure>

24
jetty-spdy/spdy-jetty-http-webapp/src/main/config/etc/jetty-spdy.xml
View File

@ -11,9 +11,7 @@
<Set name="protocol">TLSv1</Set>
</New>
<!-- Uncomment to create a ReferrerPushStrategy that can be added to the Connectors -->
<!--
<!-- Uncomment to create a ReferrerPushStrategy that can be added to the Connectors
<New id="pushStrategy" class="org.eclipse.jetty.spdy.http.ReferrerPushStrategy">
<Arg type="List">
<Array type="String">
@ -27,16 +25,19 @@
</New>
-->
<!--<Set class="org.eclipse.jetty.npn.NextProtoNego" name="debug" type="boolean">true</Set>-->
<!--
<Set class="org.eclipse.jetty.npn.NextProtoNego" name="debug" type="boolean">true</Set>
-->
<Call name="addConnector">
<Arg>
<New class="org.eclipse.jetty.spdy.http.HTTPSPDYServerConnector">
<!-- uncomment to enable to apply ReferrerPushStrategy for spdy/3
<Arg>
<Ref id="Server"/>
</Arg>
<!-- Uncomment to enable to apply ReferrerPushStrategy for spdy/3
if you want to support it in both spdy/2 and spdy/3, just replace the
value in the first map entry.
-->
<!--
<Arg name="pushStrategies">
<Map>
<Entry>
@ -58,13 +59,14 @@
<Arg>
<New class="org.eclipse.jetty.spdy.http.HTTPSPDYServerConnector">
<Arg>
<Ref id="sslContextFactory" />
<Ref id="Server"/>
</Arg>
<!-- uncomment to enable to apply ReferrerPushStrategy for spdy/3
<Arg>
<Ref id="sslContextFactory"/>
</Arg>
<!-- Uncomment to enable to apply ReferrerPushStrategy for spdy/3
if you want to support it in both spdy/2 and spdy/3, just replace the
value in the first map entry.
-->
<!--
<Arg name="pushStrategies">
<Map>
<Entry>

62
jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/AbstractHTTPSPDYServerConnector.java
View File

@ -1,62 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2012 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.http;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.spdy.SPDYServerConnector;
import org.eclipse.jetty.spdy.api.server.ServerSessionFrameListener;
public class AbstractHTTPSPDYServerConnector extends SPDYServerConnector
{
public AbstractHTTPSPDYServerConnector(Server server, ServerSessionFrameListener listener)
{
super(server, listener);
}
//TODO:
// @Override
// public void customize(EndPoint endPoint, Request request) throws IOException
// {
// super.customize(endPoint, request);
// if (getSslContextFactory() != null)
// request.setScheme(HttpSchemes.HTTPS);
// }
//
// @Override
// public boolean isConfidential(Request request)
// {
// if (getSslContextFactory() != null)
// {
// int confidentialPort = getConfidentialPort();
// return confidentialPort == 0 || confidentialPort == request.getServerPort();
// }
// return super.isConfidential(request);
// }
//
// @Override
// public boolean isIntegral(Request request)
// {
// if (getSslContextFactory() != null)
// {
// int integralPort = getIntegralPort();
// return integralPort == 0 || integralPort == request.getServerPort();
// }
// return super.isIntegral(request);
// }
}

13
jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/HTTPSPDYServerConnector.java
View File

@ -26,6 +26,7 @@ import org.eclipse.jetty.server.HttpServerConnectionFactory;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.spdy.SPDYServerConnector;
import org.eclipse.jetty.spdy.api.SPDY;
import org.eclipse.jetty.util.ssl.SslContextFactory;
public class HTTPSPDYServerConnector extends SPDYServerConnector
{
@ -34,11 +35,21 @@ public class HTTPSPDYServerConnector extends SPDYServerConnector
this(server, Collections.<Short, PushStrategy>emptyMap());
}
public HTTPSPDYServerConnector(Server server, SslContextFactory sslContextFactory)
{
this(server, sslContextFactory, Collections.<Short, PushStrategy>emptyMap());
}
public HTTPSPDYServerConnector(Server server, Map<Short, PushStrategy> pushStrategies)
{
this(server, null, pushStrategies);
}
public HTTPSPDYServerConnector(Server server, SslContextFactory sslContextFactory, Map<Short, PushStrategy> pushStrategies)
{
// We pass a null ServerSessionFrameListener because for
// HTTP over SPDY we need one that references the endPoint
super(server, null);
super(server, sslContextFactory, null);
clearConnectionFactories();
// The "spdy/3" protocol handles HTTP over SPDY
putConnectionFactory("spdy/3", new ServerHTTPSPDYAsyncConnectionFactory(SPDY.V3, getByteBufferPool(), getExecutor(), getScheduler(), this, getPushStrategy(SPDY.V3, pushStrategies)));

38
jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/proxy/HTTPSPDYProxyConnector.java
View File

@ -19,21 +19,27 @@
package org.eclipse.jetty.spdy.proxy;
public class HTTPSPDYProxyConnector //extends AbstractHTTPSPDYServerConnector
{
// public HTTPSPDYProxyConnector(ProxyEngineSelector proxyEngineSelector)
// {
// this(proxyEngineSelector, null);
// }
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.spdy.SPDYServerConnector;
import org.eclipse.jetty.spdy.ServerSPDYConnectionFactory;
import org.eclipse.jetty.spdy.api.SPDY;
import org.eclipse.jetty.util.ssl.SslContextFactory;
// public HTTPSPDYProxyConnector(ProxyEngineSelector proxyEngineSelector, SslContextFactory sslContextFactory)
// {
// super(proxyEngineSelector, sslContextFactory);
// clearAsyncConnectionFactories();
//
// putAsyncConnectionFactory("spdy/3", new ServerSPDYAsyncConnectionFactory(SPDY.V3, getByteBufferPool(), getExecutor(), getScheduler(), proxyEngineSelector));
// putAsyncConnectionFactory("spdy/2", new ServerSPDYAsyncConnectionFactory(SPDY.V2, getByteBufferPool(), getExecutor(), getScheduler(), proxyEngineSelector));
// putAsyncConnectionFactory("http/1.1", new ProxyHTTPAsyncConnectionFactory(this, SPDY.V2, proxyEngineSelector));
// setDefaultAsyncConnectionFactory(getAsyncConnectionFactory("http/1.1"));
// }
public class HTTPSPDYProxyConnector extends SPDYServerConnector
{
public HTTPSPDYProxyConnector(Server server, ProxyEngineSelector proxyEngineSelector)
{
this(server, null, proxyEngineSelector);
}
public HTTPSPDYProxyConnector(Server server, SslContextFactory sslContextFactory, ProxyEngineSelector proxyEngineSelector)
{
super(server, sslContextFactory, proxyEngineSelector);
clearConnectionFactories();
putConnectionFactory("spdy/3", new ServerSPDYConnectionFactory(SPDY.V3, getByteBufferPool(), getExecutor(), getScheduler(), proxyEngineSelector));
putConnectionFactory("spdy/2", new ServerSPDYConnectionFactory(SPDY.V2, getByteBufferPool(), getExecutor(), getScheduler(), proxyEngineSelector));
putConnectionFactory("http/1.1", new ProxyHTTPConnectionFactory(this, SPDY.V2, proxyEngineSelector));
setDefaultConnectionFactory(getConnectionFactory("http/1.1"));
}
}

5
jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/proxy/ProxyEngine.java
View File

@ -20,6 +20,7 @@
package org.eclipse.jetty.spdy.proxy;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import org.eclipse.jetty.spdy.api.Headers;
@ -73,9 +74,9 @@ public abstract class ProxyEngine
protected void addRequestProxyHeaders(Stream stream, Headers headers)
{
addViaHeader(headers);
String address = (String)stream.getSession().getAttribute("org.eclipse.jetty.spdy.remoteAddress");
InetSocketAddress address = (InetSocketAddress)stream.getSession().getAttribute("org.eclipse.jetty.spdy.remoteAddress");
if (address != null)
headers.add("X-Forwarded-For", address);
headers.add("X-Forwarded-For", address.getHostName());
}
protected void addResponseProxyHeaders(Stream stream, Headers headers)

39
jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/proxy/ProxyHTTPAsyncConnectionFactory.java
View File

@ -1,39 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2012 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.proxy;
public class ProxyHTTPAsyncConnectionFactory //extends ServerHTTPAsyncConnectionFactory
{
// private final short version;
// private final ProxyEngineSelector proxyEngineSelector;
//
// public ProxyHTTPAsyncConnectionFactory(SPDYServerConnector connector, short version, ProxyEngineSelector proxyEngineSelector)
// {
// super(connector);
// this.version = version;
// this.proxyEngineSelector = proxyEngineSelector;
// }
//
// @Override
// public Connection newAsyncConnection(SocketChannel channel, EndPoint endPoint, Object attachment)
// {
// return new ProxyHTTPSPDYAsyncConnection(getConnector(), endPoint, version, proxyEngineSelector);
// }
}

47
jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/proxy/ProxyHTTPConnectionFactory.java Normal file
View File

@ -0,0 +1,47 @@
//
// ========================================================================
// Copyright (c) 1995-2012 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.proxy;
import java.nio.channels.SocketChannel;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.server.ConnectionFactory;
import org.eclipse.jetty.server.Connector;
public class ProxyHTTPConnectionFactory implements ConnectionFactory
{
private final Connector connector;
private final short version;
private final ProxyEngineSelector proxyEngineSelector;
public ProxyHTTPConnectionFactory(Connector connector, short version, ProxyEngineSelector proxyEngineSelector)
{
this.connector = connector;
this.version = version;
this.proxyEngineSelector = proxyEngineSelector;
}
@Override
public Connection newConnection(SocketChannel channel, EndPoint endPoint, Object attachment)
{
return new ProxyHTTPSPDYConnection(connector, endPoint, version, proxyEngineSelector);
}
}

306
jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/proxy/ProxyHTTPSPDYAsyncConnection.java
View File

@ -1,306 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2012 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.proxy;
public class ProxyHTTPSPDYAsyncConnection// extends HttpConnection
{
// private final short version;
// private final Headers headers = new Headers();
// private final ProxyEngineSelector proxyEngineSelector;
// private final HttpGenerator generator;
// private final ISession session;
// private HTTPStream stream;
// private Buffer content;
//
// public ProxyHTTPSPDYAsyncConnection(SPDYServerConnector connector, EndPoint endPoint, short version, ProxyEngineSelector proxyEngineSelector)
// {
// super(new HttpConfiguration(connector.getSslContextFactory(), connector.getSslContextFactory() != null), connector, endPoint);
// this.version = version;
// this.proxyEngineSelector = proxyEngineSelector;
// this.generator = (HttpGenerator)getGenerator();
// this.session = new HTTPSession(version, connector);
// this.session.setAttribute("org.eclipse.jetty.spdy.remoteAddress", endPoint.getRemoteAddress());
// }
//
// @Override
// public EndPoint getEndPoint()
// {
// return super.getEndPoint();
// }
//
// @Override
// protected void startRequest(Buffer method, Buffer uri, Buffer httpVersion) throws IOException
// {
// SPDYServerConnector connector = (SPDYServerConnector)getConnector();
// String scheme = connector.getSslContextFactory() != null ? "https" : "http";
// headers.put(HTTPSPDYHeader.SCHEME.name(version), scheme);
// headers.put(HTTPSPDYHeader.METHOD.name(version), method.toString("UTF-8"));
// headers.put(HTTPSPDYHeader.URI.name(version), uri.toString("UTF-8"));
// headers.put(HTTPSPDYHeader.VERSION.name(version), httpVersion.toString("UTF-8"));
// }
//
// @Override
// protected void parsedHeader(Buffer name, Buffer value) throws IOException
// {
// String headerName = name.toString("UTF-8").toLowerCase();
// String headerValue = value.toString("UTF-8");
// switch (headerName)
// {
// case "host":
// headers.put(HTTPSPDYHeader.HOST.name(version), headerValue);
// break;
// default:
// headers.put(headerName, headerValue);
// break;
// }
// }
//
// @Override
// protected void headerComplete() throws IOException
// {
// }
//
// @Override
// protected void content(Buffer buffer) throws IOException
// {
// if (content == null)
// {
// stream = syn(false);
// content = buffer;
// }
// else
// {
// stream.getStreamFrameListener().onData(stream, toDataInfo(buffer, false));
// }
// }
//
// @Override
// public void messageComplete(long contentLength) throws IOException
// {
// if (stream == null)
// {
// assert content == null;
// if (headers.isEmpty())
// proxyEngineSelector.onGoAway(session, new GoAwayInfo(0, SessionStatus.OK));
// else
// syn(true);
// }
// else
// {
// stream.getStreamFrameListener().onData(stream, toDataInfo(content, true));
// }
// headers.clear();
// stream = null;
// content = null;
// }
//
// private HTTPStream syn(boolean close)
// {
// HTTPStream stream = new HTTPStream(1, (byte)0, session, null);
// StreamFrameListener streamFrameListener = proxyEngineSelector.onSyn(stream, new SynInfo(headers, close));
// stream.setStreamFrameListener(streamFrameListener);
// return stream;
// }
//
// private DataInfo toDataInfo(Buffer buffer, boolean close)
// {
// if (buffer instanceof ByteArrayBuffer)
// return new BytesDataInfo(buffer.array(), buffer.getIndex(), buffer.length(), close);
//
// if (buffer instanceof NIOBuffer)
// {
// ByteBuffer byteBuffer = ((NIOBuffer)buffer).getByteBuffer();
// byteBuffer.limit(buffer.putIndex());
// byteBuffer.position(buffer.getIndex());
// return new ByteBufferDataInfo(byteBuffer, close);
// }
//
// return new BytesDataInfo(buffer.asArray(), close);
// }
//
// private class HTTPSession extends StandardSession
// {
// private HTTPSession(short version, SPDYServerConnector connector)
// {
// super(version, connector.getByteBufferPool(), connector.getExecutor(), connector.getScheduler(), null, null, 1, proxyEngineSelector, null, null);
// }
//
// @Override
// public void rst(RstInfo rstInfo, long timeout, TimeUnit unit, Handler<Void> handler)
// {
// // Not much we can do in HTTP land: just close the connection
// goAway(timeout, unit, handler);
// }
//
// @Override
// public void goAway(long timeout, TimeUnit unit, Handler<Void> handler)
// {
// try
// {
// getEndPoint().close();
// handler.completed(null);
// }
// catch (IOException x)
// {
// handler.failed(null, x);
// }
// }
// }
//
// /**
// * <p>This stream will convert the SPDY invocations performed by the proxy into HTTP to be sent to the client.</p>
// */
// private class HTTPStream extends StandardStream
// {
// private final Pattern statusRegexp = Pattern.compile("(\\d{3})\\s*(.*)");
//
// private HTTPStream(int id, byte priority, ISession session, IStream associatedStream)
// {
// super(id, priority, session, associatedStream);
// }
//
// @Override
// public void syn(SynInfo synInfo, long timeout, TimeUnit unit, Handler<Stream> handler)
// {
// // HTTP does not support pushed streams
// handler.completed(new HTTPPushStream(2, getPriority(), getSession(), this));
// }
//
// @Override
// public void headers(HeadersInfo headersInfo, long timeout, TimeUnit unit, Handler<Void> handler)
// {
// // TODO
// throw new UnsupportedOperationException("Not Yet Implemented");
// }
//
// @Override
// public void reply(ReplyInfo replyInfo, long timeout, TimeUnit unit, Handler<Void> handler)
// {
// try
// {
// Headers headers = new Headers(replyInfo.getHeaders(), false);
//
// headers.remove(HTTPSPDYHeader.SCHEME.name(version));
//
// String status = headers.remove(HTTPSPDYHeader.STATUS.name(version)).value();
// Matcher matcher = statusRegexp.matcher(status);
// matcher.matches();
// int code = Integer.parseInt(matcher.group(1));
// String reason = matcher.group(2);
// generator.setResponse(code, reason);
//
// String httpVersion = headers.remove(HTTPSPDYHeader.VERSION.name(version)).value();
// generator.setVersion(Integer.parseInt(httpVersion.replaceAll("\\D", "")));
//
// Headers.Header host = headers.remove(HTTPSPDYHeader.HOST.name(version));
// if (host != null)
// headers.put("host", host.value());
//
// HttpFields fields = new HttpFields();
// for (Headers.Header header : headers)
// {
// String name = camelize(header.name());
// fields.put(name, header.value());
// }
// generator.completeHeader(fields, replyInfo.isClose());
//
// if (replyInfo.isClose())
// complete();
//
// handler.completed(null);
// }
// catch (IOException x)
// {
// handler.failed(null, x);
// }
// }
//
// private String camelize(String name)
// {
// char[] chars = name.toCharArray();
// chars[0] = Character.toUpperCase(chars[0]);
//
// for (int i = 0; i < chars.length; ++i)
// {
// char c = chars[i];
// int j = i + 1;
// if (c == '-' && j < chars.length)
// chars[j] = Character.toUpperCase(chars[j]);
// }
// return new String(chars);
// }
//
// @Override
// public void data(DataInfo dataInfo, long timeout, TimeUnit unit, Handler<Void> handler)
// {
// try
// {
// // Data buffer must be copied, as the ByteBuffer is pooled
// ByteBuffer byteBuffer = dataInfo.asByteBuffer(false);
//
// Buffer buffer = byteBuffer.isDirect() ?
// new DirectNIOBuffer(byteBuffer, false) :
// new IndirectNIOBuffer(byteBuffer, false);
//
// generator.addContent(buffer, dataInfo.isClose());
// generator.flush(unit.toMillis(timeout));
//
// if (dataInfo.isClose())
// complete();
//
// handler.completed(null);
// }
// catch (IOException x)
// {
// handler.failed(null, x);
// }
// }
//
// private void complete() throws IOException
// {
// generator.complete();
// // We need to call asyncDispatch() as if the HTTP request
// // has been suspended and now we complete the response
// getEndPoint().asyncDispatch();
// }
// }
//
// private class HTTPPushStream extends StandardStream
// {
// private HTTPPushStream(int id, byte priority, ISession session, IStream associatedStream)
// {
// super(id, priority, session, associatedStream);
// }
//
// @Override
// public void headers(HeadersInfo headersInfo, long timeout, TimeUnit unit, Handler<Void> handler)
// {
// // Ignore pushed headers
// handler.completed(null);
// }
//
// @Override
// public void data(DataInfo dataInfo, long timeout, TimeUnit unit, Handler<Void> handler)
// {
// // Ignore pushed data
// handler.completed(null);
// }
// }
}

330
jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/proxy/ProxyHTTPSPDYConnection.java Normal file
View File

@ -0,0 +1,330 @@
//
// ========================================================================
// Copyright (c) 1995-2012 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.proxy;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.eclipse.jetty.http.HttpFields;
import org.eclipse.jetty.http.HttpGenerator;
import org.eclipse.jetty.http.HttpHeader;
import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http.HttpParser;
import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.HttpConfiguration;
import org.eclipse.jetty.server.HttpConnection;
import org.eclipse.jetty.spdy.ISession;
import org.eclipse.jetty.spdy.IStream;
import org.eclipse.jetty.spdy.StandardSession;
import org.eclipse.jetty.spdy.StandardStream;
import org.eclipse.jetty.spdy.api.ByteBufferDataInfo;
import org.eclipse.jetty.spdy.api.DataInfo;
import org.eclipse.jetty.spdy.api.GoAwayInfo;
import org.eclipse.jetty.spdy.api.Headers;
import org.eclipse.jetty.spdy.api.HeadersInfo;
import org.eclipse.jetty.spdy.api.ReplyInfo;
import org.eclipse.jetty.spdy.api.RstInfo;
import org.eclipse.jetty.spdy.api.SessionStatus;
import org.eclipse.jetty.spdy.api.Stream;
import org.eclipse.jetty.spdy.api.StreamFrameListener;
import org.eclipse.jetty.spdy.api.SynInfo;
import org.eclipse.jetty.spdy.http.HTTPSPDYHeader;
import org.eclipse.jetty.util.Callback;
public class ProxyHTTPSPDYConnection extends HttpConnection implements HttpParser.RequestHandler<ByteBuffer>
{
private final short version;
private final Headers headers = new Headers();
private final ProxyEngineSelector proxyEngineSelector;
private final ISession session;
private HTTPStream stream;
private ByteBuffer content;
public ProxyHTTPSPDYConnection(Connector connector, EndPoint endPoint, short version, ProxyEngineSelector proxyEngineSelector)
{
super(new HttpConfiguration(connector.getSslContextFactory(), connector.getSslContextFactory() != null), connector, endPoint);
this.version = version;
this.proxyEngineSelector = proxyEngineSelector;
this.session = new HTTPSession(version, connector);
this.session.setAttribute("org.eclipse.jetty.spdy.remoteAddress", endPoint.getRemoteAddress()); // TODO: make this available through API
}
@Override
protected HttpParser.RequestHandler<ByteBuffer> newRequestHandler()
{
return this;
}
@Override
public boolean startRequest(HttpMethod method, String methodString, String uri, HttpVersion httpVersion)
{
Connector connector = getConnector();
String scheme = connector.getSslContextFactory() != null ? "https" : "http";
headers.put(HTTPSPDYHeader.SCHEME.name(version), scheme);
headers.put(HTTPSPDYHeader.METHOD.name(version), methodString);
headers.put(HTTPSPDYHeader.URI.name(version), uri);
headers.put(HTTPSPDYHeader.VERSION.name(version), httpVersion.asString());
return false;
}
@Override
public boolean parsedHeader(HttpHeader header, String headerName, String headerValue)
{
switch (headerName.toLowerCase())
{
case "host":
headers.put(HTTPSPDYHeader.HOST.name(version), headerValue);
break;
default:
headers.put(headerName, headerValue);
break;
}
return false;
}
@Override
public boolean parsedHostHeader(String host, int port)
{
return false;
}
@Override
public boolean headerComplete()
{
return false;
}
@Override
public boolean content(ByteBuffer item)
{
if (content == null)
{
stream = syn(false);
content = item;
}
else
{
stream.getStreamFrameListener().onData(stream, toDataInfo(item, false));
}
return false;
}
@Override
public boolean messageComplete(long contentLength)
{
if (stream == null)
{
assert content == null;
if (headers.isEmpty())
proxyEngineSelector.onGoAway(session, new GoAwayInfo(0, SessionStatus.OK));
else
syn(true);
}
else
{
stream.getStreamFrameListener().onData(stream, toDataInfo(content, true));
}
headers.clear();
stream = null;
content = null;
return false;
}
@Override
public boolean earlyEOF()
{
// TODO
return false;
}
@Override
public void badMessage(int status, String reason)
{
// TODO
}
private HTTPStream syn(boolean close)
{
HTTPStream stream = new HTTPStream(1, (byte)0, session, null);
StreamFrameListener streamFrameListener = proxyEngineSelector.onSyn(stream, new SynInfo(headers, close));
stream.setStreamFrameListener(streamFrameListener);
return stream;
}
private DataInfo toDataInfo(ByteBuffer buffer, boolean close)
{
return new ByteBufferDataInfo(buffer, close);
}
private class HTTPSession extends StandardSession
{
private HTTPSession(short version, Connector connector)
{
super(version, connector.getByteBufferPool(), connector.getExecutor(), connector.getScheduler(), null, null, 1, proxyEngineSelector, null, null);
}
@Override
public void rst(RstInfo rstInfo, long timeout, TimeUnit unit, Callback<Void> handler)
{
// Not much we can do in HTTP land: just close the connection
goAway(timeout, unit, handler);
}
@Override
public void goAway(long timeout, TimeUnit unit, Callback<Void> handler)
{
getEndPoint().close();
handler.completed(null);
}
}
/**
* <p>This stream will convert the SPDY invocations performed by the proxy into HTTP to be sent to the client.</p>
*/
private class HTTPStream extends StandardStream
{
private final Pattern statusRegexp = Pattern.compile("(\\d{3})\\s+(.*)");
private HTTPStream(int id, byte priority, ISession session, IStream associatedStream)
{
super(id, priority, session, associatedStream);
}
@Override
public void syn(SynInfo synInfo, long timeout, TimeUnit unit, Callback<Stream> handler)
{
// HTTP does not support pushed streams
handler.completed(new HTTPPushStream(2, getPriority(), getSession(), this));
}
@Override
public void headers(HeadersInfo headersInfo, long timeout, TimeUnit unit, Callback<Void> handler)
{
// TODO
throw new UnsupportedOperationException("Not Yet Implemented");
}
@Override
public void reply(ReplyInfo replyInfo, long timeout, TimeUnit unit, Callback<Void> handler)
{
try
{
Headers headers = new Headers(replyInfo.getHeaders(), false);
headers.remove(HTTPSPDYHeader.SCHEME.name(version));
String status = headers.remove(HTTPSPDYHeader.STATUS.name(version)).value();
Matcher matcher = statusRegexp.matcher(status);
matcher.matches();
int code = Integer.parseInt(matcher.group(1));
String reason = matcher.group(2).trim();
HttpVersion httpVersion = HttpVersion.fromString(headers.remove(HTTPSPDYHeader.VERSION.name(version)).value());
// Convert the Host header from a SPDY special header to a normal header
Headers.Header host = headers.remove(HTTPSPDYHeader.HOST.name(version));
if (host != null)
headers.put("host", host.value());
HttpFields fields = new HttpFields();
for (Headers.Header header : headers)
{
String name = camelize(header.name());
fields.put(name, header.value());
}
// TODO: handle better the HEAD last parameter
HttpGenerator.ResponseInfo info = new HttpGenerator.ResponseInfo(httpVersion, fields, -1, code, reason, false);
send(info, null, replyInfo.isClose());
if (replyInfo.isClose())
completed();
handler.completed(null);
}
catch (IOException x)
{
handler.failed(null, x);
}
}
private String camelize(String name)
{
char[] chars = name.toCharArray();
chars[0] = Character.toUpperCase(chars[0]);
for (int i = 0; i < chars.length; ++i)
{
char c = chars[i];
int j = i + 1;
if (c == '-' && j < chars.length)
chars[j] = Character.toUpperCase(chars[j]);
}
return new String(chars);
}
@Override
public void data(DataInfo dataInfo, long timeout, TimeUnit unit, Callback<Void> handler)
{
try
{
// Data buffer must be copied, as the ByteBuffer is pooled
ByteBuffer byteBuffer = dataInfo.asByteBuffer(false);
send(byteBuffer, dataInfo.isClose());
if (dataInfo.isClose())
completed();
handler.completed(null);
}
catch (IOException x)
{
handler.failed(null, x);
}
}
}
private class HTTPPushStream extends StandardStream
{
private HTTPPushStream(int id, byte priority, ISession session, IStream associatedStream)
{
super(id, priority, session, associatedStream);
}
@Override
public void headers(HeadersInfo headersInfo, long timeout, TimeUnit unit, Callback<Void> handler)
{
// Ignore pushed headers
handler.completed(null);
}
@Override
public void data(DataInfo dataInfo, long timeout, TimeUnit unit, Callback<Void> handler)
{
// Ignore pushed data
handler.completed(null);
}
}
}

959
jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/proxy/SPDYProxyEngine.java
View File

@ -19,476 +19,501 @@
package org.eclipse.jetty.spdy.proxy;
import java.net.InetSocketAddress;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.spdy.SPDYClient;
import org.eclipse.jetty.spdy.api.ByteBufferDataInfo;
import org.eclipse.jetty.spdy.api.DataInfo;
import org.eclipse.jetty.spdy.api.GoAwayInfo;
import org.eclipse.jetty.spdy.api.Headers;
import org.eclipse.jetty.spdy.api.HeadersInfo;
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.Session;
import org.eclipse.jetty.spdy.api.SessionFrameListener;
import org.eclipse.jetty.spdy.api.Stream;
import org.eclipse.jetty.spdy.api.StreamFrameListener;
import org.eclipse.jetty.spdy.api.StreamStatus;
import org.eclipse.jetty.spdy.api.SynInfo;
import org.eclipse.jetty.spdy.http.HTTPSPDYHeader;
import org.eclipse.jetty.util.Callback;
/**
* <p>{@link SPDYProxyEngine} implements a SPDY to SPDY proxy, that is, converts SPDY events received by
* clients into SPDY events for the servers.</p>
*/
public class SPDYProxyEngine //extends ProxyEngine implements StreamFrameListener
public class SPDYProxyEngine extends ProxyEngine implements StreamFrameListener
{
// private static final String STREAM_HANDLER_ATTRIBUTE = "org.eclipse.jetty.spdy.http.proxy.streamHandler";
// private static final String CLIENT_STREAM_ATTRIBUTE = "org.eclipse.jetty.spdy.http.proxy.clientStream";
//
// private final ConcurrentMap<String, Session> serverSessions = new ConcurrentHashMap<>();
// private final SessionFrameListener sessionListener = new ProxySessionFrameListener();
// private final SPDYClient.Factory factory;
// private volatile long connectTimeout = 15000;
// private volatile long timeout = 60000;
//
// public SPDYProxyEngine(SPDYClient.Factory factory)
// {
// this.factory = factory;
// }
//
// public long getConnectTimeout()
// {
// return connectTimeout;
// }
//
// public void setConnectTimeout(long connectTimeout)
// {
// this.connectTimeout = connectTimeout;
// }
//
// public long getTimeout()
// {
// return timeout;
// }
//
// public void setTimeout(long timeout)
// {
// this.timeout = timeout;
// }
//
// public StreamFrameListener proxy(final Stream clientStream, SynInfo clientSynInfo, ProxyEngineSelector.ProxyServerInfo proxyServerInfo)
// {
// Headers headers = new Headers(clientSynInfo.getHeaders(), false);
//
// short serverVersion = getVersion(proxyServerInfo.getProtocol());
// InetSocketAddress address = proxyServerInfo.getAddress();
// Session serverSession = produceSession(proxyServerInfo.getHost(), serverVersion, address);
// if (serverSession == null)
// {
// rst(clientStream);
// return null;
// }
//
// final Session clientSession = clientStream.getSession();
//
// addRequestProxyHeaders(clientStream, headers);
// customizeRequestHeaders(clientStream, headers);
// convert(clientSession.getVersion(), serverVersion, headers);
//
// SynInfo serverSynInfo = new SynInfo(headers, clientSynInfo.isClose());
// StreamFrameListener listener = new ProxyStreamFrameListener(clientStream);
// StreamHandler handler = new StreamHandler(clientStream, serverSynInfo);
// clientStream.setAttribute(STREAM_HANDLER_ATTRIBUTE, handler);
// serverSession.syn(serverSynInfo, listener, timeout, TimeUnit.MILLISECONDS, handler);
// return this;
// }
//
// private static short getVersion(String protocol)
// {
// switch (protocol)
// {
// case "spdy/2":
// return SPDY.V2;
// case "spdy/3":
// return SPDY.V3;
// default:
// throw new IllegalArgumentException("Procotol: " + protocol + " is not a known SPDY protocol");
// }
// }
//
// @Override
// public void onReply(Stream stream, ReplyInfo replyInfo)
// {
// // Servers do not receive replies
// }
//
// @Override
// public void onHeaders(Stream stream, HeadersInfo headersInfo)
// {
// // TODO
// throw new UnsupportedOperationException("Not Yet Implemented");
// }
//
// @Override
// public void onData(Stream clientStream, final DataInfo clientDataInfo)
// {
// logger.debug("C -> P {} on {}", clientDataInfo, clientStream);
//
// ByteBufferDataInfo serverDataInfo = new ByteBufferDataInfo(clientDataInfo.asByteBuffer(false), clientDataInfo.isClose())
// {
// @Override
// public void consume(int delta)
// {
// super.consume(delta);
// clientDataInfo.consume(delta);
// }
// };
//
// StreamHandler streamHandler = (StreamHandler)clientStream.getAttribute(STREAM_HANDLER_ATTRIBUTE);
// streamHandler.data(serverDataInfo);
// }
//
// private Session produceSession(String host, short version, InetSocketAddress address)
// {
// try
// {
// Session session = serverSessions.get(host);
// if (session == null)
// {
// SPDYClient client = factory.newSPDYClient(version);
// session = client.connect(address, sessionListener).get(getConnectTimeout(), TimeUnit.MILLISECONDS);
// logger.debug("Proxy session connected to {}", address);
// Session existing = serverSessions.putIfAbsent(host, session);
// if (existing != null)
// {
// session.goAway(getTimeout(), TimeUnit.MILLISECONDS, new Handler.Adapter<Void>());
// session = existing;
// }
// }
// return session;
// }
// catch (Exception x)
// {
// logger.debug(x);
// return null;
// }
// }
//
// private void convert(short fromVersion, short toVersion, Headers headers)
// {
// if (fromVersion != toVersion)
// {
// for (HTTPSPDYHeader httpHeader : HTTPSPDYHeader.values())
// {
// Headers.Header header = headers.remove(httpHeader.name(fromVersion));
// if (header != null)
// {
// String toName = httpHeader.name(toVersion);
// for (String value : header.values())
// headers.add(toName, value);
// }
// }
// }
// }
//
// private void rst(Stream stream)
// {
// RstInfo rstInfo = new RstInfo(stream.getId(), StreamStatus.REFUSED_STREAM);
// stream.getSession().rst(rstInfo, getTimeout(), TimeUnit.MILLISECONDS, new Handler.Adapter<Void>());
// }
//
// private class ProxyStreamFrameListener extends StreamFrameListener.Adapter
// {
// private final Stream clientStream;
// private volatile ReplyInfo replyInfo;
//
// public ProxyStreamFrameListener(Stream clientStream)
// {
// this.clientStream = clientStream;
// }
//
// @Override
// public void onReply(final Stream stream, ReplyInfo replyInfo)
// {
// logger.debug("S -> P {} on {}", replyInfo, stream);
//
// short serverVersion = stream.getSession().getVersion();
// Headers headers = new Headers(replyInfo.getHeaders(), false);
//
// addResponseProxyHeaders(stream, headers);
// customizeResponseHeaders(stream, headers);
// short clientVersion = this.clientStream.getSession().getVersion();
// convert(serverVersion, clientVersion, headers);
//
// this.replyInfo = new ReplyInfo(headers, replyInfo.isClose());
// if (replyInfo.isClose())
// reply(stream);
// }
//
// @Override
// public void onHeaders(Stream stream, HeadersInfo headersInfo)
// {
// // TODO
// throw new UnsupportedOperationException("Not Yet Implemented");
// }
//
// @Override
// public void onData(final Stream stream, final DataInfo dataInfo)
// {
// logger.debug("S -> P {} on {}", dataInfo, stream);
//
// if (replyInfo != null)
// {
// if (dataInfo.isClose())
// replyInfo.getHeaders().put("content-length", String.valueOf(dataInfo.available()));
// reply(stream);
// }
// data(stream, dataInfo);
// }
//
// private void reply(final Stream stream)
// {
// final ReplyInfo replyInfo = this.replyInfo;
// this.replyInfo = null;
// clientStream.reply(replyInfo, getTimeout(), TimeUnit.MILLISECONDS, new Handler<Void>()
// {
// @Override
// public void completed(Void context)
// {
// logger.debug("P -> C {} from {} to {}", replyInfo, stream, clientStream);
// }
//
// @Override
// public void failed(Void context, Throwable x)
// {
// logger.debug(x);
// rst(clientStream);
// }
// });
// }
//
// private void data(final Stream stream, final DataInfo dataInfo)
// {
// clientStream.data(dataInfo, getTimeout(), TimeUnit.MILLISECONDS, new Handler<Void>()
// {
// @Override
// public void completed(Void context)
// {
// dataInfo.consume(dataInfo.length());
// logger.debug("P -> C {} from {} to {}", dataInfo, stream, clientStream);
// }
//
// @Override
// public void failed(Void context, Throwable x)
// {
// logger.debug(x);
// rst(clientStream);
// }
// });
// }
// }
//
// /**
// * <p>{@link StreamHandler} implements the forwarding of DATA frames from the client to the server.</p>
// * <p>Instances of this class buffer DATA frames sent by clients and send them to the server.
// * The buffering happens between the send of the SYN_STREAM to the server (where DATA frames may arrive
// * from the client before the SYN_STREAM has been fully sent), and between DATA frames, if the client
// * is a fast producer and the server a slow consumer, or if the client is a SPDY v2 client (and hence
// * without flow control) while the server is a SPDY v3 server (and hence with flow control).</p>
// */
// private class StreamHandler implements Handler<Stream>
// {
// private final Queue<DataInfoHandler> queue = new LinkedList<>();
// private final Stream clientStream;
// private final SynInfo serverSynInfo;
// private Stream serverStream;
//
// private StreamHandler(Stream clientStream, SynInfo serverSynInfo)
// {
// this.clientStream = clientStream;
// this.serverSynInfo = serverSynInfo;
// }
//
// @Override
// public void completed(Stream serverStream)
// {
// logger.debug("P -> S {} from {} to {}", serverSynInfo, clientStream, serverStream);
//
// serverStream.setAttribute(CLIENT_STREAM_ATTRIBUTE, clientStream);
//
// DataInfoHandler dataInfoHandler;
// synchronized (queue)
// {
// this.serverStream = serverStream;
// dataInfoHandler = queue.peek();
// if (dataInfoHandler != null)
// {
// if (dataInfoHandler.flushing)
// {
// logger.debug("SYN completed, flushing {}, queue size {}", dataInfoHandler.dataInfo, queue.size());
// dataInfoHandler = null;
// }
// else
// {
// dataInfoHandler.flushing = true;
// logger.debug("SYN completed, queue size {}", queue.size());
// }
// }
// else
// {
// logger.debug("SYN completed, queue empty");
// }
// }
// if (dataInfoHandler != null)
// flush(serverStream, dataInfoHandler);
// }
//
// @Override
// public void failed(Stream serverStream, Throwable x)
// {
// logger.debug(x);
// rst(clientStream);
// }
//
// public void data(DataInfo dataInfo)
// {
// Stream serverStream;
// DataInfoHandler dataInfoHandler = null;
// DataInfoHandler item = new DataInfoHandler(dataInfo);
// synchronized (queue)
// {
// queue.offer(item);
// serverStream = this.serverStream;
// if (serverStream != null)
// {
// dataInfoHandler = queue.peek();
// if (dataInfoHandler.flushing)
// {
// logger.debug("Queued {}, flushing {}, queue size {}", dataInfo, dataInfoHandler.dataInfo, queue.size());
// serverStream = null;
// }
// else
// {
// dataInfoHandler.flushing = true;
// logger.debug("Queued {}, queue size {}", dataInfo, queue.size());
// }
// }
// else
// {
// logger.debug("Queued {}, SYN incomplete, queue size {}", dataInfo, queue.size());
// }
// }
// if (serverStream != null)
// flush(serverStream, dataInfoHandler);
// }
//
// private void flush(Stream serverStream, DataInfoHandler dataInfoHandler)
// {
// logger.debug("P -> S {} on {}", dataInfoHandler.dataInfo, serverStream);
// serverStream.data(dataInfoHandler.dataInfo, getTimeout(), TimeUnit.MILLISECONDS, dataInfoHandler);
// }
//
// private class DataInfoHandler implements Handler<Void>
// {
// private final DataInfo dataInfo;
// private boolean flushing;
//
// private DataInfoHandler(DataInfo dataInfo)
// {
// this.dataInfo = dataInfo;
// }
//
// @Override
// public void completed(Void context)
// {
// Stream serverStream;
// DataInfoHandler dataInfoHandler;
// synchronized (queue)
// {
// serverStream = StreamHandler.this.serverStream;
// assert serverStream != null;
// dataInfoHandler = queue.poll();
// assert dataInfoHandler == this;
// dataInfoHandler = queue.peek();
// if (dataInfoHandler != null)
// {
// assert !dataInfoHandler.flushing;
// dataInfoHandler.flushing = true;
// logger.debug("Completed {}, queue size {}", dataInfo, queue.size());
// }
// else
// {
// logger.debug("Completed {}, queue empty", dataInfo);
// }
// }
// if (dataInfoHandler != null)
// flush(serverStream, dataInfoHandler);
// }
//
// @Override
// public void failed(Void context, Throwable x)
// {
// logger.debug(x);
// rst(clientStream);
// }
// }
// }
//
// private class ProxySessionFrameListener extends SessionFrameListener.Adapter implements StreamFrameListener
// {
// @Override
// public StreamFrameListener onSyn(Stream serverStream, SynInfo serverSynInfo)
// {
// logger.debug("S -> P pushed {} on {}", serverSynInfo, serverStream);
//
// Headers headers = new Headers(serverSynInfo.getHeaders(), false);
//
// addResponseProxyHeaders(serverStream, headers);
// customizeResponseHeaders(serverStream, headers);
// Stream clientStream = (Stream)serverStream.getAssociatedStream().getAttribute(CLIENT_STREAM_ATTRIBUTE);
// convert(serverStream.getSession().getVersion(), clientStream.getSession().getVersion(), headers);
//
// StreamHandler handler = new StreamHandler(clientStream, serverSynInfo);
// serverStream.setAttribute(STREAM_HANDLER_ATTRIBUTE, handler);
// clientStream.syn(new SynInfo(headers, serverSynInfo.isClose()), getTimeout(), TimeUnit.MILLISECONDS, handler);
//
// return this;
// }
//
// @Override
// public void onRst(Session serverSession, RstInfo serverRstInfo)
// {
// Stream serverStream = serverSession.getStream(serverRstInfo.getStreamId());
// if (serverStream != null)
// {
// Stream clientStream = (Stream)serverStream.getAttribute(CLIENT_STREAM_ATTRIBUTE);
// if (clientStream != null)
// {
// Session clientSession = clientStream.getSession();
// RstInfo clientRstInfo = new RstInfo(clientStream.getId(), serverRstInfo.getStreamStatus());
// clientSession.rst(clientRstInfo, getTimeout(), TimeUnit.MILLISECONDS, new Handler.Adapter<Void>());
// }
// }
// }
//
// @Override
// public void onGoAway(Session serverSession, GoAwayInfo goAwayInfo)
// {
// serverSessions.values().remove(serverSession);
// }
//
// @Override
// public void onReply(Stream stream, ReplyInfo replyInfo)
// {
// // Push streams never send a reply
// }
//
// @Override
// public void onHeaders(Stream stream, HeadersInfo headersInfo)
// {
// throw new UnsupportedOperationException(); //TODO
// }
//
// @Override
// public void onData(Stream serverStream, final DataInfo serverDataInfo)
// {
// logger.debug("S -> P pushed {} on {}", serverDataInfo, serverStream);
//
// ByteBufferDataInfo clientDataInfo = new ByteBufferDataInfo(serverDataInfo.asByteBuffer(false), serverDataInfo.isClose())
// {
// @Override
// public void consume(int delta)
// {
// super.consume(delta);
// serverDataInfo.consume(delta);
// }
// };
//
// StreamHandler handler = (StreamHandler)serverStream.getAttribute(STREAM_HANDLER_ATTRIBUTE);
// handler.data(clientDataInfo);
// }
// }
private static final String STREAM_HANDLER_ATTRIBUTE = "org.eclipse.jetty.spdy.http.proxy.streamHandler";
private static final String CLIENT_STREAM_ATTRIBUTE = "org.eclipse.jetty.spdy.http.proxy.clientStream";
private final ConcurrentMap<String, Session> serverSessions = new ConcurrentHashMap<>();
private final SessionFrameListener sessionListener = new ProxySessionFrameListener();
private final SPDYClient.Factory factory;
private volatile long connectTimeout = 15000;
private volatile long timeout = 60000;
public SPDYProxyEngine(SPDYClient.Factory factory)
{
this.factory = factory;
}
public long getConnectTimeout()
{
return connectTimeout;
}
public void setConnectTimeout(long connectTimeout)
{
this.connectTimeout = connectTimeout;
}
public long getTimeout()
{
return timeout;
}
public void setTimeout(long timeout)
{
this.timeout = timeout;
}
public StreamFrameListener proxy(final Stream clientStream, SynInfo clientSynInfo, ProxyEngineSelector.ProxyServerInfo proxyServerInfo)
{
Headers headers = new Headers(clientSynInfo.getHeaders(), false);
short serverVersion = getVersion(proxyServerInfo.getProtocol());
InetSocketAddress address = proxyServerInfo.getAddress();
Session serverSession = produceSession(proxyServerInfo.getHost(), serverVersion, address);
if (serverSession == null)
{
rst(clientStream);
return null;
}
final Session clientSession = clientStream.getSession();
addRequestProxyHeaders(clientStream, headers);
customizeRequestHeaders(clientStream, headers);
convert(clientSession.getVersion(), serverVersion, headers);
SynInfo serverSynInfo = new SynInfo(headers, clientSynInfo.isClose());
StreamFrameListener listener = new ProxyStreamFrameListener(clientStream);
StreamHandler handler = new StreamHandler(clientStream, serverSynInfo);
clientStream.setAttribute(STREAM_HANDLER_ATTRIBUTE, handler);
serverSession.syn(serverSynInfo, listener, timeout, TimeUnit.MILLISECONDS, handler);
return this;
}
private static short getVersion(String protocol)
{
switch (protocol)
{
case "spdy/2":
return SPDY.V2;
case "spdy/3":
return SPDY.V3;
default:
throw new IllegalArgumentException("Procotol: " + protocol + " is not a known SPDY protocol");
}
}
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
// Servers do not receive replies
}
@Override
public void onHeaders(Stream stream, HeadersInfo headersInfo)
{
// TODO
throw new UnsupportedOperationException("Not Yet Implemented");
}
@Override
public void onData(Stream clientStream, final DataInfo clientDataInfo)
{
logger.debug("C -> P {} on {}", clientDataInfo, clientStream);
ByteBufferDataInfo serverDataInfo = new ByteBufferDataInfo(clientDataInfo.asByteBuffer(false), clientDataInfo.isClose())
{
@Override
public void consume(int delta)
{
super.consume(delta);
clientDataInfo.consume(delta);
}
};
StreamHandler streamHandler = (StreamHandler)clientStream.getAttribute(STREAM_HANDLER_ATTRIBUTE);
streamHandler.data(serverDataInfo);
}
private Session produceSession(String host, short version, InetSocketAddress address)
{
try
{
Session session = serverSessions.get(host);
if (session == null)
{
SPDYClient client = factory.newSPDYClient(version);
session = client.connect(address, sessionListener).get(getConnectTimeout(), TimeUnit.MILLISECONDS);
logger.debug("Proxy session connected to {}", address);
Session existing = serverSessions.putIfAbsent(host, session);
if (existing != null)
{
session.goAway(getTimeout(), TimeUnit.MILLISECONDS, new Callback.Empty<Void>());
session = existing;
}
}
return session;
}
catch (Exception x)
{
logger.debug(x);
return null;
}
}
private void convert(short fromVersion, short toVersion, Headers headers)
{
if (fromVersion != toVersion)
{
for (HTTPSPDYHeader httpHeader : HTTPSPDYHeader.values())
{
Headers.Header header = headers.remove(httpHeader.name(fromVersion));
if (header != null)
{
String toName = httpHeader.name(toVersion);
for (String value : header.values())
headers.add(toName, value);
}
}
}
}
private void rst(Stream stream)
{
RstInfo rstInfo = new RstInfo(stream.getId(), StreamStatus.REFUSED_STREAM);
stream.getSession().rst(rstInfo, getTimeout(), TimeUnit.MILLISECONDS, new Callback.Empty<Void>());
}
private class ProxyStreamFrameListener extends StreamFrameListener.Adapter
{
private final Stream clientStream;
private volatile ReplyInfo replyInfo;
public ProxyStreamFrameListener(Stream clientStream)
{
this.clientStream = clientStream;
}
@Override
public void onReply(final Stream stream, ReplyInfo replyInfo)
{
logger.debug("S -> P {} on {}", replyInfo, stream);
short serverVersion = stream.getSession().getVersion();
Headers headers = new Headers(replyInfo.getHeaders(), false);
addResponseProxyHeaders(stream, headers);
customizeResponseHeaders(stream, headers);
short clientVersion = this.clientStream.getSession().getVersion();
convert(serverVersion, clientVersion, headers);
this.replyInfo = new ReplyInfo(headers, replyInfo.isClose());
if (replyInfo.isClose())
reply(stream);
}
@Override
public void onHeaders(Stream stream, HeadersInfo headersInfo)
{
// TODO
throw new UnsupportedOperationException("Not Yet Implemented");
}
@Override
public void onData(final Stream stream, final DataInfo dataInfo)
{
logger.debug("S -> P {} on {}", dataInfo, stream);
if (replyInfo != null)
{
if (dataInfo.isClose())
replyInfo.getHeaders().put("content-length", String.valueOf(dataInfo.available()));
reply(stream);
}
data(stream, dataInfo);
}
private void reply(final Stream stream)
{
final ReplyInfo replyInfo = this.replyInfo;
this.replyInfo = null;
clientStream.reply(replyInfo, getTimeout(), TimeUnit.MILLISECONDS, new Callback<Void>()
{
@Override
public void completed(Void context)
{
logger.debug("P -> C {} from {} to {}", replyInfo, stream, clientStream);
}
@Override
public void failed(Void context, Throwable x)
{
logger.debug(x);
rst(clientStream);
}
});
}
private void data(final Stream stream, final DataInfo dataInfo)
{
clientStream.data(dataInfo, getTimeout(), TimeUnit.MILLISECONDS, new Callback<Void>()
{
@Override
public void completed(Void context)
{
dataInfo.consume(dataInfo.length());
logger.debug("P -> C {} from {} to {}", dataInfo, stream, clientStream);
}
@Override
public void failed(Void context, Throwable x)
{
logger.debug(x);
rst(clientStream);
}
});
}
}
/**
* <p>{@link StreamHandler} implements the forwarding of DATA frames from the client to the server.</p>
* <p>Instances of this class buffer DATA frames sent by clients and send them to the server.
* The buffering happens between the send of the SYN_STREAM to the server (where DATA frames may arrive
* from the client before the SYN_STREAM has been fully sent), and between DATA frames, if the client
* is a fast producer and the server a slow consumer, or if the client is a SPDY v2 client (and hence
* without flow control) while the server is a SPDY v3 server (and hence with flow control).</p>
*/
private class StreamHandler implements Callback<Stream>
{
private final Queue<DataInfoHandler> queue = new LinkedList<>();
private final Stream clientStream;
private final SynInfo serverSynInfo;
private Stream serverStream;
private StreamHandler(Stream clientStream, SynInfo serverSynInfo)
{
this.clientStream = clientStream;
this.serverSynInfo = serverSynInfo;
}
@Override
public void completed(Stream serverStream)
{
logger.debug("P -> S {} from {} to {}", serverSynInfo, clientStream, serverStream);
serverStream.setAttribute(CLIENT_STREAM_ATTRIBUTE, clientStream);
DataInfoHandler dataInfoHandler;
synchronized (queue)
{
this.serverStream = serverStream;
dataInfoHandler = queue.peek();
if (dataInfoHandler != null)
{
if (dataInfoHandler.flushing)
{
logger.debug("SYN completed, flushing {}, queue size {}", dataInfoHandler.dataInfo, queue.size());
dataInfoHandler = null;
}
else
{
dataInfoHandler.flushing = true;
logger.debug("SYN completed, queue size {}", queue.size());
}
}
else
{
logger.debug("SYN completed, queue empty");
}
}
if (dataInfoHandler != null)
flush(serverStream, dataInfoHandler);
}
@Override
public void failed(Stream serverStream, Throwable x)
{
logger.debug(x);
rst(clientStream);
}
public void data(DataInfo dataInfo)
{
Stream serverStream;
DataInfoHandler dataInfoHandler = null;
DataInfoHandler item = new DataInfoHandler(dataInfo);
synchronized (queue)
{
queue.offer(item);
serverStream = this.serverStream;
if (serverStream != null)
{
dataInfoHandler = queue.peek();
if (dataInfoHandler.flushing)
{
logger.debug("Queued {}, flushing {}, queue size {}", dataInfo, dataInfoHandler.dataInfo, queue.size());
serverStream = null;
}
else
{
dataInfoHandler.flushing = true;
logger.debug("Queued {}, queue size {}", dataInfo, queue.size());
}
}
else
{
logger.debug("Queued {}, SYN incomplete, queue size {}", dataInfo, queue.size());
}
}
if (serverStream != null)
flush(serverStream, dataInfoHandler);
}
private void flush(Stream serverStream, DataInfoHandler dataInfoHandler)
{
logger.debug("P -> S {} on {}", dataInfoHandler.dataInfo, serverStream);
serverStream.data(dataInfoHandler.dataInfo, getTimeout(), TimeUnit.MILLISECONDS, dataInfoHandler);
}
private class DataInfoHandler implements Callback<Void>
{
private final DataInfo dataInfo;
private boolean flushing;
private DataInfoHandler(DataInfo dataInfo)
{
this.dataInfo = dataInfo;
}
@Override
public void completed(Void context)
{
Stream serverStream;
DataInfoHandler dataInfoHandler;
synchronized (queue)
{
serverStream = StreamHandler.this.serverStream;
assert serverStream != null;
dataInfoHandler = queue.poll();
assert dataInfoHandler == this;
dataInfoHandler = queue.peek();
if (dataInfoHandler != null)
{
assert !dataInfoHandler.flushing;
dataInfoHandler.flushing = true;
logger.debug("Completed {}, queue size {}", dataInfo, queue.size());
}
else
{
logger.debug("Completed {}, queue empty", dataInfo);
}
}
if (dataInfoHandler != null)
flush(serverStream, dataInfoHandler);
}
@Override
public void failed(Void context, Throwable x)
{
logger.debug(x);
rst(clientStream);
}
}
}
private class ProxySessionFrameListener extends SessionFrameListener.Adapter implements StreamFrameListener
{
@Override
public StreamFrameListener onSyn(Stream serverStream, SynInfo serverSynInfo)
{
logger.debug("S -> P pushed {} on {}", serverSynInfo, serverStream);
Headers headers = new Headers(serverSynInfo.getHeaders(), false);
addResponseProxyHeaders(serverStream, headers);
customizeResponseHeaders(serverStream, headers);
Stream clientStream = (Stream)serverStream.getAssociatedStream().getAttribute(CLIENT_STREAM_ATTRIBUTE);
convert(serverStream.getSession().getVersion(), clientStream.getSession().getVersion(), headers);
StreamHandler handler = new StreamHandler(clientStream, serverSynInfo);
serverStream.setAttribute(STREAM_HANDLER_ATTRIBUTE, handler);
clientStream.syn(new SynInfo(headers, serverSynInfo.isClose()), getTimeout(), TimeUnit.MILLISECONDS, handler);
return this;
}
@Override
public void onRst(Session serverSession, RstInfo serverRstInfo)
{
Stream serverStream = serverSession.getStream(serverRstInfo.getStreamId());
if (serverStream != null)
{
Stream clientStream = (Stream)serverStream.getAttribute(CLIENT_STREAM_ATTRIBUTE);
if (clientStream != null)
{
Session clientSession = clientStream.getSession();
RstInfo clientRstInfo = new RstInfo(clientStream.getId(), serverRstInfo.getStreamStatus());
clientSession.rst(clientRstInfo, getTimeout(), TimeUnit.MILLISECONDS, new Callback.Empty<Void>());
}
}
}
@Override
public void onGoAway(Session serverSession, GoAwayInfo goAwayInfo)
{
serverSessions.values().remove(serverSession);
}
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
// Push streams never send a reply
}
@Override
public void onHeaders(Stream stream, HeadersInfo headersInfo)
{
throw new UnsupportedOperationException(); //TODO
}
@Override
public void onData(Stream serverStream, final DataInfo serverDataInfo)
{
logger.debug("S -> P pushed {} on {}", serverDataInfo, serverStream);
ByteBufferDataInfo clientDataInfo = new ByteBufferDataInfo(serverDataInfo.asByteBuffer(false), serverDataInfo.isClose())
{
@Override
public void consume(int delta)
{
super.consume(delta);
serverDataInfo.consume(delta);
}
};
StreamHandler handler = (StreamHandler)serverStream.getAttribute(STREAM_HANDLER_ATTRIBUTE);
handler.data(clientDataInfo);
}
}
}

784
jetty-spdy/spdy-jetty-http/src/test/java/org/eclipse/jetty/spdy/proxy/ProxyHTTPSPDYTest.java Normal file
View File

@ -0,0 +1,784 @@
//
// ========================================================================
// Copyright (c) 1995-2012 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.proxy;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.spdy.SPDYClient;
import org.eclipse.jetty.spdy.SPDYServerConnector;
import org.eclipse.jetty.spdy.ServerSPDYConnectionFactory;
import org.eclipse.jetty.spdy.api.BytesDataInfo;
import org.eclipse.jetty.spdy.api.DataInfo;
import org.eclipse.jetty.spdy.api.GoAwayInfo;
import org.eclipse.jetty.spdy.api.Headers;
import org.eclipse.jetty.spdy.api.PingInfo;
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.Session;
import org.eclipse.jetty.spdy.api.SessionFrameListener;
import org.eclipse.jetty.spdy.api.Stream;
import org.eclipse.jetty.spdy.api.StreamFrameListener;
import org.eclipse.jetty.spdy.api.StreamStatus;
import org.eclipse.jetty.spdy.api.SynInfo;
import org.eclipse.jetty.spdy.api.server.ServerSessionFrameListener;
import org.eclipse.jetty.spdy.http.HTTPSPDYHeader;
import org.eclipse.jetty.util.Callback;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestWatchman;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.model.FrameworkMethod;
@Ignore // TODO: make these tests pass
@RunWith(value = Parameterized.class)
public class ProxyHTTPSPDYTest
{
@Rule
public final TestWatchman testName = new TestWatchman()
{
@Override
public void starting(FrameworkMethod method)
{
super.starting(method);
System.err.printf("Running %s.%s()%n",
method.getMethod().getDeclaringClass().getName(),
method.getName());
}
};
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 SPDYServerConnector proxyConnector;
public ProxyHTTPSPDYTest(short version)
{
this.version = version;
}
protected InetSocketAddress startServer(ServerSessionFrameListener listener) throws Exception
{
server = new Server();
SPDYServerConnector serverConnector = new SPDYServerConnector(server, listener);
serverConnector.setDefaultConnectionFactory(new ServerSPDYConnectionFactory(version,
serverConnector.getByteBufferPool(), serverConnector.getExecutor(), serverConnector.getScheduler(), listener));
serverConnector.setPort(0);
server.addConnector(serverConnector);
server.start();
return new InetSocketAddress("localhost", serverConnector.getLocalPort());
}
protected InetSocketAddress startProxy(InetSocketAddress address) throws Exception
{
proxy = new Server();
ProxyEngineSelector proxyEngineSelector = new ProxyEngineSelector();
SPDYProxyEngine spdyProxyEngine = new SPDYProxyEngine(factory);
proxyEngineSelector.putProxyEngine("spdy/" + version, spdyProxyEngine);
proxyEngineSelector.putProxyServerInfo("localhost", new ProxyEngineSelector.ProxyServerInfo("spdy/" + version, address.getHostName(), address.getPort()));
proxyConnector = new HTTPSPDYProxyConnector(server, proxyEngineSelector);
proxyConnector.setPort(0);
proxy.addConnector(proxyConnector);
proxy.start();
return new InetSocketAddress("localhost", proxyConnector.getLocalPort());
}
@Before
public void init() throws Exception
{
factory = new SPDYClient.Factory();
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 testClosingClientDoesNotCloseServer() throws Exception
{
final CountDownLatch closeLatch = new CountDownLatch(1);
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Headers responseHeaders = new Headers();
responseHeaders.put(HTTPSPDYHeader.VERSION.name(version), "HTTP/1.1");
responseHeaders.put(HTTPSPDYHeader.STATUS.name(version), "200 OK");
stream.reply(new ReplyInfo(responseHeaders, true));
return null;
}
@Override
public void onGoAway(Session session, GoAwayInfo goAwayInfo)
{
closeLatch.countDown();
}
}));
Socket client = new Socket();
client.connect(proxyAddress);
OutputStream output = client.getOutputStream();
String request = "" +
"GET / 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"));
while (line.length() > 0)
line = reader.readLine();
Assert.assertFalse(reader.ready());
client.close();
// Must not close, other clients may still be connected
Assert.assertFalse(closeLatch.await(1, TimeUnit.SECONDS));
}
@Test
public void testGETThenNoContentFromTwoClients() throws Exception
{
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Assert.assertTrue(synInfo.isClose());
Headers requestHeaders = synInfo.getHeaders();
Assert.assertNotNull(requestHeaders.get("via"));
Headers responseHeaders = new Headers();
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);
return null;
}
}));
Socket client1 = new Socket();
client1.connect(proxyAddress);
OutputStream output1 = client1.getOutputStream();
String request = "" +
"GET / HTTP/1.1\r\n" +
"Host: localhost:" + proxyAddress.getPort() + "\r\n" +
"\r\n";
output1.write(request.getBytes("UTF-8"));
output1.flush();
InputStream input1 = client1.getInputStream();
BufferedReader reader1 = new BufferedReader(new InputStreamReader(input1, "UTF-8"));
String line = reader1.readLine();
Assert.assertTrue(line.contains(" 200"));
while (line.length() > 0)
line = reader1.readLine();
Assert.assertFalse(reader1.ready());
// Perform another request with another client
Socket client2 = new Socket();
client2.connect(proxyAddress);
OutputStream output2 = client2.getOutputStream();
output2.write(request.getBytes("UTF-8"));
output2.flush();
InputStream input2 = client2.getInputStream();
BufferedReader reader2 = new BufferedReader(new InputStreamReader(input2, "UTF-8"));
line = reader2.readLine();
Assert.assertTrue(line.contains(" 200"));
while (line.length() > 0)
line = reader2.readLine();
Assert.assertFalse(reader2.ready());
client1.close();
client2.close();
}
@Test
public void testGETThenSmallResponseContent() throws Exception
{
final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Assert.assertTrue(synInfo.isClose());
Headers requestHeaders = synInfo.getHeaders();
Assert.assertNotNull(requestHeaders.get("via"));
Headers responseHeaders = new Headers();
responseHeaders.put(HTTPSPDYHeader.VERSION.name(version), "HTTP/1.1");
responseHeaders.put(HTTPSPDYHeader.STATUS.name(version), "200 OK");
ReplyInfo replyInfo = new ReplyInfo(responseHeaders, false);
stream.reply(replyInfo);
stream.data(new BytesDataInfo(data, true));
return null;
}
}));
Socket client = new Socket();
client.connect(proxyAddress);
OutputStream output = client.getOutputStream();
String request = "" +
"GET / 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 "));
while (line.length() > 0)
line = reader.readLine();
for (byte datum : data)
Assert.assertEquals(datum, reader.read());
Assert.assertFalse(reader.ready());
// Perform another request so that we are sure we reset the states of parsers and generators
output.write(request.getBytes("UTF-8"));
output.flush();
line = reader.readLine();
Assert.assertTrue(line.contains(" 200"));
while (line.length() > 0)
line = reader.readLine();
for (byte datum : data)
Assert.assertEquals(datum, reader.read());
Assert.assertFalse(reader.ready());
client.close();
}
@Test
public void testPOSTWithSmallRequestContentThenRedirect() throws Exception
{
final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
return new StreamFrameListener.Adapter()
{
@Override
public void onData(Stream stream, DataInfo dataInfo)
{
dataInfo.consume(dataInfo.length());
if (dataInfo.isClose())
{
Headers headers = new Headers();
headers.put(HTTPSPDYHeader.VERSION.name(version), "HTTP/1.1");
headers.put(HTTPSPDYHeader.STATUS.name(version), "303 See Other");
stream.reply(new ReplyInfo(headers, true));
}
}
};
}
}));
Socket client = new Socket();
client.connect(proxyAddress);
OutputStream output = client.getOutputStream();
String request = "" +
"POST / HTTP/1.1\r\n" +
"Host: localhost:" + proxyAddress.getPort() + "\r\n" +
"Content-Length: " + data.length + "\r\n" +
"Content-Type: application/octet-stream\r\n" +
"\r\n";
output.write(request.getBytes("UTF-8"));
output.write(data);
output.flush();
InputStream input = client.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(input, "UTF-8"));
String line = reader.readLine();
Assert.assertTrue(line.contains(" 303"));
while (line.length() > 0)
line = reader.readLine();
Assert.assertFalse(reader.ready());
// Perform another request so that we are sure we reset the states of parsers and generators
output.write(request.getBytes("UTF-8"));
output.write(data);
output.flush();
line = reader.readLine();
Assert.assertTrue(line.contains(" 303"));
while (line.length() > 0)
line = reader.readLine();
Assert.assertFalse(reader.ready());
client.close();
}
@Test
public void testPOSTWithSmallRequestContentThenSmallResponseContent() throws Exception
{
final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
return new StreamFrameListener.Adapter()
{
@Override
public void onData(Stream stream, DataInfo dataInfo)
{
dataInfo.consume(dataInfo.length());
if (dataInfo.isClose())
{
Headers responseHeaders = new Headers();
responseHeaders.put(HTTPSPDYHeader.VERSION.name(version), "HTTP/1.1");
responseHeaders.put(HTTPSPDYHeader.STATUS.name(version), "200 OK");
ReplyInfo replyInfo = new ReplyInfo(responseHeaders, false);
stream.reply(replyInfo);
stream.data(new BytesDataInfo(data, true));
}
}
};
}
}));
Socket client = new Socket();
client.connect(proxyAddress);
OutputStream output = client.getOutputStream();
String request = "" +
"POST / HTTP/1.1\r\n" +
"Host: localhost:" + proxyAddress.getPort() + "\r\n" +
"Content-Length: " + data.length + "\r\n" +
"Content-Type: application/octet-stream\r\n" +
"\r\n";
output.write(request.getBytes("UTF-8"));
output.write(data);
output.flush();
InputStream input = client.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(input, "UTF-8"));
String line = reader.readLine();
Assert.assertTrue(line.contains(" 200"));
while (line.length() > 0)
line = reader.readLine();
for (byte datum : data)
Assert.assertEquals(datum, reader.read());
Assert.assertFalse(reader.ready());
// Perform another request so that we are sure we reset the states of parsers and generators
output.write(request.getBytes("UTF-8"));
output.write(data);
output.flush();
line = reader.readLine();
Assert.assertTrue(line.contains(" 200"));
while (line.length() > 0)
line = reader.readLine();
for (byte datum : data)
Assert.assertEquals(datum, reader.read());
Assert.assertFalse(reader.ready());
client.close();
}
@Test
public void testSYNThenREPLY() throws Exception
{
final String header = "foo";
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Headers requestHeaders = synInfo.getHeaders();
Assert.assertNotNull(requestHeaders.get("via"));
Assert.assertNotNull(requestHeaders.get(header));
Headers responseHeaders = new Headers();
responseHeaders.put(header, "baz");
stream.reply(new ReplyInfo(responseHeaders, true));
return null;
}
}));
proxyConnector.setDefaultConnectionFactory(proxyConnector.getConnectionFactory("spdy/" + version));
Session client = factory.newSPDYClient(version).connect(proxyAddress, null).get(5, TimeUnit.SECONDS);
final CountDownLatch replyLatch = new CountDownLatch(1);
Headers headers = new Headers();
headers.put(HTTPSPDYHeader.HOST.name(version), "localhost:" + proxyAddress.getPort());
headers.put(header, "bar");
client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
Headers headers = replyInfo.getHeaders();
Assert.assertNotNull(headers.get(header));
replyLatch.countDown();
}
});
Assert.assertTrue(replyLatch.await(5, TimeUnit.SECONDS));
client.goAway().get(5, TimeUnit.SECONDS);
}
@Test
public void testSYNThenREPLYAndDATA() throws Exception
{
final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
final String header = "foo";
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Headers requestHeaders = synInfo.getHeaders();
Assert.assertNotNull(requestHeaders.get("via"));
Assert.assertNotNull(requestHeaders.get(header));
Headers responseHeaders = new Headers();
responseHeaders.put(header, "baz");
stream.reply(new ReplyInfo(responseHeaders, false));
stream.data(new BytesDataInfo(data, true));
return null;
}
}));
proxyConnector.setDefaultConnectionFactory(proxyConnector.getConnectionFactory("spdy/" + version));
Session client = factory.newSPDYClient(version).connect(proxyAddress, null).get(5, TimeUnit.SECONDS);
final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1);
Headers headers = new Headers();
headers.put(HTTPSPDYHeader.HOST.name(version), "localhost:" + proxyAddress.getPort());
headers.put(header, "bar");
client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{
private final ByteArrayOutputStream result = new ByteArrayOutputStream();
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
Headers headers = replyInfo.getHeaders();
Assert.assertNotNull(headers.get(header));
replyLatch.countDown();
}
@Override
public void onData(Stream stream, DataInfo dataInfo)
{
result.write(dataInfo.asBytes(true), 0, dataInfo.length());
if (dataInfo.isClose())
{
Assert.assertArrayEquals(data, result.toByteArray());
dataLatch.countDown();
}
}
});
Assert.assertTrue(replyLatch.await(5, TimeUnit.SECONDS));
Assert.assertTrue(dataLatch.await(5, TimeUnit.SECONDS));
client.goAway().get(5, TimeUnit.SECONDS);
}
@Test
public void testGETThenSPDYPushIsIgnored() throws Exception
{
final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Headers responseHeaders = new Headers();
responseHeaders.put(HTTPSPDYHeader.VERSION.name(version), "HTTP/1.1");
responseHeaders.put(HTTPSPDYHeader.STATUS.name(version), "200 OK");
Headers pushHeaders = new Headers();
pushHeaders.put(HTTPSPDYHeader.URI.name(version), "/push");
stream.syn(new SynInfo(pushHeaders, false), 5, TimeUnit.SECONDS, new Callback.Empty<Stream>()
{
@Override
public void completed(Stream pushStream)
{
pushStream.data(new BytesDataInfo(data, true));
}
});
stream.reply(new ReplyInfo(responseHeaders, true));
return null;
}
}));
Socket client = new Socket();
client.connect(proxyAddress);
OutputStream output = client.getOutputStream();
String request = "" +
"GET / HTTP/1.1\r\n" +
"Host: localhost:" + proxyAddress.getPort() + "\r\n" +
"\r\n";
output.write(request.getBytes("UTF-8"));
output.flush();
client.setSoTimeout(1000);
InputStream input = client.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(input, "UTF-8"));
String line = reader.readLine();
Assert.assertTrue(line.contains(" 200"));
while (line.length() > 0)
line = reader.readLine();
Assert.assertFalse(reader.ready());
client.close();
}
@Test
public void testSYNThenSPDYPushIsReceived() throws Exception
{
final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Headers responseHeaders = new Headers();
responseHeaders.put(HTTPSPDYHeader.VERSION.name(version), "HTTP/1.1");
responseHeaders.put(HTTPSPDYHeader.STATUS.name(version), "200 OK");
stream.reply(new ReplyInfo(responseHeaders, false));
Headers pushHeaders = new Headers();
pushHeaders.put(HTTPSPDYHeader.URI.name(version), "/push");
stream.syn(new SynInfo(pushHeaders, false), 5, TimeUnit.SECONDS, new Callback.Empty<Stream>()
{
@Override
public void completed(Stream pushStream)
{
pushStream.data(new BytesDataInfo(data, true));
}
});
stream.data(new BytesDataInfo(data, true));
return null;
}
}));
proxyConnector.setDefaultConnectionFactory(proxyConnector.getConnectionFactory("spdy/" + version));
final CountDownLatch pushSynLatch = new CountDownLatch(1);
final CountDownLatch pushDataLatch = new CountDownLatch(1);
Session client = factory.newSPDYClient(version).connect(proxyAddress, new SessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
pushSynLatch.countDown();
return new StreamFrameListener.Adapter()
{
@Override
public void onData(Stream stream, DataInfo dataInfo)
{
dataInfo.consume(dataInfo.length());
if (dataInfo.isClose())
pushDataLatch.countDown();
}
};
}
}).get(5, TimeUnit.SECONDS);
Headers headers = new Headers();
headers.put(HTTPSPDYHeader.HOST.name(version), "localhost:" + proxyAddress.getPort());
final CountDownLatch replyLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1);
client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
{
@Override
public void onReply(Stream stream, ReplyInfo replyInfo)
{
replyLatch.countDown();
}
@Override
public void onData(Stream stream, DataInfo dataInfo)
{
dataInfo.consume(dataInfo.length());
if (dataInfo.isClose())
dataLatch.countDown();
}
});
Assert.assertTrue(replyLatch.await(5, TimeUnit.SECONDS));
Assert.assertTrue(pushSynLatch.await(5, TimeUnit.SECONDS));
Assert.assertTrue(pushDataLatch.await(5, TimeUnit.SECONDS));
Assert.assertTrue(dataLatch.await(5, TimeUnit.SECONDS));
client.goAway().get(5, TimeUnit.SECONDS);
}
@Test
public void testPING() throws Exception
{
// PING is per hop, and it does not carry the information to which server to ping to
// We just verify that it works
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()));
proxyConnector.setDefaultConnectionFactory(proxyConnector.getConnectionFactory("spdy/" + version));
final CountDownLatch pingLatch = new CountDownLatch(1);
Session client = factory.newSPDYClient(version).connect(proxyAddress, new SessionFrameListener.Adapter()
{
@Override
public void onPing(Session session, PingInfo pingInfo)
{
pingLatch.countDown();
}
}).get(5, TimeUnit.SECONDS);
client.ping().get(5, TimeUnit.SECONDS);
Assert.assertTrue(pingLatch.await(5, TimeUnit.SECONDS));
client.goAway().get(5, TimeUnit.SECONDS);
}
@Test
public void testGETThenReset() throws Exception
{
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Assert.assertTrue(synInfo.isClose());
Headers requestHeaders = synInfo.getHeaders();
Assert.assertNotNull(requestHeaders.get("via"));
stream.getSession().rst(new RstInfo(stream.getId(), StreamStatus.REFUSED_STREAM));
return null;
}
}));
Socket client = new Socket();
client.connect(proxyAddress);
OutputStream output = client.getOutputStream();
String request = "" +
"GET / 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"));
Assert.assertNull(reader.readLine());
client.close();
}
@Test
public void testSYNThenReset() throws Exception
{
InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
{
@Override
public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
{
Assert.assertTrue(synInfo.isClose());
Headers requestHeaders = synInfo.getHeaders();
Assert.assertNotNull(requestHeaders.get("via"));
stream.getSession().rst(new RstInfo(stream.getId(), StreamStatus.REFUSED_STREAM));
return null;
}
}));
proxyConnector.setDefaultConnectionFactory(proxyConnector.getConnectionFactory("spdy/" + version));
final CountDownLatch resetLatch = new CountDownLatch(1);
Session client = factory.newSPDYClient(version).connect(proxyAddress, new SessionFrameListener.Adapter()
{
@Override
public void onRst(Session session, RstInfo rstInfo)
{
resetLatch.countDown();
}
}).get(5, TimeUnit.SECONDS);
Headers headers = new Headers();
headers.put(HTTPSPDYHeader.HOST.name(version), "localhost:" + proxyAddress.getPort());
client.syn(new SynInfo(headers, true), null);
Assert.assertTrue(resetLatch.await(5, TimeUnit.SECONDS));
client.goAway().get(5, TimeUnit.SECONDS);
}
}

731
jetty-spdy/spdy-jetty-http/src/test/java/org/eclipse/jetty/spdy/proxy/ProxyHTTPSPDYv2Test.java
View File

@ -1,731 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2012 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.proxy;
public class ProxyHTTPSPDYv2Test
{
// @Rule
// public final TestWatchman testName = new TestWatchman()
// {
// @Override
// public void starting(FrameworkMethod method)
// {
// super.starting(method);
// System.err.printf("Running %s.%s()%n",
// method.getMethod().getDeclaringClass().getName(),
// method.getName());
// }
// };
//
// private SPDYClient.Factory factory;
// private Server server;
// private Server proxy;
// private SPDYServerConnector proxyConnector;
//
// protected short version()
// {
// return SPDY.V2;
// }
//
// protected InetSocketAddress startServer(ServerSessionFrameListener listener) throws Exception
// {
// server = new Server();
// SPDYServerConnector serverConnector = new SPDYServerConnector(server, listener);
// serverConnector.setDefaultConnectionFactory(new ServerSPDYAsyncConnectionFactory(version(),
// serverConnector.getByteBufferPool(), serverConnector.getExecutor(), serverConnector.getScheduler(), listener));
// serverConnector.setPort(0);
// server.addConnector(serverConnector);
// server.start();
// return new InetSocketAddress("localhost", serverConnector.getLocalPort());
// }
//
// protected InetSocketAddress startProxy(InetSocketAddress address) throws Exception
// {
// proxy = new Server();
// ProxyEngineSelector proxyEngineSelector = new ProxyEngineSelector();
// SPDYProxyEngine spdyProxyEngine = new SPDYProxyEngine(factory);
// proxyEngineSelector.putProxyEngine("spdy/" + version(), spdyProxyEngine);
// proxyEngineSelector.putProxyServerInfo("localhost", new ProxyEngineSelector.ProxyServerInfo("spdy/" + version(), address.getHostName(), address.getPort()));
// proxyConnector = new HTTPSPDYProxyConnector(proxyEngineSelector);
// proxyConnector.setPort(0);
// proxy.addConnector(proxyConnector);
// proxy.start();
// return new InetSocketAddress("localhost", proxyConnector.getLocalPort());
// }
//
// @Before
// public void init() throws Exception
// {
// factory = new SPDYClient.Factory();
// 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 testClosingClientDoesNotCloseServer() throws Exception
// {
// final CountDownLatch closeLatch = new CountDownLatch(1);
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// Headers responseHeaders = new Headers();
// responseHeaders.put(HTTPSPDYHeader.VERSION.name(version()), "HTTP/1.1");
// responseHeaders.put(HTTPSPDYHeader.STATUS.name(version()), "200 OK");
// stream.reply(new ReplyInfo(responseHeaders, true));
// return null;
// }
//
// @Override
// public void onGoAway(Session session, GoAwayInfo goAwayInfo)
// {
// closeLatch.countDown();
// }
// }));
//
// Socket client = new Socket();
// client.connect(proxyAddress);
// OutputStream output = client.getOutputStream();
//
// String request = "" +
// "GET / 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"));
// while (line.length() > 0)
// line = reader.readLine();
// Assert.assertFalse(reader.ready());
//
// client.close();
//
// // Must not close, other clients may still be connected
// Assert.assertFalse(closeLatch.await(1, TimeUnit.SECONDS));
// }
//
// @Test
// public void testGETThenNoContentFromTwoClients() throws Exception
// {
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// Assert.assertTrue(synInfo.isClose());
// Headers requestHeaders = synInfo.getHeaders();
// Assert.assertNotNull(requestHeaders.get("via"));
//
// Headers responseHeaders = new Headers();
// 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);
// return null;
// }
// }));
//
// Socket client1 = new Socket();
// client1.connect(proxyAddress);
// OutputStream output1 = client1.getOutputStream();
//
// String request = "" +
// "GET / HTTP/1.1\r\n" +
// "Host: localhost:" + proxyAddress.getPort() + "\r\n" +
// "\r\n";
// output1.write(request.getBytes("UTF-8"));
// output1.flush();
//
// InputStream input1 = client1.getInputStream();
// BufferedReader reader1 = new BufferedReader(new InputStreamReader(input1, "UTF-8"));
// String line = reader1.readLine();
// Assert.assertTrue(line.contains(" 200"));
// while (line.length() > 0)
// line = reader1.readLine();
// Assert.assertFalse(reader1.ready());
//
// // Perform another request with another client
// Socket client2 = new Socket();
// client2.connect(proxyAddress);
// OutputStream output2 = client2.getOutputStream();
//
// output2.write(request.getBytes("UTF-8"));
// output2.flush();
//
// InputStream input2 = client2.getInputStream();
// BufferedReader reader2 = new BufferedReader(new InputStreamReader(input2, "UTF-8"));
// line = reader2.readLine();
// Assert.assertTrue(line.contains(" 200"));
// while (line.length() > 0)
// line = reader2.readLine();
// Assert.assertFalse(reader2.ready());
//
// client1.close();
// client2.close();
// }
//
// @Test
// public void testGETThenSmallResponseContent() throws Exception
// {
// final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// Assert.assertTrue(synInfo.isClose());
// Headers requestHeaders = synInfo.getHeaders();
// Assert.assertNotNull(requestHeaders.get("via"));
//
// Headers responseHeaders = new Headers();
// responseHeaders.put(HTTPSPDYHeader.VERSION.name(version()), "HTTP/1.1");
// responseHeaders.put(HTTPSPDYHeader.STATUS.name(version()), "200 OK");
// ReplyInfo replyInfo = new ReplyInfo(responseHeaders, false);
// stream.reply(replyInfo);
// stream.data(new BytesDataInfo(data, true));
//
// return null;
// }
// }));
//
// Socket client = new Socket();
// client.connect(proxyAddress);
// OutputStream output = client.getOutputStream();
//
// String request = "" +
// "GET / 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"));
// while (line.length() > 0)
// line = reader.readLine();
// for (byte datum : data)
// Assert.assertEquals(datum, reader.read());
// Assert.assertFalse(reader.ready());
//
// // Perform another request so that we are sure we reset the states of parsers and generators
// output.write(request.getBytes("UTF-8"));
// output.flush();
//
// line = reader.readLine();
// Assert.assertTrue(line.contains(" 200"));
// while (line.length() > 0)
// line = reader.readLine();
// for (byte datum : data)
// Assert.assertEquals(datum, reader.read());
// Assert.assertFalse(reader.ready());
//
// client.close();
// }
//
// @Test
// public void testPOSTWithSmallRequestContentThenRedirect() throws Exception
// {
// final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// return new StreamFrameListener.Adapter()
// {
// @Override
// public void onData(Stream stream, DataInfo dataInfo)
// {
// dataInfo.consume(dataInfo.length());
// if (dataInfo.isClose())
// {
// Headers headers = new Headers();
// headers.put(HTTPSPDYHeader.VERSION.name(version()), "HTTP/1.1");
// headers.put(HTTPSPDYHeader.STATUS.name(version()), "303 See Other");
// stream.reply(new ReplyInfo(headers, true));
// }
// }
// };
// }
// }));
//
// Socket client = new Socket();
// client.connect(proxyAddress);
// OutputStream output = client.getOutputStream();
//
// String request = "" +
// "POST / HTTP/1.1\r\n" +
// "Host: localhost:" + proxyAddress.getPort() + "\r\n" +
// "Content-Length: " + data.length + "\r\n" +
// "Content-Type: application/octet-stream\r\n" +
// "\r\n";
// output.write(request.getBytes("UTF-8"));
// output.write(data);
// output.flush();
//
// InputStream input = client.getInputStream();
// BufferedReader reader = new BufferedReader(new InputStreamReader(input, "UTF-8"));
// String line = reader.readLine();
// Assert.assertTrue(line.contains(" 303"));
// while (line.length() > 0)
// line = reader.readLine();
// Assert.assertFalse(reader.ready());
//
// // Perform another request so that we are sure we reset the states of parsers and generators
// output.write(request.getBytes("UTF-8"));
// output.write(data);
// output.flush();
//
// line = reader.readLine();
// Assert.assertTrue(line.contains(" 303"));
// while (line.length() > 0)
// line = reader.readLine();
// Assert.assertFalse(reader.ready());
//
// client.close();
// }
//
// @Test
// public void testPOSTWithSmallRequestContentThenSmallResponseContent() throws Exception
// {
// final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// return new StreamFrameListener.Adapter()
// {
// @Override
// public void onData(Stream stream, DataInfo dataInfo)
// {
// dataInfo.consume(dataInfo.length());
// if (dataInfo.isClose())
// {
// Headers responseHeaders = new Headers();
// responseHeaders.put(HTTPSPDYHeader.VERSION.name(version()), "HTTP/1.1");
// responseHeaders.put(HTTPSPDYHeader.STATUS.name(version()), "200 OK");
// ReplyInfo replyInfo = new ReplyInfo(responseHeaders, false);
// stream.reply(replyInfo);
// stream.data(new BytesDataInfo(data, true));
// }
// }
// };
// }
// }));
//
// Socket client = new Socket();
// client.connect(proxyAddress);
// OutputStream output = client.getOutputStream();
//
// String request = "" +
// "POST / HTTP/1.1\r\n" +
// "Host: localhost:" + proxyAddress.getPort() + "\r\n" +
// "Content-Length: " + data.length + "\r\n" +
// "Content-Type: application/octet-stream\r\n" +
// "\r\n";
// output.write(request.getBytes("UTF-8"));
// output.write(data);
// output.flush();
//
// InputStream input = client.getInputStream();
// BufferedReader reader = new BufferedReader(new InputStreamReader(input, "UTF-8"));
// String line = reader.readLine();
// Assert.assertTrue(line.contains(" 200"));
// while (line.length() > 0)
// line = reader.readLine();
// for (byte datum : data)
// Assert.assertEquals(datum, reader.read());
// Assert.assertFalse(reader.ready());
//
// // Perform another request so that we are sure we reset the states of parsers and generators
// output.write(request.getBytes("UTF-8"));
// output.write(data);
// output.flush();
//
// line = reader.readLine();
// Assert.assertTrue(line.contains(" 200"));
// while (line.length() > 0)
// line = reader.readLine();
// for (byte datum : data)
// Assert.assertEquals(datum, reader.read());
// Assert.assertFalse(reader.ready());
//
// client.close();
// }
//
// @Test
// public void testSYNThenREPLY() throws Exception
// {
// final String header = "foo";
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// Headers requestHeaders = synInfo.getHeaders();
// Assert.assertNotNull(requestHeaders.get("via"));
// Assert.assertNotNull(requestHeaders.get(header));
//
// Headers responseHeaders = new Headers();
// responseHeaders.put(header, "baz");
// stream.reply(new ReplyInfo(responseHeaders, true));
// return null;
// }
// }));
// proxyConnector.setDefaultAsyncConnectionFactory(proxyConnector.getAsyncConnectionFactory("spdy/" + version()));
//
// Session client = factory.newSPDYClient(version()).connect(proxyAddress, null).get(5, TimeUnit.SECONDS);
//
// final CountDownLatch replyLatch = new CountDownLatch(1);
// Headers headers = new Headers();
// headers.put(HTTPSPDYHeader.HOST.name(version()), "localhost:" + proxyAddress.getPort());
// headers.put(header, "bar");
// client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
// {
// @Override
// public void onReply(Stream stream, ReplyInfo replyInfo)
// {
// Headers headers = replyInfo.getHeaders();
// Assert.assertNotNull(headers.get(header));
// replyLatch.countDown();
// }
// });
//
// Assert.assertTrue(replyLatch.await(5, TimeUnit.SECONDS));
//
// client.goAway().get(5, TimeUnit.SECONDS);
// }
//
// @Test
// public void testSYNThenREPLYAndDATA() throws Exception
// {
// final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
// final String header = "foo";
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// Headers requestHeaders = synInfo.getHeaders();
// Assert.assertNotNull(requestHeaders.get("via"));
// Assert.assertNotNull(requestHeaders.get(header));
//
// Headers responseHeaders = new Headers();
// responseHeaders.put(header, "baz");
// stream.reply(new ReplyInfo(responseHeaders, false));
// stream.data(new BytesDataInfo(data, true));
// return null;
// }
// }));
// proxyConnector.setDefaultAsyncConnectionFactory(proxyConnector.getAsyncConnectionFactory("spdy/" + version()));
//
// Session client = factory.newSPDYClient(version()).connect(proxyAddress, null).get(5, TimeUnit.SECONDS);
//
// final CountDownLatch replyLatch = new CountDownLatch(1);
// final CountDownLatch dataLatch = new CountDownLatch(1);
// Headers headers = new Headers();
// headers.put(HTTPSPDYHeader.HOST.name(version()), "localhost:" + proxyAddress.getPort());
// headers.put(header, "bar");
// client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
// {
// private final ByteArrayOutputStream result = new ByteArrayOutputStream();
//
// @Override
// public void onReply(Stream stream, ReplyInfo replyInfo)
// {
// Headers headers = replyInfo.getHeaders();
// Assert.assertNotNull(headers.get(header));
// replyLatch.countDown();
// }
//
// @Override
// public void onData(Stream stream, DataInfo dataInfo)
// {
// result.write(dataInfo.asBytes(true), 0, dataInfo.length());
// if (dataInfo.isClose())
// {
// Assert.assertArrayEquals(data, result.toByteArray());
// dataLatch.countDown();
// }
// }
// });
//
// Assert.assertTrue(replyLatch.await(5, TimeUnit.SECONDS));
// Assert.assertTrue(dataLatch.await(5, TimeUnit.SECONDS));
//
// client.goAway().get(5, TimeUnit.SECONDS);
// }
//
// @Test
// public void testGETThenSPDYPushIsIgnored() throws Exception
// {
// final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// Headers responseHeaders = new Headers();
// responseHeaders.put(HTTPSPDYHeader.VERSION.name(version()), "HTTP/1.1");
// responseHeaders.put(HTTPSPDYHeader.STATUS.name(version()), "200 OK");
//
// Headers pushHeaders = new Headers();
// pushHeaders.put(HTTPSPDYHeader.URI.name(version()), "/push");
// stream.syn(new SynInfo(pushHeaders, false), 5, TimeUnit.SECONDS, new Handler.Adapter<Stream>()
// {
// @Override
// public void completed(Stream pushStream)
// {
// pushStream.data(new BytesDataInfo(data, true));
// }
// });
//
// stream.reply(new ReplyInfo(responseHeaders, true));
// return null;
// }
// }));
//
// Socket client = new Socket();
// client.connect(proxyAddress);
// OutputStream output = client.getOutputStream();
//
// String request = "" +
// "GET / HTTP/1.1\r\n" +
// "Host: localhost:" + proxyAddress.getPort() + "\r\n" +
// "\r\n";
// output.write(request.getBytes("UTF-8"));
// output.flush();
//
// client.setSoTimeout(1000);
// InputStream input = client.getInputStream();
// BufferedReader reader = new BufferedReader(new InputStreamReader(input, "UTF-8"));
// String line = reader.readLine();
// Assert.assertTrue(line.contains(" 200"));
// while (line.length() > 0)
// line = reader.readLine();
// Assert.assertFalse(reader.ready());
//
// client.close();
// }
//
// @Test
// public void testSYNThenSPDYPushIsReceived() throws Exception
// {
// final byte[] data = "0123456789ABCDEF".getBytes("UTF-8");
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// Headers responseHeaders = new Headers();
// responseHeaders.put(HTTPSPDYHeader.VERSION.name(version()), "HTTP/1.1");
// responseHeaders.put(HTTPSPDYHeader.STATUS.name(version()), "200 OK");
// stream.reply(new ReplyInfo(responseHeaders, false));
//
// Headers pushHeaders = new Headers();
// pushHeaders.put(HTTPSPDYHeader.URI.name(version()), "/push");
// stream.syn(new SynInfo(pushHeaders, false), 5, TimeUnit.SECONDS, new Handler.Adapter<Stream>()
// {
// @Override
// public void completed(Stream pushStream)
// {
// pushStream.data(new BytesDataInfo(data, true));
// }
// });
//
// stream.data(new BytesDataInfo(data, true));
//
// return null;
// }
// }));
// proxyConnector.setDefaultAsyncConnectionFactory(proxyConnector.getAsyncConnectionFactory("spdy/" + version()));
//
// final CountDownLatch pushSynLatch = new CountDownLatch(1);
// final CountDownLatch pushDataLatch = new CountDownLatch(1);
// Session client = factory.newSPDYClient(version()).connect(proxyAddress, new SessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// pushSynLatch.countDown();
// return new StreamFrameListener.Adapter()
// {
// @Override
// public void onData(Stream stream, DataInfo dataInfo)
// {
// dataInfo.consume(dataInfo.length());
// if (dataInfo.isClose())
// pushDataLatch.countDown();
// }
// };
// }
// }).get(5, TimeUnit.SECONDS);
//
// Headers headers = new Headers();
// headers.put(HTTPSPDYHeader.HOST.name(version()), "localhost:" + proxyAddress.getPort());
// final CountDownLatch replyLatch = new CountDownLatch(1);
// final CountDownLatch dataLatch = new CountDownLatch(1);
// client.syn(new SynInfo(headers, true), new StreamFrameListener.Adapter()
// {
// @Override
// public void onReply(Stream stream, ReplyInfo replyInfo)
// {
// replyLatch.countDown();
// }
//
// @Override
// public void onData(Stream stream, DataInfo dataInfo)
// {
// dataInfo.consume(dataInfo.length());
// if (dataInfo.isClose())
// dataLatch.countDown();
// }
// });
//
// Assert.assertTrue(replyLatch.await(5, TimeUnit.SECONDS));
// Assert.assertTrue(pushSynLatch.await(5, TimeUnit.SECONDS));
// Assert.assertTrue(pushDataLatch.await(5, TimeUnit.SECONDS));
// Assert.assertTrue(dataLatch.await(5, TimeUnit.SECONDS));
//
// client.goAway().get(5, TimeUnit.SECONDS);
// }
//
// @Test
// public void testPING() throws Exception
// {
// // PING is per hop, and it does not carry the information to which server to ping to
// // We just verify that it works
//
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()));
// proxyConnector.setDefaultAsyncConnectionFactory(proxyConnector.getAsyncConnectionFactory("spdy/" + version()));
//
// final CountDownLatch pingLatch = new CountDownLatch(1);
// Session client = factory.newSPDYClient(version()).connect(proxyAddress, new SessionFrameListener.Adapter()
// {
// @Override
// public void onPing(Session session, PingInfo pingInfo)
// {
// pingLatch.countDown();
// }
// }).get(5, TimeUnit.SECONDS);
//
// client.ping().get(5, TimeUnit.SECONDS);
//
// Assert.assertTrue(pingLatch.await(5, TimeUnit.SECONDS));
//
// client.goAway().get(5, TimeUnit.SECONDS);
// }
//
// @Test
// public void testGETThenReset() throws Exception
// {
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// Assert.assertTrue(synInfo.isClose());
// Headers requestHeaders = synInfo.getHeaders();
// Assert.assertNotNull(requestHeaders.get("via"));
//
// stream.getSession().rst(new RstInfo(stream.getId(), StreamStatus.REFUSED_STREAM));
//
// return null;
// }
// }));
//
// Socket client = new Socket();
// client.connect(proxyAddress);
// OutputStream output = client.getOutputStream();
//
// String request = "" +
// "GET / 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"));
// Assert.assertNull(reader.readLine());
//
// client.close();
// }
//
// @Test
// public void testSYNThenReset() throws Exception
// {
// InetSocketAddress proxyAddress = startProxy(startServer(new ServerSessionFrameListener.Adapter()
// {
// @Override
// public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
// {
// Assert.assertTrue(synInfo.isClose());
// Headers requestHeaders = synInfo.getHeaders();
// Assert.assertNotNull(requestHeaders.get("via"));
//
// stream.getSession().rst(new RstInfo(stream.getId(), StreamStatus.REFUSED_STREAM));
//
// return null;
// }
// }));
// proxyConnector.setDefaultAsyncConnectionFactory(proxyConnector.getAsyncConnectionFactory("spdy/" + version()));
//
// final CountDownLatch resetLatch = new CountDownLatch(1);
// Session client = factory.newSPDYClient(version()).connect(proxyAddress, new SessionFrameListener.Adapter()
// {
// @Override
// public void onRst(Session session, RstInfo rstInfo)
// {
// resetLatch.countDown();
// }
// }).get(5, TimeUnit.SECONDS);
//
// Headers headers = new Headers();
// headers.put(HTTPSPDYHeader.HOST.name(version()), "localhost:" + proxyAddress.getPort());
// client.syn(new SynInfo(headers, true), null);
//
// Assert.assertTrue(resetLatch.await(5, TimeUnit.SECONDS));
//
// client.goAway().get(5, TimeUnit.SECONDS);
// }
}