Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Move processing of frames to drafts & improve handling of IOExceptions #538

Merged
merged 2 commits into from
Aug 20, 2017
Merged

Move processing of frames to drafts & improve handling of IOExceptions #538

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bfe8813
Move processing of frames to draft
marci4 Aug 16, 2017
ada0bf4
Improve handling of IOExceptions causing eot()
marci4 Aug 16, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
188 changes: 47 additions & 141 deletions src/main/java/org/java_websocket/WebSocketImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@
import org.java_websocket.framing.CloseFrame;
import org.java_websocket.framing.Framedata;
import org.java_websocket.framing.Framedata.Opcode;
import org.java_websocket.framing.FramedataImpl1;
import org.java_websocket.framing.PingFrame;
import org.java_websocket.handshake.*;
import org.java_websocket.server.WebSocketServer.WebSocketWorker;
Expand All @@ -48,10 +47,7 @@
import java.nio.channels.ByteChannel;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.SelectionKey;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

Expand All @@ -63,9 +59,9 @@
public class WebSocketImpl implements WebSocket {
public static int RCVBUF = 16384;

/**
* Activate debug mode for additional infos
*/
/**
* Activate debug mode for additional infos
*/
public static boolean DEBUG = false; // must be final in the future in order to take advantage of VM optimization

/**
Expand Down Expand Up @@ -95,26 +91,21 @@ public class WebSocketImpl implements WebSocket {
private volatile boolean flushandclosestate = false;
private READYSTATE readystate = READYSTATE.NOT_YET_CONNECTED;

/**
* A list of drafts available for this websocket
*/
/**
* A list of drafts available for this websocket
*/
private List<Draft> knownDrafts;

/**
* The draft which is used by this websocket
*/
/**
* The draft which is used by this websocket
*/
private Draft draft = null;

/**
* The role which this websocket takes in the connection
*/
/**
* The role which this websocket takes in the connection
*/
private Role role;

/**
* The frame which had the opcode Continous set
*/
private Framedata current_continuous_frame = null;

/**
* the bytes of an incomplete received handshake
*/
Expand Down Expand Up @@ -359,6 +350,22 @@ private boolean decodeHandshake( ByteBuffer socketBufferNew ) {
return false;
}

private void decodeFrames( ByteBuffer socketBuffer ) {
List<Framedata> frames;
try {
frames = draft.translateFrame( socketBuffer );
for( Framedata f : frames ) {
if( DEBUG )
System.out.println( "matched frame: " + f );
draft.processFrame( this, f );
}
} catch ( InvalidDataException e1 ) {
wsl.onWebsocketError( this, e1 );
close( e1 );
return;
}
}

/**
* Close the connection if the received handshake was not correct
* @param exception the InvalidDataException causing this problem
Expand Down Expand Up @@ -395,115 +402,7 @@ private ByteBuffer generateHttpResponseDueToError(int errorCode) {
return ByteBuffer.wrap( Charsetfunctions.asciiBytes( "HTTP/1.1 "+ errorCodeDescription +"\r\nContent-Type: text/html\nServer: TooTallNate Java-WebSocket\r\nContent-Length: " + (48 + errorCodeDescription.length()) +"\r\n\r\n<html><head></head><body><h1>" + errorCodeDescription + "</h1></body></html>" ));
}

private void decodeFrames( ByteBuffer socketBuffer ) {

List<Framedata> frames;
try {
frames = draft.translateFrame( socketBuffer );
for( Framedata f : frames ) {
if( DEBUG )
System.out.println( "matched frame: " + f );
Opcode curop = f.getOpcode();
boolean fin = f.isFin();
//Not evaluating any further frames if the connection is in READYSTATE CLOSE
if( readystate == READYSTATE.CLOSING )
return;

if( curop == Opcode.CLOSING ) {
int code = CloseFrame.NOCODE;
String reason = "";
if( f instanceof CloseFrame ) {
CloseFrame cf = ( CloseFrame ) f;
code = cf.getCloseCode();
reason = cf.getMessage();
}
if( readystate == READYSTATE.CLOSING ) {
// complete the close handshake by disconnecting
closeConnection( code, reason, true );
} else {
// echo close handshake
if( draft.getCloseHandshakeType() == CloseHandshakeType.TWOWAY )
close( code, reason, true );
else
flushAndClose( code, reason, false );
}
continue;
} else if( curop == Opcode.PING ) {
wsl.onWebsocketPing( this, f );
continue;
} else if( curop == Opcode.PONG ) {
lastPong = System.currentTimeMillis();
wsl.onWebsocketPong( this, f );
continue;
} else if( !fin || curop == Opcode.CONTINUOUS ) {
if( curop != Opcode.CONTINUOUS ) {
if( current_continuous_frame != null )
throw new InvalidDataException( CloseFrame.PROTOCOL_ERROR, "Previous continuous frame sequence not completed." );
current_continuous_frame = f;
} else if( fin ) {
if( current_continuous_frame == null )
throw new InvalidDataException( CloseFrame.PROTOCOL_ERROR, "Continuous frame sequence was not started." );
//Check if the whole payload is valid utf8, when the opcode indicates a text
if( current_continuous_frame.getOpcode() == Opcode.TEXT ) {
//Checking a bit more from the frame before this one just to make sure all the code points are correct
int off = Math.max( current_continuous_frame.getPayloadData().limit() - 64, 0 );
current_continuous_frame.append( f );
if( !Charsetfunctions.isValidUTF8( current_continuous_frame.getPayloadData(), off ) ) {
throw new InvalidDataException( CloseFrame.NO_UTF8 );
}
}
current_continuous_frame = null;
} else if( current_continuous_frame == null ) {
throw new InvalidDataException( CloseFrame.PROTOCOL_ERROR, "Continuous frame sequence was not started." );
}
//Check if the whole payload is valid utf8, when the opcode indicates a text
if( curop == Opcode.TEXT ) {
if( !Charsetfunctions.isValidUTF8( f.getPayloadData() ) ) {
throw new InvalidDataException( CloseFrame.NO_UTF8 );
}
}
//Checking if the current continous frame contains a correct payload with the other frames combined
if( curop == Opcode.CONTINUOUS && current_continuous_frame != null && current_continuous_frame.getOpcode() == Opcode.TEXT ) {
//Checking a bit more from the frame before this one just to make sure all the code points are correct
int off = Math.max( current_continuous_frame.getPayloadData().limit() - 64, 0 );
current_continuous_frame.append( f );
if( !Charsetfunctions.isValidUTF8( current_continuous_frame.getPayloadData(), off ) ) {
throw new InvalidDataException( CloseFrame.NO_UTF8 );
}
}
try {
wsl.onWebsocketMessageFragment( this, f );
} catch ( RuntimeException e ) {
wsl.onWebsocketError( this, e );
}

} else if( current_continuous_frame != null ) {
throw new InvalidDataException( CloseFrame.PROTOCOL_ERROR, "Continuous frame sequence not completed." );
} else if( curop == Opcode.TEXT ) {
try {
wsl.onWebsocketMessage( this, Charsetfunctions.stringUtf8( f.getPayloadData() ) );
} catch ( RuntimeException e ) {
wsl.onWebsocketError( this, e );
}
} else if( curop == Opcode.BINARY ) {
try {
wsl.onWebsocketMessage( this, f.getPayloadData() );
} catch ( RuntimeException e ) {
wsl.onWebsocketError( this, e );
}
} else {
throw new InvalidDataException( CloseFrame.PROTOCOL_ERROR, "non control or continious frame expected" );
}
}
} catch ( InvalidDataException e1 ) {
wsl.onWebsocketError( this, e1 );
close( e1 );
return;
}

}

private void close( int code, String message, boolean remote ) {
public void close( int code, String message, boolean remote ) {
if( readystate != READYSTATE.CLOSING && readystate != READYSTATE.CLOSED ) {
if( readystate == READYSTATE.OPEN ) {
if( code == CloseFrame.ABNORMAL_CLOSE ) {
Expand All @@ -524,13 +423,8 @@ private void close( int code, String message, boolean remote ) {
CloseFrame closeFrame = new CloseFrame();
closeFrame.setReason(message);
closeFrame.setCode(code);
try {
closeFrame.isValid();
sendFrame(closeFrame);
} catch (InvalidDataException e) {
//Rethrow invalid data exception
throw e;
}
closeFrame.isValid();
sendFrame(closeFrame);
} catch ( InvalidDataException e ) {
wsl.onWebsocketError( this, e );
flushAndClose( CloseFrame.ABNORMAL_CLOSE, "generated frame is invalid", false );
Expand Down Expand Up @@ -566,7 +460,7 @@ public void close( int code, String message ) {
* false means this endpoint decided to send the given code,<br>
* <code>remote</code> may also be true if this endpoint started the closing handshake since the other endpoint may not simply echo the <code>code</code> but close the connection the same time this endpoint does do but with an other <code>code</code>. <br>
**/
protected synchronized void closeConnection( int code, String message, boolean remote ) {
public synchronized void closeConnection( int code, String message, boolean remote ) {
if( readystate == READYSTATE.CLOSED ) {
return;
}
Expand Down Expand Up @@ -610,7 +504,7 @@ public void closeConnection( int code, String message ) {
closeConnection( code, message, false );
}

protected synchronized void flushAndClose( int code, String message, boolean remote ) {
public synchronized void flushAndClose( int code, String message, boolean remote ) {
if( flushandclosestate ) {
return;
}
Expand Down Expand Up @@ -694,7 +588,7 @@ private void send( Collection<Framedata> frames ) {
for (Framedata f : frames) {
if( DEBUG )
System.out.println( "send frame: " + f );
outgoingFrames.add( draft.createBinaryFrame( f ) );
outgoingFrames.add( draft.createBinaryFrame( f ) );
}
write( outgoingFrames );
}
Expand Down Expand Up @@ -868,4 +762,16 @@ public String getResourceDescriptor() {
long getLastPong() {
return lastPong;
}

public void setLastPong( long lastPong ) {
this.lastPong = lastPong;
}

/**
* Getter for the websocket listener
* @return the websocket listener associated with this instance
*/
public WebSocketListener getWebSocketListener() {
return wsl;
}
}
20 changes: 17 additions & 3 deletions src/main/java/org/java_websocket/client/WebSocketClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import java.util.concurrent.CountDownLatch;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLSocketFactory;

import org.java_websocket.AbstractWebSocket;
Expand Down Expand Up @@ -278,14 +279,15 @@ public void run() {
}
engine.eot();
} catch ( IOException e ) {
engine.eot();
handleIOException(e);
} catch ( RuntimeException e ) {
// this catch case covers internal errors only and indicates a bug in this websocket implementation
onError( e );
engine.closeConnection( CloseFrame.ABNORMAL_CLOSE, e.getMessage() );
}
assert ( socket.isClosed() );
}

private int getPort() {
int port = uri.getPort();
if( port == -1 ) {
Expand Down Expand Up @@ -457,13 +459,13 @@ private class WebsocketWriteThread implements Runnable {
public void run() {
Thread.currentThread().setName( "WebsocketWriteThread" );
try {
while ( !Thread.interrupted() ) {
while( !Thread.interrupted() ) {
ByteBuffer buffer = engine.outQueue.take();
ostream.write( buffer.array(), 0, buffer.limit() );
ostream.flush();
}
} catch ( IOException e ) {
engine.eot();
handleIOException(e);
} catch ( InterruptedException e ) {
// this thread is regularly terminated via an interrupt
}
Expand Down Expand Up @@ -562,4 +564,16 @@ public InetSocketAddress getRemoteSocketAddress() {
public String getResourceDescriptor() {
return uri.getPath();
}


/**
* Method to give some additional info for specific IOExceptions
* @param e the IOException causing a eot.
*/
private void handleIOException( IOException e ) {
if (e instanceof SSLException) {
onError( e );
}
engine.eot();
}
}
9 changes: 9 additions & 0 deletions src/main/java/org/java_websocket/drafts/Draft.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import java.util.Locale;

import org.java_websocket.WebSocket.Role;
import org.java_websocket.WebSocketImpl;
import org.java_websocket.exceptions.IncompleteHandshakeException;
import org.java_websocket.exceptions.InvalidDataException;
import org.java_websocket.exceptions.InvalidHandshakeException;
Expand Down Expand Up @@ -161,6 +162,14 @@ protected boolean basicAccept( Handshakedata handshakedata ) {

public abstract List<Framedata> createFrames( String text, boolean mask );


/**
* Handle the frame specific to the draft
* @param webSocketImpl the websocketimpl used for this draft
* @param frame the frame which is supposed to be handled
*/
public abstract void processFrame( WebSocketImpl webSocketImpl, Framedata frame ) throws InvalidDataException;

public List<Framedata> continuousFrame( Opcode op, ByteBuffer buffer, boolean fin ) {
if(op != Opcode.BINARY && op != Opcode.TEXT) {
throw new IllegalArgumentException( "Only Opcode.BINARY or Opcode.TEXT are allowed" );
Expand Down
5 changes: 5 additions & 0 deletions src/main/java/org/java_websocket/drafts/Draft_10.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
package org.java_websocket.drafts;

import org.java_websocket.WebSocket.Role;
import org.java_websocket.WebSocketImpl;
import org.java_websocket.exceptions.*;
import org.java_websocket.framing.*;
import org.java_websocket.framing.Framedata.Opcode;
Expand Down Expand Up @@ -170,6 +171,10 @@ public List<Framedata> createFrames( String text, boolean mask ) {
return Collections.singletonList( ( Framedata ) curframe );
}

public void processFrame( WebSocketImpl webSocketImpl, Framedata frame ) throws InvalidDataException {
throw new UnsupportedOperationException( "This draft is not supported any more. Please use Draft_6455." );
}

private byte fromOpcode( Opcode opcode ) {
if( opcode == Opcode.CONTINUOUS )
return 0;
Expand Down
Loading