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

337685 Work in progress on draft 6 websockets 

git-svn-id: svn+ssh://dev.eclipse.org/svnroot/rt/org.eclipse.jetty/jetty/trunk@2833 7e9141cc-0065-0410-87d8-b60c137991c4
This commit is contained in:
Greg Wilkins 2011-03-01 01:21:54 +00:00
parent 8f09ef1266
commit 75960baeb1
7 changed files with 250 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketConnectionD06.java
View File

@ -261,7 +261,7 @@ public class WebSocketConnectionD06 extends AbstractConnection implements WebSoc
_endp.setMaxIdleTime(maxIdleTime);
_websocket = websocket;
_generator = new WebSocketGeneratorD06(buffers, _endp,false);
_generator = new WebSocketGeneratorD06(buffers, _endp,null);
_parser = new WebSocketParserD06(buffers, endpoint, _frameHandler,true);
// TODO should these be AsyncEndPoint checks/calls?

82
jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketGeneratorD06.java
View File

@ -34,18 +34,78 @@ public class WebSocketGeneratorD06 implements WebSocketGenerator
final private WebSocketBuffers _buffers;
final private EndPoint _endp;
private Buffer _buffer;
private final boolean _masked;
private final byte[] _mask=new byte[4];
private final Random _random;
private int _m;
private boolean _opsent;
private final MaskGen _maskGen;
public WebSocketGeneratorD06(WebSocketBuffers buffers, EndPoint endp, boolean masked)
public interface MaskGen
{
void genMask(byte[] mask);
}
public static class NullMaskGen implements MaskGen
{
public void genMask(byte[] mask)
{
mask[0]=mask[1]=mask[2]=mask[3]=0;
}
}
public static class FixedMaskGen implements MaskGen
{
final byte[] _mask;
public FixedMaskGen()
{
_mask=new byte[]{(byte)0xff,(byte)0xff,(byte)0xff,(byte)0xff};
}
public FixedMaskGen(byte[] mask)
{
_mask=mask;
}
public void genMask(byte[] mask)
{
mask[0]=_mask[0];
mask[1]=_mask[1];
mask[2]=_mask[2];
mask[3]=_mask[3];
}
}
public static class RandomMaskGen implements MaskGen
{
final Random _random;
public RandomMaskGen()
{
_random=new SecureRandom();
}
public RandomMaskGen(Random random)
{
_random=random;
}
public void genMask(byte[] mask)
{
_random.nextBytes(mask);
}
}
public WebSocketGeneratorD06(WebSocketBuffers buffers, EndPoint endp)
{
_buffers=buffers;
_endp=endp;
_masked=masked;
_random=_masked?new SecureRandom():null; // TODO share the Random
_maskGen=null;
}
public WebSocketGeneratorD06(WebSocketBuffers buffers, EndPoint endp, MaskGen maskGen)
{
_buffers=buffers;
_endp=endp;
_maskGen=maskGen;
}
public synchronized void addFrame(byte opcode,byte[] content, int blockFor) throws IOException
@ -64,9 +124,9 @@ public class WebSocketGeneratorD06 implements WebSocketGenerator
public synchronized void addFragment(boolean last, byte opcode, byte[] content, int offset, int length, int blockFor) throws IOException
{
if (_buffer==null)
_buffer=_masked?_buffers.getBuffer():_buffers.getDirectBuffer();
_buffer=(_maskGen!=null)?_buffers.getBuffer():_buffers.getDirectBuffer();
int space=_masked?14:10;
int space=(_maskGen!=null)?14:10;
do
{
@ -84,9 +144,9 @@ public class WebSocketGeneratorD06 implements WebSocketGenerator
expelBuffer(blockFor);
// write mask
if (_masked)
if ((_maskGen!=null))
{
_random.nextBytes(_mask);
_maskGen.genMask(_mask);
_m=0;
_buffer.put(_mask);
}
@ -127,7 +187,7 @@ public class WebSocketGeneratorD06 implements WebSocketGenerator
_buffer.compact();
int chunk = remaining < _buffer.space() ? remaining : _buffer.space();
if (_masked)
if ((_maskGen!=null))
{
for (int i=0;i<chunk;i++)
bufferPut(content[offset+ (payload-remaining)+i]);
@ -161,7 +221,7 @@ public class WebSocketGeneratorD06 implements WebSocketGenerator
private synchronized void bufferPut(byte[] data) throws IOException
{
if (_masked)
if (_maskGen!=null)
for (int i=0;i<data.length;i++)
data[i]^=_mask[+_m++%4];
_buffer.put(data);

8
jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketParserD06.java
View File

@ -18,6 +18,7 @@ import java.io.IOException;
import org.eclipse.jetty.io.Buffer;
import org.eclipse.jetty.io.Buffers;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.util.TypeUtil;
import org.eclipse.jetty.util.Utf8StringBuilder;
import org.eclipse.jetty.util.log.Log;
@ -222,13 +223,14 @@ public class WebSocketParserD06 implements WebSocketParser
if (data.array()==null)
data=_buffer.asMutableBuffer();
byte[] array = data.array();
for (int i=data.length();i-->0;)
array[data.getIndex()+i]^=_mask[_m++%4];
final int end=data.putIndex();
for (int i=data.getIndex();i<end;i++)
array[i]^=_mask[_m++%4];
}
_handler.onFrame(!_fin,_flags, _opcode, data);
_count=0;
_state=State.OPCODE;
_state=_masked?State.MASK:State.OPCODE;
if (_buffer.length()==0)
{

14
jetty-websocket/src/test/java/org/eclipse/jetty/websocket/WebSocketGeneratorD06Test.java
View File

@ -20,6 +20,8 @@ public class WebSocketGeneratorD06Test
byte[] _mask = new byte[4];
int _m;
public WebSocketGeneratorD06.MaskGen _maskGen = new WebSocketGeneratorD06.FixedMaskGen(
new byte[]{(byte)0x00,(byte)0x00,(byte)0x0f,(byte)0xff});
@Before
public void setUp() throws Exception
@ -40,7 +42,7 @@ public class WebSocketGeneratorD06Test
@Test
public void testOneString() throws Exception
{
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,false);
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,null);
_generator.addFrame((byte)0x04,"Hell\uFF4F W\uFF4Frld",0);
@ -67,7 +69,7 @@ public class WebSocketGeneratorD06Test
@Test
public void testOneBuffer() throws Exception
{
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,false);
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,null);
String string = "Hell\uFF4F W\uFF4Frld";
byte[] bytes=string.getBytes(StringUtil.__UTF8);
@ -95,7 +97,7 @@ public class WebSocketGeneratorD06Test
@Test
public void testOneLongBuffer() throws Exception
{
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,false);
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,null);
byte[] b=new byte[150];
for (int i=0;i<b.length;i++)
@ -116,7 +118,7 @@ public class WebSocketGeneratorD06Test
@Test
public void testOneStringMasked() throws Exception
{
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,true);
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,_maskGen);
_generator.addFrame((byte)0x04,"Hell\uFF4F W\uFF4Frld",0);
_generator.flush();
@ -145,7 +147,7 @@ public class WebSocketGeneratorD06Test
@Test
public void testOneBufferMasked() throws Exception
{
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,true);
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,_maskGen);
String string = "Hell\uFF4F W\uFF4Frld";
byte[] bytes=string.getBytes(StringUtil.__UTF8);
@ -176,7 +178,7 @@ public class WebSocketGeneratorD06Test
@Test
public void testOneLongBufferMasked() throws Exception
{
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,true);
_generator = new WebSocketGeneratorD06(_buffers, _endPoint,_maskGen);
byte[] b=new byte[150];
for (int i=0;i<b.length;i++)

81
jetty-websocket/src/test/java/org/eclipse/jetty/websocket/WebSocketLoadTest.java
View File

@ -1,11 +1,9 @@
package org.eclipse.jetty.websocket;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
@ -18,6 +16,10 @@ import java.util.concurrent.TimeUnit;
import javax.servlet.http.HttpServletRequest;
import junit.framework.Assert;
import org.eclipse.jetty.io.Buffer;
import org.eclipse.jetty.io.bio.SocketEndPoint;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.handler.DefaultHandler;
@ -72,7 +74,7 @@ public class WebSocketLoadTest
public void testLoad() throws Exception
{
int count = 50;
int iterations = 10;
int iterations = 100;
ExecutorService threadPool = Executors.newCachedThreadPool();
try
@ -93,7 +95,7 @@ public class WebSocketLoadTest
long maxTimePerIteration = 5;
assertTrue(latch.await(iterations * (count / parallelism + 1) * maxTimePerIteration, TimeUnit.MILLISECONDS));
long end = System.nanoTime();
System.err.println("Elapsed: " + TimeUnit.NANOSECONDS.toMillis(end - start) + " ms");
// System.err.println("Elapsed: " + TimeUnit.NANOSECONDS.toMillis(end - start) + " ms");
for (WebSocketClient client : clients)
client.close();
@ -118,6 +120,7 @@ public class WebSocketLoadTest
{
try
{
// System.err.println(">> "+data);
outbound.sendMessage(data);
}
catch (IOException x)
@ -146,7 +149,18 @@ public class WebSocketLoadTest
private final BufferedReader input;
private final int iterations;
private final CountDownLatch latch;
private final SocketEndPoint _endp;
private final WebSocketGeneratorD06 _generator;
private final WebSocketParserD06 _parser;
private final WebSocketParser.FrameHandler _handler = new WebSocketParser.FrameHandler()
{
public void onFrame(boolean more, byte flags, byte opcode, Buffer buffer)
{
_response=buffer;
}
};
private volatile Buffer _response;
public WebSocketClient(String host, int port, int readTimeout, CountDownLatch latch, int iterations) throws IOException
{
this.latch = latch;
@ -155,19 +169,28 @@ public class WebSocketLoadTest
output = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(), "ISO-8859-1"));
input = new BufferedReader(new InputStreamReader(socket.getInputStream(), "ISO-8859-1"));
this.iterations = iterations;
_endp=new SocketEndPoint(socket);
_generator = new WebSocketGeneratorD06(new WebSocketBuffers(32*1024),_endp,new WebSocketGeneratorD06.FixedMaskGen());
_parser = new WebSocketParserD06(new WebSocketBuffers(32*1024),_endp,_handler,false);
}
private void open() throws IOException
{
output.write("GET /test HTTP/1.1\r\n" +
"Host: localhost\r\n" +
"Upgrade: WebSocket\r\n" +
"Connection: Upgrade\r\n" +
output.write("GET /chat HTTP/1.1\r\n"+
"Host: server.example.com\r\n"+
"Upgrade: websocket\r\n"+
"Connection: Upgrade\r\n"+
"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"+
"Sec-WebSocket-Origin: http://example.com\r\n"+
"Sec-WebSocket-Protocol: onConnect\r\n" +
"Sec-WebSocket-Version: 6\r\n"+
"\r\n");
output.flush();
String responseLine = input.readLine();
assertTrue(responseLine.startsWith("HTTP/1.1 101 Web Socket Protocol Handshake"));
assertTrue(responseLine.startsWith("HTTP/1.1 101 Switching Protocols"));
// Read until we find an empty line, which signals the end of the http response
String line;
while ((line = input.readLine()) != null)
@ -182,9 +205,16 @@ public class WebSocketLoadTest
String message = "0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF";
for (int i = 0; i < iterations; ++i)
{
writeString(message);
String result = readString();
assertEquals(message, result);
_generator.addFrame(WebSocket.OP_TEXT,message,10000);
_generator.flush(10000);
//System.err.println("-> "+message);
_response=null;
while(_response==null)
_parser.parseNext();
//System.err.println("<- "+_response);
Assert.assertEquals(message,_response.toString());
latch.countDown();
}
}
@ -194,31 +224,6 @@ public class WebSocketLoadTest
}
}
private void writeString(String message) throws IOException
{
output.write(0x00);
output.write(message);
output.write(0xFF);
output.flush();
}
private String readString() throws IOException
{
StringBuilder result = new StringBuilder();
int frameStart = input.read();
assertEquals(0x00, frameStart);
while (true)
{
int read = input.read();
if (read == -1)
throw new EOFException();
char c = (char)read;
if (c == 0xFF)
break;
result.append(c);
}
return result.toString();
}
public void close() throws IOException
{

63
jetty-websocket/src/test/java/org/eclipse/jetty/websocket/WebSocketMessageD06Test.java
View File

@ -11,16 +11,19 @@ import java.io.OutputStream;
import java.net.Socket;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.servlet.http.HttpServletRequest;
import org.eclipse.jetty.io.Buffer;
import org.eclipse.jetty.io.ByteArrayEndPoint;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.handler.DefaultHandler;
import org.eclipse.jetty.server.nio.SelectChannelConnector;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.TypeUtil;
import org.eclipse.jetty.util.Utf8StringBuilder;
import org.eclipse.jetty.util.log.Log;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@ -347,10 +350,62 @@ public class WebSocketMessageD06Test
catch(IOException e)
{
assertTrue(true);
}
}
}
@Test
public void testParserAndGenerator() throws Exception
{
String message = "0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF";
final AtomicReference<String> received = new AtomicReference<String>();
ByteArrayEndPoint endp = new ByteArrayEndPoint(new byte[0],4096);
WebSocketGeneratorD06 gen = new WebSocketGeneratorD06(new WebSocketBuffers(8096),endp,null);
gen.addFrame((byte)0x4,message,1000);
endp = new ByteArrayEndPoint(endp.getOut().asArray(),4096);
WebSocketParserD06 parser = new WebSocketParserD06(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler()
{
public void onFrame(boolean more, byte flags, byte opcode, Buffer buffer)
{
received.set(buffer.toString());
}
},false);
parser.parseNext();
assertEquals(message,received.get());
}
@Test
public void testParserAndGeneratorMasked() throws Exception
{
String message = "0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF";
final AtomicReference<String> received = new AtomicReference<String>();
ByteArrayEndPoint endp = new ByteArrayEndPoint(new byte[0],4096);
WebSocketGeneratorD06.MaskGen maskGen = new WebSocketGeneratorD06.RandomMaskGen();
WebSocketGeneratorD06 gen = new WebSocketGeneratorD06(new WebSocketBuffers(8096),endp,maskGen);
gen.addFrame((byte)0x4,message,1000);
endp = new ByteArrayEndPoint(endp.getOut().asArray(),4096);
WebSocketParserD06 parser = new WebSocketParserD06(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler()
{
public void onFrame(boolean more, byte flags, byte opcode, Buffer buffer)
{
received.set(buffer.toString());
}
},true);
parser.parseNext();
assertEquals(message,received.get());
}
private void lookFor(String string,InputStream in)
throws IOException
{

126
jetty-websocket/src/test/java/org/eclipse/jetty/websocket/WebSocketParserD06Test.java
View File

@ -13,6 +13,7 @@ import org.eclipse.jetty.io.BufferCache.CachedBuffer;
import org.eclipse.jetty.io.ByteArrayBuffer;
import org.eclipse.jetty.io.ByteArrayEndPoint;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.TypeUtil;
import org.eclipse.jetty.util.Utf8StringBuilder;
import org.junit.Before;
import org.junit.Test;
@ -22,11 +23,59 @@ import org.junit.Test;
*/
public class WebSocketParserD06Test
{
private ByteArrayBuffer _in;
private MaskedByteArrayBuffer _in;
private Handler _handler;
private WebSocketParser _parser;
private byte[] _mask = new byte[] {(byte)0xff,(byte)0xff,(byte)0xff,(byte)0xff};
private byte[] _mask = new byte[] {(byte)0x00,(byte)0xF0,(byte)0x0F,(byte)0xFF};
private int _m;
class MaskedByteArrayBuffer extends ByteArrayBuffer
{
MaskedByteArrayBuffer()
{
super(4096);
}
public void sendMask()
{
super.poke(putIndex(),_mask,0,4);
super.setPutIndex(putIndex()+4);
_m=0;
}
@Override
public int put(Buffer src)
{
return put(src.asArray(),0,src.length());
}
@Override
public void put(byte b)
{
super.put((byte)(b^_mask[_m++%4]));
}
@Override
public int put(byte[] b, int offset, int length)
{
byte[] mb = new byte[b.length];
final int end=offset+length;
for (int i=offset;i<end;i++)
{
mb[i]=(byte)(b[i]^_mask[_m++%4]);
}
return super.put(mb,offset,length);
}
@Override
public int put(byte[] b)
{
return put(b,0,b.length);
}
};
@Before
public void setUp() throws Exception
{
@ -34,63 +83,7 @@ public class WebSocketParserD06Test
ByteArrayEndPoint endPoint = new ByteArrayEndPoint();
_handler = new Handler();
_parser=new WebSocketParserD06(buffers, endPoint,_handler,true);
_in = new ByteArrayBuffer(2048)
{
{
// add the mask
super.put(_mask,0,4);
}
@Override
public void poke(int index, byte b)
{
super.poke(index,(byte)(b^_mask[index%4]));
}
@Override
public int poke(int index, Buffer src)
{
return poke(index,src.asArray(),0,src.length());
}
@Override
public int poke(int index, byte[] b, int offset, int length)
{
byte[] mb = new byte[b.length];
for (int i=length;i-->0;)
mb[offset+i]=(byte)(b[offset+i]^_mask[(index+i)%4]);
return super.poke(index,mb,offset,length);
}
@Override
public int put(Buffer src)
{
return put(src.asArray(),0,src.length());
}
@Override
public void put(byte b)
{
super.put((byte)(b^_mask[getIndex()%4]));
}
@Override
public int put(byte[] b, int offset, int length)
{
byte[] mb = new byte[b.length];
for (int i=length;i-->0;)
mb[offset+i]=(byte)(b[offset+i]^_mask[(getIndex()+i)%4]);
return super.put(mb,offset,length);
}
@Override
public int put(byte[] b)
{
return put(b,0,b.length);
}
};
_in = new MaskedByteArrayBuffer();
endPoint.setIn(_in);
}
@ -104,9 +97,11 @@ public class WebSocketParserD06Test
@Test
public void testShortText() throws Exception
{
_in.sendMask();
_in.put((byte)0x84);
_in.put((byte)11);
_in.put("Hello World".getBytes(StringUtil.__UTF8));
System.err.println("tosend="+TypeUtil.toHexString(_in.asArray()));
int filled =_parser.parseNext();
@ -121,7 +116,8 @@ public class WebSocketParserD06Test
{
String string = "Hell\uFF4f W\uFF4Frld";
byte[] bytes = string.getBytes("UTF-8");
_in.sendMask();
_in.put((byte)0x84);
_in.put((byte)bytes.length);
_in.put(bytes);
@ -143,7 +139,8 @@ public class WebSocketParserD06Test
string += ". The end.";
byte[] bytes = string.getBytes("UTF-8");
_in.sendMask();
_in.put((byte)0x84);
_in.put((byte)0x7E);
_in.put((byte)(bytes.length>>8));
@ -175,7 +172,8 @@ public class WebSocketParserD06Test
string += ". The end.";
byte[] bytes = string.getBytes("UTF-8");
_in.sendMask();
in.put((byte)0x84);
in.put((byte)0x7F);
in.put((byte)0x00);
@ -199,16 +197,18 @@ public class WebSocketParserD06Test
@Test
public void testShortFragmentTest() throws Exception
{
_in.sendMask();
_in.put((byte)0x04);
_in.put((byte)0x06);
_in.put("Hello ".getBytes(StringUtil.__UTF8));
_in.sendMask();
_in.put((byte)0x80);
_in.put((byte)0x05);
_in.put("World".getBytes(StringUtil.__UTF8));
int filled =_parser.parseNext();
assertEquals(19,filled);
assertEquals(23,filled);
assertEquals(0,_handler._data.size());
assertFalse(_parser.isBufferEmpty());
assertFalse(_parser.getBuffer()==null);