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

353073 WebSocketClient

This commit is contained in:
Greg Wilkins 2011-07-26 17:31:55 +10:00
parent 2be177484e
commit 5859a85919
14 changed files with 704 additions and 333 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
VERSION.txt
View File

@ -7,6 +7,7 @@ jetty-7.5.0-SNAPSHOT
+ 352421 HttpURI paths beginning with '.' + 352421 HttpURI paths beginning with '.'
+ 352684 Implemented spinning thread analyzer + 352684 Implemented spinning thread analyzer
+ 352786 GzipFilter fails to pass parameters to GzipResponseWrapper + 352786 GzipFilter fails to pass parameters to GzipResponseWrapper
+ 353073 WebSocketClient
jetty-7.4.4.v20110707 July 7th 2011 jetty-7.4.4.v20110707 July 7th 2011
+ 308851 Converted all jetty-client module tests to JUnit 4 + 308851 Converted all jetty-client module tests to JUnit 4

3
jetty-http/src/main/java/org/eclipse/jetty/http/HttpFields.java
View File

@ -57,6 +57,7 @@ import org.eclipse.jetty.util.log.Log;
public class HttpFields public class HttpFields
{ {
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
public static final String __COOKIE_DELIM="\"\\\n\r\t\f\b%+ ;=";
public static final TimeZone __GMT = TimeZone.getTimeZone("GMT"); public static final TimeZone __GMT = TimeZone.getTimeZone("GMT");
public static final BufferDateCache __dateCache = new BufferDateCache("EEE, dd MMM yyyy HH:mm:ss 'GMT'", Locale.US); public static final BufferDateCache __dateCache = new BufferDateCache("EEE, dd MMM yyyy HH:mm:ss 'GMT'", Locale.US);
@ -924,7 +925,7 @@ public class HttpFields
final boolean isHttpOnly, final boolean isHttpOnly,
int version) int version)
{ {
String delim=_maxCookieVersion==0?"":"\"\\\n\r\t\f\b%+ ;="; String delim=_maxCookieVersion==0?"":__COOKIE_DELIM;
// Check arguments // Check arguments
if (name == null || name.length() == 0) if (name == null || name.length() == 0)

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

@ -82,7 +82,7 @@ public class DeflateFrameExtension extends AbstractExtension
catch(DataFormatException e) catch(DataFormatException e)
{ {
Log.warn(e); Log.warn(e);
getConnection().close(WebSocketConnectionD7_9.CLOSE_PROTOCOL,e.toString()); getConnection().close(WebSocketConnectionD10.CLOSE_PROTOCOL,e.toString());
} }
} }

441
jetty-websocket/src/main/java/org/eclipse/jetty/websocket/TestClient.java
View File

@ -5,12 +5,19 @@ import java.io.BufferedWriter;
import java.io.IOException; import java.io.IOException;
import java.io.InputStreamReader; import java.io.InputStreamReader;
import java.io.OutputStreamWriter; import java.io.OutputStreamWriter;
import java.net.InetSocketAddress;
import java.net.Socket; import java.net.Socket;
import java.net.URI;
import java.security.SecureRandom; import java.security.SecureRandom;
import java.util.Arrays; import java.util.Arrays;
import java.util.Random; import java.util.Random;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.eclipse.jetty.io.Buffer; import org.eclipse.jetty.io.Buffer;
import org.eclipse.jetty.io.bio.SocketEndPoint; import org.eclipse.jetty.io.bio.SocketEndPoint;
@ -18,6 +25,8 @@ import org.eclipse.jetty.util.B64Code;
import org.eclipse.jetty.util.StringUtil; import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.TypeUtil; import org.eclipse.jetty.util.TypeUtil;
import org.eclipse.jetty.util.log.Log; import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.websocket.WebSocket.Connection;
import org.eclipse.jetty.websocket.WebSocket.FrameConnection;
/** /**
* @version $Revision$ $Date$ * @version $Revision$ $Date$
@ -25,250 +34,139 @@ import org.eclipse.jetty.util.log.Log;
* This is not a general purpose websocket client. * This is not a general purpose websocket client.
* It's only for testing the websocket server and is hardwired to a specific draft version of the protocol. * It's only for testing the websocket server and is hardwired to a specific draft version of the protocol.
*/ */
public class TestClient public class TestClient implements WebSocket.OnFrame
{ {
private final static Random __random = new SecureRandom(); private static WebSocketClient __client = new WebSocketClient();
private static boolean _verbose=false; private static boolean _verbose=false;
private static final Random __random = new Random();
private final String _host; private final String _host;
private final int _port; private final int _port;
private final String _protocol; private final String _protocol;
private int _size=64; private final int _timeout;
private final Socket _socket;
private final BufferedWriter _output; private static int __framesSent;
private final BufferedReader _input; private static int __messagesSent;
private final SocketEndPoint _endp; private static AtomicInteger __framesReceived=new AtomicInteger();
private final WebSocketGeneratorD7_9 _generator; private static AtomicInteger __messagesReceived=new AtomicInteger();
private final WebSocketParserD7_9 _parser;
private int _framesSent; private static AtomicLong __totalTime=new AtomicLong();
private int _messagesSent; private static AtomicLong __minDuration=new AtomicLong(Long.MAX_VALUE);
private int _framesReceived; private static AtomicLong __maxDuration=new AtomicLong(Long.MIN_VALUE);
private int _messagesReceived; private static long __start;
private long _totalTime;
private long _minDuration=Long.MAX_VALUE;
private long _maxDuration=Long.MIN_VALUE;
private long _start;
private BlockingQueue<Long> _starts = new LinkedBlockingQueue<Long>(); private BlockingQueue<Long> _starts = new LinkedBlockingQueue<Long>();
int _messageBytes; int _messageBytes;
int _frames; int _frames;
byte _opcode=-1; byte _opcode=-1;
private final WebSocketParser.FrameHandler _handler = new WebSocketParser.FrameHandler() private volatile WebSocket.FrameConnection _connection;
private final CountDownLatch _handshook = new CountDownLatch(1);
public void onOpen(Connection connection)
{ {
public synchronized void onFrame(byte flags, byte opcode, Buffer buffer) }
public void onClose(int closeCode, String message)
{
_handshook.countDown();
}
public boolean onFrame(byte flags, byte opcode, byte[] data, int offset, int length)
{
try
{ {
try __framesReceived.incrementAndGet();
{ _frames++;
_framesReceived++; _messageBytes+=length;
_frames++;
if (opcode == WebSocketConnectionD7_9.OP_CLOSE) if (_opcode==-1)
{ _opcode=opcode;
byte[] data=buffer.asArray();
// System.err.println("CLOSED: "+((0xff&data[0])*0x100+(0xff&data[1]))+" "+new String(data,2,data.length-2,StringUtil.__UTF8)); if (_connection.isControl(opcode) || _connection.isMessageComplete(flags))
_generator.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_CLOSE,data,0,data.length); {
_generator.flush(); int recv =__messagesReceived.incrementAndGet();
_socket.shutdownOutput(); Long start=_starts.poll();
_socket.close();
return;
}
else if (opcode == WebSocketConnectionD7_9.OP_PING)
{
_generator.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_PONG,buffer.array(),buffer.getIndex(),buffer.length());
_generator.flush();
}
_messageBytes+=buffer.length();
if (_opcode==-1)
_opcode=opcode;
if (WebSocketConnectionD7_9.isLastFrame(flags)) if (start!=null)
{ {
_messagesReceived++;
Long start=_starts.take();
long duration = System.nanoTime()-start.longValue(); long duration = System.nanoTime()-start.longValue();
if (duration>_maxDuration) long max=__maxDuration.get();
_maxDuration=duration; while(duration>max && !__maxDuration.compareAndSet(max,duration))
if (duration<_minDuration) max=__maxDuration.get();
_minDuration=duration; long min=__minDuration.get();
_totalTime+=duration; while(duration<min && !__minDuration.compareAndSet(min,duration))
System.out.printf("%d bytes from %s: frames=%d req=%d time=%.1fms opcode=0x%s\n",_messageBytes,_host,_frames,_messagesReceived,((double)duration/1000000.0),TypeUtil.toHexString(_opcode)); min=__minDuration.get();
_frames=0; __totalTime.addAndGet(duration);
_messageBytes=0; System.out.printf("%d bytes from %s: frames=%d req=%d time=%.1fms opcode=0x%s\n",_messageBytes,_host,_frames,recv,((double)duration/1000000.0),TypeUtil.toHexString(_opcode));
_opcode=-1;
} }
_frames=0;
_messageBytes=0;
_opcode=-1;
} }
catch(Exception e)
{
e.printStackTrace();
}
} }
catch(Exception e)
public void close(int code,String message)
{ {
e.printStackTrace();
} }
}; return false;
}
public void onHandshake(FrameConnection connection)
{
_connection=connection;
_handshook.countDown();
}
public TestClient(String host, int port,String protocol, int timeoutMS) throws IOException public TestClient(String host, int port,String protocol, int timeoutMS) throws Exception
{ {
_host=host; _host=host;
_port=port; _port=port;
_protocol=protocol; _protocol=protocol;
_socket = new Socket(host, port); _timeout=timeoutMS;
_socket.setSoTimeout(timeoutMS);
_output = new BufferedWriter(new OutputStreamWriter(_socket.getOutputStream(), "ISO-8859-1"));
_input = new BufferedReader(new InputStreamReader(_socket.getInputStream(), "ISO-8859-1"));
_endp=new SocketEndPoint(_socket);
_generator = new WebSocketGeneratorD7_9(new WebSocketBuffers(32*1024),_endp,new WebSocketGeneratorD7_9.FixedMaskGen(new byte[4]));
_parser = new WebSocketParserD7_9(new WebSocketBuffers(32*1024),_endp,_handler,false);
} }
public int getSize()
private void open() throws Exception
{ {
return _size; __client.open(new URI("ws://"+_host+":"+_port+"/"),this,_protocol,_timeout);
_handshook.await(10,TimeUnit.SECONDS);
} }
public void setSize(int size) public void ping(byte opcode,byte[] data,int fragment) throws Exception
{ {
_size = size; _starts.add(System.nanoTime());
}
private void open() throws IOException int off=0;
{ int len=data.length;
System.out.println("Jetty WebSocket PING "+_host+":"+_port+ if (fragment>0&& len>fragment)
" ("+_socket.getRemoteSocketAddress()+") " +_size+" bytes of data."); len=fragment;
byte[] key = new byte[16]; __messagesSent++;
__random.nextBytes(key); while(off<data.length)
{
__framesSent++;
_output.write("GET /chat HTTP/1.1\r\n"+ byte flags= (byte)(off+len==data.length?0x8:0);
"Host: "+_host+":"+_port+"\r\n"+ byte op=(byte)(off==0?opcode:WebSocketConnectionD10.OP_CONTINUATION);
"Upgrade: websocket\r\n"+
"Connection: Upgrade\r\n"+
"Sec-WebSocket-Key: "+new String(B64Code.encode(key))+"\r\n"+
"Sec-WebSocket-Origin: http://example.com\r\n"+
"Sec-WebSocket-Protocol: "+_protocol+"\r\n" +
"Sec-WebSocket-Version: 7\r\n"+
"\r\n");
_output.flush();
String responseLine = _input.readLine(); if (_verbose)
if(!responseLine.startsWith("HTTP/1.1 101 Switching Protocols")) System.err.printf("%s#addFrame %s|%s %s\n",this.getClass().getSimpleName(),TypeUtil.toHexString(flags),TypeUtil.toHexString(op),TypeUtil.toHexString(data,off,len));
throw new IOException(responseLine);
// Read until we find Response key
String line;
boolean accepted=false;
String protocol="";
while ((line = _input.readLine()) != null)
{
if (line.length() == 0)
break;
if (line.startsWith("Sec-WebSocket-Accept:"))
{
String accept=line.substring(21).trim();
accepted=accept.equals(WebSocketConnectionD7_9.hashKey(new String(B64Code.encode(key))));
}
else if (line.startsWith("Sec-WebSocket-Protocol:"))
{
protocol=line.substring(24).trim();
}
}
if (!accepted)
throw new IOException("Bad Sec-WebSocket-Accept");
System.out.println("handshake OK for protocol '"+protocol+"'");
new Thread()
{
public void run()
{
while (_endp.isOpen())
{
_parser.parseNext();
}
}
}.start();
}
public void ping(int count,byte opcode,int fragment) _connection.sendFrame(flags,op,data,off,len);
{
try
{
_start=System.currentTimeMillis();
for (int i=0;i<count && !_socket.isClosed();i++)
{
if (_socket.isClosed())
break;
byte data[]=null;
if (opcode==WebSocketConnectionD7_9.OP_TEXT) off+=len;
{ if(data.length-off>len)
StringBuilder b = new StringBuilder(); len=data.length-off;
while (b.length()<_size) if (fragment>0&& len>fragment)
b.append('A'+__random.nextInt(26)); len=fragment;
data=b.toString().getBytes(StringUtil.__UTF8);
}
else
{
data= new byte[_size];
__random.nextBytes(data);
}
_starts.add(System.nanoTime());
int off=0;
int len=data.length;
if (fragment>0&& len>fragment)
len=fragment;
_messagesSent++;
while(off<data.length)
{
_framesSent++;
byte flags= (byte)(off+len==data.length?0x8:0);
byte op=(byte)(off==0?opcode:WebSocketConnectionD7_9.OP_CONTINUATION);
if (_verbose)
System.err.printf("%s#addFrame %s|%s %s\n",this.getClass().getSimpleName(),TypeUtil.toHexString(flags),TypeUtil.toHexString(op),TypeUtil.toHexString(data,off,len));
_generator.addFrame(flags,op,data,off,len);
off+=len;
if(data.length-off>len)
len=data.length-off;
if (fragment>0&& len>fragment)
len=fragment;
}
_generator.flush();
Thread.sleep(1000);
}
}
catch (Exception x)
{
throw new RuntimeException(x);
} }
} }
public void dump() throws Exception public void disconnect() throws Exception
{ {
for (int i=0;i<250;i++) _connection.disconnect();
{
if (_messagesSent==_messagesReceived)
break;
_generator.flush();
Thread.sleep(100);
}
_socket.close();
long duration=System.currentTimeMillis()-_start;
System.out.println("--- "+_host+" websocket ping statistics using 1 connection ---");
System.out.println(_framesSent+" frames transmitted, "+_framesReceived+" received, "+
_messagesSent+" messages transmitted, "+_messagesReceived+" received, "+
"time "+duration+"ms");
System.out.printf("rtt min/ave/max = %.3f/%.3f/%.3f ms\n",_minDuration/1000000.0,_messagesReceived==0?0.0:(_totalTime/_messagesReceived/1000000.0),_maxDuration/1000000.0);
} }
@ -284,66 +182,109 @@ public class TestClient
System.err.println(" -s|--size n (default 64)"); System.err.println(" -s|--size n (default 64)");
System.err.println(" -f|--fragment n (default 4000) "); System.err.println(" -f|--fragment n (default 4000) ");
System.err.println(" -P|--protocol echo|echo-assemble|echo-fragment|echo-broadcast"); System.err.println(" -P|--protocol echo|echo-assemble|echo-fragment|echo-broadcast");
System.err.println(" -C|--clients n (default 1) ");
System.exit(1); System.exit(1);
} }
public static void main(String[] args) public static void main(String[] args) throws Exception
{ {
__client.start();
String host="localhost";
int port=8080;
String protocol=null;
int count=10;
int size=64;
int fragment=4000;
boolean binary=false;
int clients=1;
for (int i=0;i<args.length;i++)
{
String a=args[i];
if ("-p".equals(a)||"--port".equals(a))
port=Integer.parseInt(args[++i]);
else if ("-h".equals(a)||"--host".equals(a))
port=Integer.parseInt(args[++i]);
else if ("-c".equals(a)||"--count".equals(a))
count=Integer.parseInt(args[++i]);
else if ("-s".equals(a)||"--size".equals(a))
size=Integer.parseInt(args[++i]);
else if ("-f".equals(a)||"--fragment".equals(a))
fragment=Integer.parseInt(args[++i]);
else if ("-P".equals(a)||"--protocol".equals(a))
protocol=args[++i];
else if ("-v".equals(a)||"--verbose".equals(a))
_verbose=true;
else if ("-b".equals(a)||"--binary".equals(a))
binary=true;
else if ("-C".equals(a)||"--clients".equals(a))
clients=Integer.parseInt(args[++i]);
else if (a.startsWith("-"))
usage(args);
}
TestClient[] client = new TestClient[clients];
try try
{ {
String host="localhost"; __start=System.currentTimeMillis();
int port=8080; for (int i=0;i<clients;i++)
String protocol=null;
int count=10;
int size=64;
int fragment=4000;
boolean binary=false;
for (int i=0;i<args.length;i++)
{ {
String a=args[i]; client[i]=new TestClient(host,port,protocol==null?null:protocol,10000);
if ("-p".equals(a)||"--port".equals(a)) client[i].open();
port=Integer.parseInt(args[++i]);
else if ("-h".equals(a)||"--host".equals(a))
port=Integer.parseInt(args[++i]);
else if ("-c".equals(a)||"--count".equals(a))
count=Integer.parseInt(args[++i]);
else if ("-s".equals(a)||"--size".equals(a))
size=Integer.parseInt(args[++i]);
else if ("-f".equals(a)||"--fragment".equals(a))
fragment=Integer.parseInt(args[++i]);
else if ("-P".equals(a)||"--protocol".equals(a))
protocol=args[++i];
else if ("-v".equals(a)||"--verbose".equals(a))
_verbose=true;
else if ("-b".equals(a)||"--binary".equals(a))
binary=true;
else if (a.startsWith("-"))
usage(args);
} }
System.out.println("Jetty WebSocket PING "+host+":"+port+
" ("+ new InetSocketAddress(host,port)+") "+clients+" clients");
TestClient client = new TestClient(host,port,protocol==null?null:protocol,10000); for (int p=0;p<count;p++)
client.setSize(size);
try
{ {
client.open(); long next = System.currentTimeMillis()+1000;
if (protocol!=null && protocol.startsWith("echo"))
client.ping(count,binary?WebSocketConnectionD7_9.OP_BINARY:WebSocketConnectionD7_9.OP_TEXT,fragment); byte opcode=binary?WebSocketConnectionD10.OP_BINARY:WebSocketConnectionD10.OP_TEXT;
byte data[]=null;
if (opcode==WebSocketConnectionD10.OP_TEXT)
{
StringBuilder b = new StringBuilder();
while (b.length()<size)
b.append('A'+__random.nextInt(26));
data=b.toString().getBytes(StringUtil.__UTF8);
}
else else
client.ping(count,WebSocketConnectionD7_9.OP_PING,-1); {
data= new byte[size];
__random.nextBytes(data);
}
for (int i=0;i<clients;i++)
client[i].ping(opcode,data,opcode==WebSocketConnectionD10.OP_PING?-1:fragment);
while(System.currentTimeMillis()<next)
Thread.sleep(10);
} }
finally
{
client.dump();
}
} }
catch (Exception e) finally
{ {
Log.warn(e); for (int i=0;i<clients;i++)
if (client[i]!=null)
client[i].disconnect();
long duration=System.currentTimeMillis()-__start;
System.out.println("--- "+host+" websocket ping statistics using "+clients+" connection"+(clients>1?"s":"")+" ---");
System.out.println(__framesSent+" frames transmitted, "+__framesReceived+" received, "+
__messagesSent+" messages transmitted, "+__messagesReceived+" received, "+
"time "+duration+"ms");
System.out.printf("rtt min/ave/max = %.3f/%.3f/%.3f ms\n",__minDuration.get()/1000000.0,__messagesReceived.get()==0?0.0:(__totalTime.get()/__messagesReceived.get()/1000000.0),__maxDuration.get()/1000000.0);
__client.stop();
} }
} }

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

@ -34,7 +34,7 @@ public class WebSocketBuffers
{ {
final private int _bufferSize; final private int _bufferSize;
final private Buffers _buffers; final private Buffers _buffers;
final private int _maxBuffers=1024; final private int _maxBuffers=-1;
public WebSocketBuffers(final int bufferSize) public WebSocketBuffers(final int bufferSize)
{ {

419
jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketClient.java Normal file
View File

@ -0,0 +1,419 @@
package org.eclipse.jetty.websocket;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.List;
import java.util.Map;
import java.util.Random;
import org.eclipse.jetty.http.HttpFields;
import org.eclipse.jetty.http.HttpParser;
import org.eclipse.jetty.io.AbstractConnection;
import org.eclipse.jetty.io.Buffer;
import org.eclipse.jetty.io.Buffers;
import org.eclipse.jetty.io.ByteArrayBuffer;
import org.eclipse.jetty.io.ConnectedEndPoint;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.SimpleBuffers;
import org.eclipse.jetty.io.nio.SelectChannelEndPoint;
import org.eclipse.jetty.io.nio.SelectorManager;
import org.eclipse.jetty.util.B64Code;
import org.eclipse.jetty.util.QuotedStringTokenizer;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.component.AggregateLifeCycle;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.thread.ThreadPool;
public class WebSocketClient extends AggregateLifeCycle
{
private final static Random __random = new Random();
private final static ByteArrayBuffer __ACCEPT = new ByteArrayBuffer.CaseInsensitive("Sec-WebSocket-Accept");
private final ThreadPool _threadPool;
private final Selector _selector=new Selector();
private int _connectTimeout=30000;
private int _bufferSize=64*1024;
private WebSocketBuffers _buffers;
public WebSocketClient(ThreadPool threadpool)
{
_threadPool=threadpool;
addBean(_threadPool);
addBean(_selector);
}
public WebSocketClient()
{
this(new QueuedThreadPool());
}
public SelectorManager getSelectorManager()
{
return _selector;
}
public ThreadPool getThreadPool()
{
return _threadPool;
}
public int getConnectTimeout()
{
return _connectTimeout;
}
public void setConnectTimeout(int connectTimeout)
{
_connectTimeout = connectTimeout;
}
public int getMaxIdleTime()
{
return (int)_selector.getMaxIdleTime();
}
public void setMaxIdleTime(int maxIdleTime)
{
_selector.setMaxIdleTime(maxIdleTime);
}
public int getBufferSize()
{
return _bufferSize;
}
public void setBufferSize(int bufferSize)
{
_bufferSize = bufferSize;
}
@Override
protected void doStart() throws Exception
{
_buffers = new WebSocketBuffers(_bufferSize);
super.doStart();
for (int i=0;i<_selector.getSelectSets();i++)
{
final int id=i;
_threadPool.dispatch(new Runnable(){
public void run()
{
while(isRunning())
{
try
{
_selector.doSelect(id);
}
catch (IOException e)
{
Log.warn(e);
}
}
}
});
}
}
public void open(URI uri, WebSocket websocket) throws IOException
{
open(uri,websocket,null,(int)_selector.getMaxIdleTime(),null,null);
}
public void open(URI uri, WebSocket websocket, String protocol,int maxIdleTime) throws IOException
{
open(uri,websocket,protocol,(int)_selector.getMaxIdleTime(),null,null);
}
public void open(URI uri, WebSocket websocket, String protocol,int maxIdleTime,Map<String,String> cookies) throws IOException
{
open(uri,websocket,protocol,(int)_selector.getMaxIdleTime(),cookies,null);
}
public void open(URI uri, WebSocket websocket, String protocol,int maxIdleTime,Map<String,String> cookies,List<String> extensions) throws IOException
{
SocketChannel channel = SocketChannel.open();
channel.socket().setTcpNoDelay(true);
InetSocketAddress address=new InetSocketAddress(uri.getHost(),uri.getPort());
channel.configureBlocking(false);
channel.connect(address);
_selector.register( channel, new WebSocketHolder(websocket,uri,protocol,maxIdleTime,cookies,extensions) );
}
class Selector extends SelectorManager
{
@Override
public boolean dispatch(Runnable task)
{
return _threadPool.dispatch(task);
}
@Override
protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, final SelectionKey sKey) throws IOException
{
return new SelectChannelEndPoint(channel,selectSet,sKey);
}
@Override
protected Connection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint)
{
WebSocketHolder holder = (WebSocketHolder) endpoint.getSelectionKey().attachment();
return new HandshakeConnection(endpoint,holder);
}
@Override
protected void endPointOpened(SelectChannelEndPoint endpoint)
{
}
@Override
protected void endPointUpgraded(ConnectedEndPoint endpoint, Connection oldConnection)
{
}
@Override
protected void endPointClosed(SelectChannelEndPoint endpoint)
{
endpoint.getConnection().closed();
}
}
class HandshakeConnection extends AbstractConnection
{
private final SelectChannelEndPoint _endp;
private final WebSocketHolder _holder;
private final String _key;
private final HttpParser _parser;
private String _accept;
private String _error;
public HandshakeConnection(SelectChannelEndPoint endpoint, WebSocketHolder holder)
{
super(endpoint,System.currentTimeMillis());
_endp=endpoint;
_holder=holder;
byte[] bytes=new byte[16];
__random.nextBytes(bytes);
_key=new String(B64Code.encode(bytes));
Buffers buffers = new SimpleBuffers(_buffers.getBuffer(),null);
_parser=new HttpParser(buffers,_endp,
new HttpParser.EventHandler()
{
@Override
public void startResponse(Buffer version, int status, Buffer reason) throws IOException
{
if (status!=101)
{
_error="Bad response status "+status+" "+reason;
_endp.close();
}
}
@Override
public void parsedHeader(Buffer name, Buffer value) throws IOException
{
if (__ACCEPT.equals(name))
_accept=value.toString();
}
@Override
public void startRequest(Buffer method, Buffer url, Buffer version) throws IOException
{
_error="Bad response";
_endp.close();
}
@Override
public void content(Buffer ref) throws IOException
{
_error="Bad response";
_endp.close();
}
});
String request=
"GET "+_holder.getURI().getPath()+" HTTP/1.1\r\n"+
"Host: "+holder.getURI().getHost()+":"+_holder.getURI().getPort()+"\r\n"+
"Upgrade: websocket\r\n"+
"Connection: Upgrade\r\n"+
"Sec-WebSocket-Key: "+_key+"\r\n"+
"Sec-WebSocket-Origin: http://example.com\r\n"+
"Sec-WebSocket-Version: 8\r\n";
if (holder.getProtocol()!=null)
request+="Sec-WebSocket-Protocol: "+holder.getProtocol()+"\r\n";
if (holder.getCookies()!=null && holder.getCookies().size()>0)
{
for (String cookie : holder.getCookies().keySet())
request+="Cookie: "+QuotedStringTokenizer.quoteIfNeeded(cookie,HttpFields.__COOKIE_DELIM)+
"="+
QuotedStringTokenizer.quoteIfNeeded(holder.getCookies().get(cookie),HttpFields.__COOKIE_DELIM)+
"\r\n";
}
request+="\r\n";
// TODO extensions
try
{
ByteArrayBuffer handshake = new ByteArrayBuffer(request);
int len=handshake.length();
if (len!=_endp.flush(handshake))
throw new IOException("incomplete");
}
catch(IOException e)
{
Log.debug(e);
_holder.getWebSocket().onClose(WebSocketConnectionD10.CLOSE_PROTOCOL,"Handshake failed: "+e.toString());
}
}
public Connection handle() throws IOException
{
while (_endp.isOpen() && !_parser.isComplete())
{
switch (_parser.parseAvailable())
{
case -1:
_holder.getWebSocket().onClose(-1,"EOF");
return this;
case 0:
return this;
default:
break;
}
}
if (_error==null && WebSocketConnectionD10.hashKey(_key).equals(_accept))
{
Buffer header=_parser.getHeaderBuffer();
WebSocketConnectionD10 connection = new WebSocketConnectionD10(_holder.getWebSocket(),_endp,_buffers,System.currentTimeMillis(),_holder.getMaxIdleTime(),_holder.getProtocol(),null,10, new WebSocketGeneratorD10.RandomMaskGen());
if (header.hasContent())
connection.fillBuffersFrom(header);
_buffers.returnBuffer(header);
if (_holder.getWebSocket() instanceof WebSocket.OnFrame)
((WebSocket.OnFrame)_holder.getWebSocket()).onHandshake((WebSocket.FrameConnection)connection.getConnection());
_holder.getWebSocket().onOpen(connection.getConnection());
return connection;
}
_endp.close();
return this;
}
public boolean isIdle()
{
return false;
}
public boolean isSuspended()
{
return false;
}
public void closed()
{
_holder.getWebSocket().onClose(WebSocketConnectionD10.CLOSE_PROTOCOL,"Handshake failed "+(_error==null?"EOF":_error));
}
}
class WebSocketHolder
{
final WebSocket _websocket;;
final URI _uri;
final String _protocol;
final int _maxIdleTime;
final Map<String,String> _cookies;
final List<String> _extensions;
public WebSocketHolder(WebSocket websocket, URI uri, String protocol, int maxIdleTime, Map<String,String> cookies,List<String> extensions)
{
_websocket=websocket;
_uri=uri;
_protocol=protocol;
_maxIdleTime=maxIdleTime;
_cookies=cookies;
_extensions=extensions;
}
public Map<String,String> getCookies()
{
return _cookies;
}
public String getProtocol()
{
return _protocol;
}
public WebSocket getWebSocket()
{
return _websocket;
}
public URI getURI()
{
return _uri;
}
public int getMaxIdleTime()
{
return _maxIdleTime;
}
public String toString()
{
return "[" + _uri + ","+_websocket+"]";
}
}
public static void main(String... args) throws Exception
{
Log.getLog().setDebugEnabled(true);
WebSocketClient client = new WebSocketClient();
client.start();
client.open(new URI("ws://localhost:8080/websocket"),new WebSocket.OnTextMessage()
{
public void onOpen(org.eclipse.jetty.websocket.WebSocket.Connection connection)
{
System.err.println("onOpen "+connection);
}
public void onClose(int closeCode, String message)
{
System.err.println("onClose "+closeCode+" "+message);
}
public void onMessage(String data)
{
System.err.println("onMessage "+data);
}
});
}
}

74
jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketConnectionD7_9.java → jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketConnectionD10.java
View File

@ -37,8 +37,9 @@ import org.eclipse.jetty.websocket.WebSocket.OnFrame;
import org.eclipse.jetty.websocket.WebSocket.OnTextMessage; import org.eclipse.jetty.websocket.WebSocket.OnTextMessage;
import org.eclipse.jetty.websocket.WebSocket.OnBinaryMessage; import org.eclipse.jetty.websocket.WebSocket.OnBinaryMessage;
import org.eclipse.jetty.websocket.WebSocket.OnControl; import org.eclipse.jetty.websocket.WebSocket.OnControl;
import org.eclipse.jetty.websocket.WebSocketGeneratorD10.MaskGen;
public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSocketConnection public class WebSocketConnectionD10 extends AbstractConnection implements WebSocketConnection
{ {
final static byte OP_CONTINUATION = 0x00; final static byte OP_CONTINUATION = 0x00;
final static byte OP_TEXT = 0x01; final static byte OP_TEXT = 0x01;
@ -69,9 +70,9 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
private final static byte[] MAGIC; private final static byte[] MAGIC;
private final IdleCheck _idle; private final IdleCheck _idle;
private final List<Extension> _extensions; private final List<Extension> _extensions;
private final WebSocketParserD7_9 _parser; private final WebSocketParserD10 _parser;
private final WebSocketParser.FrameHandler _inbound; private final WebSocketParser.FrameHandler _inbound;
private final WebSocketGeneratorD7_9 _generator; private final WebSocketGeneratorD10 _generator;
private final WebSocketGenerator _outbound; private final WebSocketGenerator _outbound;
private final WebSocket _webSocket; private final WebSocket _webSocket;
private final OnFrame _onFrame; private final OnFrame _onFrame;
@ -103,11 +104,18 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
private final WebSocket.FrameConnection _connection = new FrameConnectionD07(); private final WebSocket.FrameConnection _connection = new FrameConnectionD10();
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
public WebSocketConnectionD7_9(WebSocket websocket, EndPoint endpoint, WebSocketBuffers buffers, long timestamp, int maxIdleTime, String protocol, List<Extension> extensions,int draft) public WebSocketConnectionD10(WebSocket websocket, EndPoint endpoint, WebSocketBuffers buffers, long timestamp, int maxIdleTime, String protocol, List<Extension> extensions,int draft)
throws IOException
{
this(websocket,endpoint,buffers,timestamp,maxIdleTime,protocol,extensions,draft,null);
}
/* ------------------------------------------------------------ */
public WebSocketConnectionD10(WebSocket websocket, EndPoint endpoint, WebSocketBuffers buffers, long timestamp, int maxIdleTime, String protocol, List<Extension> extensions,int draft, MaskGen maskgen)
throws IOException throws IOException
{ {
super(endpoint,timestamp); super(endpoint,timestamp);
@ -124,7 +132,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
_onTextMessage=_webSocket instanceof OnTextMessage ? (OnTextMessage)_webSocket : null; _onTextMessage=_webSocket instanceof OnTextMessage ? (OnTextMessage)_webSocket : null;
_onBinaryMessage=_webSocket instanceof OnBinaryMessage ? (OnBinaryMessage)_webSocket : null; _onBinaryMessage=_webSocket instanceof OnBinaryMessage ? (OnBinaryMessage)_webSocket : null;
_onControl=_webSocket instanceof OnControl ? (OnControl)_webSocket : null; _onControl=_webSocket instanceof OnControl ? (OnControl)_webSocket : null;
_generator = new WebSocketGeneratorD7_9(buffers, _endp,null); _generator = new WebSocketGeneratorD10(buffers, _endp,maskgen);
_extensions=extensions; _extensions=extensions;
if (_extensions!=null) if (_extensions!=null)
@ -140,10 +148,10 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
} }
} }
_outbound=_extensions.size()==0?_generator:extensions.get(extensions.size()-1); _outbound=(_extensions==null||_extensions.size()==0)?_generator:extensions.get(extensions.size()-1);
_inbound=_extensions.size()==0?_frameHandler:extensions.get(0); _inbound=(_extensions==null||_extensions.size()==0)?_frameHandler:extensions.get(0);
_parser = new WebSocketParserD7_9(buffers, endpoint,_inbound,true); _parser = new WebSocketParserD10(buffers, endpoint,_inbound,maskgen==null);
_protocol=protocol; _protocol=protocol;
@ -251,7 +259,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
@Override @Override
public void idleExpired() public void idleExpired()
{ {
closeOut(WebSocketConnectionD7_9.CLOSE_NORMAL,"Idle"); closeOut(WebSocketConnectionD10.CLOSE_NORMAL,"Idle");
} }
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
@ -263,7 +271,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
public void closed() public void closed()
{ {
_webSocket.onClose(WebSocketConnectionD7_9.CLOSE_NORMAL,""); _webSocket.onClose(WebSocketConnectionD10.CLOSE_NORMAL,"");
} }
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
@ -298,11 +306,11 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
else else
{ {
if (code<=0) if (code<=0)
code=WebSocketConnectionD7_9.CLOSE_NORMAL; code=WebSocketConnectionD10.CLOSE_NORMAL;
byte[] bytes = ("xx"+(message==null?"":message)).getBytes(StringUtil.__ISO_8859_1); byte[] bytes = ("xx"+(message==null?"":message)).getBytes(StringUtil.__ISO_8859_1);
bytes[0]=(byte)(code/0x100); bytes[0]=(byte)(code/0x100);
bytes[1]=(byte)(code%0x100); bytes[1]=(byte)(code%0x100);
_outbound.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_CLOSE,bytes,0,bytes.length); _outbound.addFrame((byte)0x8,WebSocketConnectionD10.OP_CLOSE,bytes,0,bytes.length);
} }
_outbound.flush(); _outbound.flush();
@ -335,11 +343,11 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
private class FrameConnectionD07 implements WebSocket.FrameConnection private class FrameConnectionD10 implements WebSocket.FrameConnection
{ {
volatile boolean _disconnecting; volatile boolean _disconnecting;
int _maxTextMessage=WebSocketConnectionD7_9.this._maxTextMessageSize; int _maxTextMessage=WebSocketConnectionD10.this._maxTextMessageSize;
int _maxBinaryMessage=WebSocketConnectionD7_9.this._maxBinaryMessageSize; int _maxBinaryMessage=WebSocketConnectionD10.this._maxBinaryMessageSize;
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
public synchronized void sendMessage(String content) throws IOException public synchronized void sendMessage(String content) throws IOException
@ -347,7 +355,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
if (_closedOut) if (_closedOut)
throw new IOException("closing"); throw new IOException("closing");
byte[] data = content.getBytes(StringUtil.__UTF8); byte[] data = content.getBytes(StringUtil.__UTF8);
_outbound.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_TEXT,data,0,data.length); _outbound.addFrame((byte)0x8,WebSocketConnectionD10.OP_TEXT,data,0,data.length);
checkWriteable(); checkWriteable();
_idle.access(_endp); _idle.access(_endp);
} }
@ -357,7 +365,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{ {
if (_closedOut) if (_closedOut)
throw new IOException("closing"); throw new IOException("closing");
_outbound.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_BINARY,content,offset,length); _outbound.addFrame((byte)0x8,WebSocketConnectionD10.OP_BINARY,content,offset,length);
checkWriteable(); checkWriteable();
_idle.access(_endp); _idle.access(_endp);
} }
@ -400,7 +408,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
if (_disconnecting) if (_disconnecting)
return; return;
_disconnecting=true; _disconnecting=true;
WebSocketConnectionD7_9.this.closeOut(code,message); WebSocketConnectionD10.this.closeOut(code,message);
} }
/* ------------------------------------------------------------ */ /* ------------------------------------------------------------ */
@ -525,7 +533,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{ {
boolean lastFrame = isLastFrame(flags); boolean lastFrame = isLastFrame(flags);
synchronized(WebSocketConnectionD7_9.this) synchronized(WebSocketConnectionD10.this)
{ {
// Ignore incoming after a close // Ignore incoming after a close
if (_closedIn) if (_closedIn)
@ -550,10 +558,10 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
switch(opcode) switch(opcode)
{ {
case WebSocketConnectionD7_9.OP_CONTINUATION: case WebSocketConnectionD10.OP_CONTINUATION:
{ {
// If text, append to the message buffer // If text, append to the message buffer
if (_opcode==WebSocketConnectionD7_9.OP_TEXT && _connection.getMaxTextMessageSize()>=0) if (_opcode==WebSocketConnectionD10.OP_TEXT && _connection.getMaxTextMessageSize()>=0)
{ {
if (_utf8.append(buffer.array(),buffer.getIndex(),buffer.length(),_connection.getMaxTextMessageSize())) if (_utf8.append(buffer.array(),buffer.getIndex(),buffer.length(),_connection.getMaxTextMessageSize()))
{ {
@ -568,7 +576,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
} }
else else
{ {
_connection.close(WebSocketConnectionD7_9.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars"); _connection.close(WebSocketConnectionD10.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars");
_utf8.reset(); _utf8.reset();
_opcode=-1; _opcode=-1;
} }
@ -577,7 +585,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{ {
if (_aggregate.space()<_aggregate.length()) if (_aggregate.space()<_aggregate.length())
{ {
_connection.close(WebSocketConnectionD7_9.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize()); _connection.close(WebSocketConnectionD10.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize());
_aggregate.clear(); _aggregate.clear();
_opcode=-1; _opcode=-1;
} }
@ -602,21 +610,21 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
} }
break; break;
} }
case WebSocketConnectionD7_9.OP_PING: case WebSocketConnectionD10.OP_PING:
{ {
Log.debug("PING {}",this); Log.debug("PING {}",this);
if (!_closedOut) if (!_closedOut)
_connection.sendControl(WebSocketConnectionD7_9.OP_PONG,buffer.array(),buffer.getIndex(),buffer.length()); _connection.sendControl(WebSocketConnectionD10.OP_PONG,buffer.array(),buffer.getIndex(),buffer.length());
break; break;
} }
case WebSocketConnectionD7_9.OP_PONG: case WebSocketConnectionD10.OP_PONG:
{ {
Log.debug("PONG {}",this); Log.debug("PONG {}",this);
break; break;
} }
case WebSocketConnectionD7_9.OP_CLOSE: case WebSocketConnectionD10.OP_CLOSE:
{ {
int code=-1; int code=-1;
String message=null; String message=null;
@ -631,7 +639,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
} }
case WebSocketConnectionD7_9.OP_TEXT: case WebSocketConnectionD10.OP_TEXT:
{ {
if(_onTextMessage!=null) if(_onTextMessage!=null)
{ {
@ -646,12 +654,12 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{ {
// If this is a text fragment, append to buffer // If this is a text fragment, append to buffer
if (_utf8.append(buffer.array(),buffer.getIndex(),buffer.length(),_connection.getMaxTextMessageSize())) if (_utf8.append(buffer.array(),buffer.getIndex(),buffer.length(),_connection.getMaxTextMessageSize()))
_opcode=WebSocketConnectionD7_9.OP_TEXT; _opcode=WebSocketConnectionD10.OP_TEXT;
else else
{ {
_utf8.reset(); _utf8.reset();
_opcode=-1; _opcode=-1;
_connection.close(WebSocketConnectionD7_9.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars"); _connection.close(WebSocketConnectionD10.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars");
} }
} }
} }
@ -673,7 +681,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{ {
if (buffer.length()>_connection.getMaxBinaryMessageSize()) if (buffer.length()>_connection.getMaxBinaryMessageSize())
{ {
_connection.close(WebSocketConnectionD7_9.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize()); _connection.close(WebSocketConnectionD10.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize());
if (_aggregate!=null) if (_aggregate!=null)
_aggregate.clear(); _aggregate.clear();
_opcode=-1; _opcode=-1;
@ -711,7 +719,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
public String toString() public String toString()
{ {
return WebSocketConnectionD7_9.this.toString()+"FH"; return WebSocketConnectionD10.this.toString()+"FH";
} }
} }

5
jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketFactory.java
View File

@ -166,8 +166,9 @@ public class WebSocketFactory
case 7: case 7:
case 8: case 8:
case 9: case 9:
extensions= initExtensions(extensions_requested,8-WebSocketConnectionD7_9.OP_EXT_DATA, 16-WebSocketConnectionD7_9.OP_EXT_CTRL,3); case 10:
connection = new WebSocketConnectionD7_9(websocket, endp, _buffers, http.getTimeStamp(), _maxIdleTime, protocol,extensions,draft); extensions= initExtensions(extensions_requested,8-WebSocketConnectionD10.OP_EXT_DATA, 16-WebSocketConnectionD10.OP_EXT_CTRL,3);
connection = new WebSocketConnectionD10(websocket, endp, _buffers, http.getTimeStamp(), _maxIdleTime, protocol,extensions,draft);
break; break;
default: default:
Log.warn("Unsupported Websocket version: "+draft); Log.warn("Unsupported Websocket version: "+draft);

12
jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketGeneratorD7_9.java → jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketGeneratorD10.java
View File

@ -30,7 +30,7 @@ import org.eclipse.jetty.util.TypeUtil;
* threads will call the addMessage methods while other * threads will call the addMessage methods while other
* threads are flushing the generator. * threads are flushing the generator.
*/ */
public class WebSocketGeneratorD7_9 implements WebSocketGenerator public class WebSocketGeneratorD10 implements WebSocketGenerator
{ {
final private WebSocketBuffers _buffers; final private WebSocketBuffers _buffers;
final private EndPoint _endp; final private EndPoint _endp;
@ -80,7 +80,7 @@ public class WebSocketGeneratorD7_9 implements WebSocketGenerator
final Random _random; final Random _random;
public RandomMaskGen() public RandomMaskGen()
{ {
_random=new SecureRandom(); _random=new Random();
} }
public RandomMaskGen(Random random) public RandomMaskGen(Random random)
@ -95,14 +95,14 @@ public class WebSocketGeneratorD7_9 implements WebSocketGenerator
} }
public WebSocketGeneratorD7_9(WebSocketBuffers buffers, EndPoint endp) public WebSocketGeneratorD10(WebSocketBuffers buffers, EndPoint endp)
{ {
_buffers=buffers; _buffers=buffers;
_endp=endp; _endp=endp;
_maskGen=null; _maskGen=null;
} }
public WebSocketGeneratorD7_9(WebSocketBuffers buffers, EndPoint endp, MaskGen maskGen) public WebSocketGeneratorD10(WebSocketBuffers buffers, EndPoint endp, MaskGen maskGen)
{ {
_buffers=buffers; _buffers=buffers;
_endp=endp; _endp=endp;
@ -118,14 +118,14 @@ public class WebSocketGeneratorD7_9 implements WebSocketGenerator
if (_buffer==null) if (_buffer==null)
_buffer=mask?_buffers.getBuffer():_buffers.getDirectBuffer(); _buffer=mask?_buffers.getBuffer():_buffers.getDirectBuffer();
boolean last=WebSocketConnectionD7_9.isLastFrame(flags); boolean last=WebSocketConnectionD10.isLastFrame(flags);
byte orig=opcode; byte orig=opcode;
int space=mask?14:10; int space=mask?14:10;
do do
{ {
opcode = _opsent?WebSocketConnectionD7_9.OP_CONTINUATION:opcode; opcode = _opsent?WebSocketConnectionD10.OP_CONTINUATION:opcode;
opcode=(byte)(((0xf&flags)<<4)+(0xf&opcode)); opcode=(byte)(((0xf&flags)<<4)+(0xf&opcode));
_opsent=true; _opsent=true;

14
jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketParserD7_9.java → jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketParserD10.java
View File

@ -29,7 +29,7 @@ import org.eclipse.jetty.util.log.Log;
* Parser the WebSocket protocol. * Parser the WebSocket protocol.
* *
*/ */
public class WebSocketParserD7_9 implements WebSocketParser public class WebSocketParserD10 implements WebSocketParser
{ {
public enum State { public enum State {
@ -72,7 +72,7 @@ public class WebSocketParserD7_9 implements WebSocketParser
* @param endp * @param endp
* @param handler * @param handler
*/ */
public WebSocketParserD7_9(WebSocketBuffers buffers, EndPoint endp, FrameHandler handler, boolean shouldBeMasked) public WebSocketParserD10(WebSocketBuffers buffers, EndPoint endp, FrameHandler handler, boolean shouldBeMasked)
{ {
_buffers=buffers; _buffers=buffers;
_endp=endp; _endp=endp;
@ -160,11 +160,11 @@ public class WebSocketParserD7_9 implements WebSocketParser
_opcode=(byte)(b&0xf); _opcode=(byte)(b&0xf);
_flags=(byte)(0xf&(b>>4)); _flags=(byte)(0xf&(b>>4));
if (WebSocketConnectionD7_9.isControlFrame(_opcode)&&!WebSocketConnectionD7_9.isLastFrame(_flags)) if (WebSocketConnectionD10.isControlFrame(_opcode)&&!WebSocketConnectionD10.isLastFrame(_flags))
{ {
events++; events++;
Log.warn("Fragmented Control from "+_endp); Log.warn("Fragmented Control from "+_endp);
_handler.close(WebSocketConnectionD7_9.CLOSE_PROTOCOL,"Fragmented control"); _handler.close(WebSocketConnectionD10.CLOSE_PROTOCOL,"Fragmented control");
_skip=true; _skip=true;
} }
@ -205,7 +205,7 @@ public class WebSocketParserD7_9 implements WebSocketParser
if (_length>_buffer.capacity()) if (_length>_buffer.capacity())
{ {
events++; events++;
_handler.close(WebSocketConnectionD7_9.CLOSE_LARGE,"frame size "+_length+">"+_buffer.capacity()); _handler.close(WebSocketConnectionD10.CLOSE_LARGE,"frame size "+_length+">"+_buffer.capacity());
_skip=true; _skip=true;
} }
@ -224,7 +224,7 @@ public class WebSocketParserD7_9 implements WebSocketParser
if (_length>=_buffer.capacity()) if (_length>=_buffer.capacity())
{ {
events++; events++;
_handler.close(WebSocketConnectionD7_9.CLOSE_LARGE,"frame size "+_length+">"+_buffer.capacity()); _handler.close(WebSocketConnectionD10.CLOSE_LARGE,"frame size "+_length+">"+_buffer.capacity());
_skip=true; _skip=true;
} }
@ -267,7 +267,7 @@ public class WebSocketParserD7_9 implements WebSocketParser
_buffer.skip(_bytesNeeded); _buffer.skip(_bytesNeeded);
_state=State.START; _state=State.START;
events++; events++;
_handler.close(WebSocketConnectionD7_9.CLOSE_PROTOCOL,"bad mask"); _handler.close(WebSocketConnectionD10.CLOSE_PROTOCOL,"bad mask");
} }
else else
{ {

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

@ -20,7 +20,7 @@ public class WebSocketGeneratorD7_9Test
byte[] _mask = new byte[4]; byte[] _mask = new byte[4];
int _m; int _m;
public WebSocketGeneratorD7_9.MaskGen _maskGen = new WebSocketGeneratorD7_9.FixedMaskGen( public WebSocketGeneratorD10.MaskGen _maskGen = new WebSocketGeneratorD10.FixedMaskGen(
new byte[]{(byte)0x00,(byte)0x00,(byte)0x0f,(byte)0xff}); new byte[]{(byte)0x00,(byte)0x00,(byte)0x0f,(byte)0xff});
@Before @Before
@ -42,7 +42,7 @@ public class WebSocketGeneratorD7_9Test
@Test @Test
public void testOneString() throws Exception public void testOneString() throws Exception
{ {
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,null); _generator = new WebSocketGeneratorD10(_buffers, _endPoint,null);
byte[] data = "Hell\uFF4F W\uFF4Frld".getBytes(StringUtil.__UTF8); byte[] data = "Hell\uFF4F W\uFF4Frld".getBytes(StringUtil.__UTF8);
_generator.addFrame((byte)0x8,(byte)0x04,data,0,data.length); _generator.addFrame((byte)0x8,(byte)0x04,data,0,data.length);
@ -69,7 +69,7 @@ public class WebSocketGeneratorD7_9Test
@Test @Test
public void testOneBuffer() throws Exception public void testOneBuffer() throws Exception
{ {
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,null); _generator = new WebSocketGeneratorD10(_buffers, _endPoint,null);
String string = "Hell\uFF4F W\uFF4Frld"; String string = "Hell\uFF4F W\uFF4Frld";
byte[] bytes=string.getBytes(StringUtil.__UTF8); byte[] bytes=string.getBytes(StringUtil.__UTF8);
@ -97,7 +97,7 @@ public class WebSocketGeneratorD7_9Test
@Test @Test
public void testOneLongBuffer() throws Exception public void testOneLongBuffer() throws Exception
{ {
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,null); _generator = new WebSocketGeneratorD10(_buffers, _endPoint,null);
byte[] b=new byte[150]; byte[] b=new byte[150];
for (int i=0;i<b.length;i++) for (int i=0;i<b.length;i++)
@ -118,7 +118,7 @@ public class WebSocketGeneratorD7_9Test
@Test @Test
public void testOneStringMasked() throws Exception public void testOneStringMasked() throws Exception
{ {
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,_maskGen); _generator = new WebSocketGeneratorD10(_buffers, _endPoint,_maskGen);
byte[] data = "Hell\uFF4F W\uFF4Frld".getBytes(StringUtil.__UTF8); byte[] data = "Hell\uFF4F W\uFF4Frld".getBytes(StringUtil.__UTF8);
_generator.addFrame((byte)0x8,(byte)0x04,data,0,data.length); _generator.addFrame((byte)0x8,(byte)0x04,data,0,data.length);
@ -147,7 +147,7 @@ public class WebSocketGeneratorD7_9Test
@Test @Test
public void testOneBufferMasked() throws Exception public void testOneBufferMasked() throws Exception
{ {
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,_maskGen); _generator = new WebSocketGeneratorD10(_buffers, _endPoint,_maskGen);
String string = "Hell\uFF4F W\uFF4Frld"; String string = "Hell\uFF4F W\uFF4Frld";
byte[] bytes=string.getBytes(StringUtil.__UTF8); byte[] bytes=string.getBytes(StringUtil.__UTF8);
@ -177,7 +177,7 @@ public class WebSocketGeneratorD7_9Test
@Test @Test
public void testOneLongBufferMasked() throws Exception public void testOneLongBufferMasked() throws Exception
{ {
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,_maskGen); _generator = new WebSocketGeneratorD10(_buffers, _endPoint,_maskGen);
byte[] b=new byte[150]; byte[] b=new byte[150];
for (int i=0;i<b.length;i++) for (int i=0;i<b.length;i++)

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

@ -142,8 +142,8 @@ public class WebSocketLoadD7_9Test
private final int iterations; private final int iterations;
private final CountDownLatch latch; private final CountDownLatch latch;
private final SocketEndPoint _endp; private final SocketEndPoint _endp;
private final WebSocketGeneratorD7_9 _generator; private final WebSocketGeneratorD10 _generator;
private final WebSocketParserD7_9 _parser; private final WebSocketParserD10 _parser;
private final WebSocketParser.FrameHandler _handler = new WebSocketParser.FrameHandler() private final WebSocketParser.FrameHandler _handler = new WebSocketParser.FrameHandler()
{ {
public void onFrame(byte flags, byte opcode, Buffer buffer) public void onFrame(byte flags, byte opcode, Buffer buffer)
@ -167,8 +167,8 @@ public class WebSocketLoadD7_9Test
this.iterations = iterations; this.iterations = iterations;
_endp=new SocketEndPoint(socket); _endp=new SocketEndPoint(socket);
_generator = new WebSocketGeneratorD7_9(new WebSocketBuffers(32*1024),_endp,new WebSocketGeneratorD7_9.FixedMaskGen()); _generator = new WebSocketGeneratorD10(new WebSocketBuffers(32*1024),_endp,new WebSocketGeneratorD10.FixedMaskGen());
_parser = new WebSocketParserD7_9(new WebSocketBuffers(32*1024),_endp,_handler,false); _parser = new WebSocketParserD10(new WebSocketBuffers(32*1024),_endp,_handler,false);
} }
@ -202,7 +202,7 @@ public class WebSocketLoadD7_9Test
for (int i = 0; i < iterations; ++i) for (int i = 0; i < iterations; ++i)
{ {
byte[] data = message.getBytes(StringUtil.__UTF8); byte[] data = message.getBytes(StringUtil.__UTF8);
_generator.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_TEXT,data,0,data.length); _generator.addFrame((byte)0x8,WebSocketConnectionD10.OP_TEXT,data,0,data.length);
_generator.flush(); _generator.flush();
//System.err.println("-> "+message); //System.err.println("-> "+message);

34
jetty-websocket/src/test/java/org/eclipse/jetty/websocket/WebSocketMessageD7_9Test.java
View File

@ -75,7 +75,7 @@ public class WebSocketMessageD7_9Test
@Test @Test
public void testHash() public void testHash()
{ {
assertEquals("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=",WebSocketConnectionD7_9.hashKey("dGhlIHNhbXBsZSBub25jZQ==")); assertEquals("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=",WebSocketConnectionD10.hashKey("dGhlIHNhbXBsZSBub25jZQ=="));
} }
@Test @Test
@ -116,7 +116,7 @@ public class WebSocketMessageD7_9Test
String data=message.toString(); String data=message.toString();
_serverWebSocket.connection.sendMessage(data); _serverWebSocket.connection.sendMessage(data);
assertEquals(WebSocketConnectionD7_9.OP_TEXT,input.read()); assertEquals(WebSocketConnectionD10.OP_TEXT,input.read());
assertEquals(0x7e,input.read()); assertEquals(0x7e,input.read());
assertEquals(0x1f,input.read()); assertEquals(0x1f,input.read());
assertEquals(0xf6,input.read()); assertEquals(0xf6,input.read());
@ -322,7 +322,7 @@ public class WebSocketMessageD7_9Test
output.write(buf,0,l+3); output.write(buf,0,l+3);
output.flush(); output.flush();
assertEquals(0x40+WebSocketConnectionD7_9.OP_TEXT,input.read()); assertEquals(0x40+WebSocketConnectionD10.OP_TEXT,input.read());
assertEquals(0x20+3,input.read()); assertEquals(0x20+3,input.read());
assertEquals(0x7e,input.read()); assertEquals(0x7e,input.read());
assertEquals(0x02,input.read()); assertEquals(0x02,input.read());
@ -490,7 +490,7 @@ public class WebSocketMessageD7_9Test
output.write(bytes[i]^0xff); output.write(bytes[i]^0xff);
output.flush(); output.flush();
assertEquals(0x80|WebSocketConnectionD7_9.OP_CLOSE,input.read()); assertEquals(0x80|WebSocketConnectionD10.OP_CLOSE,input.read());
assertEquals(30,input.read()); assertEquals(30,input.read());
int code=(0xff&input.read())*0x100+(0xff&input.read()); int code=(0xff&input.read())*0x100+(0xff&input.read());
assertEquals(1004,code); assertEquals(1004,code);
@ -541,7 +541,7 @@ public class WebSocketMessageD7_9Test
assertEquals(0x80|WebSocketConnectionD7_9.OP_CLOSE,input.read()); assertEquals(0x80|WebSocketConnectionD10.OP_CLOSE,input.read());
assertEquals(30,input.read()); assertEquals(30,input.read());
int code=(0xff&input.read())*0x100+(0xff&input.read()); int code=(0xff&input.read())*0x100+(0xff&input.read());
assertEquals(1004,code); assertEquals(1004,code);
@ -577,7 +577,7 @@ public class WebSocketMessageD7_9Test
assertNotNull(_serverWebSocket.connection); assertNotNull(_serverWebSocket.connection);
_serverWebSocket.getConnection().setMaxBinaryMessageSize(1024); _serverWebSocket.getConnection().setMaxBinaryMessageSize(1024);
output.write(WebSocketConnectionD7_9.OP_BINARY); output.write(WebSocketConnectionD10.OP_BINARY);
output.write(0x8a); output.write(0x8a);
output.write(0xff); output.write(0xff);
output.write(0xff); output.write(0xff);
@ -598,7 +598,7 @@ public class WebSocketMessageD7_9Test
output.write(bytes[i]^0xff); output.write(bytes[i]^0xff);
output.flush(); output.flush();
assertEquals(0x80+WebSocketConnectionD7_9.OP_BINARY,input.read()); assertEquals(0x80+WebSocketConnectionD10.OP_BINARY,input.read());
assertEquals(20,input.read()); assertEquals(20,input.read());
lookFor("01234567890123456789",input); lookFor("01234567890123456789",input);
} }
@ -655,7 +655,7 @@ public class WebSocketMessageD7_9Test
output.flush(); output.flush();
assertEquals(0x80|WebSocketConnectionD7_9.OP_CLOSE,input.read()); assertEquals(0x80|WebSocketConnectionD10.OP_CLOSE,input.read());
assertEquals(19,input.read()); assertEquals(19,input.read());
int code=(0xff&input.read())*0x100+(0xff&input.read()); int code=(0xff&input.read())*0x100+(0xff&input.read());
assertEquals(1004,code); assertEquals(1004,code);
@ -704,7 +704,7 @@ public class WebSocketMessageD7_9Test
output.write(bytes[i]^0xff); output.write(bytes[i]^0xff);
output.flush(); output.flush();
assertEquals(0x80|WebSocketConnectionD7_9.OP_CLOSE,input.read()); assertEquals(0x80|WebSocketConnectionD10.OP_CLOSE,input.read());
assertEquals(19,input.read()); assertEquals(19,input.read());
int code=(0xff&input.read())*0x100+(0xff&input.read()); int code=(0xff&input.read())*0x100+(0xff&input.read());
assertEquals(1004,code); assertEquals(1004,code);
@ -830,14 +830,14 @@ public class WebSocketMessageD7_9Test
final AtomicReference<String> received = new AtomicReference<String>(); final AtomicReference<String> received = new AtomicReference<String>();
ByteArrayEndPoint endp = new ByteArrayEndPoint(new byte[0],4096); ByteArrayEndPoint endp = new ByteArrayEndPoint(new byte[0],4096);
WebSocketGeneratorD7_9 gen = new WebSocketGeneratorD7_9(new WebSocketBuffers(8096),endp,null); WebSocketGeneratorD10 gen = new WebSocketGeneratorD10(new WebSocketBuffers(8096),endp,null);
byte[] data = message.getBytes(StringUtil.__UTF8); byte[] data = message.getBytes(StringUtil.__UTF8);
gen.addFrame((byte)0x8,(byte)0x4,data,0,data.length); gen.addFrame((byte)0x8,(byte)0x4,data,0,data.length);
endp = new ByteArrayEndPoint(endp.getOut().asArray(),4096); endp = new ByteArrayEndPoint(endp.getOut().asArray(),4096);
WebSocketParserD7_9 parser = new WebSocketParserD7_9(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler() WebSocketParserD10 parser = new WebSocketParserD10(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler()
{ {
public void onFrame(byte flags, byte opcode, Buffer buffer) public void onFrame(byte flags, byte opcode, Buffer buffer)
{ {
@ -862,15 +862,15 @@ public class WebSocketMessageD7_9Test
final AtomicReference<String> received = new AtomicReference<String>(); final AtomicReference<String> received = new AtomicReference<String>();
ByteArrayEndPoint endp = new ByteArrayEndPoint(new byte[0],4096); ByteArrayEndPoint endp = new ByteArrayEndPoint(new byte[0],4096);
WebSocketGeneratorD7_9.MaskGen maskGen = new WebSocketGeneratorD7_9.RandomMaskGen(); WebSocketGeneratorD10.MaskGen maskGen = new WebSocketGeneratorD10.RandomMaskGen();
WebSocketGeneratorD7_9 gen = new WebSocketGeneratorD7_9(new WebSocketBuffers(8096),endp,maskGen); WebSocketGeneratorD10 gen = new WebSocketGeneratorD10(new WebSocketBuffers(8096),endp,maskGen);
byte[] data = message.getBytes(StringUtil.__UTF8); byte[] data = message.getBytes(StringUtil.__UTF8);
gen.addFrame((byte)0x8,(byte)0x1,data,0,data.length); gen.addFrame((byte)0x8,(byte)0x1,data,0,data.length);
endp = new ByteArrayEndPoint(endp.getOut().asArray(),4096); endp = new ByteArrayEndPoint(endp.getOut().asArray(),4096);
WebSocketParserD7_9 parser = new WebSocketParserD7_9(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler() WebSocketParserD10 parser = new WebSocketParserD10(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler()
{ {
public void onFrame(byte flags, byte opcode, Buffer buffer) public void onFrame(byte flags, byte opcode, Buffer buffer)
{ {
@ -993,9 +993,9 @@ public class WebSocketMessageD7_9Test
{ {
switch(opcode) switch(opcode)
{ {
case WebSocketConnectionD7_9.OP_CLOSE: case WebSocketConnectionD10.OP_CLOSE:
case WebSocketConnectionD7_9.OP_PING: case WebSocketConnectionD10.OP_PING:
case WebSocketConnectionD7_9.OP_PONG: case WebSocketConnectionD10.OP_PONG:
break; break;
default: default:

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

@ -87,7 +87,7 @@ public class WebSocketParserD7_9Test
ByteArrayEndPoint endPoint = new ByteArrayEndPoint(); ByteArrayEndPoint endPoint = new ByteArrayEndPoint();
endPoint.setNonBlocking(true); endPoint.setNonBlocking(true);
_handler = new Handler(); _handler = new Handler();
_parser=new WebSocketParserD7_9(buffers, endPoint,_handler,true); _parser=new WebSocketParserD10(buffers, endPoint,_handler,true);
_in = new MaskedByteArrayBuffer(); _in = new MaskedByteArrayBuffer();
endPoint.setIn(_in); endPoint.setIn(_in);
@ -187,7 +187,7 @@ public class WebSocketParserD7_9Test
{ {
WebSocketBuffers buffers = new WebSocketBuffers(0x20000); WebSocketBuffers buffers = new WebSocketBuffers(0x20000);
ByteArrayEndPoint endPoint = new ByteArrayEndPoint(); ByteArrayEndPoint endPoint = new ByteArrayEndPoint();
WebSocketParser parser=new WebSocketParserD7_9(buffers, endPoint,_handler,false); WebSocketParser parser=new WebSocketParserD10(buffers, endPoint,_handler,false);
ByteArrayBuffer in = new ByteArrayBuffer(0x20000); ByteArrayBuffer in = new ByteArrayBuffer(0x20000);
endPoint.setIn(in); endPoint.setIn(in);
@ -261,7 +261,7 @@ public class WebSocketParserD7_9Test
assertTrue(progress>0); assertTrue(progress>0);
assertEquals(WebSocketConnectionD7_9.CLOSE_LARGE,_handler._code); assertEquals(WebSocketConnectionD10.CLOSE_LARGE,_handler._code);
for (int i=0;i<2048;i++) for (int i=0;i<2048;i++)
_in.put((byte)'a'); _in.put((byte)'a');
progress =_parser.parseNext(); progress =_parser.parseNext();