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

Extracting IBlockheadServerConnection for refactoring prep

This commit is contained in:
Joakim Erdfelt 2015-09-25 14:35:29 -07:00
parent b865b50cb6
commit 8946badbed
17 changed files with 752 additions and 692 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
jetty-websocket/javax-websocket-client-impl/src/test/java/org/eclipse/jetty/websocket/jsr356/DecoderReaderManySmallTest.java
View File

@ -42,7 +42,7 @@ import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.common.frames.TextFrame;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -113,7 +113,7 @@ public class DecoderReaderManySmallTest
private static class EventIdServer implements Runnable
{
private BlockheadServer server;
private ServerConnection sconnection;
private IBlockheadServerConnection sconnection;
private CountDownLatch connectLatch = new CountDownLatch(1);
public EventIdServer(BlockheadServer server)

6
jetty-websocket/javax-websocket-client-impl/src/test/java/org/eclipse/jetty/websocket/jsr356/DecoderReaderTest.java
View File

@ -18,7 +18,7 @@
package org.eclipse.jetty.websocket.jsr356;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.*;
import java.io.BufferedReader;
import java.io.File;
@ -50,7 +50,7 @@ import org.eclipse.jetty.websocket.common.WebSocketFrame;
import org.eclipse.jetty.websocket.common.frames.ContinuationFrame;
import org.eclipse.jetty.websocket.common.frames.TextFrame;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -155,7 +155,7 @@ public class DecoderReaderTest
private static class QuoteServer implements Runnable
{
private BlockheadServer server;
private ServerConnection sconnection;
private IBlockheadServerConnection sconnection;
private CountDownLatch connectLatch = new CountDownLatch(1);
public QuoteServer(BlockheadServer server)

4
jetty-websocket/javax-websocket-client-impl/src/test/java/org/eclipse/jetty/websocket/jsr356/EncoderTest.java
View File

@ -45,7 +45,7 @@ import org.eclipse.jetty.toolchain.test.TestTracker;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -58,7 +58,7 @@ public class EncoderTest
{
private Thread thread;
private BlockheadServer server;
private ServerConnection sconnection;
private IBlockheadServerConnection sconnection;
private CountDownLatch connectLatch = new CountDownLatch(1);
public EchoServer(BlockheadServer server)

6
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/BadNetworkTest.java
View File

@ -26,7 +26,7 @@ import org.eclipse.jetty.toolchain.test.TestTracker;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.StatusCode;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.eclipse.jetty.websocket.common.test.LeakTrackingBufferPoolRule;
import org.junit.After;
import org.junit.Before;
@ -82,7 +82,7 @@ public class BadNetworkTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection ssocket = server.accept();
IBlockheadServerConnection ssocket = server.accept();
ssocket.upgrade();
// Validate that we are connected
@ -110,7 +110,7 @@ public class BadNetworkTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection ssocket = server.accept();
IBlockheadServerConnection ssocket = server.accept();
ssocket.upgrade();
// Validate that we are connected

33
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/ClientCloseTest.java
View File

@ -18,14 +18,7 @@
package org.eclipse.jetty.websocket.client;
import static org.hamcrest.Matchers.allOf;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.Matchers.*;
import java.io.IOException;
import java.lang.reflect.Field;
@ -65,7 +58,7 @@ import org.eclipse.jetty.websocket.common.WebSocketSession;
import org.eclipse.jetty.websocket.common.frames.TextFrame;
import org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.eclipse.jetty.websocket.common.test.IncomingFramesCapture;
import org.eclipse.jetty.websocket.common.test.RawFrameBuilder;
import org.hamcrest.Matcher;
@ -194,7 +187,7 @@ public class ClientCloseTest
private BlockheadServer server;
private WebSocketClient client;
private void confirmConnection(CloseTrackingSocket clientSocket, Future<Session> clientFuture, ServerConnection serverConn) throws Exception
private void confirmConnection(CloseTrackingSocket clientSocket, Future<Session> clientFuture, IBlockheadServerConnection serverConns) throws Exception
{
// Wait for client connect on via future
clientFuture.get(500,TimeUnit.MILLISECONDS);
@ -212,7 +205,7 @@ public class ClientCloseTest
testFut.get(500,TimeUnit.MILLISECONDS);
// Read Frame on server side
IncomingFramesCapture serverCapture = serverConn.readFrames(1,500,TimeUnit.MILLISECONDS);
IncomingFramesCapture serverCapture = serverConns.readFrames(1,500,TimeUnit.MILLISECONDS);
serverCapture.assertNoErrors();
serverCapture.assertFrameCount(1);
WebSocketFrame frame = serverCapture.getFrames().poll();
@ -220,7 +213,7 @@ public class ClientCloseTest
Assert.assertThat("Server received frame payload",frame.getPayloadAsUTF8(),is(echoMsg));
// Server send echo reply
serverConn.write(new TextFrame().setPayload(echoMsg));
serverConns.write(new TextFrame().setPayload(echoMsg));
// Wait for received echo
clientSocket.messageQueue.awaitEventCount(1,1,TimeUnit.SECONDS);
@ -238,7 +231,7 @@ public class ClientCloseTest
}
}
private void confirmServerReceivedCloseFrame(ServerConnection serverConn, int expectedCloseCode, Matcher<String> closeReasonMatcher) throws IOException,
private void confirmServerReceivedCloseFrame(IBlockheadServerConnection serverConn, int expectedCloseCode, Matcher<String> closeReasonMatcher) throws IOException,
TimeoutException
{
IncomingFramesCapture serverCapture = serverConn.readFrames(1,500,TimeUnit.MILLISECONDS);
@ -355,7 +348,7 @@ public class ClientCloseTest
Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
// Server accepts connect
ServerConnection serverConn = server.accept();
IBlockheadServerConnection serverConn = server.accept();
serverConn.upgrade();
// client confirms connection via echo
@ -404,7 +397,7 @@ public class ClientCloseTest
Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
// Server accepts connect
ServerConnection serverConn = server.accept();
IBlockheadServerConnection serverConn = server.accept();
serverConn.upgrade();
// client confirms connection via echo
@ -455,7 +448,7 @@ public class ClientCloseTest
Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
// Server accepts connect
ServerConnection serverConn = server.accept();
IBlockheadServerConnection serverConn = server.accept();
serverConn.upgrade();
// client confirms connection via echo
@ -503,7 +496,7 @@ public class ClientCloseTest
Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
// Server accepts connect
ServerConnection serverConn = server.accept();
IBlockheadServerConnection serverConn = server.accept();
serverConn.upgrade();
// client confirms connection via echo
@ -539,7 +532,7 @@ public class ClientCloseTest
Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
// Server accepts connect
ServerConnection serverConn = server.accept();
IBlockheadServerConnection serverConn = server.accept();
serverConn.upgrade();
// client confirms connection via echo
@ -571,7 +564,7 @@ public class ClientCloseTest
int clientCount = 3;
CloseTrackingSocket clientSockets[] = new CloseTrackingSocket[clientCount];
ServerConnection serverConns[] = new ServerConnection[clientCount];
IBlockheadServerConnection serverConns[] = new IBlockheadServerConnection[clientCount];
// Connect Multiple Clients
for (int i = 0; i < clientCount; i++)
@ -617,7 +610,7 @@ public class ClientCloseTest
Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
// Server accepts connect
ServerConnection serverConn = server.accept();
IBlockheadServerConnection serverConn = server.accept();
serverConn.upgrade();
// client confirms connection via echo

21
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/ClientConnectTest.java
View File

@ -18,10 +18,7 @@
package org.eclipse.jetty.websocket.client;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.*;
import java.io.IOException;
import java.net.ConnectException;
@ -39,7 +36,7 @@ import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.UpgradeException;
import org.eclipse.jetty.websocket.common.AcceptHash;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.eclipse.jetty.websocket.common.test.LeakTrackingBufferPoolRule;
import org.junit.After;
import org.junit.Assert;
@ -121,7 +118,7 @@ public class ClientConnectTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection connection = server.accept();
IBlockheadServerConnection connection = server.accept();
connection.readRequest();
// no upgrade, just fail with a 404 error
connection.respond("HTTP/1.1 404 NOT FOUND\r\n\r\n");
@ -150,7 +147,7 @@ public class ClientConnectTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection connection = server.accept();
IBlockheadServerConnection connection = server.accept();
connection.readRequest();
// Send OK to GET but not upgrade
connection.respond("HTTP/1.1 200 OK\r\n\r\n");
@ -179,7 +176,7 @@ public class ClientConnectTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection connection = server.accept();
IBlockheadServerConnection connection = server.accept();
List<String> requestLines = connection.readRequestLines();
String key = connection.parseWebSocketKey(requestLines);
@ -215,7 +212,7 @@ public class ClientConnectTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection connection = server.accept();
IBlockheadServerConnection connection = server.accept();
List<String> requestLines = connection.readRequestLines();
String key = connection.parseWebSocketKey(requestLines);
@ -251,7 +248,7 @@ public class ClientConnectTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection connection = server.accept();
IBlockheadServerConnection connection = server.accept();
List<String> requestLines = connection.readRequestLines();
String key = connection.parseWebSocketKey(requestLines);
@ -287,7 +284,7 @@ public class ClientConnectTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection connection = server.accept();
IBlockheadServerConnection connection = server.accept();
connection.readRequest();
// Upgrade badly
connection.respond("HTTP/1.1 101 Upgrade\r\n" + "Sec-WebSocket-Accept: rubbish\r\n" + "\r\n");
@ -381,7 +378,7 @@ public class ClientConnectTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection ssocket = server.accept();
IBlockheadServerConnection ssocket = server.accept();
Assert.assertNotNull(ssocket);
// Intentionally don't upgrade
// ssocket.upgrade();

12
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/CookieTest.java
View File

@ -18,8 +18,8 @@
package org.eclipse.jetty.websocket.client;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertThat;
import static org.hamcrest.Matchers.*;
import static org.junit.Assert.*;
import java.net.CookieManager;
import java.net.HttpCookie;
@ -37,7 +37,7 @@ import org.eclipse.jetty.websocket.api.WebSocketAdapter;
import org.eclipse.jetty.websocket.api.util.QuoteUtil;
import org.eclipse.jetty.websocket.common.frames.TextFrame;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -119,7 +119,7 @@ public class CookieTest
Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri());
// Server accepts connect
ServerConnection serverConn = server.accept();
IBlockheadServerConnection serverConn = server.accept();
// client confirms upgrade and receipt of frame
String serverCookies = confirmClientUpgradeAndCookies(clientSocket,clientConnectFuture,serverConn);
@ -144,7 +144,7 @@ public class CookieTest
Future<Session> clientConnectFuture = client.connect(clientSocket,server.getWsUri(),request);
// Server accepts connect
ServerConnection serverConn = server.accept();
IBlockheadServerConnection serverConn = server.accept();
// client confirms upgrade and receipt of frame
String serverCookies = confirmClientUpgradeAndCookies(clientSocket,clientConnectFuture,serverConn);
@ -152,7 +152,7 @@ public class CookieTest
Assert.assertThat("Cookies seen at server side",serverCookies,containsString("hello=\"world\""));
}
private String confirmClientUpgradeAndCookies(CookieTrackingSocket clientSocket, Future<Session> clientConnectFuture, ServerConnection serverConn)
private String confirmClientUpgradeAndCookies(CookieTrackingSocket clientSocket, Future<Session> clientConnectFuture, IBlockheadServerConnection serverConn)
throws Exception
{
// Server upgrades

10
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/ServerReadThread.java
View File

@ -18,7 +18,7 @@
package org.eclipse.jetty.websocket.client;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.*;
import java.io.IOException;
import java.nio.ByteBuffer;
@ -34,22 +34,22 @@ import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.common.CloseInfo;
import org.eclipse.jetty.websocket.common.OpCode;
import org.eclipse.jetty.websocket.common.WebSocketFrame;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.Assert;
public class ServerReadThread extends Thread
{
private static final int BUFFER_SIZE = 8192;
private static final Logger LOG = Log.getLogger(ServerReadThread.class);
private final ServerConnection conn;
private final IBlockheadServerConnection conn;
private boolean active = true;
private int slowness = -1; // disabled is default
private final AtomicInteger frameCount = new AtomicInteger();
private final CountDownLatch expectedMessageCount;
public ServerReadThread(ServerConnection conn, int expectedMessages)
public ServerReadThread(IBlockheadServerConnection sconnection, int expectedMessages)
{
this.conn = conn;
this.conn = sconnection;
this.expectedMessageCount = new CountDownLatch(expectedMessages);
}

6
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/ServerWriteThread.java
View File

@ -25,17 +25,17 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.common.frames.TextFrame;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
public class ServerWriteThread extends Thread
{
private static final Logger LOG = Log.getLogger(ServerWriteThread.class);
private final ServerConnection conn;
private final IBlockheadServerConnection conn;
private int slowness = -1;
private int messageCount = 100;
private String message = "Hello";
public ServerWriteThread(ServerConnection conn)
public ServerWriteThread(IBlockheadServerConnection conn)
{
this.conn = conn;
}

7
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/SessionTest.java
View File

@ -18,8 +18,7 @@
package org.eclipse.jetty.websocket.client;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.*;
import java.net.URI;
import java.util.Set;
@ -31,7 +30,7 @@ import org.eclipse.jetty.websocket.api.RemoteEndpoint;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.common.WebSocketSession;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -70,7 +69,7 @@ public class SessionTest
request.setSubProtocols("echo");
Future<Session> future = client.connect(cliSock,wsUri,request);
final ServerConnection srvSock = server.accept();
final IBlockheadServerConnection srvSock = server.accept();
srvSock.upgrade();
Session sess = future.get(500,TimeUnit.MILLISECONDS);

6
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/SlowClientTest.java
View File

@ -18,7 +18,7 @@
package org.eclipse.jetty.websocket.client;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.*;
import java.net.URI;
import java.util.concurrent.Future;
@ -29,7 +29,7 @@ import org.eclipse.jetty.toolchain.test.annotation.Slow;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.StatusCode;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -81,7 +81,7 @@ public class SlowClientTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(tsocket, wsUri);
ServerConnection sconnection = server.accept();
IBlockheadServerConnection sconnection = server.accept();
sconnection.setSoTimeout(60000);
sconnection.upgrade();

8
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/SlowServerTest.java
View File

@ -18,7 +18,7 @@
package org.eclipse.jetty.websocket.client;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.*;
import java.net.URI;
import java.util.concurrent.Future;
@ -30,7 +30,7 @@ import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.StatusCode;
import org.eclipse.jetty.websocket.client.masks.ZeroMasker;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -83,7 +83,7 @@ public class SlowServerTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(tsocket,wsUri);
ServerConnection sconnection = server.accept();
IBlockheadServerConnection sconnection = server.accept();
sconnection.setSoTimeout(60000);
sconnection.upgrade();
@ -130,7 +130,7 @@ public class SlowServerTest
URI wsUri = server.getWsUri();
Future<Session> clientConnectFuture = client.connect(clientSocket,wsUri);
ServerConnection serverConn = server.accept();
IBlockheadServerConnection serverConn = server.accept();
serverConn.setSoTimeout(60000);
serverConn.upgrade();

8
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/TomcatServerQuirksTest.java
View File

@ -18,7 +18,6 @@
package org.eclipse.jetty.websocket.client;
import java.io.IOException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.Arrays;
@ -29,7 +28,7 @@ import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.WebSocketAdapter;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.Assert;
import org.junit.Test;
@ -92,7 +91,7 @@ public class TomcatServerQuirksTest
client.connect(websocket,wsURI);
// Accept incoming connection
ServerConnection socket = server.accept();
IBlockheadServerConnection socket = server.accept();
socket.setSoTimeout(2000); // timeout
// Issue upgrade
@ -114,8 +113,7 @@ public class TomcatServerQuirksTest
serverFrame.put((byte)(payload.length & 0xFF)); // second length byte
serverFrame.put(payload);
BufferUtil.flipToFlush(serverFrame,0);
byte buf[] = BufferUtil.toArray(serverFrame);
socket.write(buf,0,buf.length);
socket.write(serverFrame);
socket.flush();
Assert.assertTrue(websocket.dataLatch.await(1000,TimeUnit.SECONDS));

16
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/WebSocketClientTest.java
View File

@ -40,7 +40,7 @@ import org.eclipse.jetty.websocket.api.UpgradeRequest;
import org.eclipse.jetty.websocket.common.frames.TextFrame;
import org.eclipse.jetty.websocket.common.io.FutureWriteCallback;
import org.eclipse.jetty.websocket.common.test.BlockheadServer;
import org.eclipse.jetty.websocket.common.test.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -106,7 +106,7 @@ public class WebSocketClientTest
request.setSubProtocols("echo");
Future<Session> future = client.connect(cliSock,wsUri,request);
final ServerConnection srvSock = server.accept();
final IBlockheadServerConnection srvSock = server.accept();
srvSock.upgrade();
Session sess = future.get(500,TimeUnit.MILLISECONDS);
@ -152,7 +152,7 @@ public class WebSocketClientTest
request.setSubProtocols("echo");
Future<Session> future = client.connect(cliSock,wsUri,request);
final ServerConnection srvSock = server.accept();
final IBlockheadServerConnection srvSock = server.accept();
srvSock.upgrade();
Session sess = future.get(500,TimeUnit.MILLISECONDS);
@ -188,7 +188,7 @@ public class WebSocketClientTest
Future<Session> future = client.connect(wsocket,server.getWsUri());
// Server
final ServerConnection srvSock = server.accept();
final IBlockheadServerConnection srvSock = server.accept();
srvSock.upgrade();
// Validate connect
@ -226,7 +226,7 @@ public class WebSocketClientTest
URI wsUri = server.getWsUri();
Future<Session> future = fact.connect(wsocket,wsUri);
ServerConnection ssocket = server.accept();
IBlockheadServerConnection ssocket = server.accept();
ssocket.upgrade();
future.get(500,TimeUnit.MILLISECONDS);
@ -266,7 +266,7 @@ public class WebSocketClientTest
URI wsUri = server.getWsUri();
Future<Session> future = factSmall.connect(wsocket,wsUri);
ServerConnection ssocket = server.accept();
IBlockheadServerConnection ssocket = server.accept();
ssocket.upgrade();
future.get(500,TimeUnit.MILLISECONDS);
@ -304,7 +304,7 @@ public class WebSocketClientTest
URI wsUri = server.getWsUri();
Future<Session> future = client.connect(wsocket,wsUri);
ServerConnection ssocket = server.accept();
IBlockheadServerConnection ssocket = server.accept();
ssocket.upgrade();
wsocket.awaitConnect(1,TimeUnit.SECONDS);
@ -346,7 +346,7 @@ public class WebSocketClientTest
URI wsUri = server.getWsUri().resolve("/test?snack=cashews&amount=handful&brand=off");
Future<Session> future = fact.connect(wsocket,wsUri);
ServerConnection ssocket = server.accept();
IBlockheadServerConnection ssocket = server.accept();
ssocket.upgrade();
future.get(500,TimeUnit.MILLISECONDS);

615
jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/test/BlockheadServer.java
View File

@ -18,56 +18,17 @@
package org.eclipse.jetty.websocket.common.test;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.*;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.io.MappedByteBufferPool;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.IO;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.api.BatchMode;
import org.eclipse.jetty.websocket.api.WebSocketPolicy;
import org.eclipse.jetty.websocket.api.WriteCallback;
import org.eclipse.jetty.websocket.api.extensions.ExtensionConfig;
import org.eclipse.jetty.websocket.api.extensions.Frame;
import org.eclipse.jetty.websocket.api.extensions.Frame.Type;
import org.eclipse.jetty.websocket.api.extensions.IncomingFrames;
import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
import org.eclipse.jetty.websocket.common.AcceptHash;
import org.eclipse.jetty.websocket.common.CloseInfo;
import org.eclipse.jetty.websocket.common.Generator;
import org.eclipse.jetty.websocket.common.OpCode;
import org.eclipse.jetty.websocket.common.Parser;
import org.eclipse.jetty.websocket.common.WebSocketFrame;
import org.eclipse.jetty.websocket.common.extensions.ExtensionStack;
import org.eclipse.jetty.websocket.common.extensions.WebSocketExtensionFactory;
import org.eclipse.jetty.websocket.common.frames.CloseFrame;
import org.eclipse.jetty.websocket.common.scopes.SimpleContainerScope;
import org.junit.Assert;
/**
@ -77,550 +38,16 @@ import org.junit.Assert;
*/
public class BlockheadServer
{
public static class ServerConnection implements IncomingFrames, OutgoingFrames, Runnable
{
private final int BUFFER_SIZE = 8192;
private final Socket socket;
private final ByteBufferPool bufferPool;
private final WebSocketPolicy policy;
private final IncomingFramesCapture incomingFrames;
private final Parser parser;
private final Generator generator;
private final AtomicInteger parseCount;
private final WebSocketExtensionFactory extensionRegistry;
private final AtomicBoolean echoing = new AtomicBoolean(false);
private Thread echoThread;
/** Set to true to disable timeouts (for debugging reasons) */
private boolean debug = false;
private OutputStream out;
private InputStream in;
private Map<String, String> extraResponseHeaders = new HashMap<>();
private OutgoingFrames outgoing = this;
public ServerConnection(Socket socket)
{
this.socket = socket;
this.incomingFrames = new IncomingFramesCapture();
this.policy = WebSocketPolicy.newServerPolicy();
this.policy.setMaxBinaryMessageSize(100000);
this.policy.setMaxTextMessageSize(100000);
// This is a blockhead server connection, no point tracking leaks on this object.
this.bufferPool = new MappedByteBufferPool(BUFFER_SIZE);
this.parser = new Parser(policy,bufferPool);
this.parseCount = new AtomicInteger(0);
this.generator = new Generator(policy,bufferPool,false);
this.extensionRegistry = new WebSocketExtensionFactory(new SimpleContainerScope(policy,bufferPool));
}
/**
* Add an extra header for the upgrade response (from the server). No extra work is done to ensure the key and value are sane for http.
* @param rawkey the raw key
* @param rawvalue the raw value
*/
public void addResponseHeader(String rawkey, String rawvalue)
{
extraResponseHeaders.put(rawkey,rawvalue);
}
public void close() throws IOException
{
write(new CloseFrame());
flush();
}
public void close(int statusCode) throws IOException
{
CloseInfo close = new CloseInfo(statusCode);
write(close.asFrame());
flush();
}
public void disconnect()
{
LOG.debug("disconnect");
IO.close(in);
IO.close(out);
if (socket != null)
{
try
{
socket.close();
}
catch (IOException ignore)
{
/* ignore */
}
}
}
public void echoMessage(int expectedFrames, int timeoutDuration, TimeUnit timeoutUnit) throws IOException, TimeoutException
{
LOG.debug("Echo Frames [expecting {}]",expectedFrames);
IncomingFramesCapture cap = readFrames(expectedFrames,timeoutDuration,timeoutUnit);
// now echo them back.
for (Frame frame : cap.getFrames())
{
write(WebSocketFrame.copy(frame).setMasked(false));
}
}
public void flush() throws IOException
{
getOutputStream().flush();
}
public ByteBufferPool getBufferPool()
{
return bufferPool;
}
public IncomingFramesCapture getIncomingFrames()
{
return incomingFrames;
}
public InputStream getInputStream() throws IOException
{
if (in == null)
{
in = socket.getInputStream();
}
return in;
}
private OutputStream getOutputStream() throws IOException
{
if (out == null)
{
out = socket.getOutputStream();
}
return out;
}
public Parser getParser()
{
return parser;
}
public WebSocketPolicy getPolicy()
{
return policy;
}
@Override
public void incomingError(Throwable e)
{
incomingFrames.incomingError(e);
}
@Override
public void incomingFrame(Frame frame)
{
LOG.debug("incoming({})",frame);
int count = parseCount.incrementAndGet();
if ((count % 10) == 0)
{
LOG.info("Server parsed {} frames",count);
}
incomingFrames.incomingFrame(WebSocketFrame.copy(frame));
if (frame.getOpCode() == OpCode.CLOSE)
{
CloseInfo close = new CloseInfo(frame);
LOG.debug("Close frame: {}",close);
}
Type type = frame.getType();
if (echoing.get() && (type.isData() || type.isContinuation()))
{
try
{
write(WebSocketFrame.copy(frame).setMasked(false));
}
catch (IOException e)
{
LOG.warn(e);
}
}
}
@Override
public void outgoingFrame(Frame frame, WriteCallback callback, BatchMode batchMode)
{
ByteBuffer headerBuf = generator.generateHeaderBytes(frame);
if (LOG.isDebugEnabled())
{
LOG.debug("writing out: {}",BufferUtil.toDetailString(headerBuf));
}
try
{
BufferUtil.writeTo(headerBuf,out);
if (frame.hasPayload())
BufferUtil.writeTo(frame.getPayload(),out);
out.flush();
if (callback != null)
{
callback.writeSuccess();
}
if (frame.getOpCode() == OpCode.CLOSE)
{
disconnect();
}
}
catch (Throwable t)
{
if (callback != null)
{
callback.writeFailed(t);
}
}
}
public List<ExtensionConfig> parseExtensions(List<String> requestLines)
{
List<ExtensionConfig> extensionConfigs = new ArrayList<>();
List<String> hits = regexFind(requestLines, "^Sec-WebSocket-Extensions: (.*)$");
for (String econf : hits)
{
// found extensions
ExtensionConfig config = ExtensionConfig.parse(econf);
extensionConfigs.add(config);
}
return extensionConfigs;
}
public String parseWebSocketKey(List<String> requestLines)
{
List<String> hits = regexFind(requestLines,"^Sec-WebSocket-Key: (.*)$");
if (hits.size() <= 0)
{
return null;
}
Assert.assertThat("Number of Sec-WebSocket-Key headers", hits.size(), is(1));
String key = hits.get(0);
return key;
}
public int read(ByteBuffer buf) throws IOException
{
int len = 0;
while ((in.available() > 0) && (buf.remaining() > 0))
{
buf.put((byte)in.read());
len++;
}
return len;
}
public IncomingFramesCapture readFrames(int expectedCount, int timeoutDuration, TimeUnit timeoutUnit) throws IOException, TimeoutException
{
LOG.debug("Read: waiting for {} frame(s) from client",expectedCount);
int startCount = incomingFrames.size();
ByteBuffer buf = bufferPool.acquire(BUFFER_SIZE,false);
BufferUtil.clearToFill(buf);
try
{
long msDur = TimeUnit.MILLISECONDS.convert(timeoutDuration,timeoutUnit);
long now = System.currentTimeMillis();
long expireOn = now + msDur;
LOG.debug("Now: {} - expireOn: {} ({} ms)",now,expireOn,msDur);
int len = 0;
while (incomingFrames.size() < (startCount + expectedCount))
{
BufferUtil.clearToFill(buf);
len = read(buf);
if (len > 0)
{
LOG.debug("Read {} bytes",len);
BufferUtil.flipToFlush(buf,0);
parser.parse(buf);
}
try
{
TimeUnit.MILLISECONDS.sleep(20);
}
catch (InterruptedException gnore)
{
/* ignore */
}
if (!debug && (System.currentTimeMillis() > expireOn))
{
incomingFrames.dump();
throw new TimeoutException(String.format("Timeout reading all %d expected frames. (managed to only read %d frame(s))",expectedCount,
incomingFrames.size()));
}
}
}
finally
{
bufferPool.release(buf);
}
return incomingFrames;
}
public String readRequest() throws IOException
{
LOG.debug("Reading client request");
StringBuilder request = new StringBuilder();
BufferedReader in = new BufferedReader(new InputStreamReader(getInputStream()));
for (String line = in.readLine(); line != null; line = in.readLine())
{
if (line.length() == 0)
{
break;
}
request.append(line).append("\r\n");
LOG.debug("read line: {}",line);
}
LOG.debug("Client Request:{}{}","\n",request);
return request.toString();
}
public List<String> readRequestLines() throws IOException
{
LOG.debug("Reading client request header");
List<String> lines = new ArrayList<>();
BufferedReader in = new BufferedReader(new InputStreamReader(getInputStream()));
for (String line = in.readLine(); line != null; line = in.readLine())
{
if (line.length() == 0)
{
break;
}
lines.add(line);
}
return lines;
}
public List<String> regexFind(List<String> lines, String pattern)
{
List<String> hits = new ArrayList<>();
Pattern patKey = Pattern.compile(pattern,Pattern.CASE_INSENSITIVE);
Matcher mat;
for (String line : lines)
{
mat = patKey.matcher(line);
if (mat.matches())
{
if (mat.groupCount() >= 1)
{
hits.add(mat.group(1));
}
else
{
hits.add(mat.group(0));
}
}
}
return hits;
}
public void respond(String rawstr) throws IOException
{
LOG.debug("respond(){}{}","\n",rawstr);
getOutputStream().write(rawstr.getBytes());
flush();
}
@Override
public void run()
{
LOG.debug("Entering echo thread");
ByteBuffer buf = bufferPool.acquire(BUFFER_SIZE,false);
BufferUtil.clearToFill(buf);
long readBytes = 0;
try
{
while (echoing.get())
{
BufferUtil.clearToFill(buf);
long len = read(buf);
if (len > 0)
{
readBytes += len;
LOG.debug("Read {} bytes",len);
BufferUtil.flipToFlush(buf,0);
parser.parse(buf);
}
try
{
TimeUnit.MILLISECONDS.sleep(20);
}
catch (InterruptedException gnore)
{
/* ignore */
}
}
}
catch (IOException e)
{
LOG.debug("Exception during echo loop",e);
}
finally
{
LOG.debug("Read {} bytes",readBytes);
bufferPool.release(buf);
}
}
public void setSoTimeout(int ms) throws SocketException
{
socket.setSoTimeout(ms);
}
public void startEcho()
{
if (echoThread != null)
{
throw new IllegalStateException("Echo thread already declared!");
}
echoThread = new Thread(this,"BlockheadServer/Echo");
echoing.set(true);
echoThread.start();
}
public void stopEcho()
{
echoing.set(false);
}
public List<String> upgrade() throws IOException
{
List<String> requestLines = readRequestLines();
List<ExtensionConfig> extensionConfigs = parseExtensions(requestLines);
String key = parseWebSocketKey(requestLines);
LOG.debug("Client Request Extensions: {}",extensionConfigs);
LOG.debug("Client Request Key: {}",key);
Assert.assertThat("Request: Sec-WebSocket-Key",key,notNullValue());
// collect extensions configured in response header
ExtensionStack extensionStack = new ExtensionStack(extensionRegistry);
extensionStack.negotiate(extensionConfigs);
// Start with default routing
extensionStack.setNextIncoming(this);
extensionStack.setNextOutgoing(this);
// Configure Parser / Generator
extensionStack.configure(parser);
extensionStack.configure(generator);
// Start Stack
try
{
extensionStack.start();
}
catch (Exception e)
{
throw new IOException("Unable to start Extension Stack");
}
// Configure Parser
parser.setIncomingFramesHandler(extensionStack);
// Setup Response
StringBuilder resp = new StringBuilder();
resp.append("HTTP/1.1 101 Upgrade\r\n");
resp.append("Connection: upgrade\r\n");
resp.append("Sec-WebSocket-Accept: ");
resp.append(AcceptHash.hashKey(key)).append("\r\n");
if (extensionStack.hasNegotiatedExtensions())
{
// Respond to used extensions
resp.append("Sec-WebSocket-Extensions: ");
boolean delim = false;
for (ExtensionConfig ext : extensionStack.getNegotiatedExtensions())
{
if (delim)
{
resp.append(", ");
}
resp.append(ext.getParameterizedName());
delim = true;
}
resp.append("\r\n");
}
if (extraResponseHeaders.size() > 0)
{
for (Map.Entry<String, String> xheader : extraResponseHeaders.entrySet())
{
resp.append(xheader.getKey());
resp.append(": ");
resp.append(xheader.getValue());
resp.append("\r\n");
}
}
resp.append("\r\n");
// Write Response
LOG.debug("Response: {}",resp.toString());
write(resp.toString().getBytes());
return requestLines;
}
private void write(byte[] bytes) throws IOException
{
getOutputStream().write(bytes);
}
public void write(byte[] buf, int offset, int length) throws IOException
{
getOutputStream().write(buf,offset,length);
}
public void write(Frame frame) throws IOException
{
LOG.debug("write(Frame->{}) to {}",frame,outgoing);
outgoing.outgoingFrame(frame,null,BatchMode.OFF);
}
public void write(int b) throws IOException
{
getOutputStream().write(b);
}
public void write(ByteBuffer buf) throws IOException
{
byte arr[] = BufferUtil.toArray(buf);
if ((arr != null) && (arr.length > 0))
{
getOutputStream().write(arr);
}
}
}
private static final Logger LOG = Log.getLogger(BlockheadServer.class);
private ServerSocket serverSocket;
private URI wsUri;
public ServerConnection accept() throws IOException
public IBlockheadServerConnection accept() throws IOException
{
LOG.debug(".accept()");
assertIsStarted();
Socket socket = serverSocket.accept();
return new ServerConnection(socket);
return new BlockheadServerConnection(socket);
}
private void assertIsStarted()
@ -637,42 +64,6 @@ public class BlockheadServer
return wsUri;
}
public void respondToClient(Socket connection, String serverResponse) throws IOException
{
InputStream in = null;
InputStreamReader isr = null;
BufferedReader buf = null;
OutputStream out = null;
try
{
in = connection.getInputStream();
isr = new InputStreamReader(in);
buf = new BufferedReader(isr);
String line;
while ((line = buf.readLine()) != null)
{
// System.err.println(line);
if (line.length() == 0)
{
// Got the "\r\n" line.
break;
}
}
// System.out.println("[Server-Out] " + serverResponse);
out = connection.getOutputStream();
out.write(serverResponse.getBytes());
out.flush();
}
finally
{
IO.close(buf);
IO.close(isr);
IO.close(in);
IO.close(out);
}
}
public void start() throws IOException
{
InetAddress addr = InetAddress.getByName("localhost");

614
jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/test/BlockheadServerConnection.java Normal file
View File

@ -0,0 +1,614 @@
//
// ========================================================================
// Copyright (c) 1995-2015 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.websocket.common.test;
import static org.hamcrest.Matchers.*;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.io.MappedByteBufferPool;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.IO;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.api.BatchMode;
import org.eclipse.jetty.websocket.api.WebSocketPolicy;
import org.eclipse.jetty.websocket.api.WriteCallback;
import org.eclipse.jetty.websocket.api.extensions.ExtensionConfig;
import org.eclipse.jetty.websocket.api.extensions.Frame;
import org.eclipse.jetty.websocket.api.extensions.IncomingFrames;
import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
import org.eclipse.jetty.websocket.api.extensions.Frame.Type;
import org.eclipse.jetty.websocket.common.AcceptHash;
import org.eclipse.jetty.websocket.common.CloseInfo;
import org.eclipse.jetty.websocket.common.Generator;
import org.eclipse.jetty.websocket.common.OpCode;
import org.eclipse.jetty.websocket.common.Parser;
import org.eclipse.jetty.websocket.common.WebSocketFrame;
import org.eclipse.jetty.websocket.common.extensions.ExtensionStack;
import org.eclipse.jetty.websocket.common.extensions.WebSocketExtensionFactory;
import org.eclipse.jetty.websocket.common.frames.CloseFrame;
import org.eclipse.jetty.websocket.common.scopes.SimpleContainerScope;
import org.junit.Assert;
public class BlockheadServerConnection implements IncomingFrames, OutgoingFrames, Runnable, IBlockheadServerConnection
{
private static final Logger LOG = Log.getLogger(BlockheadServerConnection.class);
private final int BUFFER_SIZE = 8192;
private final Socket socket;
private final ByteBufferPool bufferPool;
private final WebSocketPolicy policy;
private final IncomingFramesCapture incomingFrames;
private final Parser parser;
private final Generator generator;
private final AtomicInteger parseCount;
private final WebSocketExtensionFactory extensionRegistry;
private final AtomicBoolean echoing = new AtomicBoolean(false);
private Thread echoThread;
/** Set to true to disable timeouts (for debugging reasons) */
private boolean debug = false;
private OutputStream out;
private InputStream in;
private Map<String, String> extraResponseHeaders = new HashMap<>();
private OutgoingFrames outgoing = this;
public BlockheadServerConnection(Socket socket)
{
this.socket = socket;
this.incomingFrames = new IncomingFramesCapture();
this.policy = WebSocketPolicy.newServerPolicy();
this.policy.setMaxBinaryMessageSize(100000);
this.policy.setMaxTextMessageSize(100000);
// This is a blockhead server connection, no point tracking leaks on this object.
this.bufferPool = new MappedByteBufferPool(BUFFER_SIZE);
this.parser = new Parser(policy,bufferPool);
this.parseCount = new AtomicInteger(0);
this.generator = new Generator(policy,bufferPool,false);
this.extensionRegistry = new WebSocketExtensionFactory(new SimpleContainerScope(policy,bufferPool));
}
/**
* Add an extra header for the upgrade response (from the server). No extra work is done to ensure the key and value are sane for http.
* @param rawkey the raw key
* @param rawvalue the raw value
*/
public void addResponseHeader(String rawkey, String rawvalue)
{
extraResponseHeaders.put(rawkey,rawvalue);
}
/* (non-Javadoc)
* @see org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection#close()
*/
@Override
public void close() throws IOException
{
write(new CloseFrame());
flush();
}
/* (non-Javadoc)
* @see org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection#close(int)
*/
@Override
public void close(int statusCode) throws IOException
{
CloseInfo close = new CloseInfo(statusCode);
write(close.asFrame());
flush();
}
public void disconnect()
{
LOG.debug("disconnect");
IO.close(in);
IO.close(out);
if (socket != null)
{
try
{
socket.close();
}
catch (IOException ignore)
{
/* ignore */
}
}
}
public void echoMessage(int expectedFrames, int timeoutDuration, TimeUnit timeoutUnit) throws IOException, TimeoutException
{
LOG.debug("Echo Frames [expecting {}]",expectedFrames);
IncomingFramesCapture cap = readFrames(expectedFrames,timeoutDuration,timeoutUnit);
// now echo them back.
for (Frame frame : cap.getFrames())
{
write(WebSocketFrame.copy(frame).setMasked(false));
}
}
public void flush() throws IOException
{
getOutputStream().flush();
}
public ByteBufferPool getBufferPool()
{
return bufferPool;
}
public IncomingFramesCapture getIncomingFrames()
{
return incomingFrames;
}
public InputStream getInputStream() throws IOException
{
if (in == null)
{
in = socket.getInputStream();
}
return in;
}
private OutputStream getOutputStream() throws IOException
{
if (out == null)
{
out = socket.getOutputStream();
}
return out;
}
public Parser getParser()
{
return parser;
}
public WebSocketPolicy getPolicy()
{
return policy;
}
@Override
public void incomingError(Throwable e)
{
incomingFrames.incomingError(e);
}
@Override
public void incomingFrame(Frame frame)
{
LOG.debug("incoming({})",frame);
int count = parseCount.incrementAndGet();
if ((count % 10) == 0)
{
LOG.info("Server parsed {} frames",count);
}
incomingFrames.incomingFrame(WebSocketFrame.copy(frame));
if (frame.getOpCode() == OpCode.CLOSE)
{
CloseInfo close = new CloseInfo(frame);
LOG.debug("Close frame: {}",close);
}
Type type = frame.getType();
if (echoing.get() && (type.isData() || type.isContinuation()))
{
try
{
write(WebSocketFrame.copy(frame).setMasked(false));
}
catch (IOException e)
{
LOG.warn(e);
}
}
}
@Override
public void outgoingFrame(Frame frame, WriteCallback callback, BatchMode batchMode)
{
ByteBuffer headerBuf = generator.generateHeaderBytes(frame);
if (LOG.isDebugEnabled())
{
LOG.debug("writing out: {}",BufferUtil.toDetailString(headerBuf));
}
try
{
BufferUtil.writeTo(headerBuf,out);
if (frame.hasPayload())
BufferUtil.writeTo(frame.getPayload(),out);
out.flush();
if (callback != null)
{
callback.writeSuccess();
}
if (frame.getOpCode() == OpCode.CLOSE)
{
disconnect();
}
}
catch (Throwable t)
{
if (callback != null)
{
callback.writeFailed(t);
}
}
}
public List<ExtensionConfig> parseExtensions(List<String> requestLines)
{
List<ExtensionConfig> extensionConfigs = new ArrayList<>();
List<String> hits = regexFind(requestLines, "^Sec-WebSocket-Extensions: (.*)$");
for (String econf : hits)
{
// found extensions
ExtensionConfig config = ExtensionConfig.parse(econf);
extensionConfigs.add(config);
}
return extensionConfigs;
}
public String parseWebSocketKey(List<String> requestLines)
{
List<String> hits = regexFind(requestLines,"^Sec-WebSocket-Key: (.*)$");
if (hits.size() <= 0)
{
return null;
}
Assert.assertThat("Number of Sec-WebSocket-Key headers", hits.size(), is(1));
String key = hits.get(0);
return key;
}
public int read(ByteBuffer buf) throws IOException
{
int len = 0;
while ((in.available() > 0) && (buf.remaining() > 0))
{
buf.put((byte)in.read());
len++;
}
return len;
}
public IncomingFramesCapture readFrames(int expectedCount, int timeoutDuration, TimeUnit timeoutUnit) throws IOException, TimeoutException
{
LOG.debug("Read: waiting for {} frame(s) from client",expectedCount);
int startCount = incomingFrames.size();
ByteBuffer buf = bufferPool.acquire(BUFFER_SIZE,false);
BufferUtil.clearToFill(buf);
try
{
long msDur = TimeUnit.MILLISECONDS.convert(timeoutDuration,timeoutUnit);
long now = System.currentTimeMillis();
long expireOn = now + msDur;
LOG.debug("Now: {} - expireOn: {} ({} ms)",now,expireOn,msDur);
int len = 0;
while (incomingFrames.size() < (startCount + expectedCount))
{
BufferUtil.clearToFill(buf);
len = read(buf);
if (len > 0)
{
LOG.debug("Read {} bytes",len);
BufferUtil.flipToFlush(buf,0);
parser.parse(buf);
}
try
{
TimeUnit.MILLISECONDS.sleep(20);
}
catch (InterruptedException gnore)
{
/* ignore */
}
if (!debug && (System.currentTimeMillis() > expireOn))
{
incomingFrames.dump();
throw new TimeoutException(String.format("Timeout reading all %d expected frames. (managed to only read %d frame(s))",expectedCount,
incomingFrames.size()));
}
}
}
finally
{
bufferPool.release(buf);
}
return incomingFrames;
}
public String readRequest() throws IOException
{
LOG.debug("Reading client request");
StringBuilder request = new StringBuilder();
BufferedReader in = new BufferedReader(new InputStreamReader(getInputStream()));
for (String line = in.readLine(); line != null; line = in.readLine())
{
if (line.length() == 0)
{
break;
}
request.append(line).append("\r\n");
LOG.debug("read line: {}",line);
}
LOG.debug("Client Request:{}{}","\n",request);
return request.toString();
}
public List<String> readRequestLines() throws IOException
{
LOG.debug("Reading client request header");
List<String> lines = new ArrayList<>();
BufferedReader in = new BufferedReader(new InputStreamReader(getInputStream()));
for (String line = in.readLine(); line != null; line = in.readLine())
{
if (line.length() == 0)
{
break;
}
lines.add(line);
}
return lines;
}
public List<String> regexFind(List<String> lines, String pattern)
{
List<String> hits = new ArrayList<>();
Pattern patKey = Pattern.compile(pattern,Pattern.CASE_INSENSITIVE);
Matcher mat;
for (String line : lines)
{
mat = patKey.matcher(line);
if (mat.matches())
{
if (mat.groupCount() >= 1)
{
hits.add(mat.group(1));
}
else
{
hits.add(mat.group(0));
}
}
}
return hits;
}
public void respond(String rawstr) throws IOException
{
LOG.debug("respond(){}{}","\n",rawstr);
getOutputStream().write(rawstr.getBytes());
flush();
}
@Override
public void run()
{
LOG.debug("Entering echo thread");
ByteBuffer buf = bufferPool.acquire(BUFFER_SIZE,false);
BufferUtil.clearToFill(buf);
long readBytes = 0;
try
{
while (echoing.get())
{
BufferUtil.clearToFill(buf);
long len = read(buf);
if (len > 0)
{
readBytes += len;
LOG.debug("Read {} bytes",len);
BufferUtil.flipToFlush(buf,0);
parser.parse(buf);
}
try
{
TimeUnit.MILLISECONDS.sleep(20);
}
catch (InterruptedException gnore)
{
/* ignore */
}
}
}
catch (IOException e)
{
LOG.debug("Exception during echo loop",e);
}
finally
{
LOG.debug("Read {} bytes",readBytes);
bufferPool.release(buf);
}
}
public void setSoTimeout(int ms) throws SocketException
{
socket.setSoTimeout(ms);
}
public void startEcho()
{
if (echoThread != null)
{
throw new IllegalStateException("Echo thread already declared!");
}
echoThread = new Thread(this,"BlockheadServer/Echo");
echoing.set(true);
echoThread.start();
}
public void stopEcho()
{
echoing.set(false);
}
public List<String> upgrade() throws IOException
{
List<String> requestLines = readRequestLines();
List<ExtensionConfig> extensionConfigs = parseExtensions(requestLines);
String key = parseWebSocketKey(requestLines);
LOG.debug("Client Request Extensions: {}",extensionConfigs);
LOG.debug("Client Request Key: {}",key);
Assert.assertThat("Request: Sec-WebSocket-Key",key,notNullValue());
// collect extensions configured in response header
ExtensionStack extensionStack = new ExtensionStack(extensionRegistry);
extensionStack.negotiate(extensionConfigs);
// Start with default routing
extensionStack.setNextIncoming(this);
extensionStack.setNextOutgoing(this);
// Configure Parser / Generator
extensionStack.configure(parser);
extensionStack.configure(generator);
// Start Stack
try
{
extensionStack.start();
}
catch (Exception e)
{
throw new IOException("Unable to start Extension Stack");
}
// Configure Parser
parser.setIncomingFramesHandler(extensionStack);
// Setup Response
StringBuilder resp = new StringBuilder();
resp.append("HTTP/1.1 101 Upgrade\r\n");
resp.append("Connection: upgrade\r\n");
resp.append("Sec-WebSocket-Accept: ");
resp.append(AcceptHash.hashKey(key)).append("\r\n");
if (extensionStack.hasNegotiatedExtensions())
{
// Respond to used extensions
resp.append("Sec-WebSocket-Extensions: ");
boolean delim = false;
for (ExtensionConfig ext : extensionStack.getNegotiatedExtensions())
{
if (delim)
{
resp.append(", ");
}
resp.append(ext.getParameterizedName());
delim = true;
}
resp.append("\r\n");
}
if (extraResponseHeaders.size() > 0)
{
for (Map.Entry<String, String> xheader : extraResponseHeaders.entrySet())
{
resp.append(xheader.getKey());
resp.append(": ");
resp.append(xheader.getValue());
resp.append("\r\n");
}
}
resp.append("\r\n");
// Write Response
LOG.debug("Response: {}",resp.toString());
write(resp.toString().getBytes());
return requestLines;
}
private void write(byte[] bytes) throws IOException
{
getOutputStream().write(bytes);
}
public void write(byte[] buf, int offset, int length) throws IOException
{
getOutputStream().write(buf,offset,length);
}
/* (non-Javadoc)
* @see org.eclipse.jetty.websocket.common.test.IBlockheadServerConnection#write(org.eclipse.jetty.websocket.api.extensions.Frame)
*/
@Override
public void write(Frame frame) throws IOException
{
LOG.debug("write(Frame->{}) to {}",frame,outgoing);
outgoing.outgoingFrame(frame,null,BatchMode.OFF);
}
public void write(int b) throws IOException
{
getOutputStream().write(b);
}
public void write(ByteBuffer buf) throws IOException
{
byte arr[] = BufferUtil.toArray(buf);
if ((arr != null) && (arr.length > 0))
{
getOutputStream().write(arr);
}
}
}

68
jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/test/IBlockheadServerConnection.java Normal file
View File

@ -0,0 +1,68 @@
//
// ========================================================================
// Copyright (c) 1995-2015 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.websocket.common.test;
import java.io.IOException;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.websocket.api.extensions.Frame;
import org.eclipse.jetty.websocket.common.Parser;
public interface IBlockheadServerConnection
{
public void close() throws IOException;
public void close(int statusCode) throws IOException;
public void write(Frame frame) throws IOException;
public List<String> upgrade() throws IOException;
public void disconnect();
public IncomingFramesCapture readFrames(int expectedCount, int timeoutDuration, TimeUnit timeoutUnit) throws IOException, TimeoutException;
public void write(ByteBuffer buf) throws IOException;
public List<String> readRequestLines() throws IOException;
public String parseWebSocketKey(List<String> requestLines);
public void respond(String rawstr) throws IOException;
public String readRequest() throws IOException;
public List<String> regexFind(List<String> lines, String pattern);
public void echoMessage(int expectedFrames, int timeoutDuration, TimeUnit timeoutUnit) throws IOException, TimeoutException;
public void setSoTimeout(int ms) throws SocketException;
public ByteBufferPool getBufferPool();
public int read(ByteBuffer buf) throws IOException;
public Parser getParser();
public IncomingFramesCapture getIncomingFrames();
public void flush() throws IOException;
public void write(int b) throws IOException;
public void startEcho();
public void stopEcho();
/**
* Add an extra header for the upgrade response (from the server). No extra work is done to ensure the key and value are sane for http.
* @param rawkey the raw key
* @param rawvalue the raw value
*/
public void addResponseHeader(String rawkey, String rawvalue);
}