diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index 5bd097c111..2e75fc5b32 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -158,7 +158,6 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection return connection; } } - return doObtain(singleUse); } catch (RuntimeException e) { @@ -190,7 +189,6 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection if (!singleUse) { setTheConnection(connection); } - connection.publishConnectionOpenEvent(); return connection; } 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 c51dfeb53c..c642f13ad3 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 @@ -616,6 +616,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport connection.registerSender(wrapper); } connection.setWrapped(true); + connection.setWrapper(wrapper); connection = wrapper; } return connection; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index c338db8092..f0fe5ba91a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java @@ -98,6 +98,8 @@ public abstract class TcpConnectionSupport implements TcpConnection { private boolean wrapped; + private TcpConnectionSupport wrapper; + /* * This boolean is to avoid looking for a temporary listener when not needed * to avoid a CPU cache flush. This does not have to be volatile because it @@ -415,11 +417,21 @@ public abstract class TcpConnectionSupport implements TcpConnection { /** * Set to true if intercepted. * @param wrapped true if wrapped. + * @since 5.4.5 */ public void setWrapped(boolean wrapped) { this.wrapped = wrapped; } + /** + * Set the wrapper. + * @param wrapper the wrapper. + * @since 5.4.6 + */ + public void setWrapper(TcpConnectionSupport wrapper) { + this.wrapper = wrapper; + } + public String getConnectionFactoryName() { return this.connectionFactoryName; } @@ -443,15 +455,30 @@ public abstract class TcpConnectionSupport implements TcpConnection { } protected void publishConnectionOpenEvent() { - doPublish(new TcpConnectionOpenEvent(this, getConnectionFactoryName())); + if (this.wrapper != null) { + this.wrapper.publishConnectionOpenEvent(); + } + else { + doPublish(new TcpConnectionOpenEvent(this, getConnectionFactoryName())); + } } protected void publishConnectionCloseEvent() { - doPublish(new TcpConnectionCloseEvent(this, getConnectionFactoryName())); + if (this.wrapper != null) { + this.wrapper.publishConnectionCloseEvent(); + } + else { + doPublish(new TcpConnectionCloseEvent(this, getConnectionFactoryName())); + } } protected void publishConnectionExceptionEvent(Throwable t) { - doPublish(new TcpConnectionExceptionEvent(this, getConnectionFactoryName(), t)); + if (this.wrapper != null) { + this.wrapper.publishConnectionExceptionEvent(t); + } + else { + doPublish(new TcpConnectionExceptionEvent(this, getConnectionFactoryName(), t)); + } } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index dffca3229b..6d7c01853c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2020 the original author or authors. + * Copyright 2001-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -189,9 +189,10 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling */ @Override public void run() { - if (logger.isDebugEnabled()) { - logger.debug(getConnectionId() + " Reading..."); + if (this.logger.isDebugEnabled()) { + this.logger.debug(getConnectionId() + " Reading..."); } + publishConnectionOpenEvent(); while (true) { if (!receiveAndProcessMessage()) { break; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index 3590ab2d53..b16d61075e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java @@ -112,59 +112,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto try { setupServerSocket(); while (true) { - final Socket socket; - /* - * User hooks in the TcpSocketSupport may have set the server socket SO_TIMEOUT. - * Not fatal. - */ - try { - if (this.serverSocket == null) { - logger.debug(() -> this + " stopped before accept"); - throw new IOException(this + " stopped before accept"); - } - else { - socket = this.serverSocket.accept(); - } - } - catch (@SuppressWarnings("unused") SocketTimeoutException ste) { - logger.debug("Timed out on accept; continuing"); - continue; - } - if (isShuttingDown()) { - logger.info(() -> "New connection from " + socket.getInetAddress().getHostAddress() - + ":" + socket.getPort() - + " rejected; the server is in the process of shutting down."); - socket.close(); - } - else { - logger.debug(() -> "Accepted connection from " + socket.getInetAddress().getHostAddress() - + ":" + socket.getPort()); - try { - setSocketAttributes(socket); - TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, true, - isLookupHost(), getApplicationEventPublisher(), getComponentName()); - TcpConnectionSupport wrapped = wrapConnection(connection); - if (!wrapped.equals(connection)) { - connection.setSenders(getSenders()); - connection = wrapped; - } - initializeConnection(connection, socket); - getTaskExecutor().execute(connection); - harvestClosedConnections(); - connection.publishConnectionOpenEvent(); - } - catch (RuntimeException ex) { - this.logger.error(ex, () -> - "Failed to create and configure a TcpConnection for the new socket: " - + socket.getInetAddress().getHostAddress() + ":" + socket.getPort()); - try { - socket.close(); - } - catch (@SuppressWarnings("unused") IOException e1) { // NOSONAR - exception as flow control - // empty - } - } - } + acceptConnectionAndExecute(); } } catch (IOException ex) { // NOSONAR flow control via exceptions @@ -200,6 +148,57 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto publishServerListeningEvent(getPort()); } + private void acceptConnectionAndExecute() throws IOException { + final Socket socket; + /* + * User hooks in the TcpSocketSupport may have set the server socket SO_TIMEOUT. + * Not fatal. + */ + try { + if (this.serverSocket == null) { + logger.debug(() -> this + " stopped before accept"); + throw new IOException(this + " stopped before accept"); + } + else { + socket = this.serverSocket.accept(); + } + } + catch (@SuppressWarnings("unused") SocketTimeoutException ste) { + logger.debug("Timed out on accept; continuing"); + return; + } + if (isShuttingDown()) { + logger.info(() -> "New connection from " + socket.getInetAddress().getHostAddress() + + ":" + socket.getPort() + + " rejected; the server is in the process of shutting down."); + socket.close(); + } + else { + logger.debug(() -> "Accepted connection from " + socket.getInetAddress().getHostAddress() + + ":" + socket.getPort()); + try { + setSocketAttributes(socket); + TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, true, + isLookupHost(), getApplicationEventPublisher(), getComponentName()); + connection = wrapConnection(connection); + initializeConnection(connection, socket); + getTaskExecutor().execute(connection); + harvestClosedConnections(); + } + catch (RuntimeException ex) { + this.logger.error(ex, () -> + "Failed to create and configure a TcpConnection for the new socket: " + + socket.getInetAddress().getHostAddress() + ":" + socket.getPort()); + try { + socket.close(); + } + catch (@SuppressWarnings("unused") IOException e1) { // NOSONAR - exception as flow control + // empty + } + } + } + } + /** * Create a new {@link ServerSocket}. This default implementation uses the default * {@link ServerSocketFactory}. Override to use some other mechanism 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 f31e8d5aa3..6764db6d34 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 @@ -106,6 +106,7 @@ public class TcpNioClientConnectionFactory extends connection.setLastRead(System.currentTimeMillis()); } this.channelMap.put(socketChannel, connection); + wrappedConnection.publishConnectionOpenEvent(); this.newChannels.add(socketChannel); this.selector.wakeup(); return wrappedConnection; @@ -125,9 +126,9 @@ public class TcpNioClientConnectionFactory extends boolean connected = socketChannel.finishConnect(); long timeLeft = getConnectTimeout().toMillis(); while (!connected && timeLeft > 0) { - Thread.sleep(50); // NOSONAR Magic # + Thread.sleep(5); // NOSONAR Magic # connected = socketChannel.finishConnect(); - timeLeft -= 50; // NOSONAR Magic # + timeLeft -= 5; // NOSONAR Magic # } if (!connected) { throw new IOException("Not connected after connectTimeout"); 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 578ed8ce0c..e73a6fb275 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 @@ -258,7 +258,6 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto } this.channelMap.put(channel, connection); channel.register(selectorForNewSocket, SelectionKey.OP_READ, connection); - connection.publishConnectionOpenEvent(); } } catch (IOException ex) { @@ -281,6 +280,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto connection.setSenders(getSenders()); } initializeConnection(wrappedConnection, socketChannel.socket()); + wrappedConnection.publishConnectionOpenEvent(); return connection; } catch (Exception ex) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java index a462f1a3cc..ca668b07af 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -108,6 +108,7 @@ public class ConnectionToConnectionTests { @SuppressWarnings("unchecked") private void testConnectGuts(AbstractClientConnectionFactory client, AbstractServerConnectionFactory server, String gatewayName, boolean expectExceptionOnClose) throws Exception { + TestingUtilities.waitListening(server, null); client.setPort(server.getPort()); client.start(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml index a4863bc65e..0e74cd199b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml @@ -45,6 +45,28 @@ interceptor-factory-chain="helloWorldInterceptors" /> + + + + @@ -71,6 +93,16 @@ + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests.java index 4484d8ae6a..703e141eef 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,8 +18,10 @@ package org.springframework.integration.ip.tcp; import static org.assertj.core.api.Assertions.assertThat; -import org.junit.Test; -import org.junit.runner.RunWith; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationListener; @@ -30,20 +32,20 @@ import org.springframework.integration.ip.tcp.connection.AbstractClientConnectio import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptor; import org.springframework.integration.ip.tcp.connection.TcpConnectionOpenEvent; +import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gary Russell * @since 2.0 */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@SpringJUnitConfig @DirtiesContext public class InterceptedSharedConnectionTests { @@ -56,6 +58,12 @@ public class InterceptedSharedConnectionTests { @Autowired AbstractClientConnectionFactory client; + @Autowired + AbstractServerConnectionFactory netServer; + + @Autowired + AbstractClientConnectionFactory netClient; + @Autowired Listener listener; @@ -70,7 +78,7 @@ public class InterceptedSharedConnectionTests { * @throws Exception */ @Test - public void test1() throws Exception { + void test1() throws Exception { TestingUtilities.waitListening(this.server, null); this.client.setPort(this.server.getPort()); this.ctx.getBeansOfType(ConsumerEndpointFactoryBean.class).values().forEach(c -> c.start()); @@ -82,18 +90,49 @@ public class InterceptedSharedConnectionTests { assertThat(message).isNotNull(); assertThat(message.getPayload()).isEqualTo("Test"); } - assertThat(this.listener.openEvent).isNotNull(); - assertThat(this.listener.openEvent.getConnectionFactoryName()).isEqualTo("client"); + assertThat(this.listener.clientOpenEvent).isNotNull(); + assertThat(this.listener.clientOpenEvent.getConnectionFactoryName()).isEqualTo("client"); + assertThat(this.listener.serverOpenEvent).isNotNull(); + assertThat(this.listener.serverOpenEvent.getConnectionFactoryName()).isEqualTo("server"); + } + + @Test + void correctOpenNetEventPublished() throws InterruptedException { + TestingUtilities.waitListening(this.netServer, null); + this.listener.clientOpenEvent = null; + this.listener.serverOpenEvent = null; + this.netClient.setPort(this.netServer.getPort()); + this.netClient.start(); + TcpConnectionSupport conn = this.netClient.getConnection(); + conn.send(new GenericMessage<>("foo")); + conn.close(); + assertThat(this.listener.latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.listener.clientOpenEvent).isNotNull(); + assertThat(this.listener.clientOpenEvent.getConnectionFactoryName()).isEqualTo("netClient"); + assertThat(this.listener.serverOpenEvent).isNotNull(); + assertThat(this.listener.serverOpenEvent.getConnectionFactoryName()).isEqualTo("netServer"); } public static class Listener implements ApplicationListener { - private volatile TcpConnectionOpenEvent openEvent; + final CountDownLatch latch = new CountDownLatch(2); + + volatile TcpConnectionOpenEvent clientOpenEvent; + + volatile TcpConnectionOpenEvent serverOpenEvent; @Override public void onApplicationEvent(TcpConnectionOpenEvent event) { if (event.getSource() instanceof HelloWorldInterceptor) { - this.openEvent = event; + if (event.getConnectionFactoryName().startsWith("net")) { + this.latch.countDown(); + } + if (event.getConnectionFactoryName().contains("lient")) { + this.clientOpenEvent = event; + } + else { + this.serverOpenEvent = event; + } } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java index 6e4a4d490b..a873a0d429 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java @@ -29,7 +29,9 @@ import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.IOException; import java.net.InetSocketAddress; +import java.net.Socket; import java.net.SocketAddress; import java.time.Duration; import java.util.ArrayList; @@ -43,6 +45,8 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; +import javax.net.SocketFactory; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.junit.jupiter.api.Test; @@ -78,6 +82,36 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; */ public class ConnectionFactoryTests { + @Test + void netOpenEventOnReadThread() throws InterruptedException, IOException { + TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(0); + AtomicReference readThread = new AtomicReference<>(); + AtomicReference openEventThread = new AtomicReference<>(); + CountDownLatch latch1 = new CountDownLatch(1); + CountDownLatch latch2 = new CountDownLatch(1); + server.registerListener(msg -> { + readThread.set(Thread.currentThread()); + latch2.countDown(); + return false; + }); + server.setApplicationEventPublisher(event -> { + if (event instanceof TcpConnectionServerListeningEvent) { + latch1.countDown(); + } + if (event instanceof TcpConnectionOpenEvent) { + openEventThread.set(Thread.currentThread()); + } + }); + server.afterPropertiesSet(); + server.start(); + assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue(); + Socket socket = SocketFactory.getDefault().createSocket("localhost", server.getPort()); + socket.getOutputStream().write("test\r\n".getBytes()); + socket.close(); + assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(readThread.get()).isSameAs(openEventThread.get()); + } + @Test public void factoryBeanTests() { TcpConnectionFactoryFactoryBean fb = new TcpConnectionFactoryFactoryBean("client");