wcrisman/Brainstorm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merged as many changes as possible in WebServer from the master branch without including any that would likely cause problems.

Browse Source
This commit is contained in:
wcrisman
2014-12-07 20:19:51 -08:00
parent 3d08ad5810
commit d0360ad252
1 changed files with 365 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

496
Foundation Web Core/src/com/foundation/web/server/WebServer.java
View File

@@ -17,6 +17,7 @@ import java.io.UnsupportedEncodingException;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.URLDecoder; import java.net.URLDecoder;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.nio.CharBuffer; import java.nio.CharBuffer;
import java.nio.channels.ClosedChannelException; import java.nio.channels.ClosedChannelException;
@@ -278,13 +279,16 @@ public class WebServer {
return result; return result;
}//loadBuffer()// }//loadBuffer()//
/** Gets the next message buffer (only used for pass through sockets). */
public MessageBuffer getNext() {return next;} public MessageBuffer getNext() {return next;}
/** Sets the next message buffer (only used for pass through sockets). */
public void setNext(MessageBuffer next) {this.next = next;} public void setNext(MessageBuffer next) {this.next = next;}
/** Gets the response object that created the message. This will be null for pass through sockets. */
public Response getResponse() {return response;} public Response getResponse() {return response;}
}//MessageBuffer// }//MessageBuffer//
private abstract class AbstractSocketContext extends ChannelContext { private abstract class AbstractSocketContext extends ChannelContext {
/** The key that represents the connection between the channel (socket) and the selector used to multiplex the listener. */ /** The key that represents the connection between the channel (socket) and the selector used to multiplex the listener. The code must synchronize on this attribute when accessing the isUsed functionality, or when interacting with the key's interestOps. */
public SelectionKey key = null; public SelectionKey key = null;
/** Whether the socket is currently being used by a thread designated by the network listener thread to read or write to the socket. Currently the socket type we use in Java only allows one thread to read and write at a time. Note: Always synchronize on <code>key</code> before using this attribute. */ /** Whether the socket is currently being used by a thread designated by the network listener thread to read or write to the socket. Currently the socket type we use in Java only allows one thread to read and write at a time. Note: Always synchronize on <code>key</code> before using this attribute. */
private boolean isUsed = false; private boolean isUsed = false;
@@ -323,6 +327,10 @@ public class WebServer {
* @return The related socket context, or null if none exists (data not forwarded to a remote server). * @return The related socket context, or null if none exists (data not forwarded to a remote server).
*/ */
protected AbstractSocketContext getRelatedSocketContext() {return relatedSocketContext;} protected AbstractSocketContext getRelatedSocketContext() {return relatedSocketContext;}
/**
* Determines whether the socket has a pending write operation.
*/
protected abstract boolean hasPendingWrite();
/** /**
* Updates the write flag status. * Updates the write flag status.
* @param requiresWrite Whether a write is required. * @param requiresWrite Whether a write is required.
@@ -529,7 +537,13 @@ public class WebServer {
linkedClientContext.key.selector().wakeup(); linkedClientContext.key.selector().wakeup();
runnable.waitForRun(); runnable.waitForRun();
key = runnable.getKey(); key = runnable.getKey();
flagReadOnly();
//Set the initial interest ops to read.//
synchronized(key) {
key.interestOps(SelectionKey.OP_READ);
}//synchronized//
key.selector().wakeup();
}//PassThroughSocketContext()// }//PassThroughSocketContext()//
/* (non-Javadoc) /* (non-Javadoc)
* @see com.foundation.web.server.WebServer.AbstractSocketContext#getLock() * @see com.foundation.web.server.WebServer.AbstractSocketContext#getLock()
@@ -562,12 +576,8 @@ public class WebServer {
//Keep sending encrypted frames until the output buffer is full, or we run out of message to send.// //Keep sending encrypted frames until the output buffer is full, or we run out of message to send.//
while(result && (pendingMessageBuffer != null) && pendingMessageBuffer.getBuffer().hasRemaining()) { while(result && (pendingMessageBuffer != null) && pendingMessageBuffer.getBuffer().hasRemaining()) {
//TODO: Comment me.
//int rem = pendingMessageBuffer.getBuffer().remaining();
//Write the bytes to the stream.// //Write the bytes to the stream.//
((SocketChannel) key.channel()).write(pendingMessageBuffer.getBuffer()); ((SocketChannel) key.channel()).write(pendingMessageBuffer.getBuffer());
//TODO: Comment me.
//Debug.log("Sent " + (rem - pendingMessageBuffer.getBuffer().remaining()) + " bytes to Git.");
//If not all the bytes could be written then we will need to wait until we can write more.// //If not all the bytes could be written then we will need to wait until we can write more.//
if(pendingMessageBuffer.getBuffer().hasRemaining()) { if(pendingMessageBuffer.getBuffer().hasRemaining()) {
@@ -614,14 +624,10 @@ public class WebServer {
socketReadBuffer.flip(); socketReadBuffer.flip();
if(count == -1) { if(count == -1) {
//TODO: Comment me.
//Debug.log("Git Closed Socket");
//The socket has been closed by the client.// //The socket has been closed by the client.//
try {relatedSocketContext.close();} catch(Throwable e) {} try {relatedSocketContext.close();} catch(Throwable e) {}
}//if// }//if//
else if(socketReadBuffer.hasRemaining()) { else if(socketReadBuffer.hasRemaining()) {
//TODO: Comment me.
//Debug.log("Git Sent " + count + " bytes.");
result = relatedSocketContext.passThrough(socketReadBuffer); result = relatedSocketContext.passThrough(socketReadBuffer);
socketReadBuffer.compact(); socketReadBuffer.compact();
}//else// }//else//
@@ -648,15 +654,10 @@ public class WebServer {
//Chain the message into the linked list. //Chain the message into the linked list.
if(lastAddedMessageBuffer == null) { if(lastAddedMessageBuffer == null) {
//TODO: Comment me.
//Debug.log("Posting request content to Git.");
pendingMessageBuffer = lastAddedMessageBuffer = message; pendingMessageBuffer = lastAddedMessageBuffer = message;
flagReadWrite(); flagReadWrite();
// flagWrite(true);
}//if// }//if//
else { else {
//TODO: Comment me.
//Debug.log("Queuing request content to Git.");
lastAddedMessageBuffer.setNext(message); lastAddedMessageBuffer.setNext(message);
lastAddedMessageBuffer = message; lastAddedMessageBuffer = message;
}//else// }//else//
@@ -667,6 +668,12 @@ public class WebServer {
try {if(key != null && key.channel() != null) key.channel().close();} catch(Throwable e) {} try {if(key != null && key.channel() != null) key.channel().close();} catch(Throwable e) {}
try {if(key != null) key.cancel();} catch(Throwable e) {} try {if(key != null) key.cancel();} catch(Throwable e) {}
}//close()// }//close()//
/* (non-Javadoc)
* @see com.foundation.web.server.WebServer.AbstractSocketContext#hasPendingWrite()
*/
protected boolean hasPendingWrite() {
return pendingMessageBuffer != null;
}//hasPendingWrite()//
}//PassThroughSocketContext// }//PassThroughSocketContext//
/** /**
@@ -1165,14 +1172,10 @@ public class WebServer {
//Chain the message into the linked list. //Chain the message into the linked list.
if(lastAddedMessageBuffer == null || pendingOutboundMessage == null) { if(lastAddedMessageBuffer == null || pendingOutboundMessage == null) {
//TODO: Comment me.
//Debug.log("Posting outbound message to client from git. Nothing currently in the queue.");
pendingOutboundMessage = lastAddedMessageBuffer = message; pendingOutboundMessage = lastAddedMessageBuffer = message;
flagWrite(true); flagWrite(true);
}//if// }//if//
else { else {
//TODO: Comment me.
//Debug.log("Posting outbound message to client from git at the end of the queue.");
lastAddedMessageBuffer.setNext(message); lastAddedMessageBuffer.setNext(message);
lastAddedMessageBuffer = message; lastAddedMessageBuffer = message;
}//else// }//else//
@@ -1189,7 +1192,6 @@ public class WebServer {
writeClientResponse(); writeClientResponse();
//Set the write flag if we have more to write.// //Set the write flag if we have more to write.//
flagReadWrite(true, pendingOutboundMessage != null); flagReadWrite(true, pendingOutboundMessage != null);
// flagWrite(pendingOutboundMessage != null);
}//synchronized// }//synchronized//
}//if// }//if//
else if(isWebsocket) { else if(isWebsocket) {
@@ -1202,15 +1204,7 @@ public class WebServer {
}//else if// }//else if//
else { else {
//Go directly to writing the client response if we are just passing everything through to another process.// //Go directly to writing the client response if we are just passing everything through to another process.//
boolean receive = internalProcessResponses(); if(internalProcessResponses()) flagReadOnly(); else flagReadWrite();
if(receive) {
flagReadOnly();
}//if//
else {
flagReadWrite();
// flagWriteOnly();
}//else//
}//else// }//else//
}//processCurrentResponse()// }//processCurrentResponse()//
/** /**
@@ -1341,31 +1335,23 @@ public class WebServer {
* @return * @return
*/ */
private synchronized boolean internalProcessResponses() { private synchronized boolean internalProcessResponses() {
boolean receive = true; boolean finishedSending = true;
//Keep sending responses while the buffers are not full and there is another response to send.// //Keep sending responses while the buffers are not full and there is another response to send.//
while(receive && currentResponse != null) { while(finishedSending && currentResponse != null) {
//If the socket is open then send the next buffer of data.// //If the socket is open then send the next buffer of data.//
if(key.channel().isOpen()) { if(key.channel().isOpen()) {
//Send the pending response object's prepared buffer of data.// //Send the pending response object's prepared buffer of data.//
receive = writeClientResponse(); finishedSending = writeClientResponse();
//DEBUG//
// if(result && debug) {
// Debug.log("--==-- BEGIN MESSAGE --==--\n" + debugBuffer.toString() + "\n--==-- END MESSAGE --==--");
// debugBuffer.setLength(0);
// }//if//
// else if(debug) {
// Debug.log("Couldn't complete sending the message. Will wait before trying to send the rest.");
// }//else//
}//if// }//if//
//Close the response if successfully sent, or if the socket is closed.// //Close the response if successfully sent, or if the socket is closed.//
if(receive || !key.channel().isOpen()) { if(finishedSending || !key.channel().isOpen()) {
try {currentResponse.close();} catch(Throwable e) {} try {currentResponse.close();} catch(Throwable e) {}
}//if// }//if//
if(receive) { //If we finished sending the current response then load the next one.//
if(finishedSending) {
currentResponse = currentResponse.getNextResponse(); currentResponse = currentResponse.getNextResponse();
if(currentResponse == null) { if(currentResponse == null) {
@@ -1388,14 +1374,14 @@ public class WebServer {
}//if// }//if//
}//while// }//while//
return receive; return finishedSending;
}//processCurrentResponse()// }//processCurrentResponse()//
/** /**
* Sends a response to the client. * Sends a response to the client.
* @return Whether the response could be fully sent. This will be false if there is still more data to be written when the call returns. * @return Whether the response could be fully sent. This will be false if there is still more data to be written when the call returns.
*/ */
private boolean writeClientResponse() { private boolean writeClientResponse() {
boolean result = true; boolean sendMore = true;
// if(debug) { // if(debug) {
// debugBuffer.append("Starting a write cycle.\n"); // debugBuffer.append("Starting a write cycle.\n");
@@ -1417,95 +1403,89 @@ public class WebServer {
//Check to see if we failed to send the whole frame.// //Check to see if we failed to send the whole frame.//
if(encryptedWriteBuffer.hasRemaining()) { if(encryptedWriteBuffer.hasRemaining()) {
result = false; sendMore = false;
}//if// }//if//
}//if// }//if//
if(result && sslNeedsWrap) { while(sendMore && sslNeedsWrap) {
// if(debug) { SSLEngineResult handshakeResult;
// debugBuffer.append("Begin Handshaking SSL\n");
// }//if//
while(result && sslNeedsWrap) { //Reset the encrypted write buffer - note that since we will never read while waiting to write data, this should always be empty.//
SSLEngineResult handshakeResult; encryptedWriteBuffer.position(0);
encryptedWriteBuffer.limit(encryptedWriteBuffer.capacity());
//Reset the encrypted write buffer - note that since we will never read while waiting to write data, this should always be empty.// //Generate the handshake message.//
encryptedWriteBuffer.position(0); handshakeResult = sslEngine.wrap(ByteBuffer.allocate(0), encryptedWriteBuffer);
encryptedWriteBuffer.limit(encryptedWriteBuffer.capacity()); encryptedWriteBuffer.flip();
//Generate the handshake message.//
handshakeResult = sslEngine.wrap(ByteBuffer.allocate(0), encryptedWriteBuffer); if(handshakeResult.getStatus() == Status.BUFFER_OVERFLOW) {
encryptedWriteBuffer.flip(); //Should never happen.//
Debug.log(new RuntimeException("Unexpected ssl engine buffer overflow."));
if(handshakeResult.getStatus() == Status.BUFFER_OVERFLOW) { }//if//
else if(handshakeResult.getStatus() == Status.BUFFER_UNDERFLOW) {
//Should never happen.//
Debug.log(new RuntimeException("Unexpected ssl engine buffer underflow."));
}//else if//
else if(handshakeResult.getStatus() == Status.CLOSED) {
//Should never happen.//
Debug.log(new RuntimeException("Unexpected ssl engine closed."));
//TODO: Handle this closure without an infinate loop...
//Close the socket.//
try {key.channel().close();}catch(Throwable e2) {}
}//else if//
else if(handshakeResult.getStatus() == Status.OK) {
if(handshakeResult.getHandshakeStatus() == HandshakeStatus.NEED_TASK) {
//Should never happen.// //Should never happen.//
Debug.log(new RuntimeException("Unexpected ssl engine buffer overflow.")); Debug.log(new RuntimeException("Unexpected ssl engine task."));
}//if// }//if//
else if(handshakeResult.getStatus() == Status.BUFFER_UNDERFLOW) { else if(encryptedWriteBuffer.hasRemaining()) {
//Should never happen.// int remaining = encryptedWriteBuffer.remaining();
Debug.log(new RuntimeException("Unexpected ssl engine buffer underflow."));
}//else if// //Write the bytes to the stream.//
else if(handshakeResult.getStatus() == Status.CLOSED) { ((SocketChannel) key.channel()).write(encryptedWriteBuffer);
//Should never happen.//
Debug.log(new RuntimeException("Unexpected ssl engine closed.")); // if(debug) {
//TODO: Handle this closure without an infinate loop... // debugBuffer.append("Sent " + (remaining - encryptedWriteBuffer.remaining()) + " encrypted bytes.\n");
//Close the socket.// // }//if//
try {key.channel().close();}catch(Throwable e2) {}
}//else if// //If not all the bytes could be written then we will need to wait until we can write more.//
else if(handshakeResult.getStatus() == Status.OK) { if(encryptedWriteBuffer.hasRemaining()) {
if(handshakeResult.getHandshakeStatus() == HandshakeStatus.NEED_TASK) {
//Should never happen.//
Debug.log(new RuntimeException("Unexpected ssl engine task."));
}//if//
else if(encryptedWriteBuffer.hasRemaining()) {
int remaining = encryptedWriteBuffer.remaining();
//Write the bytes to the stream.//
((SocketChannel) key.channel()).write(encryptedWriteBuffer);
// if(debug) { // if(debug) {
// debugBuffer.append("Sent " + (remaining - encryptedWriteBuffer.remaining()) + " encrypted bytes.\n"); // debugBuffer.append("Pausing due to a partially sent packet (while ssl handshaking). Bytes actually sent: " + encryptedWriteBuffer.position() + ". Bytes remaining: " + encryptedWriteBuffer.remaining() + ".\n");
// }//if// // }//if//
//If not all the bytes could be written then we will need to wait until we can write more.// //Leave the data in the encrypted write buffer for the writing operation to send it.//
if(encryptedWriteBuffer.hasRemaining()) { sendMore = false;
// if(debug) { }//if//
// debugBuffer.append("Pausing due to a partially sent packet (while ssl handshaking). Bytes actually sent: " + encryptedWriteBuffer.position() + ". Bytes remaining: " + encryptedWriteBuffer.remaining() + ".\n");
// }//if// //Update the SSL needs wrap flag.//
if(handshakeResult.getHandshakeStatus() != HandshakeStatus.NEED_WRAP) {
//Leave the data in the encrypted write buffer for the writing operation to send it.//
result = false;
}//if//
//Update the SSL needs wrap flag.//
if(handshakeResult.getHandshakeStatus() != HandshakeStatus.NEED_WRAP) {
sslNeedsWrap = false;
}//if//
}//else if//
else {
sslNeedsWrap = false; sslNeedsWrap = false;
}//else// }//if//
}//else if// }//else if//
else { else {
//Should never happen.// sslNeedsWrap = false;
Debug.log(new RuntimeException("Unexpected ssl engine status code."));
}//else// }//else//
}//while// }//else if//
else {
// if(debug) { //Should never happen.//
// debugBuffer.append("End Handshaking SSL\n"); Debug.log(new RuntimeException("Unexpected ssl engine status code."));
// }//if// }//else//
}//if// }//while//
// if(debug) {
// debugBuffer.append("End Handshaking SSL\n");
// }//if//
}//if// }//if//
if(result && pendingOutboundMessage != null) { if(sendMore && pendingOutboundMessage != null) {
//Check to see if the outbound message is prepared to send more content. For chunked transfers the outbound message may be waiting for additional content from another stream and we should return later.// //Check to see if the outbound message is prepared to send more content. For chunked transfers the outbound message may be waiting for additional content from another stream and we should return later.//
if(result && !pendingOutboundMessage.getBuffer().hasRemaining()) { if(!pendingOutboundMessage.getBuffer().hasRemaining()) {
if(!pendingOutboundMessage.loadBuffer()) { if(!pendingOutboundMessage.loadBuffer()) {
if(pendingOutboundMessage.getBuffer() == null && pendingOutboundMessage.getNext() != null) { if(pendingOutboundMessage.getBuffer() == null && pendingOutboundMessage.getNext() != null) {
pendingOutboundMessage = pendingOutboundMessage.getNext(); pendingOutboundMessage = pendingOutboundMessage.getNext();
}//if// }//if//
else { else {
result = false; sendMore = false;
}//else// }//else//
}//if// }//if//
@@ -1516,10 +1496,10 @@ public class WebServer {
}//if// }//if//
//If we have an application response pending then send it now.// //If we have an application response pending then send it now.//
if(result && pendingOutboundMessage.getBuffer().hasRemaining()) { if(sendMore && pendingOutboundMessage.getBuffer().hasRemaining()) {
if(sslEngine != null) { if(sslEngine != null) {
//Keep sending encrypted frames until the output buffer is full, or we run out of message to send.// //Keep sending encrypted frames until the output buffer is full, or we run out of message to send.//
while(key.channel().isOpen() && result && (pendingOutboundMessage != null) && pendingOutboundMessage.getBuffer().hasRemaining()) { while(key.channel().isOpen() && sendMore && (pendingOutboundMessage != null) && pendingOutboundMessage.getBuffer().hasRemaining()) {
SSLEngineResult encryptResult; SSLEngineResult encryptResult;
// int offset = pendingOutboundMessage.getBuffer().position(); // int offset = pendingOutboundMessage.getBuffer().position();
//TODO: Comment me. //TODO: Comment me.
@@ -1570,7 +1550,7 @@ public class WebServer {
//If not all the bytes could be written then we will need to wait until we can write more.// //If not all the bytes could be written then we will need to wait until we can write more.//
if(encryptedWriteBuffer.hasRemaining()) { if(encryptedWriteBuffer.hasRemaining()) {
//Leave the data in the encrypted write buffer for the writing operation to send it.// //Leave the data in the encrypted write buffer for the writing operation to send it.//
result = false; sendMore = false;
// if(debug) { // if(debug) {
// debugBuffer.append("Pausing due to a partially sent packet. Bytes actually sent: " + encryptedWriteBuffer.position() + ". Bytes remaining: " + encryptedWriteBuffer.remaining() + ".\n"); // debugBuffer.append("Pausing due to a partially sent packet. Bytes actually sent: " + encryptedWriteBuffer.position() + ". Bytes remaining: " + encryptedWriteBuffer.remaining() + ".\n");
@@ -1592,7 +1572,7 @@ public class WebServer {
}//if// }//if//
else { else {
//Wait until additional message bytes are available.// //Wait until additional message bytes are available.//
result = false; sendMore = false;
}//else// }//else//
}//if// }//if//
@@ -1606,7 +1586,7 @@ public class WebServer {
}//if// }//if//
else { else {
//Keep sending encrypted frames until the output buffer is full, or we run out of message to send.// //Keep sending encrypted frames until the output buffer is full, or we run out of message to send.//
while(result && (pendingOutboundMessage != null) && pendingOutboundMessage.getBuffer().hasRemaining()) { while(sendMore && (pendingOutboundMessage != null) && pendingOutboundMessage.getBuffer().hasRemaining()) {
//Write the bytes to the stream.// //Write the bytes to the stream.//
((SocketChannel) key.channel()).write(pendingOutboundMessage.getBuffer()); ((SocketChannel) key.channel()).write(pendingOutboundMessage.getBuffer());
@@ -1617,7 +1597,7 @@ public class WebServer {
//If not all the bytes could be written then we will need to wait until we can write more.// //If not all the bytes could be written then we will need to wait until we can write more.//
if(pendingOutboundMessage.getBuffer().hasRemaining()) { if(pendingOutboundMessage.getBuffer().hasRemaining()) {
result = false; sendMore = false;
}//if// }//if//
else { else {
if(!pendingOutboundMessage.loadBuffer()) { if(!pendingOutboundMessage.loadBuffer()) {
@@ -1627,7 +1607,7 @@ public class WebServer {
}//if// }//if//
else { else {
//Wait until additional message bytes are available.// //Wait until additional message bytes are available.//
result = false; sendMore = false;
}//else// }//else//
}//if// }//if//
@@ -1660,7 +1640,7 @@ public class WebServer {
close(); close();
}//catch// }//catch//
return result; return sendMore;
}//writeClientResponse()// }//writeClientResponse()//
/* (non-Javadoc) /* (non-Javadoc)
* @see com.foundation.web.server.WebServer.AbstractSocketContext#processRequest() * @see com.foundation.web.server.WebServer.AbstractSocketContext#processRequest()
@@ -1850,7 +1830,12 @@ public class WebServer {
if(sslResult.bytesProduced() > 0) { if(sslResult.bytesProduced() > 0) {
//If we are not passing all content to another process then handle it by calling processClientRequest, otherwise pass it through.// //If we are not passing all content to another process then handle it by calling processClientRequest, otherwise pass it through.//
if(getPassThroughSocketContext() == null) { if(getPassThroughSocketContext() == null) {
requiresRead = WebServer.this.processClientRequest((SocketContext) this, unencryptedReadBuffer, key); if(isWebsocket) {
requiresRead = WebServer.this.processWebsocketFrame((SocketContext) this, unencryptedReadBuffer, key);
}//if//
else {
requiresRead = WebServer.this.processClientRequest((SocketContext) this, unencryptedReadBuffer, key);
}//else//
}//if// }//if//
else { else {
//TODO: Comment me. //TODO: Comment me.
@@ -1918,9 +1903,24 @@ public class WebServer {
}//if// }//if//
else if(relatedSocketContext != null) { else if(relatedSocketContext != null) {
relatedSocketContext.passThrough(socketReadBuffer); relatedSocketContext.passThrough(socketReadBuffer);
if(socketReadBuffer.remaining() > 0) {
Debug.log(new RuntimeException("Remaining bytes found on the read buffer!"));
}//if//
}//else if//
else if(isWebsocket) {
requiresRead = WebServer.this.processWebsocketFrame((SocketContext) this, socketReadBuffer, key);
if(socketReadBuffer.remaining() > 0) {
Debug.log(new RuntimeException("Remaining bytes found on the read buffer!"));
}//if//
}//else if// }//else if//
else if(socketReadBuffer.hasRemaining()) { else if(socketReadBuffer.hasRemaining()) {
requiresRead = WebServer.this.processClientRequest((SocketContext) this, socketReadBuffer, key); requiresRead = WebServer.this.processClientRequest((SocketContext) this, socketReadBuffer, key);
if(socketReadBuffer.remaining() > 0) {
Debug.log(new RuntimeException("Remaining bytes found on the read buffer!"));
}//if//
}//else// }//else//
else { else {
break; break;
@@ -1928,13 +1928,7 @@ public class WebServer {
}//while// }//while//
}//else// }//else//
if(requiresRead) { if(requiresRead) flagReadOnly(); else flagReadWrite();
flagReadOnly();
}//if//
else {
flagReadWrite();
// flagWriteOnly();
}//else//
}//processRequest()// }//processRequest()//
/* (non-Javadoc) /* (non-Javadoc)
* @see com.foundation.web.server.WebServer.AbstractSocketContext#hasPendingWrite() * @see com.foundation.web.server.WebServer.AbstractSocketContext#hasPendingWrite()
@@ -2211,6 +2205,7 @@ public class WebServer {
//Not allowing either reads or writes to continue until all processing of this message is done.// //Not allowing either reads or writes to continue until all processing of this message is done.//
((AbstractSocketContext) context).flags = key.interestOps(); ((AbstractSocketContext) context).flags = key.interestOps();
((AbstractSocketContext) context).isUsed = true;
key.interestOps(0); key.interestOps(0);
//The problem with this is that we'd have to use AsynchronousSocketChannel which would appear to require a complete rewrite of everything since it operates completely differently.// //The problem with this is that we'd have to use AsynchronousSocketChannel which would appear to require a complete rewrite of everything since it operates completely differently.//
@@ -2256,8 +2251,15 @@ public class WebServer {
//Set the new ops for the selection key and notify the selector that ops have changed.// //Set the new ops for the selection key and notify the selector that ops have changed.//
synchronized(selectionKey) { synchronized(selectionKey) {
if(selectionKey.isValid()) { if(selectionKey.isValid()) {
//Always flag the socket for reading, only flag the socket for writing if a pending write operation exists.//
//selectionKey.interestOps(SelectionKey.OP_READ | (((AbstractSocketContext) context).hasPendingWrite() ? SelectionKey.OP_WRITE : 0));
selectionKey.interestOps(((AbstractSocketContext) context).flags); selectionKey.interestOps(((AbstractSocketContext) context).flags);
}//if// }//if//
else {
Debug.log(new RuntimeException("Woops! Somehow the selection key isn't valid, but the socket isn't closed either!"));
}//else//
((AbstractSocketContext) context).isUsed = false;
}//synchronized// }//synchronized//
selector.wakeup(); selector.wakeup();
@@ -2269,6 +2271,10 @@ public class WebServer {
else if(channel != null && (!channel.isOpen() || socketClosed) && channel instanceof SocketChannel && context instanceof SocketContext) { else if(channel != null && (!channel.isOpen() || socketClosed) && channel instanceof SocketChannel && context instanceof SocketContext) {
cleanupClientChannel((SocketContext) context, (SocketChannel) channel); cleanupClientChannel((SocketContext) context, (SocketChannel) channel);
}//else if// }//else if//
else {
//This shouldn't be called I don't think.//
Debug.log(new RuntimeException("Woops! Somehow we aren't closed and we didn't setup the interestOps for the HTTP socket!"));
}//else//
}//finally// }//finally//
}//run()// }//run()//
}); });
@@ -2899,6 +2905,220 @@ private boolean parseFirstTlsMessage(SocketContext context, SocketChannel channe
return result; return result;
}//parseFirstTlsMessage()// }//parseFirstTlsMessage()//
/**
* Processes a single websocket frame if there is enough data in the fragment.
* <br/>Will not return until all data is read from the frame or the socket is closed.
* @param context The context for the socket.
* @param fragment The fragment of data off the socket.
* @param key The key for the socket.
* @return Whether more data is required in order to process a single frame.
* @throws IOException
*/
private boolean processWebsocketFrame(SocketContext context, ByteBuffer fragment, SelectionKey key) throws IOException {
boolean isFrameFullyRead = false;
while(fragment.remaining() > 0) {
byte[] frameHeader = context.websocketFrameHeader;
byte[] maskKey = context.websocketMessageMaskKey;
final int OP_CODE_INDEX = 0;
final int LENGTH_INDEX = 1;
final int EXTENDED_LENGTH_INDEX = 2;
isFrameFullyRead = false;
//Read the frame header (unless we are waiting on the remaining content of a previously read frame header).//
if(context.websocketFrameRemainder == 0) {
long length = -1;
try {
boolean mask = false;
int expectedHeaderLength = 6;
int readLength = Math.min(2 - context.websocketFrameHeaderIndex, fragment.remaining());
if(readLength > 0) {
fragment.get(frameHeader, context.websocketFrameHeaderIndex, readLength);
context.websocketFrameHeaderIndex += readLength;
}//if//
//Handle a partial header by checking the first two bytes to find the header's overall length (header length varies based on the message length and options used).//
if(context.websocketFrameHeaderIndex > 1) {
context.websocketIsLastFrameInMessage = (frameHeader[OP_CODE_INDEX] & 0x80) > 0;
mask = (frameHeader[LENGTH_INDEX] & 0x80) > 0;
length = frameHeader[LENGTH_INDEX] & 0x7F;
if(length == 127) {
expectedHeaderLength = 8;
}//if//
else if(length == 128) {
expectedHeaderLength = 14;
}//else if//
}//if//
readLength = Math.min(expectedHeaderLength - context.websocketFrameHeaderIndex, fragment.remaining());
fragment.get(frameHeader, context.websocketFrameHeaderIndex, readLength);
context.websocketFrameHeaderIndex += readLength;
//Ensure we have the full header before proceeding to read it.//
if(context.websocketFrameHeaderIndex == expectedHeaderLength) {
int maskKeyIndex = 2;
if(length == 127) {
length = (int) (StreamSupport.readShort(frameHeader, EXTENDED_LENGTH_INDEX) & 0xFFFF);
maskKeyIndex = 4;
}//if//
else if(length == 128) {
//Note: Cannot handle unsigned longs in java. Assume negative numbers are over the length maximum for the server (since they most certainly should be - we'd run out of ram).//
length = StreamSupport.readLong(frameHeader, EXTENDED_LENGTH_INDEX);
maskKeyIndex = 10;
}//if//
//Ensure we don't exceed the maximum message length applied by the web application when the socket was upgraded. Also ensure that we don't exceed the server's limitations on frame size (the highest possible positive value an integer can hold).//
if(length < 0 || length > Integer.MAX_VALUE || length + (context.websocketPartialReceivedMessage == null ? 0 : context.websocketPartialReceivedMessage.getSize()) > context.websocketMaxMessageLength) {
//TODO: Reply with a closed message?
//Kill the socket!//
context.close();
//Notify ourselves via the log.//
Debug.log(new RuntimeException("Killed the client websocket due to excess frame size (Integer.MAX_VALUE) or the message exceeding the maximum size allowed by the web application (as specified when the websocket was accepted)."));
}//if//
else if(!mask) {
//Kill the socket - should never not have the mask set for the client calling the server.//
context.close();
Debug.log(new RuntimeException("Killed the client websocket due to the mask not being set."));
}//else if//
else {
//If this is the first frame in a message then parse the header for info.//
if(context.websocketMessageOpCode == 0) {
//Read the 4 bytes of the mask key.//
for(int maskIndex = 0, headerIndex = maskKeyIndex; maskIndex < maskKey.length; maskIndex++, headerIndex++) {
maskKey[maskIndex] = frameHeader[headerIndex];
}//for//
//The op code shouldn't be zero because zero is for a continuation frame (another frame in a message, not the first frame).//
context.websocketMessageOpCode = frameHeader[OP_CODE_INDEX] & 0x0F;
}//if//
//If it is a content message then begin reading it.//
context.websocketFrameRemainder = (int) length;
}//else//
}//if//
}//try//
catch(BufferUnderflowException e) {
//Ignore - shouldn't occure.//
}//catch//
//If we could read the next frame header then reset the frame header index to zero for the next frame header read.//
if(length != -1) {
context.websocketFrameHeaderIndex = 0;
}//if//
else if(context.websocketFrameRemainder == 0) {
isFrameFullyRead = true;
}//else if//
}//if//
//Read the frame content.//
if(context.websocketFrameRemainder > 0) {
//If the whole message is here then read it.//
if(fragment.remaining() > 0) {
int transferCount = 0;
int streamPosition = 0;
int previousReadPosition;
if(context.websocketPartialReceivedMessage == null) {
context.websocketPartialReceivedMessage = new StreamBuffer();
}//if//
else {
streamPosition = context.websocketPartialReceivedMessage.getSize();
}//else//
//Retain the index of the last previously read byte so we start decoding using the correct mask offset.//
previousReadPosition = context.websocketPartialReceivedMessage.getSize();
//Transfer either all the bytes in the fragment read from the socket, or transfer all the remaining bytes belonging to the current frame.//
transferCount = Math.max((int) context.websocketFrameRemainder, fragment.remaining());
//Move the bytes into our partial message buffer.//
context.websocketPartialReceivedMessage.writeBytes(fragment, transferCount);
//Decode the message bytes with the mask key. The encoding is to protect intermediate servers that cache the contents from cache attacks. The mask key must be set by the client browser and must be random for each frame.//
for(int index = streamPosition, frameIndex = previousReadPosition, maxIndex = streamPosition + transferCount; index < maxIndex; index++, frameIndex++) {
//XOR each byte read with the next byte in the mask key (loops over the mask key starting at zero for the whole frame).//
context.websocketPartialReceivedMessage.putByte(index, context.websocketPartialReceivedMessage.getByte(index) ^ context.websocketMessageMaskKey[frameIndex % 4]);
}//for//
//Track the remaining bytes to be read for the current frame.//
context.websocketFrameRemainder -= transferCount;
if(context.websocketFrameRemainder == 0) {
//Mark the frame as is being fully read.//
isFrameFullyRead = true;
}//if//
}//if//
}//if//
//If we finished reading this frame, and this is the last frame in the message, then handle it.//
if(isFrameFullyRead && context.websocketIsLastFrameInMessage) {
//Handle the completed message.//
switch(context.websocketMessageOpCode) {
case 0x01: //Text Frame//
//Convert the data into text and send to the application.//
try {
context.websocketHandler.receiveTextMessage(new String(context.websocketPartialReceivedMessage.toByteArray(), "UTF-8"));
}//try//
catch(Throwable e) {
Debug.log(e);
}//catch//
break;
case 0x02: //Binary Frame//
//Send binary data to the application.//
try {
context.websocketHandler.receiveBinaryMessage(context.websocketPartialReceivedMessage.toByteArray());
}//try//
catch(Throwable e) {
Debug.log(e);
}//catch//
break;
case 0x08: //Socket Close//
context.close();
break;
case 0x09: //Ping//
/*
context.sendWebsocketPong();
//Notify the listener of the ping.//
try {
context.websocketHandler.receivePing();
}//try//
catch(Throwable e) {
Debug.log(e);
}//catch//
*/
Debug.log(new RuntimeException("Received PING from the client. The client shouldn't ever send PING messages (only PONG)."));
break;
case 0x0A: //Pong//
//Notify the listener of the pong.//
try {
context.websocketHandler.receivePong();
}//try//
catch(Throwable e) {
Debug.log(e);
}//catch//
break;
default: //Unsupported//
context.close();
}//switch//
//Clean up after the message.//
context.websocketFrameRemainder = 0;
context.websocketMessageOpCode = 0;
context.websocketIsLastFrameInMessage = false;
if(context.websocketPartialReceivedMessage != null) {
context.websocketPartialReceivedMessage.release();
context.websocketPartialReceivedMessage = null;
}//if//
}//if//
}//while//
return !isFrameFullyRead;
}//processWebsocketFrame()//
/** /**
* Determines whether we have a complete HTML header in the string buffer. * Determines whether we have a complete HTML header in the string buffer.
* @param buffer The buffer containing the HTML ASCII header characters. * @param buffer The buffer containing the HTML ASCII header characters.
@@ -3595,7 +3815,8 @@ private boolean processClientRequest(SocketContext context, final Request reques
//Create the response.// //Create the response.//
response = new Response(request, session, application); response = new Response(request, session, application);
//Call the web application to handle result processing.//
//Move to the next step in the request processing.//
result = internalProcessClientRequest(context, key, application, request, response, session, allowSecureAccess, clientHadBadSession); result = internalProcessClientRequest(context, key, application, request, response, session, allowSecureAccess, clientHadBadSession);
}//if// }//if//
else { else {
@@ -3666,7 +3887,20 @@ private boolean internalProcessClientRequest(final SocketContext context, Select
}//if// }//if//
//Send the request to the application to be processed if we are not dealing with an error.// //Send the request to the application to be processed if we are not dealing with an error.//
else if(application != null) { else if(application != null) {
application.processRequest(request, response, session, allowSecureAccess, clientHadBadSession, context); String upgrade = request.getHeaderFieldValue("Upgrade");
if(upgrade != null && "websocket".equalsIgnoreCase(upgrade)) {
String connection = request.getHeaderFieldValue("Connection");
String secWebSocketKey = request.getHeaderFieldValue("Sec-WebSocket-Key");
String secWebSocketProtocol = request.getHeaderFieldValue("Sec-WebSocket-Protocol");
String secWebSocketVersion = request.getHeaderFieldValue("Sec-WebSocket-Version");
String origin = request.getHeaderFieldValue("Origin");
application.handleWebSocketUpgrade(request, response, session, allowSecureAccess, clientHadBadSession, context, connection, secWebSocketKey, secWebSocketProtocol, secWebSocketVersion, origin);
}//if//
else {
application.processRequest(request, response, session, allowSecureAccess, clientHadBadSession, context);
}//else//
}//else if// }//else if//
//Convert the response into a byte stream and send it via the socket.// //Convert the response into a byte stream and send it via the socket.//