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();