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

Fixing testBlockReceiving

This commit is contained in:
Joakim Erdfelt 2012-08-22 11:06:57 -07:00
parent 4ed72ad585
commit 28ae24fc35
4 changed files with 185 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/TrackingSocket.java
View File

@ -22,7 +22,9 @@ import static org.hamcrest.Matchers.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.eclipse.jetty.util.BlockingArrayQueue;
import org.eclipse.jetty.util.log.Log;
@ -39,6 +41,7 @@ public class TrackingSocket extends WebSocketAdapter
private static final Logger LOG = Log.getLogger(TrackingSocket.class);
public int closeCode = -1;
public Exchanger<String> messageExchanger;
public StringBuilder closeMessage = new StringBuilder();
public CountDownLatch openLatch = new CountDownLatch(1);
public CountDownLatch closeLatch = new CountDownLatch(1);
@ -59,7 +62,7 @@ public class TrackingSocket extends WebSocketAdapter
private void assertCloseReason(String expectedReason)
{
Assert.assertThat("Close Reaosn",closeMessage.toString(),is(expectedReason));
Assert.assertThat("Close Reason",closeMessage.toString(),is(expectedReason));
}
public void assertIsOpen() throws InterruptedException
@ -89,6 +92,32 @@ public class TrackingSocket extends WebSocketAdapter
Assert.assertThat("Was Opened",openLatch.await(500,TimeUnit.MILLISECONDS),is(true));
}
public void awaitMessage(int expectedMessageCount, TimeUnit timeoutUnit, int timeoutDuration) throws TimeoutException
{
int startCount = messageQueue.size();
long msDur = TimeUnit.MILLISECONDS.convert(timeoutDuration,timeoutUnit);
long now = System.currentTimeMillis();
long expireOn = now + msDur;
LOG.debug("Now: {} - expireOn: {} ({} ms)",now,expireOn,msDur);
while (messageQueue.size() < (startCount + expectedMessageCount))
{
try
{
TimeUnit.MILLISECONDS.sleep(20);
}
catch (InterruptedException gnore)
{
/* ignore */
}
if (!LOG.isDebugEnabled() && (System.currentTimeMillis() > expireOn))
{
throw new TimeoutException(String.format("Timeout reading all %d expected messages. (managed to only read %d messages)",expectedMessageCount,
messageQueue.size()));
}
}
}
@Override
public void onWebSocketBinary(byte[] payload, int offset, int len)
{
@ -99,6 +128,7 @@ public class TrackingSocket extends WebSocketAdapter
@Override
public void onWebSocketClose(int statusCode, String reason)
{
LOG.debug("onWebSocketClose({},{})",statusCode,reason);
super.onWebSocketClose(statusCode,reason);
closeCode = statusCode;
closeMessage.append(reason);
@ -116,8 +146,20 @@ public class TrackingSocket extends WebSocketAdapter
public void onWebSocketText(String message)
{
LOG.debug("onWebSocketText({})",message);
messageQueue.add(message);
messageQueue.offer(message);
dataLatch.countDown();
if (messageExchanger != null)
{
try
{
messageExchanger.exchange(message);
}
catch (InterruptedException e)
{
LOG.debug(e);
}
}
}
public void waitForMessage(TimeUnit timeoutUnit, int timeoutDuration) throws InterruptedException

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

@ -22,13 +22,11 @@ import static org.hamcrest.Matchers.*;
import java.net.ConnectException;
import java.net.URI;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
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.concurrent.atomic.AtomicLong;
@ -37,19 +35,14 @@ import org.eclipse.jetty.websocket.api.StatusCode;
import org.eclipse.jetty.websocket.api.UpgradeException;
import org.eclipse.jetty.websocket.api.UpgradeRequest;
import org.eclipse.jetty.websocket.api.UpgradeResponse;
import org.eclipse.jetty.websocket.api.WebSocketAdapter;
import org.eclipse.jetty.websocket.api.WebSocketConnection;
import org.eclipse.jetty.websocket.api.WebSocketListener;
import org.eclipse.jetty.websocket.client.blockhead.BlockheadServer;
import org.eclipse.jetty.websocket.client.blockhead.BlockheadServer.ServerConnection;
import org.eclipse.jetty.websocket.protocol.WebSocketFrame;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
@Ignore("Work in Progress")
public class WebSocketClientTest
{
private BlockheadServer server;
@ -187,50 +180,19 @@ public class WebSocketClientTest
// Verify connect
future.get(500,TimeUnit.MILLISECONDS);
wsocket.assertWasOpened();
wsocket.awaitMessage(1,TimeUnit.MILLISECONDS,500);
wsocket.assertMessage("Hello world");
wsocket.assertMessage("Hello World");
}
@Test
public void testBlockReceiving() throws Exception
{
final AtomicBoolean open = new AtomicBoolean(false);
final AtomicInteger close = new AtomicInteger();
final CountDownLatch _latch = new CountDownLatch(1);
final StringBuilder closeMessage = new StringBuilder();
final Exchanger<String> exchanger = new Exchanger<String>();
WebSocketListener socket = new WebSocketAdapter()
{
@Override
public void onWebSocketClose(int statusCode, String reason)
{
close.set(statusCode);
closeMessage.append(reason);
_latch.countDown();
}
@Override
public void onWebSocketConnect(WebSocketConnection connection)
{
open.set(true);
}
@Override
public void onWebSocketText(String message)
{
try
{
exchanger.exchange(message);
}
catch (InterruptedException e)
{
// e.printStackTrace();
}
}
};
WebSocketClient client = factory.newWebSocketClient(socket);
TrackingSocket tsocket = new TrackingSocket();
tsocket.messageExchanger = exchanger;
WebSocketClient client = factory.newWebSocketClient(tsocket);
client.getPolicy().setIdleTimeout(60000);
URI wsUri = server.getWsUri();
@ -238,8 +200,9 @@ public class WebSocketClientTest
ServerConnection sconnection = server.accept();
sconnection.setSoTimeout(60000);
sconnection.upgrade();
UpgradeResponse resp = future.get(250,TimeUnit.MILLISECONDS);
future.get(500,TimeUnit.MILLISECONDS);
// define some messages to send server to client
byte[] send = new byte[]
@ -305,15 +268,10 @@ public class WebSocketClientTest
Assert.assertEquals(m.get(),messages);
// Close with code
start = System.currentTimeMillis();
sconnection.write(new byte[]
{ (byte)0x88, (byte)0x02, (byte)4, (byte)87 },0,4);
sconnection.flush();
sconnection.close(StatusCode.NORMAL);
_latch.await(10,TimeUnit.SECONDS);
Assert.assertTrue((System.currentTimeMillis() - start) < 5000);
Assert.assertEquals(1002,close.get());
Assert.assertEquals("Invalid close code 1111",closeMessage.toString());
Assert.assertTrue("Client Socket Closed",tsocket.closeLatch.await(10,TimeUnit.SECONDS));
tsocket.assertCloseCode(StatusCode.NORMAL);
}
@Test
@ -329,7 +287,7 @@ public class WebSocketClientTest
final ServerConnection ssocket = server.accept();
ssocket.upgrade();
UpgradeResponse resp = future.get(250,TimeUnit.MILLISECONDS);
future.get(250,TimeUnit.MILLISECONDS);
final int messages = 200000;
final AtomicLong totalB = new AtomicLong();
@ -592,9 +550,8 @@ public class WebSocketClientTest
wsocket.assertIsOpen();
ssocket.close();
wsocket.openLatch.await(10,TimeUnit.SECONDS);
ssocket.disconnect();
wsocket.assertCloseCode(StatusCode.NO_CLOSE);
Assert.assertThat("Close should have been detected",wsocket.closeLatch.await(10,TimeUnit.SECONDS),is(true));
}
}

136
jetty-websocket/websocket-client/src/test/java/org/eclipse/jetty/websocket/client/blockhead/BlockheadServer.java
View File

@ -32,9 +32,15 @@ import java.net.Socket;
import java.net.SocketException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
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.StandardByteBufferPool;
@ -43,11 +49,15 @@ import org.eclipse.jetty.util.Callback;
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.Extension;
import org.eclipse.jetty.websocket.api.WebSocketException;
import org.eclipse.jetty.websocket.api.WebSocketPolicy;
import org.eclipse.jetty.websocket.extensions.WebSocketExtensionRegistry;
import org.eclipse.jetty.websocket.io.IncomingFrames;
import org.eclipse.jetty.websocket.io.OutgoingFrames;
import org.eclipse.jetty.websocket.protocol.AcceptHash;
import org.eclipse.jetty.websocket.protocol.CloseInfo;
import org.eclipse.jetty.websocket.protocol.ExtensionConfig;
import org.eclipse.jetty.websocket.protocol.Generator;
import org.eclipse.jetty.websocket.protocol.OpCode;
import org.eclipse.jetty.websocket.protocol.Parser;
@ -70,6 +80,7 @@ public class BlockheadServer
private final Parser parser;
private final Generator generator;
private final AtomicInteger parseCount;
private final WebSocketExtensionRegistry extensionRegistry;
/** Set to true to disable timeouts (for debugging reasons) */
private boolean debug = false;
@ -87,12 +98,23 @@ public class BlockheadServer
this.bufferPool = new StandardByteBufferPool(policy.getBufferSize());
this.parser = new Parser(policy);
this.parseCount = new AtomicInteger(0);
this.generator = new Generator(policy,bufferPool);
this.generator = new Generator(policy,bufferPool,false);
this.extensionRegistry = new WebSocketExtensionRegistry(policy,bufferPool);
}
public void close() throws IOException
{
this.socket.close();
write(new WebSocketFrame(OpCode.CLOSE));
flush();
disconnect();
}
public void close(int statusCode) throws IOException
{
CloseInfo close = new CloseInfo(statusCode);
write(close.asFrame());
flush();
disconnect();
}
public void disconnect()
@ -150,13 +172,20 @@ public class BlockheadServer
@Override
public void incoming(WebSocketException e)
{
// TODO Auto-generated method stub
incomingFrames.incoming(e);
}
@Override
public void incoming(WebSocketFrame frame)
{
// TODO Auto-generated method stub
LOG.debug("incoming({})",frame);
int count = parseCount.incrementAndGet();
if ((count % 10) == 0)
{
LOG.info("Server parsed {} frames",count);
}
WebSocketFrame copy = new WebSocketFrame(frame);
incomingFrames.incoming(copy);
}
@Override
@ -269,6 +298,12 @@ public class BlockheadServer
public void upgrade() throws IOException
{
List<ExtensionConfig> extensionConfigs = new ArrayList<>();
Pattern patExts = Pattern.compile("^Sec-WebSocket-Extensions: (.*)$",Pattern.CASE_INSENSITIVE);
Pattern patKey = Pattern.compile("^Sec-WebSocket-Key: (.*)$",Pattern.CASE_INSENSITIVE);
Matcher mat;
String key = "not sent";
BufferedReader in = new BufferedReader(new InputStreamReader(getInputStream()));
for (String line = in.readLine(); line != null; line = in.readLine())
@ -277,27 +312,104 @@ public class BlockheadServer
{
break;
}
if (line.startsWith("Sec-WebSocket-Key:"))
// Check for extensions
mat = patExts.matcher(line);
if (mat.matches())
{
key = line.substring(18).trim();
// found extensions
String econf = mat.group(1);
ExtensionConfig config = ExtensionConfig.parse(econf);
extensionConfigs.add(config);
continue;
}
// Check for Key
mat = patKey.matcher(line);
if (mat.matches())
{
key = mat.group(1);
}
}
// TODO: parse extensions
// Init extensions
List<Extension> extensions = new ArrayList<>();
for (ExtensionConfig config : extensionConfigs)
{
Extension ext = extensionRegistry.newInstance(config);
extensions.add(ext);
}
// TODO: setup extensions
// Start with default routing
incoming = this;
outgoing = this;
// Connect extensions
if (!extensions.isEmpty())
{
Iterator<Extension> extIter;
// Connect outgoings
extIter = extensions.iterator();
while (extIter.hasNext())
{
Extension ext = extIter.next();
ext.setNextOutgoingFrames(outgoing);
outgoing = ext;
// Handle RSV reservations
if (ext.useRsv1())
{
generator.setRsv1InUse(true);
}
if (ext.useRsv2())
{
generator.setRsv2InUse(true);
}
if (ext.useRsv3())
{
generator.setRsv3InUse(true);
}
}
// Connect incomings
Collections.reverse(extensions);
extIter = extensions.iterator();
while (extIter.hasNext())
{
Extension ext = extIter.next();
ext.setNextIncomingFrames(incoming);
incoming = ext;
}
}
// Configure Parser
parser.setIncomingFramesHandler(incoming);
// Setup Response
StringBuilder resp = new StringBuilder();
resp.append("HTTP/1.1 101 Upgrade\r\n");
resp.append("Sec-WebSocket-Accept: ");
resp.append(AcceptHash.hashKey(key)).append("\r\n");
// TODO: respond to used extensions
if (!extensions.isEmpty())
{
// Respond to used extensions
resp.append("Sec-WebSocket-Extensions: ");
boolean delim = false;
for (Extension ext : extensions)
{
if (delim)
{
resp.append(", ");
}
resp.append(ext.getParameterizedName());
delim = true;
}
resp.append("\r\n");
}
resp.append("\r\n");
// Write Response
write(resp.toString().getBytes());
// Configure Parser
parser.setIncomingFramesHandler(incomingFrames);
}
private void write(byte[] bytes) throws IOException

5
jetty-websocket/websocket-client/src/test/resources/jetty-logging.properties
View File

@ -1,8 +1,9 @@
org.eclipse.jetty.util.log.class=org.eclipse.jetty.util.log.StdErrLog
# org.eclipse.jetty.LEVEL=DEBUG
# org.eclipse.jetty.io.ChannelEndPoint.LEVEL=INFO
# org.eclipse.jetty.websocket.LEVEL=WARN
org.eclipse.jetty.websocket.LEVEL=DEBUG
org.eclipse.jetty.websocket.LEVEL=WARN
# org.eclipse.jetty.websocket.LEVEL=DEBUG
org.eclipse.jetty.websocket.client.TrackingSocket.LEVEL=DEBUG
# See the read/write traffic
# org.eclipse.jetty.websocket.io.Frames.LEVEL=DEBUG
# org.eclipse.jetty.websocket.io.LEVEL=DEBUG