}
}
private void readMessage(SelectionKey sk, SocketChannel readChannel,
TcpAddress incomingAddress) throws IOException {
SocketEntry entry = (SocketEntry) sockets.get(incomingAddress);
if (entry != null) {
// note that socket has been used
entry.used();
ByteBuffer readBuffer = entry.getReadBuffer();
if (readBuffer != null) {
readChannel.read(readBuffer);
if (readBuffer.hasRemaining()) {
entry.addRegistration(selector, SelectionKey.OP_READ);
}
else {
entry.setReadBuffer(null); // <== set read buffer of entry to null
dispatchMessage(incomingAddress, readBuffer, readBuffer.capacity());
}
return;
}
}
ByteBuffer byteBuffer = ByteBuffer.wrap(buf);
byteBuffer.limit(messageLengthDecoder.getMinHeaderLength());
if (!readChannel.isOpen()) {
sk.cancel();
if (logger.isDebugEnabled()) {
logger.debug("Read channel not open, no bytes read from " +
incomingAddress);
}
return;
}
long bytesRead = 0;
try {
bytesRead = readChannel.read(byteBuffer);
if (logger.isDebugEnabled()) {
logger.debug("Reading header " + bytesRead + " bytes from " +
incomingAddress);
}
}
catch (ClosedChannelException ccex) {
sk.cancel();
if (logger.isDebugEnabled()) {
logger.debug("Read channel not open, no bytes read from " +
incomingAddress);
}
return;
}
MessageLength messageLength = new MessageLength(0, Integer.MIN_VALUE);
if (bytesRead == messageLengthDecoder.getMinHeaderLength()) {
messageLength =
messageLengthDecoder.getMessageLength(ByteBuffer.wrap(buf));
if (logger.isDebugEnabled()) {
logger.debug("Message length is "+messageLength);
}
if ((messageLength.getMessageLength() > getMaxInboundMessageSize()) ||
(messageLength.getMessageLength() <= 0)) {
logger.error("Received message length "+messageLength+
" is greater than inboundBufferSize "+
getMaxInboundMessageSize());
if (entry != null) {
Socket s = entry.getSocket();
if (s != null) {
s.close();
logger.info("Socket to " + entry.getPeerAddress() +
" closed due to an error");
}
}
}
else {
byteBuffer.limit(messageLength.getMessageLength());
bytesRead += readChannel.read(byteBuffer);
if (bytesRead == messageLength.getMessageLength()) {
dispatchMessage(incomingAddress, byteBuffer, bytesRead);
}
else {
byte[] message = new byte[byteBuffer.limit()];
int buflen = byteBuffer.limit() - byteBuffer.remaining();
byteBuffer.flip();
byteBuffer.get(message, 0, buflen);
ByteBuffer newBuffer = ByteBuffer.wrap(message);
newBuffer.position(buflen);
entry.setReadBuffer(newBuffer);
}
entry.addRegistration(selector, SelectionKey.OP_READ);
}
}
else if (bytesRead < 0) {
logger.debug("Socket closed remotely");
sk.cancel();
readChannel.close();
TransportStateEvent e =
new TransportStateEvent(DefaultTcpTransportMapping.this,
incomingAddress,
TransportStateEvent.
STATE_DISCONNECTED_REMOTELY,
null);
fireConnectionStateChanged(e);
}
else {
entry.addRegistration(selector, SelectionKey.OP_READ);
}
}