From a7bda6bbb2bf6c3b3e9b295ee1728764361b9bbb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 14 Oct 2011 12:48:49 -0400 Subject: [PATCH] INT-1770 Support Client Mode For TCP Endpoints Add client mode for inbound endpoints, and for the outbound adapter. INT-1770 Allow Inbound Adapter to Open Connection Normally, inbound adapters use server sockets and wait for incoming connection. There are use cases where the adapter should establish the connection and wait for inbound messages. INT-1770 Open Connection on Start Allow configuration of outbound adapter to permit connection establishment when the adapter is started rather than when the first message arrives. INT-1770 Allow Inbound Gateway to Open Connection Normally, inbound gateways listen for connections. There are use cases where an inbound gateway might open the connection and then wait for incoming requests. INT-1770 Parsers Update parsers and tests to support attributes for setting endpoints in client-mode. INT-1770 Docs Update reference with client-mode information. INT-1770 Add Control Bus for Client Mode Enable control bus commands to check status and to attempt connection establishment. --- docs/src/reference/docbook/ip.xml | 165 +++++++++++++++++- spring-integration-ip/.springBeans | 1 + .../ip/config/IpAdapterParserUtils.java | 14 +- .../TcpInboundChannelAdapterParser.java | 6 + .../ip/config/TcpInboundGatewayParser.java | 8 +- .../TcpOutboundChannelAdapterParser.java | 6 + .../integration/ip/tcp/TcpInboundGateway.java | 165 ++++++++++++++++-- .../ip/tcp/TcpReceivingChannelAdapter.java | 149 ++++++++++++++-- .../ip/tcp/TcpSendingMessageHandler.java | 160 +++++++++++++++-- .../AbstractClientConnectionFactory.java | 24 +++ .../ip/tcp/connection/ClientModeCapable.java | 58 ++++++ .../ClientModeConnectionManager.java | 73 ++++++++ .../TcpNetClientConnectionFactory.java | 14 +- .../TcpNioClientConnectionFactory.java | 20 +-- .../ip/config/spring-integration-ip-2.1.xsd | 58 ++++++ .../ip/config/ParserUnitTests-context.xml | 50 ++++++ .../ip/config/ParserUnitTests.java | 66 ++++++- .../tcp/ClientModeControlBusTests-context.xml | 51 ++++++ .../ip/tcp/ClientModeControlBusTests.java | 62 +++++++ .../ip/tcp/TcpInboundGatewayTests.java | 59 +++++++ .../tcp/TcpReceivingChannelAdapterTests.java | 88 ++++++++-- .../ip/tcp/TcpSendingMessageHandlerTests.java | 91 ++++++++-- 22 files changed, 1279 insertions(+), 109 deletions(-) create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeCapable.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests-context.xml create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java diff --git a/docs/src/reference/docbook/ip.xml b/docs/src/reference/docbook/ip.xml index 369bbcd980..6891402580 100644 --- a/docs/src/reference/docbook/ip.xml +++ b/docs/src/reference/docbook/ip.xml @@ -331,7 +331,7 @@
- Tcp Connection Interceptors + TCP Connection Interceptors Connection factories can be configured with a reference to a TcpConnectionInterceptorFactoryChain. Interceptors can be used @@ -403,7 +403,7 @@ TCP Adapters TCP inbound and outbound channel adapters that utilize the above connection - factories are provided. These adapters have just 2 attributes + factories are provided. These adapters have attributes connection-factory and channel. The channel attribute specifies the channel on which messages arrive at an outbound adapter and on which messages are placed by an inbound adapter. @@ -471,6 +471,48 @@ 'inboundClient' and deposited in channel 'replies'. Java serialization is used on the wire. + + Normally, inbound adapters use a type="server" connection + factory, which listens for incoming connection requests. + In some cases, it is desireable to establish the connection + in reverse, whereby the inbound adapter connects to an + external server and then waits for inbound messages on that + connection. + + + This topology is supported by using client-mode="true" + on the inbound adapter. In this case, the connection factory must + be of type 'client' and must have single-use + set to false. + + + Two additional attributes are used to support this mechanism: + retry-interval specifies (in milliseconds) + how often the framework will attempt to reconnect after a + connection failure. scheduler is used to + supply a TaskScheduler used to + schedule the connection attempts, and to test that the connection is + still active. + + + For an outbound adapter, the connection is normally established + when the first message is sent. client-mode="true" + on an outbound adapter will cause the connection to be established + when the adapter is started. Adapters are automatically started + by default. Again, the connection factory must be of type client + and have single-use set to false and + retry-interval and + scheduler are also supported. If a connection + fails, it will be re-established either by the scheduler or + when the next message is sent. + + + For both inbound and outbound, + if the adapter is started, you may force the adapter to establish + a connection by sending a <control-bus /> command: + @adapter_id.retryConnection() and examine the + current state with @adapter_id.isConnected(). +
TCP Gateways @@ -497,6 +539,36 @@ + + As with inbound adapters, inbound gateways normally use a + type="server" connection + factory, which listens for incoming connection requests. + In some cases, it is desireable to establish the connection + in reverse, whereby the inbound gateway connects to an + external server and then waits for, and replies to, inbound + messages on that connection. + + + This topology is supported by using client-mode="true" + on the inbound gateway. In this case, the connection factory must + be of type 'client' and must have single-use + set to false. + + + Two additional attributes are used to support this mechanism: + retry-interval specifies (in milliseconds) + how often the framework will attempt to reconnect after a + connection failure. scheduler is used to + supply a TaskScheduler used to + schedule the connection attempts, and to test that the connection is + still active. + + + If the gateway is started, you may force the gateway to establish + a connection by sending a <control-bus /> command: + @adapter_id.retryConnection() and examine the + current state with @adapter_id.isConnected(). + The outbound gateway, after sending a message over the connection, waits for a response and constructs a response message and puts in on the reply channel. @@ -1164,6 +1236,35 @@ component, the MessagingException message containing the exception and failed message is sent to this channel. + + client-mode + true, false + + When true, the inbound adapter will act as a client, with respect to establishing + the connection and then receive incoming messages on that connection. Default = false. + Also see retry-interval and scheduler. + The connection factory must be of type 'client' and have single-use + set to false. + + + + retry-interval + + + When in client-mode, specifies the number of milliseconds + to wait between connection attempts, or after a connection failure. Default 60,000 + (60 seconds). + + + + scheduler + true, false + + Specifies a TaskScheduler to use for managing the + client-mode connection. Defaults to a + ThreadPoolTaskScheduler with a pool size of `. + + @@ -1196,6 +1297,36 @@ to correlate messages to the connection on which an original inbound message was received. + + client-mode + true, false + + When true, the outbound adapter will attempt to establish the connection as + soon as it is started. When false, the connection is established when + the first message is sent. Default = false. + Also see retry-interval and scheduler. + The connection factory must be of type 'client' and have single-use + set to false. + + + + retry-interval + + + When in client-mode, specifies the number of milliseconds + to wait between connection attempts, or after a connection failure. Default 60,000 + (60 seconds). + + + + scheduler + true, false + + Specifies a TaskScheduler to use for managing the + client-mode connection. Defaults to a + ThreadPoolTaskScheduler with a pool size of `. + + @@ -1246,6 +1377,36 @@ any reply from that flow will then be returned as a response by the gateway. + + client-mode + true, false + + When true, the inbound gateway will act as a client, with respect to establishing + the connection and then receive (and reply to) incoming messages on that connection. + Default = false. + Also see retry-interval and scheduler. + The connection factory must be of type 'client' and have single-use + set to false. + + + + retry-interval + + + When in client-mode, specifies the number of milliseconds + to wait between connection attempts, or after a connection failure. Default 60,000 + (60 seconds). + + + + scheduler + true, false + + Specifies a TaskScheduler to use for managing the + client-mode connection. Defaults to a + ThreadPoolTaskScheduler with a pool size of `. + + diff --git a/spring-integration-ip/.springBeans b/spring-integration-ip/.springBeans index 253efca1a7..2dfc932cdc 100644 --- a/spring-integration-ip/.springBeans +++ b/spring-integration-ip/.springBeans @@ -9,6 +9,7 @@ src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml src/test/java/org/springframework/integration/ip/tcp/AutoStartTests-context.xml + src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests-context.xml diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index a5cee11ea0..9742fe065f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -98,21 +98,27 @@ public abstract class IpAdapterParserUtils { static final String TCP_CONNECTION_FACTORY = "connection-factory"; public static final String INTERCEPTOR_FACTORY_CHAIN = "interceptor-factory-chain"; - + public static final String REQUEST_TIMEOUT = "request-timeout"; - + public static final String REPLY_TIMEOUT = "reply-timeout"; - + public static final String REPLY_CHANNEL = "reply-channel"; public static final String LOOKUP_HOST = "lookup-host"; - + public static final String AUTO_STARTUP = "auto-startup"; public static final String PHASE = "phase"; public static final String APPLY_SEQUENCE = "apply-sequence"; + public static final String CLIENT_MODE = "client-mode"; + + public static final String RETRY_INTERVAL = "retry-interval"; + + public static final String SCHEDULER = "scheduler"; + /** * Adds a constructor-arg to the provided bean definition builder * with the value of the attribute whose name is provided if that diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java index fbf0c35081..a70b4d2770 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java @@ -46,6 +46,12 @@ public class TcpInboundChannelAdapterParser extends AbstractChannelAdapterParser IpAdapterParserUtils.AUTO_STARTUP); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.PHASE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.CLIENT_MODE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.RETRY_INTERVAL); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.SCHEDULER); return builder.getBeanDefinition(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundGatewayParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundGatewayParser.java index e72e1ecaf1..4633324394 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundGatewayParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundGatewayParser.java @@ -37,12 +37,16 @@ public class TcpInboundGatewayParser extends AbstractInboundGatewayParser { @Override protected boolean isEligibleAttribute(String attributeName) { return !attributeName.equals(IpAdapterParserUtils.TCP_CONNECTION_FACTORY) + && !attributeName.equals(IpAdapterParserUtils.SCHEDULER) && super.isEligibleAttribute(attributeName); } @Override protected void doPostProcess(BeanDefinitionBuilder builder, Element element) { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, - IpAdapterParserUtils.TCP_CONNECTION_FACTORY); } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.TCP_CONNECTION_FACTORY); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.SCHEDULER); + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java index 590c7f2f55..79cc16475d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java @@ -40,6 +40,12 @@ public class TcpOutboundChannelAdapterParser extends AbstractOutboundChannelAdap IpAdapterParserUtils.AUTO_STARTUP); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.PHASE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.CLIENT_MODE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.RETRY_INTERVAL); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.SCHEDULER); return builder.getBeanDefinition(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java index 8cf4d50b99..a32b511e11 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java @@ -17,14 +17,22 @@ package org.springframework.integration.ip.tcp; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; import org.springframework.integration.Message; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.ClientModeCapable; +import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager; import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpListener; import org.springframework.integration.ip.tcp.connection.TcpSender; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.Assert; /** * Inbound Gateway using a server connection factory - threading is controlled by the @@ -38,12 +46,27 @@ import org.springframework.integration.ip.tcp.connection.TcpSender; * @since 2.0 * */ -public class TcpInboundGateway extends MessagingGatewaySupport implements TcpListener, TcpSender { +public class TcpInboundGateway extends MessagingGatewaySupport implements + TcpListener, TcpSender, ClientModeCapable { + + private volatile AbstractServerConnectionFactory serverConnectionFactory; + + private volatile AbstractClientConnectionFactory clientConnectionFactory; - private AbstractServerConnectionFactory connectionFactory; - private Map connections = new ConcurrentHashMap(); + private volatile boolean isClientMode; + + private volatile TaskScheduler scheduler; + + private volatile long retryInterval = 60000; + + private volatile ScheduledFuture scheduledFuture; + + private volatile ClientModeConnectionManager clientModeConnectionManager; + + private volatile boolean active; + public boolean onMessage(Message message) { Message reply = this.sendAndReceiveMessage(message); if (reply == null) { @@ -73,16 +96,25 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements TcpLis * @return true if the associated connection factory is listening. */ public boolean isListening() { - return connectionFactory.isListening(); - + return this.serverConnectionFactory == null ? false + : this.serverConnectionFactory.isListening(); } /** - * + * Must be {@link AbstractClientConnectionFactory} or {@link AbstractServerConnectionFactory}. + * * @param connectionFactory the Connection Factory */ - public void setConnectionFactory(AbstractServerConnectionFactory connectionFactory) { - this.connectionFactory = connectionFactory; + public void setConnectionFactory(AbstractConnectionFactory connectionFactory) { + Assert.notNull(connectionFactory, "Connection factory must not be null"); + if (connectionFactory instanceof AbstractServerConnectionFactory) { + this.serverConnectionFactory = (AbstractServerConnectionFactory) connectionFactory; + } else if (connectionFactory instanceof AbstractClientConnectionFactory) { + this.clientConnectionFactory = (AbstractClientConnectionFactory) connectionFactory; + } else { + throw new IllegalArgumentException("Connection factory must be either an " + + "AbstractServerConnectionFactory or an AbstractClientConnectionFactory"); + } connectionFactory.registerListener(this); connectionFactory.registerSender(this); } @@ -98,10 +130,117 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements TcpLis return "ip:tcp-inbound-gateway"; } - /** - * @return the connectionFactory - */ - protected AbstractServerConnectionFactory getConnectionFactory() { - return connectionFactory; + @Override + protected void onInit() throws Exception { + super.onInit(); + if (this.isClientMode) { + Assert.notNull(this.clientConnectionFactory, + "For client-mode, connection factory must be type='client'"); + Assert.isTrue(!this.clientConnectionFactory.isSingleUse(), + "For client-mode, connection factory must have single-use='false'"); + } } + + @Override // protected by super#lifecycleLock + protected void doStart() { + super.doStart(); + if (!this.active) { + this.active = true; + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.start(); + } + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.start(); + } + if (this.isClientMode) { + ClientModeConnectionManager manager = new ClientModeConnectionManager( + this.clientConnectionFactory); + this.clientModeConnectionManager = manager; + this.scheduledFuture = this.getScheduler().scheduleAtFixedRate(manager, this.retryInterval); + } + } + } + + @Override // protected by super#lifecycleLock + protected void doStop() { + super.doStop(); + if (this.active) { + this.active = false; + if (this.scheduledFuture != null) { + this.scheduledFuture.cancel(true); + } + this.clientModeConnectionManager = null; + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.stop(); + } + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.stop(); + } + } + } + + /** + * @return the isClientMode + */ + public boolean isClientMode() { + return isClientMode; + } + + /** + * @param isClientMode + * the isClientMode to set + */ + public void setClientMode(boolean isClientMode) { + this.isClientMode = isClientMode; + } + + /** + * @return the scheduler + */ + protected TaskScheduler getScheduler() { + if (this.scheduler == null) { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + this.scheduler = scheduler; + } + return scheduler; + } + + /** + * @param scheduler + * the scheduler to set + */ + public void setScheduler(TaskScheduler scheduler) { + this.scheduler = scheduler; + } + + /** + * @return the retryInterval + */ + public long getRetryInterval() { + return retryInterval; + } + + /** + * @param retryInterval + * the retryInterval to set + */ + public void setRetryInterval(long retryInterval) { + this.retryInterval = retryInterval; + } + + public boolean isClientModeConnected() { + if (this.isClientMode && this.clientModeConnectionManager != null) { + return this.clientModeConnectionManager.isConnected(); + } else { + return false; + } + } + + public void retryConnection() { + if (this.active && this.isClientMode && this.clientModeConnectionManager != null) { + this.clientModeConnectionManager.run(); + } + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java index 9cb093a452..736d546ef2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -15,13 +15,20 @@ */ package org.springframework.integration.ip.tcp; +import java.util.concurrent.ScheduledFuture; + import org.springframework.integration.Message; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.ClientModeCapable; +import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager; import org.springframework.integration.ip.tcp.connection.ConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpListener; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.Assert; /** * Tcp inbound channel adapter using a TcpConnection to @@ -34,34 +41,75 @@ import org.springframework.integration.ip.tcp.connection.TcpListener; * */ public class TcpReceivingChannelAdapter - extends MessageProducerSupport implements TcpListener { + extends MessageProducerSupport implements TcpListener, ClientModeCapable { + + private AbstractConnectionFactory clientConnectionFactory; + + private AbstractConnectionFactory serverConnectionFactory; + + private volatile boolean isClientMode; + + private volatile TaskScheduler scheduler; + + private volatile long retryInterval = 60000; + + private volatile ScheduledFuture scheduledFuture; + + private volatile ClientModeConnectionManager clientModeConnectionManager; + + private volatile boolean active; - private ConnectionFactory clientConnectionFactory; - - private ConnectionFactory serverConnectionFactory; - public boolean onMessage(Message message) { sendMessage(message); return false; } - + @Override - protected void doStart() { - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.start(); - } - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.start(); + protected void onInit() { + super.onInit(); + if (this.isClientMode) { + Assert.notNull(this.clientConnectionFactory, + "For client-mode, connection factory must be type='client'"); + Assert.isTrue(!this.clientConnectionFactory.isSingleUse(), + "For client-mode, connection factory must have single-use='false'"); } } - @Override - protected void doStop() { - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.stop(); + @Override // protected by super#lifecycleLock + protected void doStart() { + super.doStart(); + if (!this.active) { + this.active = true; + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.start(); + } + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.start(); + } + if (this.isClientMode) { + ClientModeConnectionManager manager = new ClientModeConnectionManager( + this.clientConnectionFactory); + this.clientModeConnectionManager = manager; + this.scheduledFuture = this.getScheduler().scheduleAtFixedRate(manager, this.retryInterval); + } } - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.stop(); + } + + @Override // protected by super#lifecycleLock + protected void doStop() { + super.doStop(); + if (this.active) { + this.active = false; + if (this.scheduledFuture != null) { + this.scheduledFuture.cancel(true); + } + this.clientModeConnectionManager = null; + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.stop(); + } + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.stop(); + } } } @@ -108,4 +156,69 @@ public class TcpReceivingChannelAdapter protected ConnectionFactory getServerConnectionFactory() { return serverConnectionFactory; } + + /** + * @return the isClientMode + */ + public boolean isClientMode() { + return this.isClientMode; + } + + /** + * @param isClientMode + * the isClientMode to set + */ + public void setClientMode(boolean isClientMode) { + this.isClientMode = isClientMode; + } + + /** + * @return the scheduler + */ + protected TaskScheduler getScheduler() { + if (this.scheduler == null) { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + this.scheduler = scheduler; + } + return this.scheduler; + } + + /** + * @param scheduler + * the scheduler to set + */ + public void setScheduler(TaskScheduler scheduler) { + this.scheduler = scheduler; + } + + /** + * @return the retryInterval + */ + public long getRetryInterval() { + return this.retryInterval; + } + + /** + * @param retryInterval + * the retryInterval to set + */ + public void setRetryInterval(long retryInterval) { + this.retryInterval = retryInterval; + } + + public boolean isClientModeConnected() { + if (this.isClientMode && this.clientModeConnectionManager != null) { + return this.clientModeConnectionManager.isConnected(); + } else { + return false; + } + } + + public void retryConnection() { + if (this.active && this.isClientMode && this.clientModeConnectionManager != null) { + this.clientModeConnectionManager.run(); + } + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index 1370e92ed8..bb5119e2cb 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -30,10 +31,15 @@ import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.integration.ip.tcp.connection.ClientModeCapable; +import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager; import org.springframework.integration.ip.tcp.connection.ConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpSender; import org.springframework.integration.mapping.MessageMappingException; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.Assert; /** * Tcp outbound channel adapter using a TcpConnection to @@ -44,13 +50,14 @@ import org.springframework.integration.mapping.MessageMappingException; * @since 2.0 * */ -public class TcpSendingMessageHandler extends AbstractMessageHandler implements TcpSender, SmartLifecycle { +public class TcpSendingMessageHandler extends AbstractMessageHandler implements + TcpSender, SmartLifecycle, ClientModeCapable { protected final Log logger = LogFactory.getLog(this.getClass()); - private volatile ConnectionFactory clientConnectionFactory; + private volatile AbstractConnectionFactory clientConnectionFactory; - private volatile ConnectionFactory serverConnectionFactory; + private volatile AbstractConnectionFactory serverConnectionFactory; private Map connections = new ConcurrentHashMap(); @@ -58,6 +65,20 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements private volatile int phase; + private volatile boolean isClientMode; + + private volatile TaskScheduler scheduler; + + private volatile long retryInterval = 60000; + + private volatile ScheduledFuture scheduledFuture; + + private volatile ClientModeConnectionManager clientModeConnectionManager; + + protected final Object lifecycleMonitor = new Object(); + + private volatile boolean active; + protected TcpConnection getConnection() { TcpConnection connection = null; if (this.clientConnectionFactory == null) { @@ -167,21 +188,51 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements return "ip:tcp-outbound-channel-adapter"; } - public void start() { - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.start(); + @Override + protected void onInit() throws Exception { + super.onInit(); + if (this.isClientMode) { + Assert.notNull(this.clientConnectionFactory, + "For client-mode, connection factory must be type='client'"); + Assert.isTrue(!this.clientConnectionFactory.isSingleUse(), + "For client-mode, connection factory must have single-use='false'"); } - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.start(); + } + + public void start() { + synchronized (this.lifecycleMonitor) { + if (!this.active) { + this.active = true; + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.start(); + } + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.start(); + } + if (this.isClientMode) { + ClientModeConnectionManager manager = new ClientModeConnectionManager( + this.clientConnectionFactory); + this.clientModeConnectionManager = manager; + this.scheduledFuture = this.getScheduler().scheduleAtFixedRate(manager, this.retryInterval); + } + } } } public void stop() { - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.stop(); - } - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.stop(); + synchronized (this.lifecycleMonitor) { + if (this.active) { + this.active = false; + if (this.scheduledFuture != null) { + this.scheduledFuture.cancel(true); + } + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.stop(); + } + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.stop(); + } + } } } @@ -200,11 +251,20 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements } public void stop(Runnable callback) { - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.stop(callback); - } - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.stop(callback); + synchronized (this.lifecycleMonitor) { + if (this.active) { + this.active = false; + if (this.scheduledFuture != null) { + this.scheduledFuture.cancel(true); + } + this.clientModeConnectionManager = null; + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.stop(callback); + } + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.stop(callback); + } + } } } @@ -237,4 +297,68 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements return connections; } + /** + * @return the isClientMode + */ + public boolean isClientMode() { + return this.isClientMode; + } + + /** + * @param isClientMode + * the isClientMode to set + */ + public void setClientMode(boolean isClientMode) { + this.isClientMode = isClientMode; + } + + /** + * @return the scheduler + */ + protected TaskScheduler getScheduler() { + if (this.scheduler == null) { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + this.scheduler = scheduler; + } + return this.scheduler; + } + + /** + * @param scheduler + * the scheduler to set + */ + public void setScheduler(TaskScheduler scheduler) { + this.scheduler = scheduler; + } + + /** + * @return the retryInterval + */ + public long getRetryInterval() { + return this.retryInterval; + } + + /** + * @param retryInterval + * the retryInterval to set + */ + public void setRetryInterval(long retryInterval) { + this.retryInterval = retryInterval; + } + + public boolean isClientModeConnected() { + if (this.isClientMode && this.clientModeConnectionManager != null) { + return this.clientModeConnectionManager.isConnected(); + } else { + return false; + } + } + + public void retryConnection() { + if (this.active && this.isClientMode && this.clientModeConnectionManager != null) { + this.clientModeConnectionManager.run(); + } + } + } 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 259d47bb36..c1d3d14ac1 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 @@ -39,6 +39,26 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection super(host, port); } + /** + * Obtains a connection - if {@link #setSingleUse(boolean)} was called with + * true, a new connection is returned; otherwise a single connection is + * reused for all requests while the connection remains open. + */ + public TcpConnection getConnection() throws Exception { + this.checkActive(); + if (this.isSingleUse()) { + return getOrMakeConnection(); + } else { + synchronized(this) { + TcpConnection connection = getOrMakeConnection(); + this.setTheConnection(connection); + return connection; + } + } + } + + protected abstract TcpConnection getOrMakeConnection() throws Exception; + /** * Transfers attributes such as (de)serializers, singleUse etc to a new connection. * When the connection factory has a reference to a TCPListener (to read @@ -62,6 +82,10 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection } } } + TcpSender sender = this.getSender(); + if (sender != null) { + connection.registerSender(sender); + } connection.setMapper(this.getMapper()); connection.setDeserializer(this.getDeserializer()); connection.setSerializer(this.getSerializer()); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeCapable.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeCapable.java new file mode 100644 index 0000000000..3ee6ef0c13 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeCapable.java @@ -0,0 +1,58 @@ +/* + * Copyright 2002-2011 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp.connection; + +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedOperation; + +/** + * Edpoints implementing this interface are capable + * of running in client-mode. For inbound endpoints, + * this means that the endpoint establishes the connection + * and then receives incoming data. + *

+ * For an outbound adapter, it means that the adapter + * will establish the connection rather than waiting + * for a message to cause the connection to be + * established. + * + * @author Gary Russell + * @since 2.1 + * + */ +public interface ClientModeCapable { + + /** + * @return true if the endpoint is running in + * client mode. + */ + @ManagedAttribute + boolean isClientMode(); + + /** + * @return true if the endpoint is running in + * client mode. + */ + @ManagedAttribute + boolean isClientModeConnected(); + + /** + * Immediately attempt to establish the connection. + */ + @ManagedOperation + void retryConnection(); + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java new file mode 100644 index 0000000000..94d574f99f --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java @@ -0,0 +1,73 @@ +/* + * Copyright 2002-2011 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp.connection; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.util.Assert; + +/** + * Intended to be run on a schedule, simply gets the connection + * from a client connection factory each time it is run. + * If no connection exists (or it has been closed), the + * connection factory will create a new one (if possible). + * + * @author Gary Russell + * @since 2.1 + * + */ +public class ClientModeConnectionManager implements Runnable { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final AbstractConnectionFactory clientConnectionFactory; + + private volatile TcpConnection lastConnection; + + /** + * @param clientConnectionFactory + */ + public ClientModeConnectionManager( + AbstractConnectionFactory clientConnectionFactory) { + Assert.notNull(clientConnectionFactory, "Connection factory cannot be null"); + this.clientConnectionFactory = clientConnectionFactory; + } + + public void run() { + synchronized (this.clientConnectionFactory) { + try { + TcpConnection connection = this.clientConnectionFactory.getConnection(); + if (connection != lastConnection) { + if (logger.isDebugEnabled()) { + logger.debug("Connection " + connection.getConnectionId() + " established"); + } + lastConnection = connection; + } else { + if (logger.isTraceEnabled()) { + logger.trace("Connection " + connection.getConnectionId() + " still OK"); + } + } + } catch (Exception e) { + logger.error("Could not establish connection using " + this.clientConnectionFactory, e); + } + } + } + + public boolean isConnected() { + return this.lastConnection == null ? false : this.lastConnection.isOpen(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java index 46e84ee55a..82664d1a03 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java @@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp.connection; import java.io.IOException; import java.net.Socket; +import java.net.SocketException; import javax.net.SocketFactory; @@ -40,12 +41,12 @@ public class TcpNetClientConnectionFactory extends } /** - * Obtains a connection - if {@link #setSingleUse(boolean)} was called with - * true, a new connection is returned; otherwise a single connection is - * reused for all requests while the connection remains open. + * @return + * @throws IOException + * @throws SocketException + * @throws Exception */ - public TcpConnection getConnection() throws Exception { - this.checkActive(); + protected TcpConnection getOrMakeConnection() throws Exception { TcpConnection theConnection = this.getTheConnection(); if (theConnection != null && theConnection.isOpen()) { return theConnection; @@ -59,9 +60,6 @@ public class TcpNetClientConnectionFactory extends connection = wrapConnection(connection); initializeConnection(connection, socket); this.getTaskExecutor().execute(connection); - if (!this.isSingleUse()) { - this.setTheConnection(connection); - } this.harvestClosedConnections(); return connection; } 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 1676e54735..4bb7e0fa6a 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 @@ -16,7 +16,9 @@ package org.springframework.integration.ip.tcp.connection; +import java.io.IOException; import java.net.InetSocketAddress; +import java.net.SocketException; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; @@ -55,12 +57,12 @@ public class TcpNioClientConnectionFactory extends } /** - * Obtains a connection - if {@link #setSingleUse(boolean)} was called with - * true, a new connection is returned; otherwise a single connection is - * reused for all requests while the connection remains open. + * @return + * @throws Exception + * @throws IOException + * @throws SocketException */ - public TcpConnection getConnection() throws Exception { - this.checkActive(); + protected TcpConnection getOrMakeConnection() throws Exception { int n = 0; while (this.selector == null) { try { @@ -72,8 +74,9 @@ public class TcpNioClientConnectionFactory extends throw new Exception("Factory failed to start"); } } - if (this.getTheConnection() != null && this.getTheConnection().isOpen()) { - return this.getTheConnection(); + TcpConnection theConnection = this.getTheConnection(); + if (theConnection != null && theConnection.isOpen()) { + return theConnection; } if (logger.isDebugEnabled()) { logger.debug("Opening new socket channel connection to " + this.getHost() + ":" + this.getPort()); @@ -92,9 +95,6 @@ public class TcpNioClientConnectionFactory extends this.connections.put(socketChannel, connection); newChannels.add(socketChannel); selector.wakeup(); - if (!this.isSingleUse()) { - this.setTheConnection(wrappedConnection); - } return wrappedConnection; } diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd index 104296f0eb..af597cb7b1 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd @@ -156,6 +156,17 @@ task executors such as a WorkManagerTaskExecutor. + + + + If set to true, causes the adapter to act as a client with respect to + establishing the connection, rather than listening for incoming connections. + Requires a type="client" connection factory, with single-use set to false. + Defaults to true. + + + + @@ -198,6 +209,17 @@ task executors such as a WorkManagerTaskExecutor. + + + + If set to true, causes the adapter to establish a connection when started, + rather than when the first message is sent. + Requires a type="client" connection factory, with single-use set to false. + Defaults to true. + + + + @@ -256,6 +278,17 @@ task executors such as a WorkManagerTaskExecutor. + + + + If set to true, causes the gateway to act as a client with respect to + establishing the connection, rather than listening for incoming connections. + Requires a type="client" connection factory, with single-use set to false. + Defaults to true. + + + + @@ -523,4 +556,29 @@ default is 0. Values can be negative. See SmartLifeCycle. + + + + + When in client mode, specifies the retry interval, in milliseconds, if a connection + cannot be established. Defaults to 60000. + + + + + + + When in client mode, + provide a reference to the TaskScheduler instance to + be used for establishing connections. If not provided, the default + will use a thread pool of size 1. + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 0d1b9de721..a8d47e1bc5 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -278,4 +278,54 @@ channel="tcpChannel" connection-factory="server1" /> + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index 07d701316b..ac5c9a7673 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -56,6 +56,7 @@ import org.springframework.integration.ip.udp.MulticastSendingMessageHandler; import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter; import org.springframework.integration.ip.udp.UnicastSendingMessageHandler; import org.springframework.integration.test.util.TestUtils; +import org.springframework.scheduling.TaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -128,6 +129,15 @@ public class ParserUnitTests { @Autowired AbstractConnectionFactory cfC2; + @Autowired + AbstractConnectionFactory cfC3; + + @Autowired + AbstractConnectionFactory cfC4; + + @Autowired + AbstractConnectionFactory cfC5; + @Autowired Serializer serializer; @@ -172,6 +182,19 @@ public class ParserUnitTests { @Autowired private DirectChannel tcpChannel; + @Autowired + TcpReceivingChannelAdapter tcpInClientMode; + + @Autowired + TcpInboundGateway inGatewayClientMode; + + @Autowired + TaskScheduler sched; + + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.TcpSendingMessageHandler#3") + TcpSendingMessageHandler tcpOutClientMode; + @Test public void testInUdp() { DirectFieldAccessor dfa = new DirectFieldAccessor(udpIn); @@ -297,7 +320,7 @@ public class ParserUnitTests { @Test public void testInGateway1() { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInboundGateway1); - assertSame(cfS2, dfa.getPropertyValue("connectionFactory")); + assertSame(cfS2, dfa.getPropertyValue("serverConnectionFactory")); assertEquals(456L, dfa.getPropertyValue("replyTimeout")); assertEquals("inGateway1",tcpInboundGateway1.getComponentName()); assertEquals("ip:tcp-inbound-gateway", tcpInboundGateway1.getComponentType()); @@ -312,11 +335,14 @@ public class ParserUnitTests { @Test public void testInGateway2() { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInboundGateway2); - assertSame(cfS3, dfa.getPropertyValue("connectionFactory")); + assertSame(cfS3, dfa.getPropertyValue("serverConnectionFactory")); assertEquals(456L, dfa.getPropertyValue("replyTimeout")); assertEquals("inGateway2",tcpInboundGateway2.getComponentName()); assertEquals("ip:tcp-inbound-gateway", tcpInboundGateway2.getComponentType()); assertNull(dfa.getPropertyValue("errorChannel")); + assertEquals(Boolean.FALSE, dfa.getPropertyValue("isClientMode")); + assertNull(dfa.getPropertyValue("scheduler")); + assertEquals(60000L, dfa.getPropertyValue("retryInterval")); } @Test @@ -418,6 +444,9 @@ public class ParserUnitTests { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewOut1); assertSame(client1, dfa.getPropertyValue("clientConnectionFactory")); assertEquals(25, dfa.getPropertyValue("order")); + assertEquals(Boolean.FALSE, dfa.getPropertyValue("isClientMode")); + assertNull(dfa.getPropertyValue("scheduler")); + assertEquals(60000L, dfa.getPropertyValue("retryInterval")); } @Test @@ -432,6 +461,9 @@ public class ParserUnitTests { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewIn1); assertSame(client1, dfa.getPropertyValue("clientConnectionFactory")); assertNull(dfa.getPropertyValue("errorChannel")); + assertEquals(Boolean.FALSE, dfa.getPropertyValue("isClientMode")); + assertNull(dfa.getPropertyValue("scheduler")); + assertEquals(60000L, dfa.getPropertyValue("retryInterval")); } @Test @@ -456,4 +488,34 @@ public class ParserUnitTests { assertSame(this.tcpOut, iterator.next()); //35 } + @Test + public void testInClientMode() { + DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInClientMode); + assertSame(cfC3, dfa.getPropertyValue("clientConnectionFactory")); + assertNull(dfa.getPropertyValue("serverConnectionFactory")); + assertEquals(Boolean.TRUE, dfa.getPropertyValue("isClientMode")); + assertSame(sched, dfa.getPropertyValue("scheduler")); + assertEquals(123L, dfa.getPropertyValue("retryInterval")); + } + + @Test + public void testOutClientMode() { + DirectFieldAccessor dfa = new DirectFieldAccessor(tcpOutClientMode); + assertSame(cfC4, dfa.getPropertyValue("clientConnectionFactory")); + assertNull(dfa.getPropertyValue("serverConnectionFactory")); + assertEquals(Boolean.TRUE, dfa.getPropertyValue("isClientMode")); + assertSame(sched, dfa.getPropertyValue("scheduler")); + assertEquals(124L, dfa.getPropertyValue("retryInterval")); + } + + @Test + public void testInGatewayClientMode() { + DirectFieldAccessor dfa = new DirectFieldAccessor(inGatewayClientMode); + assertSame(cfC5, dfa.getPropertyValue("clientConnectionFactory")); + assertNull(dfa.getPropertyValue("serverConnectionFactory")); + assertEquals(Boolean.TRUE, dfa.getPropertyValue("isClientMode")); + assertSame(sched, dfa.getPropertyValue("scheduler")); + assertEquals(125L, dfa.getPropertyValue("retryInterval")); + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests-context.xml new file mode 100644 index 0000000000..d4ad03c6d0 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests-context.xml @@ -0,0 +1,51 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java new file mode 100644 index 0000000000..25663cef2f --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java @@ -0,0 +1,62 @@ +/* + * Copyright 2002-2011 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp; + +import static org.junit.Assert.*; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 2.1 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class ClientModeControlBusTests { + + @Autowired + ControlBus controlBus; + + @Autowired + TcpReceivingChannelAdapter tcpIn; + + @Test + public void test() throws Exception { + assertTrue(controlBus.boolResult("@tcpIn.isClientMode()")); + int n = 0; + while (!controlBus.boolResult("@tcpIn.isClientModeConnected()")) { + Thread.sleep(100); + n += 100; + if (n > 10000) { + fail("Connection never established"); + } + } + assertTrue(controlBus.boolResult("@tcpIn.isRunning()")); + controlBus.voidResult("@tcpIn.retryConnection()"); + } + + public static interface ControlBus { + + boolean boolResult(String command); + + void voidResult(String command); + } +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java index 76909ae693..dcd2f000ef 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java @@ -22,10 +22,16 @@ import static org.junit.Assert.fail; import java.io.IOException; import java.io.InputStream; +import java.net.ServerSocket; import java.net.Socket; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.net.ServerSocketFactory; import javax.net.SocketFactory; import org.junit.Test; @@ -37,7 +43,9 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.handler.ServiceActivatingHandler; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; import org.springframework.integration.ip.util.SocketTestUtils; @@ -112,6 +120,57 @@ public class TcpInboundGatewayTests { assertEquals("Echo:Test2\r\n", new String(bytes)); } + @Test + public void testNetClientMode() throws Exception { + final int port = SocketTestUtils.findAvailableServerSocket(); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ccf.setSingleUse(false); + TcpInboundGateway gateway = new TcpInboundGateway(); + gateway.setConnectionFactory(ccf); + final QueueChannel channel = new QueueChannel(); + gateway.setRequestChannel(channel); + gateway.setClientMode(true); + gateway.setRetryInterval(10000); + gateway.afterPropertiesSet(); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service()); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 10); + latch1.countDown(); + Socket socket = server.accept(); + socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes()); + byte[] bytes = new byte[12]; + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test1\r\n", new String(bytes)); + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test2\r\n", new String(bytes)); + latch2.await(); + socket.close(); + server.close(); + done.set(true); + latch3.countDown(); + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + gateway.start(); + handler.handleMessage(channel.receive()); + handler.handleMessage(channel.receive()); + latch2.countDown(); + assertTrue(latch3.await(10, TimeUnit.SECONDS)); + assertTrue(done.get()); + } + @Test public void testNioSingle() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index f682ed66c2..93e43ca9ab 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -25,14 +25,19 @@ import java.io.IOException; import java.io.InputStream; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; +import java.net.ServerSocket; import java.net.Socket; import java.util.HashSet; import java.util.LinkedList; import java.util.List; import java.util.Set; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.net.ServerSocketFactory; import javax.net.SocketFactory; import org.junit.Test; @@ -43,10 +48,12 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.handler.ServiceActivatingHandler; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory; import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory; import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain; +import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; @@ -58,7 +65,7 @@ import org.springframework.integration.ip.util.SocketTestUtils; public class TcpReceivingChannelAdapterTests { @Test - public void newTestNet() throws Exception { + public void testNet() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -76,6 +83,7 @@ public class TcpReceivingChannelAdapterTests { } QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); + adapter.afterPropertiesSet(); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); socket.getOutputStream().write("Test1\r\n".getBytes()); socket.getOutputStream().write("Test2\r\n".getBytes()); @@ -88,7 +96,57 @@ public class TcpReceivingChannelAdapterTests { } @Test - public void newTestNio() throws Exception { + public void testNetClientMode() throws Exception { + final int port = SocketTestUtils.findAvailableServerSocket(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 10); + latch1.countDown(); + Socket socket = server.accept(); + socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes()); + latch2.await(); + socket.close(); + server.close(); + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); + ccf.setSerializer(serializer); + ccf.setDeserializer(serializer); + ccf.setSoTimeout(Integer.MAX_VALUE); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + adapter.setClientMode(true); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + adapter.afterPropertiesSet(); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + adapter.setRetryInterval(10000); + adapter.start(); + Message message = channel.receive(10000); + assertNotNull(message); + assertEquals("Test1", new String((byte[]) message.getPayload())); + message = channel.receive(10000); + assertNotNull(message); + assertEquals("Test2", new String((byte[]) message.getPayload())); + adapter.stop(); + adapter.start(); + adapter.stop(); + latch2.countDown(); + } + + @Test + public void testNio() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -123,7 +181,7 @@ public class TcpReceivingChannelAdapterTests { } @Test - public void newTestNetShared() throws Exception { + public void testNetShared() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -161,7 +219,7 @@ public class TcpReceivingChannelAdapterTests { } @Test - public void newTestNioShared() throws Exception { + public void testNioShared() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -199,7 +257,7 @@ public class TcpReceivingChannelAdapterTests { } @Test - public void newTestNetSingleNoOutbound() throws Exception { + public void testNetSingleNoOutbound() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -235,7 +293,7 @@ public class TcpReceivingChannelAdapterTests { } @Test - public void newTestNioSingleNoOutbound() throws Exception { + public void testNioSingleNoOutbound() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -281,7 +339,7 @@ public class TcpReceivingChannelAdapterTests { } @Test - public void newTestNetSingleShared() throws Exception { + public void testNetSingleShared() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -322,7 +380,7 @@ public class TcpReceivingChannelAdapterTests { } @Test - public void newTestNioSingleShared() throws Exception { + public void testNioSingleShared() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -363,7 +421,7 @@ public class TcpReceivingChannelAdapterTests { } @Test - public void newTestNioSingleSharedMany() throws Exception { + public void testNioSingleSharedMany() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -407,42 +465,42 @@ public class TcpReceivingChannelAdapterTests { } @Test - public void newTestNetInterceptors() throws Exception { + public void testNetInterceptors() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); interceptorsGuts(port, scf); } @Test - public void newTestNetSingleNoOutboundInterceptors() throws Exception { + public void testNetSingleNoOutboundInterceptors() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); singleNoOutboundInterceptorsGuts(port, scf); } @Test - public void newTestNetSingleSharedInterceptors() throws Exception { + public void testNetSingleSharedInterceptors() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); singleSharedInterceptorsGuts(port, scf); } @Test - public void newTestNioInterceptors() throws Exception { + public void testNioInterceptors() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); interceptorsGuts(port, scf); } @Test - public void newTestNioSingleNoOutboundInterceptors() throws Exception { + public void testNioSingleNoOutboundInterceptors() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); singleNoOutboundInterceptorsGuts(port, scf); } @Test - public void newTestNioSingleSharedInterceptors() throws Exception { + public void testNioSingleSharedInterceptors() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); singleSharedInterceptorsGuts(port, scf); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index b4b21dbdee..23528c5125 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -76,7 +76,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNetCrLf() throws Exception { + public void testNetCrLf() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -126,7 +126,64 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNioCrLf() throws Exception { + public void testNetCrLfClientMode() throws Exception { + final int port = SocketTestUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + b = ("Reply" + (++i) + "\r\n").getBytes(); + socket.getOutputStream().write(b); + } + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); + ccf.setSerializer(serializer); + ccf.setDeserializer(serializer); + ccf.setSoTimeout(Integer.MAX_VALUE); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.setClientMode(true); + handler.setRetryInterval(10000); + handler.afterPropertiesSet(); + handler.start(); + adapter.start(); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", new String((byte[]) mOut.getPayload())); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", new String((byte[]) mOut.getPayload())); + done.set(true); + handler.stop(); + handler.start(); + handler.stop(); + } + + @Test + public void testNioCrLf() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -178,7 +235,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNetStxEtx() throws Exception { + public void testNetStxEtx() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -227,7 +284,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNioStxEtx() throws Exception { + public void testNioStxEtx() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -279,7 +336,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNetLength() throws Exception { + public void testNetLength() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -331,7 +388,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNioLength() throws Exception { + public void testNioLength() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -386,7 +443,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNetSerial() throws Exception { + public void testNetSerial() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -434,7 +491,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNioSerial() throws Exception { + public void testNioSerial() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -485,7 +542,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNetSingleUseNoInbound() throws Exception { + public void testNetSingleUseNoInbound() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); @@ -526,7 +583,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNioSingleUseNoInbound() throws Exception { + public void testNioSingleUseNoInbound() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); @@ -567,7 +624,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNetSingleUseWithInbound() throws Exception { + public void testNetSingleUseWithInbound() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); @@ -622,7 +679,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNioSingleUseWithInbound() throws Exception { + public void testNioSingleUseWithInbound() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); @@ -677,7 +734,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNioSingleUseWithInboundMany() throws Exception { + public void testNioSingleUseWithInboundMany() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final Semaphore semaphore = new Semaphore(0); @@ -743,7 +800,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNetNegotiate() throws Exception { + public void testNetNegotiate() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -810,7 +867,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNioNegotiate() throws Exception { + public void testNioNegotiate() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -875,7 +932,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNetNegotiateSingleNoListen() throws Exception { + public void testNetNegotiateSingleNoListen() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -932,7 +989,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNioNegotiateSingleNoListen() throws Exception { + public void testNioNegotiateSingleNoListen() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean();