From 0dfbfa0677ae2a34d07a775d1545f5a1ab811e5b Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 26 Apr 2021 13:02:58 -0400 Subject: [PATCH] Reduce Code Complexity in TCP Methods --- .../connection/AbstractConnectionFactory.java | 183 ++++++++++-------- .../ip/tcp/connection/TcpNioConnection.java | 100 +++++----- .../tcp/connection/TcpNioSSLConnection.java | 73 ++++--- .../integration/test/mail/TestMailServer.java | 6 +- 4 files changed, 203 insertions(+), 159 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index c642f13ad3..409eea4e5e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -648,33 +648,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport this.nextCheckForClosedNioConnections = now + this.nioHarvestInterval; Iterator> 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 connectionMap, final long now, + Iterator> 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 diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index 74802a099e..1d1844c08f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -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; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java index 3460b4dbe3..3137e63095 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java @@ -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. diff --git a/spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java b/spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java index 2dcb21d407..09e1ab8555 100644 --- a/spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java +++ b/spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java @@ -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) {