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 fd06b42aed..b8ebbb995d 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 @@ -78,9 +78,9 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport protected final Lock lifecycleMonitor = new ReentrantLock(); // NOSONAR final - private final Map connections = new ConcurrentHashMap<>(); + protected final Map connections = new ConcurrentHashMap<>(); // NOSONAR final - private final Lock connectionsMonitor = new ReentrantLock(); + protected final Lock connectionsMonitor = new ReentrantLock(); // NOSONAR final private final BlockingQueue delayedReads = new LinkedBlockingQueue<>(); @@ -977,7 +977,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport this.connectionsMonitor.lock(); try { boolean closed = false; - TcpConnectionSupport connection = removeConnection(connectionId); + TcpConnectionSupport connection = this.connections.remove(connectionId); if (connection != null) { try { connection.close(); @@ -996,11 +996,6 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport } } - @Nullable - protected TcpConnectionSupport removeConnection(String connectionId) { - return this.connections.remove(connectionId); - } - @Override public String toString() { return super.toString() diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index c5f9f65dfa..1e5f8594ac 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -33,6 +33,7 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ErrorMessage; +import org.springframework.util.Assert; /** * Connection factory that caches connections from the underlying target factory. The underlying @@ -359,8 +360,31 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact } @Override - protected TcpConnectionSupport removeConnection(String connectionId) { - return this.targetConnectionFactory.removeConnection(connectionId.replaceFirst("Cached:", "")); + public boolean closeConnection(String connectionId) { + Assert.notNull(connectionId, "'connectionId' to close must not be null"); + String targetConnectionId = connectionId.replaceFirst("Cached:", ""); + this.connectionsMonitor.lock(); + try { + TcpConnectionSupport targetConnection = this.targetConnectionFactory.connections.get(targetConnectionId); + if (targetConnection != null) { + /* + * If the delegate is stopped, actually close the connection, but still release + * it to the pool, it will be discarded/renewed the next time it is retrieved. + */ + if (!isRunning()) { + logger.debug(() -> "Factory not running - closing " + connectionId); + super.closeConnection(targetConnectionId); + } + CachingClientConnectionFactory.this.pool.releaseItem(targetConnection); + return true; + } + else { + return false; + } + } + finally { + this.connectionsMonitor.unlock(); + } } @Override diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java index db18e206d5..a243d6b468 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java @@ -35,6 +35,7 @@ import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.EventListener; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; @@ -51,6 +52,7 @@ import org.springframework.integration.ip.tcp.TcpSendingMessageHandler; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnectionOpenEvent; import org.springframework.integration.ip.tcp.connection.TcpConnectionServerListeningEvent; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; @@ -261,6 +263,7 @@ public class IpIntegrationTests { IntStream.rangeClosed('a', 'z') .mapToObj((characterCode) -> (char) characterCode) .map((character) -> "" + character) + .parallel() .peek((character) -> this.outboundFlowInput.send(new GenericMessage<>(character))) .map(String::toUpperCase) .toList(); @@ -274,6 +277,8 @@ public class IpIntegrationTests { } assertThat(replies).containsAll(expected); + + assertThat(config.openEvents).hasSizeLessThanOrEqualTo(5); } @Configuration @@ -431,9 +436,18 @@ public class IpIntegrationTests { return Tcp.netClient("localhost", 0); } + final List openEvents = new ArrayList<>(); + + @EventListener + void connectionOpened(TcpConnectionOpenEvent tcpConnectionOpenEvent) { + if ("client3".equals(tcpConnectionOpenEvent.getConnectionFactoryName())) { + this.openEvents.add(tcpConnectionOpenEvent); + } + } + @Bean CachingClientConnectionFactory cachingClient(TcpNetClientConnectionFactory client3) { - var cachingClientConnectionFactory = new CachingClientConnectionFactory(client3, 10); + var cachingClientConnectionFactory = new CachingClientConnectionFactory(client3, 5); cachingClientConnectionFactory.setConnectionWaitTimeout(10_000); return cachingClientConnectionFactory; } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index b339edafe9..ca0e3a2550 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -389,12 +389,7 @@ public class CachingClientConnectionFactoryTests { private TcpConnectionSupport mockedTcpNioConnection() throws Exception { SocketChannel socketChannel = mock(SocketChannel.class); - if (System.getProperty("java.version").startsWith("1.8")) { - new DirectFieldAccessor(socketChannel).setPropertyValue("open", false); - } - else { - new DirectFieldAccessor(socketChannel).setPropertyValue("closed", true); - } + new DirectFieldAccessor(socketChannel).setPropertyValue("closed", true); doThrow(new IOException("Foo")).when(socketChannel).write(Mockito.any(ByteBuffer.class)); when(socketChannel.socket()).thenReturn(mock(Socket.class)); TcpNioConnection conn = new TcpNioConnection(socketChannel, false, false, event -> {