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

Modified SocketContext to use a queue and an active reference to the outbound messages instead of multiple linked lists.

Browse Source
This commit is contained in:
wcrisman
2014-12-28 10:43:25 -08:00
parent 3e1a531fc3
commit 06cbff7cf0
5 changed files with 495 additions and 509 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
Foundation Web Core/src/com/foundation/web/server/HttpMessageBuffer.java Normal file
View File

@@ -0,0 +1,28 @@
package com.foundation.web.server;
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.text.SimpleDateFormat;
import java.util.Date;
import com.common.debug.Debug;
import com.common.util.LiteHashMap;
import com.common.util.LiteList;
import com.foundation.web.interfaces.IContent;
import com.foundation.web.interfaces.IMimeType;
/**
* A message buffer for HTTP messages.
*/
public class HttpMessageBuffer extends MessageBuffer {
protected final WebServer webServer;
private Response response;
private Request request;
public HttpMessageBuffer(WebServer webServer, Response response, Request request) {
this.webServer = webServer;
this.response = response;
this.request = request;
}//HttpMessageBuffer()//
}//HttpMessageBuffer//

10
Foundation Web Core/src/com/foundation/web/server/MessageBuffer.java
View File

@@ -11,7 +11,7 @@ class MessageBuffer {
/** The actual underlying buffer containing the bytes to be sent. Will be null if the message buffer needs initializing or has finished. */ /** The actual underlying buffer containing the bytes to be sent. Will be null if the message buffer needs initializing or has finished. */
private ByteBuffer buffer = null; private ByteBuffer buffer = null;
/** The ability to chain message buffers into a linked list. */ /** The ability to chain message buffers into a linked list. */
private MessageBuffer next = null; // private MessageBuffer next = null;
/** The optional response the message is based upon. */ /** The optional response the message is based upon. */
private Response response = null; private Response response = null;
@@ -119,10 +119,10 @@ class MessageBuffer {
return result; return result;
}//loadBuffer()// }//loadBuffer()//
/** Gets the next message buffer (only used for pass through sockets). */ ///** 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). */ ///** 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. */ /** 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//

115
Foundation Web Core/src/com/foundation/web/server/PassThroughSocketContext.java
View File

@@ -5,6 +5,8 @@ import java.net.InetSocketAddress;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey; import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel; import java.nio.channels.SocketChannel;
import com.common.util.Queue;
import com.foundation.web.server.WebServer.RegisterKeyRunnable; import com.foundation.web.server.WebServer.RegisterKeyRunnable;
/** /**
@@ -12,8 +14,8 @@ import com.foundation.web.server.WebServer.RegisterKeyRunnable;
* Allows the web server to act as an SSL front to another web server or service. * Allows the web server to act as an SSL front to another web server or service.
*/ */
public class PassThroughSocketContext extends AbstractSocketContext { public class PassThroughSocketContext extends AbstractSocketContext {
private MessageBuffer pendingMessageBuffer = null; private MessageBuffer outboundMessage = null;
private MessageBuffer lastAddedMessageBuffer = null; private Queue outboundMessages = new Queue(8);
/** The byte buffer used to read data from the socket. */ /** The byte buffer used to read data from the socket. */
public ByteBuffer socketReadBuffer = ByteBuffer.allocate(BUFFER_SIZE); public ByteBuffer socketReadBuffer = ByteBuffer.allocate(BUFFER_SIZE);
/** /**
@@ -59,47 +61,35 @@ protected Object getLock() {
*/ */
protected synchronized void writeOutgoingMessages() throws IOException { protected synchronized void writeOutgoingMessages() throws IOException {
//Actually this is called when a request is being sent via the pass through socket (sending the request to the remote server).// //Actually this is called when a request is being sent via the pass through socket (sending the request to the remote server).//
//Synchronized to avoid accessing the pendingMessageBuffer and lastAddedMessageBuffer at the same time as a thread that is calling passThrough(ByteBuffer) which also accesses these variables.// // //Synchronized to avoid accessing the pendingMessageBuffer and lastAddedMessageBuffer at the same time as a thread that is calling passThrough(ByteBuffer) which also accesses these variables.//
boolean result = true; // boolean result = true;
//
if(result && pendingMessageBuffer != null) { // if(result && pendingMessageBuffer != null) {
//Check to see if the outbound message is prepared to send more content.// // //Check to see if the outbound message is prepared to send more content.//
if(!pendingMessageBuffer.getBuffer().hasRemaining()) { // if(!pendingMessageBuffer.getBuffer().hasRemaining()) {
//Load the next pending outbound message in the chain.// // //Wait until additional message bytes are available.//
if(pendingMessageBuffer.getNext() != null) { // result = false;
pendingMessageBuffer = pendingMessageBuffer.getNext(); // pendingMessageBuffer = null;
}//if// // lastAddedMessageBuffer = null;
else { // }//if//
//Wait until additional message bytes are available.// //
result = false; // //Keep sending encrypted frames until the output buffer is full, or we run out of message to send.//
pendingMessageBuffer = null; // while(result && (pendingMessageBuffer != null) && pendingMessageBuffer.getBuffer().hasRemaining()) {
lastAddedMessageBuffer = null; // //Write the bytes to the stream.//
}//else// // ((SocketChannel) key.channel()).write(pendingMessageBuffer.getBuffer());
}//if// //
// //If not all the bytes could be written then we will need to wait until we can write more.//
//Keep sending encrypted frames until the output buffer is full, or we run out of message to send.// // if(pendingMessageBuffer.getBuffer().hasRemaining()) {
while(result && (pendingMessageBuffer != null) && pendingMessageBuffer.getBuffer().hasRemaining()) { // result = false;
//Write the bytes to the stream.// // }//if//
((SocketChannel) key.channel()).write(pendingMessageBuffer.getBuffer()); // else {
// //Wait until additional message bytes are available.//
//If not all the bytes could be written then we will need to wait until we can write more.// // result = false;
if(pendingMessageBuffer.getBuffer().hasRemaining()) { // pendingMessageBuffer = null;
result = false; // lastAddedMessageBuffer = null;
}//if// // }//else//
else { // }//while//
//Load the next pending outbound message in the chain.// // }//if//
if(pendingMessageBuffer.getNext() != null) {
pendingMessageBuffer = pendingMessageBuffer.getNext();
}//if//
else {
//Wait until additional message bytes are available.//
result = false;
pendingMessageBuffer = null;
lastAddedMessageBuffer = null;
}//else//
}//else//
}//while//
}//if//
}//processResponses()// }//processResponses()//
/* (non-Javadoc) /* (non-Javadoc)
* @see com.foundation.web.server.WebServer.AbstractSocketContext#processRequest() * @see com.foundation.web.server.WebServer.AbstractSocketContext#processRequest()
@@ -136,23 +126,25 @@ protected void readIncomingMessages() throws IOException {
* @see com.foundation.web.server.WebServer.AbstractSocketContext#passThrough(java.nio.ByteBuffer) * @see com.foundation.web.server.WebServer.AbstractSocketContext#passThrough(java.nio.ByteBuffer)
*/ */
protected synchronized boolean passThrough(ByteBuffer buffer) { protected synchronized boolean passThrough(ByteBuffer buffer) {
ByteBuffer messageBytes = ByteBuffer.allocate(buffer.remaining()); // ByteBuffer messageBytes = ByteBuffer.allocate(buffer.remaining());
MessageBuffer message; // MessageBuffer message;
//
//Create a new buffer to hold the data so we don't modify the passed buffer (other than to update its position).// // //Create a new buffer to hold the data so we don't modify the passed buffer (other than to update its position).//
messageBytes = ByteBuffer.allocate(buffer.remaining()); // messageBytes = ByteBuffer.allocate(buffer.remaining());
messageBytes.put(buffer); // messageBytes.put(buffer);
message = new MessageBuffer(messageBytes); // message = new MessageBuffer(messageBytes);
//
//Chain the message into the linked list. //
if(lastAddedMessageBuffer == null) { // //Chain the message into the linked list.
pendingMessageBuffer = lastAddedMessageBuffer = message; // if(lastAddedMessageBuffer == null) {
}//if// // pendingMessageBuffer = lastAddedMessageBuffer = message;
else { // }//if//
lastAddedMessageBuffer.setNext(message); // else {
lastAddedMessageBuffer = message; // lastAddedMessageBuffer.setNext(message);
}//else// // lastAddedMessageBuffer = message;
// }//else//
//
// return true;
return true; return true;
}//passThrough()// }//passThrough()//
protected synchronized void close() { protected synchronized void close() {
@@ -163,6 +155,7 @@ protected synchronized void close() {
* @see com.foundation.web.server.WebServer.AbstractSocketContext#hasPendingWrite() * @see com.foundation.web.server.WebServer.AbstractSocketContext#hasPendingWrite()
*/ */
protected boolean hasPendingWrite() { protected boolean hasPendingWrite() {
return pendingMessageBuffer != null; // return pendingMessageBuffer != null;
return false;
}//hasPendingWrite()// }//hasPendingWrite()//
}//PassThroughSocketContext// }//PassThroughSocketContext//

16
Foundation Web Core/src/com/foundation/web/server/Response.java
View File

@@ -26,8 +26,6 @@ import com.foundation.web.interfaces.*;
public class Response implements IResponse { public class Response implements IResponse {
/** The sequence number for the request. */ /** The sequence number for the request. */
private int requestNumber = 0; private int requestNumber = 0;
/** Used to chain responses together when sending. */
private Response nextResponse = null;
/** The session associated with the response. */ /** The session associated with the response. */
private ISession session = null; private ISession session = null;
/** The application that is serving the response. */ /** The application that is serving the response. */
@@ -242,20 +240,6 @@ public String getCustomHeader() {
public void setCustomHeader(String customHeader) { public void setCustomHeader(String customHeader) {
this.customHeader = customHeader; this.customHeader = customHeader;
}//getCustomHeader()// }//getCustomHeader()//
/**
* Gets the next response in the chain of responses.
* @return The next response to be sent.
*/
public Response getNextResponse() {
return nextResponse;
}//getNextResponse()//
/**
* Sets the next response in the chain of responses.
* @param nextResponse The response to send after this one.
*/
public void setNextResponse(Response nextResponse) {
this.nextResponse = nextResponse;
}//setNextResponse()//
/** /**
* Gets the ordered list of header field names. * Gets the ordered list of header field names.
* @return The optional field names for the header. If provided then the server shouldn't generate a header for the response. * @return The optional field names for the header. If provided then the server shouldn't generate a header for the response.

397
Foundation Web Core/src/com/foundation/web/server/SocketContext.java
View File

@@ -70,10 +70,14 @@ public class SocketContext extends AbstractSocketContext implements IWebApplicat
public ContentPart currentPart = null; public ContentPart currentPart = null;
/** Whether the final part boundary has already been found. If this is true then the message bytes should keep getting read until the partReadCount == contentLength. */ /** Whether the final part boundary has already been found. If this is true then the message bytes should keep getting read until the partReadCount == contentLength. */
public boolean endPartBoundaryFound = false; public boolean endPartBoundaryFound = false;
/** The bytes containing the unencrypted outbound message that is waiting for the socket to allow a write. */ // /** The bytes containing the unencrypted outbound message that is waiting for the socket to allow a write. */
public MessageBuffer/*ByteBuffer*/ currentOutboundMessage = null; // public MessageBuffer/*ByteBuffer*/ currentOutboundMessage = null;
/** The last message buffer added to the pending outbound message chain (linked list). Used only for pass through contexts currently since locally handled messages link the reponses together into a list. */ // /** The last message buffer added to the pending outbound message chain (linked list). Used only for pass through contexts currently since locally handled messages link the reponses together into a list. */
private MessageBuffer lastOutboundMessage = null; // private MessageBuffer lastOutboundMessage = null;
/** The queue of ready to be sent outbound messages. */
private Queue outboundMessages = new Queue(6);
/** The currently active outbound message (pulled from the outboundMessages queue). */
private MessageBuffer outboundMessage = null;
/** The byte buffer used to read data from the socket. This must be null if a SSL engine is provided. */ /** The byte buffer used to read data from the socket. This must be null if a SSL engine is provided. */
public ByteBuffer socketReadBuffer = null; public ByteBuffer socketReadBuffer = null;
/** The buffer used to store the initial data in a SSL/TLS connection. The buffering is necessary to allow us to pre-read the client's hello message to gather the domain the client is connecting to - allowing for the correct SSL engine to be used. */ /** The buffer used to store the initial data in a SSL/TLS connection. The buffering is necessary to allow us to pre-read the client's hello message to gather the domain the client is connecting to - allowing for the correct SSL engine to be used. */
@@ -94,10 +98,10 @@ public class SocketContext extends AbstractSocketContext implements IWebApplicat
public ByteBuffer encryptedWriteBuffer = null; public ByteBuffer encryptedWriteBuffer = null;
/** The last used request number which identifies the sequence for the requests. */ /** The last used request number which identifies the sequence for the requests. */
private int lastRequestNumber = 0; private int lastRequestNumber = 0;
/** The response we are currently processing. */ // /** The response we are currently processing. */
private Response currentResponse = null; // private Response currentResponse = null;
/** The response we are will process last. */ // /** The response we are will process last. */
private Response lastResponse = null; // private Response lastResponse = null;
/** Tracks the number of bytes sent from the current response. This is only used when debugging. */ /** Tracks the number of bytes sent from the current response. This is only used when debugging. */
private int sentBytes = 0; private int sentBytes = 0;
/** Tracks the getWebServer().debug output for the current request/response cycle. This is only used when debugging. */ /** Tracks the getWebServer().debug output for the current request/response cycle. This is only used when debugging. */
@@ -189,7 +193,7 @@ protected synchronized void close() {
try {if(key != null) key.cancel();} catch(Throwable e) {} try {if(key != null) key.cancel();} catch(Throwable e) {}
//Clean up after the response and request.// //Clean up after the response and request.//
//try {while(currentOutboundMessage != null) {currentOutboundMessage.close(); currentOutboundMessage = currentOutboundMessage.getNext();}} catch(Throwable e2) {} //try {while(currentOutboundMessage != null) {currentOutboundMessage.close(); currentOutboundMessage = currentOutboundMessage.getNext();}} catch(Throwable e2) {}
try {if(currentResponse != null) currentResponse.close();} catch(Throwable e2) {} // try {if(currentResponse != null) currentResponse.close();} catch(Throwable e2) {}
if(getPassThroughSocketContext() != null) { if(getPassThroughSocketContext() != null) {
getPassThroughSocketContext().close(); getPassThroughSocketContext().close();
@@ -232,22 +236,14 @@ private void queueOutboundClientMessage(MessageBuffer messageBuffer) {
boolean notify = false; boolean notify = false;
synchronized(this) { synchronized(this) {
if(currentOutboundMessage == null) { outboundMessages.enqueue(messageBuffer);
lastOutboundMessage = currentOutboundMessage = messageBuffer;
notify = true;
}//if//
else {
lastOutboundMessage.setNext(messageBuffer);
lastOutboundMessage = messageBuffer;
}//else//
}//synchronized()// }//synchronized()//
if(notify) { if(notify) {
notifyListenerOfPendingWrite(); notifyListenerOfPendingWrite();
}//if// }//if//
}//queueOutboundClientMessage()// }//queueOutboundClientMessage()//
private void writeSessionCookies(PrintStream pout) { private void writeSessionCookies(Response response, PrintStream pout) {
Response response = currentResponse;
ISession session = response.getSession(); ISession session = response.getSession();
if(session != null) { if(session != null) {
@@ -271,8 +267,7 @@ private void writeSessionCookies(PrintStream pout) {
* <p>Note: The caller must synchronize on this context to prevent multiple threads from accessing the context at the same time.</p> * <p>Note: The caller must synchronize on this context to prevent multiple threads from accessing the context at the same time.</p>
* @result Whether request is in a receive state. Will be false if the request generated a response that could not be completely transmitted. * @result Whether request is in a receive state. Will be false if the request generated a response that could not be completely transmitted.
*/ */
private void prepareResponse() { private void prepareResponse(Response response) {
Response response = currentResponse;
Request request = (Request) response.getRequest(); Request request = (Request) response.getRequest();
byte[] headerBytes = null; byte[] headerBytes = null;
IContent content = null; IContent content = null;
@@ -308,7 +303,7 @@ private void prepareResponse() {
}//for// }//for//
//Write out any cookies necessary to retain our session.// //Write out any cookies necessary to retain our session.//
writeSessionCookies(pout); writeSessionCookies(response, pout);
//End the header.// //End the header.//
pout.print("\r\n"); pout.print("\r\n");
@@ -331,7 +326,7 @@ private void prepareResponse() {
pout.print("HTTP/1.1 302 Moved Temporarily\r\n"); pout.print("HTTP/1.1 302 Moved Temporarily\r\n");
//}//else// //}//else//
writeSessionCookies(pout); writeSessionCookies(response, pout);
pout.print("Location: " + response.getForwardUri() + "\r\n"); pout.print("Location: " + response.getForwardUri() + "\r\n");
@@ -423,7 +418,7 @@ private void prepareResponse() {
}//if// }//if//
}//if// }//if//
writeSessionCookies(pout); writeSessionCookies(response, pout);
pout.print("Server: DE/1.0\r\n"); pout.print("Server: DE/1.0\r\n");
//TODO: IE has a problem with caching and forwarding/redirecting. A page that redirects to another page that was previously cached does not result in IE sending a request for the forwarded content.// //TODO: IE has a problem with caching and forwarding/redirecting. A page that redirects to another page that was previously cached does not result in IE sending a request for the forwarded content.//
@@ -487,7 +482,7 @@ private void prepareResponse() {
pout.print("HTTP/1.1 200 OK\r\n"); pout.print("HTTP/1.1 200 OK\r\n");
}//else// }//else//
writeSessionCookies(pout); writeSessionCookies(response, pout);
pout.print("Content-Length: 0\r\n"); pout.print("Content-Length: 0\r\n");
pout.print("Server: DE/1.0\r\n"); pout.print("Server: DE/1.0\r\n");
pout.print("\r\n"); pout.print("\r\n");
@@ -516,7 +511,7 @@ private void prepareResponse() {
// }//if// // }//if//
//Save the buffer as the current pending outbound message for this socket context.// //Save the buffer as the current pending outbound message for this socket context.//
currentOutboundMessage = new MessageBuffer(buffer, response, content != null && request.getRequestType() != Request.TYPE_HEAD ? content : null); queueOutboundClientMessage(new MessageBuffer(buffer, response, content != null && request.getRequestType() != Request.TYPE_HEAD ? content : null));
}//try// }//try//
catch(Throwable e) { catch(Throwable e) {
Debug.log("Fatal Error: Failed to build and send the response message due to an exception.", e); Debug.log("Fatal Error: Failed to build and send the response message due to an exception.", e);
@@ -533,18 +528,7 @@ private void prepareResponse() {
* @result Whether request is in a receive state. Will be false if the request generated a response that could not be completely transmitted. * @result Whether request is in a receive state. Will be false if the request generated a response that could not be completely transmitted.
*/ */
public synchronized boolean sendHttpResponse(Response response) { public synchronized boolean sendHttpResponse(Response response) {
if(currentResponse != null) { prepareResponse(response);
lastResponse.setNextResponse(response);
lastResponse = response;
}//if//
else {
lastResponse = currentResponse = response;
sentBytes = 0;
prepareResponse();
//Note: Not going to process the response on this thread. Allow the flag to be set for writing to the socket, and have the next thread in the network listener handle the write. This allows for cleaner code and pipelining without all the synchronizing.
// result = internalProcessResponses();
}//else//
request = null; request = null;
@@ -562,14 +546,7 @@ protected synchronized boolean passThrough(ByteBuffer buffer) {
messageBytes.put(buffer); messageBytes.put(buffer);
message = new MessageBuffer(messageBytes); message = new MessageBuffer(messageBytes);
//Chain the message into the linked list. queueOutboundClientMessage(message);
if(lastOutboundMessage == null || currentOutboundMessage == null) {
currentOutboundMessage = lastOutboundMessage = message;
}//if//
else {
lastOutboundMessage.setNext(message);
lastOutboundMessage = message;
}//else//
return true; return true;
}//passThrough()// }//passThrough()//
@@ -584,12 +561,12 @@ protected void writeOutgoingMessages() throws IOException {
}//synchronized// }//synchronized//
}//if// }//if//
else if(isWebsocket) { else if(isWebsocket) {
//Right after upgrading the socket we have one last HTTP response to process.// // //Right after upgrading the socket we have one last HTTP response to process.//
if(currentResponse != null) { // if(currentResponse != null) {
internalProcessResponses(); // internalProcessResponses();
}//if// // }//if//
//
internalProcessWebsocketMessages(); // internalProcessWebsocketMessages();
}//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.//
@@ -601,30 +578,30 @@ protected void writeOutgoingMessages() throws IOException {
* If a message could only be partially sent then the next call will attempt to finish sending it. * If a message could only be partially sent then the next call will attempt to finish sending it.
*/ */
private void internalProcessWebsocketMessages() { private void internalProcessWebsocketMessages() {
if(websocketSendingMessage == null) { // if(websocketSendingMessage == null) {
loadNextWebsocketMessage(); // loadNextWebsocketMessage();
}//if// // }//if//
//
while(websocketSendingMessage != null) { // while(websocketSendingMessage != 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()) {
if(currentOutboundMessage != null) { // if(currentOutboundMessage != null) {
//Put the sending message in a MessageBuffer (pendingOutboundMessage).// // //Put the sending message in a MessageBuffer (pendingOutboundMessage).//
currentOutboundMessage = new MessageBuffer(websocketSendingMessage); // currentOutboundMessage = new MessageBuffer(websocketSendingMessage);
}//if// // }//if//
//
//Write the pendingOutboundMessage to the socket.// // //Write the pendingOutboundMessage to the socket.//
if(writeClientBoundMessage()) { // if(writeClientBoundMessage()) {
websocketSendingMessage = null; // websocketSendingMessage = null;
currentOutboundMessage = null; // currentOutboundMessage = null;
}//if// // }//if//
}//if// // }//if//
//
//If we finished sending the message then load the next one.// // //If we finished sending the message then load the next one.//
if(websocketSendingMessage == null) { // if(websocketSendingMessage == null) {
loadNextWebsocketMessage(); // loadNextWebsocketMessage();
}//if// // }//if//
}//while// // }//while//
}//internalProcessWebsocketMessages()// }//internalProcessWebsocketMessages()//
/** /**
* Loads and prepares the next websocket message from the queue of pending messages. * Loads and prepares the next websocket message from the queue of pending messages.
@@ -726,42 +703,46 @@ private void loadNextWebsocketMessage() {
private synchronized void internalProcessResponses() { private synchronized void internalProcessResponses() {
boolean finishedSending = true; boolean finishedSending = true;
//Keep sending responses while the buffers are not full and there is another response to send.// while(finishedSending) {
while(finishedSending && currentResponse != null) {
//If the socket is open then send the next buffer of data.//
if(key.channel().isOpen()) {
//Send the pending response object's prepared buffer of data.//
finishedSending = writeClientBoundMessage(); finishedSending = writeClientBoundMessage();
}//if//
//Close the response if successfully sent, or if the socket is closed.//
if(finishedSending || !key.channel().isOpen()) {
try {currentResponse.close();} catch(Throwable e) {}
}//if//
//If we finished sending the current response then load the next one.//
if(finishedSending) {
currentResponse = currentResponse.getNextResponse();
if(currentResponse == null) {
lastResponse = null;
}//if//
else if(key.channel().isOpen()) {
//Prep the next response object for sending.//
prepareResponse();
}//else//
else {
//Clean up after all the left over responses.//
while(currentResponse != null) {
currentResponse.close();
currentResponse = currentResponse.getNextResponse();
}//while// }//while//
currentResponse = null; // //Keep sending responses while the buffers are not full and there is another response to send.//
lastResponse = null; // while(finishedSending && currentResponse != null) {
}//else// // //If the socket is open then send the next buffer of data.//
}//if// // if(key.channel().isOpen()) {
}//while// // //Send the pending response object's prepared buffer of data.//
// finishedSending = writeClientBoundMessage();
// }//if//
//
// //Close the response if successfully sent, or if the socket is closed.//
// if(finishedSending || !key.channel().isOpen()) {
// try {currentResponse.close();} catch(Throwable e) {}
// }//if//
//
// //If we finished sending the current response then load the next one.//
// if(finishedSending) {
// currentResponse = currentResponse.getNextResponse();
//
// if(currentResponse == null) {
// lastResponse = null;
// }//if//
// else if(key.channel().isOpen()) {
// //Prep the next response object for sending.//
// prepareResponse(currentResponse);
// }//else//
// else {
// //Clean up after all the left over responses.//
// while(currentResponse != null) {
// currentResponse.close();
// currentResponse = currentResponse.getNextResponse();
// }//while//
//
// currentResponse = null;
// lastResponse = null;
// }//else//
// }//if//
// }//while//
}//processCurrentResponse()// }//processCurrentResponse()//
/** /**
* Sends a response to the client. * Sends a response to the client.
@@ -770,13 +751,24 @@ private synchronized void internalProcessResponses() {
private boolean writeClientBoundMessage() { private boolean writeClientBoundMessage() {
boolean sendMore = true; boolean sendMore = true;
// if(getWebServer().debug) {
// debugBuffer.append("Starting a write cycle.\n");
// }//if//
try {
//Process SSL output first.// //Process SSL output first.//
if(sslEngine != null) { if(sslEngine != null) {
sendMore = writeClientBoundSslMessage();
}//if//
else {
sendMore = writeClientBoundPlainMessage();
}//else//
return sendMore;
}//writeClientBoundMessage()//
/**
* 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.
*/
private boolean writeClientBoundSslMessage() {
boolean sendMore = true;
try {
//If we have part of an SSL frame then try to send it first.// //If we have part of an SSL frame then try to send it first.//
if(encryptedWriteBuffer.hasRemaining()) { if(encryptedWriteBuffer.hasRemaining()) {
int remaining = encryptedWriteBuffer.remaining(); int remaining = encryptedWriteBuffer.remaining();
@@ -784,10 +776,6 @@ private boolean writeClientBoundMessage() {
//Write the bytes to the stream.// //Write the bytes to the stream.//
((SocketChannel) key.channel()).write(encryptedWriteBuffer); ((SocketChannel) key.channel()).write(encryptedWriteBuffer);
// if(getWebServer().debug) {
// debugBuffer.append("Wrote " + (remaining - encryptedWriteBuffer.remaining()) + " encrypted bytes to the stream. " + encryptedWriteBuffer.remaining() + " remain.\n");
// }//if//
//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()) {
sendMore = false; sendMore = false;
@@ -859,50 +847,31 @@ private boolean writeClientBoundMessage() {
}//else// }//else//
}//while// }//while//
// if(getWebServer().debug) { if(sendMore) {
// debugBuffer.append("End Handshaking SSL\n"); //Prepare a new message for sending.//
// }//if// if(outboundMessage == null && outboundMessages.getSize() > 0) {
outboundMessage = (MessageBuffer) outboundMessages.dequeue();
}//if// }//if//
if(sendMore && currentOutboundMessage != 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(!currentOutboundMessage.getBuffer().hasRemaining()) { if(!outboundMessage.getBuffer().hasRemaining()) {
if(!currentOutboundMessage.loadBuffer()) { if(!outboundMessage.loadBuffer()) {
if(currentOutboundMessage.getBuffer() == null && currentOutboundMessage.getNext() != null) {
currentOutboundMessage = currentOutboundMessage.getNext();
}//if//
else {
sendMore = false; sendMore = false;
}//else//
}//if// }//if//
if(currentOutboundMessage.getBuffer() == null) { if(outboundMessage.getBuffer() == null) {
currentOutboundMessage = null; outboundMessage = null;
lastOutboundMessage = null;
}//if// }//if//
}//if// }//if//
//If we have an application response pending then send it now.//
if(sendMore && currentOutboundMessage.getBuffer().hasRemaining()) {
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() && sendMore && (currentOutboundMessage != null) && currentOutboundMessage.getBuffer().hasRemaining()) { while(key.channel().isOpen() && sendMore && outboundMessage != null && outboundMessage.getBuffer().hasRemaining()) {
SSLEngineResult encryptResult; SSLEngineResult encryptResult;
// int offset = pendingOutboundMessage.getBuffer().position();
//TODO: Comment me.
//int rem = pendingOutboundMessage.getBuffer().remaining();
//Reset the encrypted write buffer.// //Reset the encrypted write buffer.//
encryptedWriteBuffer.compact(); encryptedWriteBuffer.compact();
//Encrypt the next message frame.// //Encrypt the next message frame.//
encryptResult = sslEngine.wrap(currentOutboundMessage.getBuffer(), encryptedWriteBuffer); encryptResult = sslEngine.wrap(outboundMessage.getBuffer(), encryptedWriteBuffer);
encryptedWriteBuffer.flip(); encryptedWriteBuffer.flip();
//TODO: Comment me.
//Debug.log("Encrypting/Sending to client from Git " + (rem - pendingOutboundMessage.getBuffer().remaining()) + " bytes.");
// if(getWebServer().debug) {
// sentBytes += (pendingOutboundMessage.position() - offset);
// debugBuffer.append("Encrypted: " + (pendingOutboundMessage.position() - offset) + ". Total Encrypted: " + sentBytes + ". Encrypted size: " + encryptedWriteBuffer.limit() + ".\n");
// }//if//
if(encryptResult.getStatus() == Status.BUFFER_OVERFLOW) { if(encryptResult.getStatus() == Status.BUFFER_OVERFLOW) {
//Should never happen.// //Should never happen.//
@@ -913,9 +882,6 @@ private boolean writeClientBoundMessage() {
Debug.log(new RuntimeException("Unexpected ssl engine buffer underflow.")); Debug.log(new RuntimeException("Unexpected ssl engine buffer underflow."));
}//else if// }//else if//
else if(encryptResult.getStatus() == Status.CLOSED) { else if(encryptResult.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.// //Close the socket.//
try {key.channel().close();} catch(Throwable e2) {} try {key.channel().close();} catch(Throwable e2) {}
}//else if// }//else if//
@@ -925,10 +891,6 @@ private boolean writeClientBoundMessage() {
int remaining = encryptedWriteBuffer.remaining(); int remaining = encryptedWriteBuffer.remaining();
((SocketChannel) key.channel()).write(encryptedWriteBuffer); ((SocketChannel) key.channel()).write(encryptedWriteBuffer);
// if(getWebServer().debug) {
// debugBuffer.append("Sent " + (remaining - encryptedWriteBuffer.remaining()) + " encrypted bytes.\n");
// }//if//
}//try// }//try//
catch(IOException e) { catch(IOException e) {
//Caught if the channel is forcably closed by the client. We will ignore it.// //Caught if the channel is forcably closed by the client. We will ignore it.//
@@ -938,10 +900,6 @@ private boolean writeClientBoundMessage() {
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.//
sendMore = false; sendMore = false;
// if(getWebServer().debug) {
// debugBuffer.append("Pausing due to a partially sent packet. Bytes actually sent: " + encryptedWriteBuffer.position() + ". Bytes remaining: " + encryptedWriteBuffer.remaining() + ".\n");
// }//if//
}//if// }//if//
}//else if// }//else if//
else { else {
@@ -951,63 +909,19 @@ private boolean writeClientBoundMessage() {
//Add more content to the buffer.// //Add more content to the buffer.//
//Note: Do this even if the last encrypted write buffer could not be fully sent - so that when it is sent there will be outbound message content.// //Note: Do this even if the last encrypted write buffer could not be fully sent - so that when it is sent there will be outbound message content.//
if(key.channel().isOpen() && currentOutboundMessage != null) { if(key.channel().isOpen() && outboundMessage != null) {
if(!currentOutboundMessage.loadBuffer()) { if(!outboundMessage.loadBuffer()) {
//Load the next pending outbound message in the chain. This is currently only used for content being passed through to another process via a second socket.//
if(currentOutboundMessage.getBuffer() == null && currentOutboundMessage.getNext() != null) {
currentOutboundMessage = currentOutboundMessage.getNext();
}//if//
else {
//Wait until additional message bytes are available.// //Wait until additional message bytes are available.//
sendMore = false; sendMore = false;
}//else//
}//if// }//if//
//If the message end has been reached then the buffer will be null.// //If the message end has been reached then the buffer will be null.//
if(currentOutboundMessage.getBuffer() == null) { if(outboundMessage.getBuffer() == null) {
currentOutboundMessage = null; outboundMessage = null;
lastOutboundMessage = null;
}//if// }//if//
}//if// }//if//
}//while// }//while//
}//if// }//if//
else {
//Keep sending encrypted frames until the output buffer is full, or we run out of message to send.//
while(sendMore && (currentOutboundMessage != null) && currentOutboundMessage.getBuffer().hasRemaining()) {
//Write the bytes to the stream.//
((SocketChannel) key.channel()).write(currentOutboundMessage.getBuffer());
// if(getWebServer().debug) {
// sentBytes += pendingOutboundMessage.position();
// debugBuffer.append("Wrote " + pendingOutboundMessage.position() + " bytes to the client. Total sent: " + sentBytes + "\n");
// }//if//
//If not all the bytes could be written then we will need to wait until we can write more.//
if(currentOutboundMessage.getBuffer().hasRemaining()) {
sendMore = false;
}//if//
else {
if(!currentOutboundMessage.loadBuffer()) {
//Load the next pending outbound message in the chain. This is currently only used for content being passed through to another process via a second socket.//
if(currentOutboundMessage.getBuffer() == null && currentOutboundMessage.getNext() != null) {
currentOutboundMessage = currentOutboundMessage.getNext();
}//if//
else {
//Wait until additional message bytes are available.//
sendMore = false;
}//else//
}//if//
//If the message end has been reached then the buffer will be null.//
if(currentOutboundMessage.getBuffer() == null) {
currentOutboundMessage = null;
lastOutboundMessage = null;
}//if//
}//else//
}//while//
}//else//
}//if//
}//if//
}//try// }//try//
catch(ClosedChannelException e) { catch(ClosedChannelException e) {
close(); close();
@@ -1028,7 +942,73 @@ private boolean writeClientBoundMessage() {
}//catch// }//catch//
return sendMore; return sendMore;
}//writeClientBoundMessage()// }//writeClientBoundSslMessage()//
/**
* 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.
*/
private boolean writeClientBoundPlainMessage() {
boolean sendMore = true;
try {
//Prepare a new message for sending.//
if(outboundMessage == null && outboundMessages.getSize() > 0) {
outboundMessage = (MessageBuffer) outboundMessages.dequeue();
}//if//
//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(!outboundMessage.getBuffer().hasRemaining()) {
if(!outboundMessage.loadBuffer()) {
sendMore = false;
}//if//
if(outboundMessage.getBuffer() == null) {
outboundMessage = null;
}//if//
}//if//
//Keep sending encrypted frames until the output buffer is full, or we run out of message to send.//
while(sendMore && outboundMessage != null && outboundMessage.getBuffer().hasRemaining()) {
//Write the bytes to the stream.//
((SocketChannel) key.channel()).write(outboundMessage.getBuffer());
//If not all the bytes could be written then we will need to wait until we can write more.//
if(outboundMessage.getBuffer().hasRemaining()) {
sendMore = false;
}//if//
else {
if(!outboundMessage.loadBuffer()) {
//Wait until additional message bytes are available.//
sendMore = false;
}//if//
//If the message end has been reached then the buffer will be null.//
if(outboundMessage.getBuffer() == null) {
outboundMessage = null;
}//if//
}//else//
}//while//
}//try//
catch(ClosedChannelException e) {
close();
}//catch//
catch(SSLException e) {
if(getWebServer().debug) {
Debug.log(e);
}//if//
close();
}//catch//
catch(IOException e) {
if(getWebServer().debug) {
Debug.log(e);
}//if//
close();
}//catch//
return sendMore;
}//writeClientBoundPlainMessage()//
/* (non-Javadoc) /* (non-Javadoc)
* @see com.foundation.web.server.WebServer.AbstractSocketContext#processRequest() * @see com.foundation.web.server.WebServer.AbstractSocketContext#processRequest()
*/ */
@@ -2543,7 +2523,8 @@ private int indexOf(byte[] source, byte[] pattern, int fromOffset) {
* @see com.foundation.web.server.WebServer.AbstractSocketContext#hasPendingWrite() * @see com.foundation.web.server.WebServer.AbstractSocketContext#hasPendingWrite()
*/ */
protected boolean hasPendingWrite() { protected boolean hasPendingWrite() {
return currentOutboundMessage != null || (encryptedWriteBuffer != null && encryptedWriteBuffer.hasRemaining()); //return currentOutboundMessage != null || (encryptedWriteBuffer != null && encryptedWriteBuffer.hasRemaining());
return outboundMessage != null || outboundMessages.getSize() > 0 || (encryptedWriteBuffer != null && encryptedWriteBuffer.hasRemaining());
}//hasPendingWrite()// }//hasPendingWrite()//
/* (non-Javadoc) /* (non-Javadoc)
* @see com.foundation.web.interfaces.IConnectionContext#upgradeToWebsocket(java.lang.String, long, com.foundation.web.interfaces.WebsocketHandler) * @see com.foundation.web.interfaces.IConnectionContext#upgradeToWebsocket(java.lang.String, long, com.foundation.web.interfaces.WebsocketHandler)