Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Reduce Code Complexity in TCP Methods

Browse Source
This commit is contained in:
Gary Russell
2021-04-26 13:02:58 -04:00
parent 0ea9d700de
commit 0dfbfa0677
4 changed files with 203 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -648,33 +648,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
this.nextCheckForClosedNioConnections = now + this.nioHarvestInterval;
Iterator<Entry<SocketChannel, TcpNioConnection>> it = connectionMap.entrySet().iterator();
while (it.hasNext()) {
SocketChannel channel = it.next().getKey();
if (!channel.isOpen()) {
logger.debug("Removing closed channel");
it.remove();
}
else if (this.soTimeout > 0) {
TcpNioConnection connection = connectionMap.get(channel);
if (now - connection.getLastRead() >= this.soTimeout) {
/*
* For client connections, we have to wait for 2 timeouts if the last
* send was within the current timeout.
*/
if (!connection.isServer() &&
now - connection.getLastSend() < this.soTimeout &&
now - connection.getLastRead() < this.soTimeout * 2) {
logger.debug(() -> "Skipping a connection timeout because we have a recent send " +
connection.getConnectionId());
}
else {
logger.warn(() -> "Timing out TcpNioConnection " + connection.getConnectionId());
Exception exception = new SocketTimeoutException("Timing out connection");
connection.publishConnectionExceptionEvent(exception);
connection.timeout();
connection.sendExceptionToListener(exception);
}
}
}
checkChannel(connectionMap, now, it);
}
}
harvestClosedConnections();
@@ -695,61 +669,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
final SelectionKey key = iterator.next();
iterator.remove();
try {
if (!key.isValid()) {
logger.debug("Selection key no longer valid");
}
else if (key.isReadable()) {
key.interestOps(key.interestOps() - SelectionKey.OP_READ);
final TcpNioConnection connection;
connection = (TcpNioConnection) key.attachment();
connection.setLastRead(System.currentTimeMillis());
try {
this.taskExecutor.execute(() -> {
boolean delayed = false;
try {
connection.readPacket();
}
catch (@SuppressWarnings(UNUSED) RejectedExecutionException e1) {
delayRead(selector, now, key);
delayed = true;
}
catch (Exception e2) {
if (connection.isOpen()) {
logger.error(() -> "Exception on read " +
connection.getConnectionId() + " " +
e2.getMessage());
connection.close();
}
else {
logger.debug("Connection closed");
}
}
if (!delayed) {
if (key.channel().isOpen()) {
key.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
else {
connection.sendExceptionToListener(new EOFException("Connection is closed"));
}
}
});
}
catch (@SuppressWarnings(UNUSED) RejectedExecutionException e) {
delayRead(selector, now, key);
}
}
else if (key.isAcceptable()) {
try {
doAccept(selector, server, now);
}
catch (Exception ex) {
logger.error(ex, "Exception accepting new connection(s)");
}
}
else {
logger.error("Unexpected key: " + key);
}
handleKey(selector, server, now, key);
}
catch (@SuppressWarnings(UNUSED) CancelledKeyException e) {
logger.debug(() -> "Selection key " + key + " cancelled");
@@ -761,6 +681,105 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
}
}
private void checkChannel(Map<SocketChannel, TcpNioConnection> connectionMap, final long now,
Iterator<Entry<SocketChannel, TcpNioConnection>> it) {
SocketChannel channel = it.next().getKey();
if (!channel.isOpen()) {
logger.debug("Removing closed channel");
it.remove();
}
else if (this.soTimeout > 0) {
TcpNioConnection connection = connectionMap.get(channel);
if (now - connection.getLastRead() >= this.soTimeout) {
/*
* For client connections, we have to wait for 2 timeouts if the last
* send was within the current timeout.
*/
if (!connection.isServer() &&
now - connection.getLastSend() < this.soTimeout &&
now - connection.getLastRead() < this.soTimeout * 2) {
logger.debug(() -> "Skipping a connection timeout because we have a recent send " +
connection.getConnectionId());
}
else {
logger.warn(() -> "Timing out TcpNioConnection " + connection.getConnectionId());
Exception exception = new SocketTimeoutException("Timing out connection");
connection.publishConnectionExceptionEvent(exception);
connection.timeout();
connection.sendExceptionToListener(exception);
}
}
}
}
private void handleKey(final Selector selector, ServerSocketChannel server, final long now,
final SelectionKey key) {
if (!key.isValid()) {
logger.debug("Selection key no longer valid");
}
else if (key.isReadable()) {
keyReadable(selector, now, key);
}
else if (key.isAcceptable()) {
keyAcceptable(selector, server, now);
}
else {
logger.error("Unexpected key: " + key);
}
}
private void keyReadable(final Selector selector, final long now, final SelectionKey key) {
key.interestOps(key.interestOps() - SelectionKey.OP_READ);
final TcpNioConnection connection;
connection = (TcpNioConnection) key.attachment();
connection.setLastRead(System.currentTimeMillis());
try {
this.taskExecutor.execute(() -> {
boolean delayed = false;
try {
connection.readPacket();
}
catch (@SuppressWarnings(UNUSED) RejectedExecutionException e1) {
delayRead(selector, now, key);
delayed = true;
}
catch (Exception e2) {
if (connection.isOpen()) {
logger.error(() -> "Exception on read " +
connection.getConnectionId() + " " +
e2.getMessage());
connection.close();
}
else {
logger.debug("Connection closed");
}
}
if (!delayed) {
if (key.channel().isOpen()) {
key.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
else {
connection.sendExceptionToListener(new EOFException("Connection is closed"));
}
}
});
}
catch (@SuppressWarnings(UNUSED) RejectedExecutionException e) {
delayRead(selector, now, key);
}
}
private void keyAcceptable(final Selector selector, ServerSocketChannel server, final long now) {
try {
doAccept(selector, server, now);
}
catch (Exception ex) {
logger.error(ex, "Exception accepting new connection(s)");
}
}
protected void delayRead(Selector selector, long now, final SelectionKey key) {
TcpNioConnection connection = (TcpNioConnection) key.attachment();
if (!this.delayedReads.add(new PendingIO(now, key))) { // should never happen - unbounded queue

100
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -274,60 +274,70 @@ public class TcpNioConnection extends TcpConnectionSupport {
this.executionControl.decrementAndGet();
}
}
catch (Exception e) {
if (logger.isTraceEnabled()) {
logger.error("Read exception " + getConnectionId(), e);
}
else if (!isNoReadErrorOnClose()) {
logger.error("Read exception " +
getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Read exception " +
getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
}
closeConnection(true);
sendExceptionToListener(e);
catch (Exception ex) {
wrapUp(ex);
return;
}
}
finally {
moreDataAvailable = false;
// Final check in case new data came in and the
// timing was such that we were the last assembler and
// a new one wasn't run
if (dataAvailable()) {
synchronized (this.executionControl) {
if (this.executionControl.incrementAndGet() <= 1) {
// only continue if we don't already have another assembler running
this.executionControl.set(1);
moreDataAvailable = true;
moreDataAvailable = checkForMoreData();
}
}
}
private void wrapUp(Exception ex) {
if (logger.isTraceEnabled()) {
logger.error("Read exception " + getConnectionId(), ex);
}
else if (!isNoReadErrorOnClose()) {
logger.error("Read exception " +
getConnectionId() + " " +
ex.getClass().getSimpleName() +
":" + ex.getCause() + ":" + ex.getMessage());
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Read exception " +
getConnectionId() + " " +
ex.getClass().getSimpleName() +
":" + ex.getCause() + ":" + ex.getMessage());
}
}
closeConnection(true);
sendExceptionToListener(ex);
}
private boolean checkForMoreData() {
boolean moreDataAvailable;
moreDataAvailable = false;
// Final check in case new data came in and the
// timing was such that we were the last assembler and
// a new one wasn't run
if (dataAvailable()) {
synchronized (this.executionControl) {
if (this.executionControl.incrementAndGet() <= 1) {
// only continue if we don't already have another assembler running
this.executionControl.set(1);
moreDataAvailable = true;
}
else {
this.executionControl.decrementAndGet();
}
}
}
if (moreDataAvailable) {
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " Nio message assembler continuing...");
}
}
else {
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " Nio message assembler exiting... avail: "
+ this.channelInputStream.available());
}
this.executionControl.decrementAndGet();
}
}
}
if (moreDataAvailable) {
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " Nio message assembler continuing...");
}
}
else {
if (logger.isTraceEnabled()) {
logger.trace(getConnectionId() + " Nio message assembler exiting... avail: "
+ this.channelInputStream.available());
}
}
return moreDataAvailable;
}
private boolean dataAvailable() {
@@ -723,7 +733,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
}
int bite;
bite = this.currentBuffer[this.currentOffset++] & 0xff; // N0SONAR
bite = this.currentBuffer[this.currentOffset++] & 0xff; // NOSONAR
this.available.decrementAndGet();
if (this.currentOffset >= this.currentBuffer.length) {
this.currentBuffer = null;

73
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java
View File

@@ -27,6 +27,7 @@ import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLEngineResult.HandshakeStatus;
import javax.net.ssl.SSLEngineResult.Status;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLHandshakeException;
import javax.net.ssl.SSLSession;
@@ -152,8 +153,8 @@ public class TcpNioSSLConnection extends TcpNioConnection {
*/
@SuppressWarnings("fallthrough")
private SSLEngineResult decode(ByteBuffer networkBuffer) throws IOException {
SSLEngineResult result = new SSLEngineResult(Status.OK, this.sslEngine.getHandshakeStatus(), 0, 0);
HandshakeStatus handshakeStatus = this.sslEngine.getHandshakeStatus();
SSLEngineResult result = new SSLEngineResult(Status.OK, handshakeStatus, 0, 0);
switch (handshakeStatus) {
case NEED_TASK:
runTasks();
@@ -161,36 +162,10 @@ public class TcpNioSSLConnection extends TcpNioConnection {
case NEED_UNWRAP:
case FINISHED:
case NOT_HANDSHAKING:
((Buffer) this.decoded).clear();
result = this.sslEngine.unwrap(networkBuffer, this.decoded);
if (logger.isDebugEnabled()) {
logger.debug("After unwrap: " + resultToString(result));
}
Status status = result.getStatus();
if (status == Status.BUFFER_OVERFLOW) {
this.decoded =
this.allocateEncryptionBuffer(this.sslEngine.getSession().getApplicationBufferSize());
}
if (result.bytesProduced() > 0) {
((Buffer) this.decoded).flip();
super.sendToPipe(this.decoded);
}
result = checkBytesProduced(networkBuffer);
break;
case NEED_WRAP:
if (!resumeWriterIfNeeded()) {
((Buffer) this.encoded).clear();
result = this.sslEngine.wrap(networkBuffer, this.encoded);
if (logger.isDebugEnabled()) {
logger.debug("After wrap: " + resultToString(result));
}
if (result.getStatus() == Status.BUFFER_OVERFLOW) {
this.encoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getPacketBufferSize());
}
else {
((Buffer) this.encoded).flip();
getSSLChannelOutputStream().writeEncoded(this.encoded);
}
}
result = needWrap(networkBuffer, result);
break;
default:
}
@@ -208,6 +183,46 @@ public class TcpNioSSLConnection extends TcpNioConnection {
return result;
}
private SSLEngineResult checkBytesProduced(ByteBuffer networkBuffer) throws SSLException, IOException {
SSLEngineResult result;
((Buffer) this.decoded).clear();
result = this.sslEngine.unwrap(networkBuffer, this.decoded);
if (logger.isDebugEnabled()) {
logger.debug("After unwrap: " + resultToString(result));
}
Status status = result.getStatus();
if (status == Status.BUFFER_OVERFLOW) {
this.decoded =
this.allocateEncryptionBuffer(this.sslEngine.getSession().getApplicationBufferSize());
}
if (result.bytesProduced() > 0) {
((Buffer) this.decoded).flip();
super.sendToPipe(this.decoded);
}
return result;
}
private SSLEngineResult needWrap(ByteBuffer networkBuffer, SSLEngineResult result)
throws SSLException, IOException {
SSLEngineResult engineResult = result;
if (!resumeWriterIfNeeded()) {
((Buffer) this.encoded).clear();
engineResult = this.sslEngine.wrap(networkBuffer, this.encoded);
if (logger.isDebugEnabled()) {
logger.debug("After wrap: " + resultToString(engineResult));
}
if (engineResult.getStatus() == Status.BUFFER_OVERFLOW) {
this.encoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getPacketBufferSize());
}
else {
((Buffer) this.encoded).flip();
getSSLChannelOutputStream().writeEncoded(this.encoded);
}
}
return engineResult;
}
/**
* Handshake sends are handled by the initiator.
* @return false if we are the initiator.

6
spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java
View File

@@ -95,8 +95,8 @@ public final class TestMailServer {
super(socket);
}
@Override
void doRun() {
@Override // NOSONAR
void doRun() { // NOSONAR
try {
write("220 foo SMTP");
while (!socket.isClosed()) {
@@ -391,7 +391,7 @@ public final class TestMailServer {
LOGGER.error(IO_EXCEPTION, e);
}
}
} // NOSONAR
}
void searchReply(String tag) throws IOException {
if (seen) {