From d2853be1cada8c706c03354c54b2a469737c5e56 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Sat, 30 Mar 2019 09:05:09 -0400 Subject: [PATCH] Fix new Sonar smells in TCP module; polishing --- .../TcpNioClientConnectionFactory.java | 102 +++++++++--------- .../TcpNioServerConnectionFactory.java | 17 +-- 2 files changed, 62 insertions(+), 57 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index 3ffd48dd5b..ab365f08c5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -36,24 +36,26 @@ import org.springframework.util.Assert; /** * A client connection factory that creates {@link TcpNioConnection}s. + * * @author Gary Russell * @author Artem Bilan + * * @since 2.0 * */ public class TcpNioClientConnectionFactory extends AbstractClientConnectionFactory implements SchedulingAwareRunnable { - private volatile boolean usingDirectBuffers; + private final Map channelMap = new ConcurrentHashMap<>(); + + private final BlockingQueue newChannels = new LinkedBlockingQueue<>(); + + private boolean usingDirectBuffers; + + private TcpNioConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioConnectionSupport(); private volatile Selector selector; - private final Map channelMap = new ConcurrentHashMap(); - - private final BlockingQueue newChannels = new LinkedBlockingQueue(); - - private volatile TcpNioConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioConnectionSupport(); - /** * Creates a TcpNioClientConnectionFactory for connections to the host and port. * @param host the host @@ -69,12 +71,12 @@ public class TcpNioClientConnectionFactory extends int n = 0; while (this.selector == null) { try { - Thread.sleep(100); + Thread.sleep(100); // NOSONAR magic number } catch (@SuppressWarnings("unused") InterruptedException e) { Thread.currentThread().interrupt(); } - if (n++ > 600) { + if (n++ > 600) { // NOSONAR magic number throw new UncheckedIOException(new IOException("Factory failed to start")); } } @@ -85,17 +87,19 @@ public class TcpNioClientConnectionFactory extends try { SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(getHost(), getPort())); setSocketAttributes(socketChannel.socket()); - TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection( - socketChannel, false, this.isLookupHost(), this.getApplicationEventPublisher(), getComponentName()); + TcpNioConnection connection = + this.tcpNioConnectionSupport.createNewConnection(socketChannel, false, isLookupHost(), + getApplicationEventPublisher(), getComponentName()); connection.setUsingDirectBuffers(this.usingDirectBuffers); - connection.setTaskExecutor(this.getTaskExecutor()); - if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) { - ((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout()); + connection.setTaskExecutor(getTaskExecutor()); + Integer sslHandshakeTimeout = getSslHandshakeTimeout(); + if (sslHandshakeTimeout != null && connection instanceof TcpNioSSLConnection) { + ((TcpNioSSLConnection) connection).setHandshakeTimeout(sslHandshakeTimeout); } TcpConnectionSupport wrappedConnection = wrapConnection(connection); initializeConnection(wrappedConnection, socketChannel.socket()); socketChannel.configureBlocking(false); - if (this.getSoTimeout() > 0) { + if (getSoTimeout() > 0) { connection.setLastRead(System.currentTimeMillis()); } this.channelMap.put(socketChannel, connection); @@ -144,9 +148,9 @@ public class TcpNioClientConnectionFactory extends @Override public void start() { synchronized (this.lifecycleMonitor) { - if (!this.isActive()) { - this.setActive(true); - this.getTaskExecutor().execute(this); + if (!isActive()) { + setActive(true); + getTaskExecutor().execute(this); } } super.start(); @@ -155,53 +159,53 @@ public class TcpNioClientConnectionFactory extends @Override public void run() { if (logger.isDebugEnabled()) { - logger.debug("Read selector running for connections to " + this.getHost() + ":" + this.getPort()); + logger.debug("Read selector running for connections to " + getHost() + ":" + getPort()); } try { this.selector = Selector.open(); - while (this.isActive()) { - SocketChannel newChannel; - int soTimeout = this.getSoTimeout(); - int selectionCount = 0; - try { - long timeout = soTimeout < 0 ? 0 : soTimeout; - if (getDelayedReads().size() > 0 && (timeout == 0 || getReadDelay() < timeout)) { - timeout = getReadDelay(); - } - selectionCount = this.selector.select(timeout); - } - catch (@SuppressWarnings("unused") CancelledKeyException cke) { - if (logger.isDebugEnabled()) { - logger.debug("CancelledKeyException during Selector.select()"); - } - } - while ((newChannel = this.newChannels.poll()) != null) { - try { - newChannel.register(this.selector, SelectionKey.OP_READ, this.channelMap.get(newChannel)); - } - catch (@SuppressWarnings("unused") ClosedChannelException cce) { - if (logger.isDebugEnabled()) { - logger.debug("Channel closed before registering with selector for reading"); - } - } - } - processNioSelections(selectionCount, this.selector, null, this.channelMap); + while (isActive()) { + processSelectorWhileActive(); } } catch (ClosedSelectorException cse) { - if (this.isActive()) { + if (isActive()) { logger.error("Selector closed", cse); } } catch (Exception e) { logger.error("Exception in read selector thread", e); - this.setActive(false); + setActive(false); } if (logger.isDebugEnabled()) { - logger.debug("Read selector exiting for connections to " + this.getHost() + ":" + this.getPort()); + logger.debug("Read selector exiting for connections to " + getHost() + ":" + getPort()); } } + private void processSelectorWhileActive() throws IOException { + SocketChannel newChannel; + int soTimeout = getSoTimeout(); + int selectionCount = 0; + try { + long timeout = soTimeout < 0 ? 0 : soTimeout; + if (getDelayedReads().size() > 0 && (timeout == 0 || getReadDelay() < timeout)) { + timeout = getReadDelay(); + } + selectionCount = this.selector.select(timeout); + } + catch (@SuppressWarnings("unused") CancelledKeyException cke) { + logger.debug("CancelledKeyException during Selector.select()"); + } + while ((newChannel = this.newChannels.poll()) != null) { + try { + newChannel.register(this.selector, SelectionKey.OP_READ, this.channelMap.get(newChannel)); + } + catch (@SuppressWarnings("unused") ClosedChannelException cce) { + logger.debug("Channel closed before registering with selector for reading"); + } + } + processNioSelections(selectionCount, this.selector, null, this.channelMap); + } + /** * @return the usingDirectBuffers */ diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index e9be671de1..f85caf7794 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -47,18 +47,18 @@ import org.springframework.util.Assert; */ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFactory { + private final Map channelMap = new HashMap<>(); + + private TcpNioConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioConnectionSupport(); + private boolean multiAccept = true; + private boolean usingDirectBuffers; + private volatile ServerSocketChannel serverChannel; - private volatile boolean usingDirectBuffers; - - private final Map channelMap = new HashMap<>(); - private volatile Selector selector; - private volatile TcpNioConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioConnectionSupport(); - /** * Listens for incoming connections on the port. * @param port The port. @@ -256,8 +256,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto if (connection != null) { connection.setTaskExecutor(getTaskExecutor()); connection.setLastRead(now); - if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) { - ((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout()); + Integer sslHandshakeTimeout = getSslHandshakeTimeout(); + if (sslHandshakeTimeout != null && connection instanceof TcpNioSSLConnection) { + ((TcpNioSSLConnection) connection).setHandshakeTimeout(sslHandshakeTimeout); } this.channelMap.put(channel, connection); channel.register(selectorForNewSocket, SelectionKey.OP_READ, connection);