andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fixed formatting.

Browse Source
This commit is contained in:
Veebs 2011-11-29 10:12:42 +11:00
parent 48addae927
commit 181355665b
1 changed files with 280 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

560
src/main/java/org/jboss/netty/handler/codec/http/websocketx/WebSocket08FrameDecoder.java
View File

@ -59,325 +59,325 @@ import org.jboss.netty.logging.InternalLoggerFactory;
*/ */
public class WebSocket08FrameDecoder extends ReplayingDecoder<WebSocket08FrameDecoder.State> { public class WebSocket08FrameDecoder extends ReplayingDecoder<WebSocket08FrameDecoder.State> {
private static final InternalLogger logger = InternalLoggerFactory.getInstance(WebSocket08FrameDecoder.class); private static final InternalLogger logger = InternalLoggerFactory.getInstance(WebSocket08FrameDecoder.class);
private static final byte OPCODE_CONT = 0x0; private static final byte OPCODE_CONT = 0x0;
private static final byte OPCODE_TEXT = 0x1; private static final byte OPCODE_TEXT = 0x1;
private static final byte OPCODE_BINARY = 0x2; private static final byte OPCODE_BINARY = 0x2;
private static final byte OPCODE_CLOSE = 0x8; private static final byte OPCODE_CLOSE = 0x8;
private static final byte OPCODE_PING = 0x9; private static final byte OPCODE_PING = 0x9;
private static final byte OPCODE_PONG = 0xA; private static final byte OPCODE_PONG = 0xA;
private UTF8Output fragmentedFramesText = null; private UTF8Output fragmentedFramesText = null;
private int fragmentedFramesCount = 0; private int fragmentedFramesCount = 0;
private boolean frameFinalFlag; private boolean frameFinalFlag;
private int frameRsv; private int frameRsv;
private int frameOpcode; private int frameOpcode;
private long framePayloadLength; private long framePayloadLength;
private ChannelBuffer framePayload = null; private ChannelBuffer framePayload = null;
private int framePayloadBytesRead = 0; private int framePayloadBytesRead = 0;
private ChannelBuffer maskingKey; private ChannelBuffer maskingKey;
private boolean allowExtensions = false; private boolean allowExtensions = false;
private boolean maskedPayload = false; private boolean maskedPayload = false;
private boolean receivedClosingHandshake = false; private boolean receivedClosingHandshake = false;
public enum State { public enum State {
FRAME_START, MASKING_KEY, PAYLOAD, CORRUPT FRAME_START, MASKING_KEY, PAYLOAD, CORRUPT
} }
/** /**
* Constructor * Constructor
* *
* @param maskedPayload * @param maskedPayload
* Web socket servers must set this to true processed incoming * Web socket servers must set this to true processed incoming
* masked payload. Client implementations must set this to false. * masked payload. Client implementations must set this to false.
* @param allowExtensions * @param allowExtensions
* Flag to allow reserved extension bits to be used or not * Flag to allow reserved extension bits to be used or not
*/ */
public WebSocket08FrameDecoder(boolean maskedPayload, boolean allowExtensions) { public WebSocket08FrameDecoder(boolean maskedPayload, boolean allowExtensions) {
super(State.FRAME_START); super(State.FRAME_START);
this.maskedPayload = maskedPayload; this.maskedPayload = maskedPayload;
this.allowExtensions = allowExtensions; this.allowExtensions = allowExtensions;
} }
@Override @Override
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, State state) protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, State state)
throws Exception { throws Exception {
// Discard all data received if closing handshake was received before. // Discard all data received if closing handshake was received before.
if (receivedClosingHandshake) { if (receivedClosingHandshake) {
buffer.skipBytes(actualReadableBytes()); buffer.skipBytes(actualReadableBytes());
return null; return null;
} }
switch (state) { switch (state) {
case FRAME_START: case FRAME_START:
framePayloadBytesRead = 0; framePayloadBytesRead = 0;
framePayloadLength = -1; framePayloadLength = -1;
framePayload = null; framePayload = null;
// FIN, RSV, OPCODE // FIN, RSV, OPCODE
byte b = buffer.readByte(); byte b = buffer.readByte();
frameFinalFlag = (b & 0x80) != 0; frameFinalFlag = (b & 0x80) != 0;
frameRsv = (b & 0x70) >> 4; frameRsv = (b & 0x70) >> 4;
frameOpcode = (b & 0x0F); frameOpcode = (b & 0x0F);
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("Decoding WebSocket Frame opCode=" + frameOpcode); logger.debug("Decoding WebSocket Frame opCode=" + frameOpcode);
} }
// MASK, PAYLOAD LEN 1 // MASK, PAYLOAD LEN 1
b = buffer.readByte(); b = buffer.readByte();
boolean frameMasked = (b & 0x80) != 0; boolean frameMasked = (b & 0x80) != 0;
int framePayloadLen1 = (b & 0x7F); int framePayloadLen1 = (b & 0x7F);
if (frameRsv != 0 && !this.allowExtensions) { if (frameRsv != 0 && !this.allowExtensions) {
protocolViolation(channel, "RSV != 0 and no extension negotiated, RSV:" + frameRsv); protocolViolation(channel, "RSV != 0 and no extension negotiated, RSV:" + frameRsv);
return null; return null;
} }
if (this.maskedPayload && !frameMasked) { if (this.maskedPayload && !frameMasked) {
protocolViolation(channel, "unmasked client to server frame"); protocolViolation(channel, "unmasked client to server frame");
return null; return null;
} }
if (frameOpcode > 7) { // control frame (have MSB in opcode set) if (frameOpcode > 7) { // control frame (have MSB in opcode set)
// control frames MUST NOT be fragmented // control frames MUST NOT be fragmented
if (!frameFinalFlag) { if (!frameFinalFlag) {
protocolViolation(channel, "fragmented control frame"); protocolViolation(channel, "fragmented control frame");
return null; return null;
} }
// control frames MUST have payload 125 octets or less // control frames MUST have payload 125 octets or less
if (framePayloadLen1 > 125) { if (framePayloadLen1 > 125) {
protocolViolation(channel, "control frame with payload length > 125 octets"); protocolViolation(channel, "control frame with payload length > 125 octets");
return null; return null;
} }
// check for reserved control frame opcodes // check for reserved control frame opcodes
if (!(frameOpcode == OPCODE_CLOSE || frameOpcode == OPCODE_PING || frameOpcode == OPCODE_PONG)) { if (!(frameOpcode == OPCODE_CLOSE || frameOpcode == OPCODE_PING || frameOpcode == OPCODE_PONG)) {
protocolViolation(channel, "control frame using reserved opcode " + frameOpcode); protocolViolation(channel, "control frame using reserved opcode " + frameOpcode);
return null; return null;
} }
// close frame : if there is a body, the first two bytes of the // close frame : if there is a body, the first two bytes of the
// body MUST be a 2-byte unsigned integer (in network byte // body MUST be a 2-byte unsigned integer (in network byte
// order) representing a status code // order) representing a status code
if (frameOpcode == 8 && framePayloadLen1 == 1) { if (frameOpcode == 8 && framePayloadLen1 == 1) {
protocolViolation(channel, "received close control frame with payload len 1"); protocolViolation(channel, "received close control frame with payload len 1");
return null; return null;
} }
} else { // data frame } else { // data frame
// check for reserved data frame opcodes // check for reserved data frame opcodes
if (!(frameOpcode == OPCODE_CONT || frameOpcode == OPCODE_TEXT || frameOpcode == OPCODE_BINARY)) { if (!(frameOpcode == OPCODE_CONT || frameOpcode == OPCODE_TEXT || frameOpcode == OPCODE_BINARY)) {
protocolViolation(channel, "data frame using reserved opcode " + frameOpcode); protocolViolation(channel, "data frame using reserved opcode " + frameOpcode);
return null; return null;
} }
// check opcode vs message fragmentation state 1/2 // check opcode vs message fragmentation state 1/2
if (fragmentedFramesCount == 0 && frameOpcode == OPCODE_CONT) { if (fragmentedFramesCount == 0 && frameOpcode == OPCODE_CONT) {
protocolViolation(channel, "received continuation data frame outside fragmented message"); protocolViolation(channel, "received continuation data frame outside fragmented message");
return null; return null;
} }
// check opcode vs message fragmentation state 2/2 // check opcode vs message fragmentation state 2/2
if (fragmentedFramesCount != 0 && frameOpcode != OPCODE_CONT && frameOpcode != OPCODE_PING) { if (fragmentedFramesCount != 0 && frameOpcode != OPCODE_CONT && frameOpcode != OPCODE_PING) {
protocolViolation(channel, "received non-continuation data frame while inside fragmented message"); protocolViolation(channel, "received non-continuation data frame while inside fragmented message");
return null; return null;
} }
} }
// Read frame payload length // Read frame payload length
if (framePayloadLen1 == 126) { if (framePayloadLen1 == 126) {
framePayloadLength = buffer.readUnsignedShort(); framePayloadLength = buffer.readUnsignedShort();
if (framePayloadLength < 126) { if (framePayloadLength < 126) {
protocolViolation(channel, "invalid data frame length (not using minimal length encoding)"); protocolViolation(channel, "invalid data frame length (not using minimal length encoding)");
return null; return null;
} }
} else if (framePayloadLen1 == 127) { } else if (framePayloadLen1 == 127) {
framePayloadLength = buffer.readLong(); framePayloadLength = buffer.readLong();
// TODO: check if it's bigger than 0x7FFFFFFFFFFFFFFF, Maybe // TODO: check if it's bigger than 0x7FFFFFFFFFFFFFFF, Maybe
// just check if it's negative? // just check if it's negative?
if (framePayloadLength < 65536) { if (framePayloadLength < 65536) {
protocolViolation(channel, "invalid data frame length (not using minimal length encoding)"); protocolViolation(channel, "invalid data frame length (not using minimal length encoding)");
return null; return null;
} }
} else { } else {
framePayloadLength = framePayloadLen1; framePayloadLength = framePayloadLen1;
} }
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("Decoding WebSocket Frame length=" + framePayloadLength); logger.debug("Decoding WebSocket Frame length=" + framePayloadLength);
} }
checkpoint(State.MASKING_KEY); checkpoint(State.MASKING_KEY);
case MASKING_KEY: case MASKING_KEY:
if (this.maskedPayload) { if (this.maskedPayload) {
maskingKey = buffer.readBytes(4); maskingKey = buffer.readBytes(4);
} }
checkpoint(State.PAYLOAD); checkpoint(State.PAYLOAD);
case PAYLOAD: case PAYLOAD:
// Sometimes, the payload may not be delivered in 1 nice packet // Sometimes, the payload may not be delivered in 1 nice packet
// We need to accumulate the data until we have it all // We need to accumulate the data until we have it all
int rbytes = actualReadableBytes(); int rbytes = actualReadableBytes();
ChannelBuffer payloadBuffer = null; ChannelBuffer payloadBuffer = null;
int willHaveReadByteCount = framePayloadBytesRead + rbytes; int willHaveReadByteCount = framePayloadBytesRead + rbytes;
// logger.debug("Frame rbytes=" + rbytes + " willHaveReadByteCount=" // logger.debug("Frame rbytes=" + rbytes + " willHaveReadByteCount="
// + willHaveReadByteCount + " framePayloadLength=" + // + willHaveReadByteCount + " framePayloadLength=" +
// framePayloadLength); // framePayloadLength);
if (willHaveReadByteCount == framePayloadLength) { if (willHaveReadByteCount == framePayloadLength) {
// We have all our content so proceed to process // We have all our content so proceed to process
payloadBuffer = buffer.readBytes(rbytes); payloadBuffer = buffer.readBytes(rbytes);
} else if (willHaveReadByteCount < framePayloadLength) { } else if (willHaveReadByteCount < framePayloadLength) {
// We don't have all our content so accumulate payload. // We don't have all our content so accumulate payload.
// Returning null means we will get called back // Returning null means we will get called back
payloadBuffer = buffer.readBytes(rbytes); payloadBuffer = buffer.readBytes(rbytes);
if (framePayload == null) { if (framePayload == null) {
framePayload = channel.getConfig().getBufferFactory().getBuffer(toFrameLength(framePayloadLength)); framePayload = channel.getConfig().getBufferFactory().getBuffer(toFrameLength(framePayloadLength));
} }
framePayload.writeBytes(payloadBuffer); framePayload.writeBytes(payloadBuffer);
framePayloadBytesRead = framePayloadBytesRead + rbytes; framePayloadBytesRead = framePayloadBytesRead + rbytes;
// Return null to wait for more bytes to arrive // Return null to wait for more bytes to arrive
return null; return null;
} else if (willHaveReadByteCount > framePayloadLength) { } else if (willHaveReadByteCount > framePayloadLength) {
// We have more than what we need so read up to the end of frame // We have more than what we need so read up to the end of frame
// Leave the remainder in the buffer for next frame // Leave the remainder in the buffer for next frame
payloadBuffer = buffer.readBytes(toFrameLength(framePayloadLength - framePayloadBytesRead)); payloadBuffer = buffer.readBytes(toFrameLength(framePayloadLength - framePayloadBytesRead));
} }
// Now we have all the data, the next checkpoint must be the next // Now we have all the data, the next checkpoint must be the next
// frame // frame
checkpoint(State.FRAME_START); checkpoint(State.FRAME_START);
// Take the data that we have in this packet // Take the data that we have in this packet
if (framePayload == null) { if (framePayload == null) {
framePayload = payloadBuffer; framePayload = payloadBuffer;
} else { } else {
framePayload.writeBytes(payloadBuffer); framePayload.writeBytes(payloadBuffer);
} }
// Unmask data if needed // Unmask data if needed
if (this.maskedPayload) { if (this.maskedPayload) {
unmask(framePayload); unmask(framePayload);
} }
// Processing ping/pong/close frames because they cannot be // Processing ping/pong/close frames because they cannot be
// fragmented // fragmented
if (frameOpcode == OPCODE_PING) { if (frameOpcode == OPCODE_PING) {
return new PingWebSocketFrame(frameFinalFlag, frameRsv, framePayload); return new PingWebSocketFrame(frameFinalFlag, frameRsv, framePayload);
} else if (frameOpcode == OPCODE_PONG) { } else if (frameOpcode == OPCODE_PONG) {
return new PongWebSocketFrame(frameFinalFlag, frameRsv, framePayload); return new PongWebSocketFrame(frameFinalFlag, frameRsv, framePayload);
} else if (frameOpcode == OPCODE_CLOSE) { } else if (frameOpcode == OPCODE_CLOSE) {
this.receivedClosingHandshake = true; this.receivedClosingHandshake = true;
return new CloseWebSocketFrame(frameFinalFlag, frameRsv); return new CloseWebSocketFrame(frameFinalFlag, frameRsv);
} }
// Processing for possible fragmented messages for text and binary // Processing for possible fragmented messages for text and binary
// frames // frames
String aggregatedText = null; String aggregatedText = null;
if (frameFinalFlag) { if (frameFinalFlag) {
// Final frame of the sequence. Apparently ping frames are // Final frame of the sequence. Apparently ping frames are
// allowed in the middle of a fragmented message // allowed in the middle of a fragmented message
if (frameOpcode != OPCODE_PING) { if (frameOpcode != OPCODE_PING) {
fragmentedFramesCount = 0; fragmentedFramesCount = 0;
// Check text for UTF8 correctness // Check text for UTF8 correctness
if (frameOpcode == OPCODE_TEXT || fragmentedFramesText != null) { if (frameOpcode == OPCODE_TEXT || fragmentedFramesText != null) {
// Check UTF-8 correctness for this payload // Check UTF-8 correctness for this payload
checkUTF8String(channel, framePayload.array()); checkUTF8String(channel, framePayload.array());
// This does a second check to make sure UTF-8 // This does a second check to make sure UTF-8
// correctness for entire text message // correctness for entire text message
aggregatedText = fragmentedFramesText.toString(); aggregatedText = fragmentedFramesText.toString();
fragmentedFramesText = null; fragmentedFramesText = null;
} }
} }
} else { } else {
// Not final frame so we can expect more frames in the // Not final frame so we can expect more frames in the
// fragmented sequence // fragmented sequence
if (fragmentedFramesCount == 0) { if (fragmentedFramesCount == 0) {
// First text or binary frame for a fragmented set // First text or binary frame for a fragmented set
fragmentedFramesText = null; fragmentedFramesText = null;
if (frameOpcode == OPCODE_TEXT) { if (frameOpcode == OPCODE_TEXT) {
checkUTF8String(channel, framePayload.array()); checkUTF8String(channel, framePayload.array());
} }
} else { } else {
// Subsequent frames - only check if init frame is text // Subsequent frames - only check if init frame is text
if (fragmentedFramesText != null) { if (fragmentedFramesText != null) {
checkUTF8String(channel, framePayload.array()); checkUTF8String(channel, framePayload.array());
} }
} }
// Increment counter // Increment counter
fragmentedFramesCount++; fragmentedFramesCount++;
} }
// Return the frame // Return the frame
if (frameOpcode == OPCODE_TEXT) { if (frameOpcode == OPCODE_TEXT) {
return new TextWebSocketFrame(frameFinalFlag, frameRsv, framePayload); return new TextWebSocketFrame(frameFinalFlag, frameRsv, framePayload);
} else if (frameOpcode == OPCODE_BINARY) { } else if (frameOpcode == OPCODE_BINARY) {
return new BinaryWebSocketFrame(frameFinalFlag, frameRsv, framePayload); return new BinaryWebSocketFrame(frameFinalFlag, frameRsv, framePayload);
} else if (frameOpcode == OPCODE_CONT) { } else if (frameOpcode == OPCODE_CONT) {
return new ContinuationWebSocketFrame(frameFinalFlag, frameRsv, framePayload, aggregatedText); return new ContinuationWebSocketFrame(frameFinalFlag, frameRsv, framePayload, aggregatedText);
} else { } else {
throw new UnsupportedOperationException("Cannot decode web socket frame with opcode: " + frameOpcode); throw new UnsupportedOperationException("Cannot decode web socket frame with opcode: " + frameOpcode);
} }
case CORRUPT: case CORRUPT:
// If we don't keep reading Netty will throw an exception saying // If we don't keep reading Netty will throw an exception saying
// we can't return null if no bytes read and state not changed. // we can't return null if no bytes read and state not changed.
buffer.readByte(); buffer.readByte();
return null; return null;
default: default:
throw new Error("Shouldn't reach here."); throw new Error("Shouldn't reach here.");
} }
} }
private void unmask(ChannelBuffer frame) { private void unmask(ChannelBuffer frame) {
byte[] bytes = frame.array(); byte[] bytes = frame.array();
for (int i = 0; i < bytes.length; i++) { for (int i = 0; i < bytes.length; i++) {
frame.setByte(i, frame.getByte(i) ^ maskingKey.getByte(i % 4)); frame.setByte(i, frame.getByte(i) ^ maskingKey.getByte(i % 4));
} }
} }
private void protocolViolation(Channel channel, String reason) throws CorruptedFrameException { private void protocolViolation(Channel channel, String reason) throws CorruptedFrameException {
checkpoint(State.CORRUPT); checkpoint(State.CORRUPT);
if (channel.isConnected()) { if (channel.isConnected()) {
channel.write(ChannelBuffers.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE); channel.write(ChannelBuffers.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
channel.close().awaitUninterruptibly(); channel.close().awaitUninterruptibly();
} }
throw new CorruptedFrameException(reason); throw new CorruptedFrameException(reason);
} }
private int toFrameLength(long l) throws TooLongFrameException { private int toFrameLength(long l) throws TooLongFrameException {
if (l > Integer.MAX_VALUE) { if (l > Integer.MAX_VALUE) {
throw new TooLongFrameException("Length:" + l); throw new TooLongFrameException("Length:" + l);
} else { } else {
return (int) l; return (int) l;
} }
} }
private void checkUTF8String(Channel channel, byte[] bytes) throws CorruptedFrameException { private void checkUTF8String(Channel channel, byte[] bytes) throws CorruptedFrameException {
try { try {
// StringBuilder sb = new StringBuilder("UTF8 " + bytes.length + // StringBuilder sb = new StringBuilder("UTF8 " + bytes.length +
// " bytes: "); // " bytes: ");
// for (byte b : bytes) { // for (byte b : bytes) {
// sb.append(Integer.toHexString(b)).append(" "); // sb.append(Integer.toHexString(b)).append(" ");
// } // }
// logger.debug(sb.toString()); // logger.debug(sb.toString());
if (fragmentedFramesText == null) { if (fragmentedFramesText == null) {
fragmentedFramesText = new UTF8Output(bytes); fragmentedFramesText = new UTF8Output(bytes);
} else { } else {
fragmentedFramesText.write(bytes); fragmentedFramesText.write(bytes);
} }
} catch (UTF8Exception ex) { } catch (UTF8Exception ex) {
protocolViolation(channel, "invalid UTF-8 bytes"); protocolViolation(channel, "invalid UTF-8 bytes");
} }
} }
} }