Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #127 from garyrussell/INT-1770

Browse Source 
Support Client Mode For TCP Endpoints
This commit is contained in:
Mark Fisher
2011-10-15 13:52:22 -04:00
parent a8e752e486 a7bda6bbb2
commit 0c15ee1516
22 changed files with 1279 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
docs/src/reference/docbook/ip.xml
View File

@@ -331,7 +331,7 @@
</para>
</section>
<section id="ip-interceptors">
<title>Tcp Connection Interceptors</title>
<title>TCP Connection Interceptors</title>
<para>
Connection factories can be configured with a reference to a
<classname>TcpConnectionInterceptorFactoryChain</classname>. Interceptors can be used
@@ -403,7 +403,7 @@
<title>TCP Adapters</title>
<para>
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
<classname>connection-factory</classname> and <classname>channel</classname>.
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.
</para>
<para>
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.
</para>
<para>
This topology is supported by using <emphasis>client-mode="true"</emphasis>
on the inbound adapter. In this case, the connection factory must
be of type 'client' and must have <emphasis>single-use</emphasis>
set to false.
</para>
<para>
Two additional attributes are used to support this mechanism:
<emphasis>retry-interval</emphasis> specifies (in milliseconds)
how often the framework will attempt to reconnect after a
connection failure. <emphasis>scheduler</emphasis> is used to
supply a <classname>TaskScheduler</classname> used to
schedule the connection attempts, and to test that the connection is
still active.
</para>
<para>
For an outbound adapter, the connection is normally established
when the first message is sent. <emphasis>client-mode="true"</emphasis>
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 <emphasis>single-use</emphasis> set to false and
<emphasis>retry-interval</emphasis> and
<emphasis>scheduler</emphasis> are also supported. If a connection
fails, it will be re-established either by the scheduler or
when the next message is sent.
</para>
<para>
For both inbound and outbound,
if the adapter is started, you may force the adapter to establish
a connection by sending a &lt;control-bus /&gt; command:
<classname>@adapter_id.retryConnection()</classname> and examine the
current state with <classname>@adapter_id.isConnected()</classname>.
</para>
</section>
<section id="tcp-gateways">
<title>TCP Gateways</title>
@@ -497,6 +539,36 @@
</para>
</note>
</para>
<para>
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.
</para>
<para>
This topology is supported by using <emphasis>client-mode="true"</emphasis>
on the inbound gateway. In this case, the connection factory must
be of type 'client' and must have <emphasis>single-use</emphasis>
set to false.
</para>
<para>
Two additional attributes are used to support this mechanism:
<emphasis>retry-interval</emphasis> specifies (in milliseconds)
how often the framework will attempt to reconnect after a
connection failure. <emphasis>scheduler</emphasis> is used to
supply a <classname>TaskScheduler</classname> used to
schedule the connection attempts, and to test that the connection is
still active.
</para>
<para>
If the gateway is started, you may force the gateway to establish
a connection by sending a &lt;control-bus /&gt; command:
<classname>@adapter_id.retryConnection()</classname> and examine the
current state with <classname>@adapter_id.isConnected()</classname>.
</para>
<para>
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.</entry>
</row>
<row>
<entry>client-mode</entry>
<entry>true, false</entry>
<entry>
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 <emphasis>retry-interval</emphasis> and <emphasis>scheduler</emphasis>.
The connection factory must be of type 'client' and have <emphasis>single-use</emphasis>
set to false.
</entry>
</row>
<row>
<entry>retry-interval</entry>
<entry></entry>
<entry>
When in <emphasis>client-mode</emphasis>, specifies the number of milliseconds
to wait between connection attempts, or after a connection failure. Default 60,000
(60 seconds).
</entry>
</row>
<row>
<entry>scheduler</entry>
<entry>true, false</entry>
<entry>
Specifies a <classname>TaskScheduler</classname> to use for managing the
<emphasis>client-mode</emphasis> connection. Defaults to a
<classname>ThreadPoolTaskScheduler</classname> with a pool size of `.
</entry>
</row>
</tbody>
</tgroup>
</table>
@@ -1196,6 +1297,36 @@
to correlate messages to the connection on which an
original inbound message was received. </entry>
</row>
<row>
<entry>client-mode</entry>
<entry>true, false</entry>
<entry>
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 <emphasis>retry-interval</emphasis> and <emphasis>scheduler</emphasis>.
The connection factory must be of type 'client' and have <emphasis>single-use</emphasis>
set to false.
</entry>
</row>
<row>
<entry>retry-interval</entry>
<entry></entry>
<entry>
When in <emphasis>client-mode</emphasis>, specifies the number of milliseconds
to wait between connection attempts, or after a connection failure. Default 60,000
(60 seconds).
</entry>
</row>
<row>
<entry>scheduler</entry>
<entry>true, false</entry>
<entry>
Specifies a <classname>TaskScheduler</classname> to use for managing the
<emphasis>client-mode</emphasis> connection. Defaults to a
<classname>ThreadPoolTaskScheduler</classname> with a pool size of `.
</entry>
</row>
</tbody>
</tgroup>
</table>
@@ -1246,6 +1377,36 @@
any reply from that flow will then be returned as a response by
the gateway.</entry>
</row>
<row>
<entry>client-mode</entry>
<entry>true, false</entry>
<entry>
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 <emphasis>retry-interval</emphasis> and <emphasis>scheduler</emphasis>.
The connection factory must be of type 'client' and have <emphasis>single-use</emphasis>
set to false.
</entry>
</row>
<row>
<entry>retry-interval</entry>
<entry></entry>
<entry>
When in <emphasis>client-mode</emphasis>, specifies the number of milliseconds
to wait between connection attempts, or after a connection failure. Default 60,000
(60 seconds).
</entry>
</row>
<row>
<entry>scheduler</entry>
<entry>true, false</entry>
<entry>
Specifies a <classname>TaskScheduler</classname> to use for managing the
<emphasis>client-mode</emphasis> connection. Defaults to a
<classname>ThreadPoolTaskScheduler</classname> with a pool size of `.
</entry>
</row>
</tbody>
</tgroup>
</table>

1
spring-integration-ip/.springBeans
View File

@@ -9,6 +9,7 @@
<configs>
<config>src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml</config>
<config>src/test/java/org/springframework/integration/ip/tcp/AutoStartTests-context.xml</config>
<config>src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests-context.xml</config>
</configs>
<configSets>
</configSets>

14
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -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

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java
View File

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

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundGatewayParser.java
View File

@@ -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);
}
}

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java
View File

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

165
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java
View File

@@ -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<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
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();
}
}
}

149
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java
View File

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

160
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java
View File

@@ -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<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
@@ -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();
}
}
}

24
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java
View File

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

58
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeCapable.java Normal file
View File

@@ -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.
* <p/>
* 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();
}

73
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java Normal file
View File

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

14
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java
View File

@@ -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;
}

20
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java
View File

@@ -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;
}

58
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd
View File

@@ -156,6 +156,17 @@ task executors such as a WorkManagerTaskExecutor.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="client-mode" type="xsd:string" use="optional" default="false">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="clientModeAttributeGroup" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -198,6 +209,17 @@ task executors such as a WorkManagerTaskExecutor.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="client-mode" type="xsd:string" use="optional" default="false">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="clientModeAttributeGroup" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -256,6 +278,17 @@ task executors such as a WorkManagerTaskExecutor.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="client-mode" type="xsd:string" use="optional" default="false">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="clientModeAttributeGroup" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -523,4 +556,29 @@ default is 0. Values can be negative. See SmartLifeCycle.
</xsd:attribute>
</xsd:complexType>
<xsd:attributeGroup name="clientModeAttributeGroup">
<xsd:attribute name="retry-interval" type="xsd:string" use="optional" default="60000">
<xsd:annotation>
<xsd:documentation>
When in client mode, specifies the retry interval, in milliseconds, if a connection
cannot be established. Defaults to 60000.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="scheduler" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.scheduling.TaskScheduler" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:attributeGroup>
</xsd:schema>

50
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -278,4 +278,54 @@
channel="tcpChannel"
connection-factory="server1" />
<ip:tcp-connection-factory id="cfC3"
type="client"
port="#{tcpIpUtils.findAvailableServerSocket(6120)}"
host="localhost"
lookup-host="false"
apply-sequence="false"
/>
<ip:tcp-inbound-channel-adapter id="tcpInClientMode"
channel="tcpChannel"
connection-factory="cfC3"
client-mode="true"
retry-interval="123"
scheduler="sched" />
<ip:tcp-connection-factory id="cfC4"
type="client"
port="#{tcpIpUtils.findAvailableServerSocket(6140)}"
host="localhost"
lookup-host="false"
apply-sequence="false"
/>
<ip:tcp-outbound-channel-adapter id="tcpOutClientMode"
channel="tcpChannel"
connection-factory="cfC4"
client-mode="true"
retry-interval="124"
scheduler="sched" />
<ip:tcp-connection-factory id="cfC5"
type="client"
port="#{tcpIpUtils.findAvailableServerSocket(6160)}"
host="localhost"
lookup-host="false"
apply-sequence="false"
/>
<ip:tcp-inbound-gateway id="inGatewayClientMode"
request-channel="tcpChannel"
reply-channel="replyChannel"
connection-factory="cfC5"
reply-timeout="456"
client-mode="true"
retry-interval="125"
scheduler="sched"
/>
<task:scheduler id="sched"/>
</beans>

66
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

@@ -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"));
}
}

51
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests-context.xml Normal file
View File

@@ -0,0 +1,51 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-ip="http://www.springframework.org/schema/integration/ip"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip-2.1.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketTestUtils" />
<int:channel id="in"/>
<int:channel id="dummy">
<int:queue />
</int:channel>
<int-ip:tcp-connection-factory id="server"
type="server"
using-nio="true"
port="#{tcpIpUtils.findAvailableServerSocket(12000)}"
lookup-host="false"
so-timeout="20000"
/>
<int-ip:tcp-connection-factory id="client1"
type="client"
host="localhost"
port="#{server.port}"
lookup-host="false"
so-timeout="100000"
/>
<int-ip:tcp-inbound-gateway id="servergw"
request-channel="dummy"
connection-factory="server"/>
<int-ip:tcp-inbound-channel-adapter
id="tcpIn"
connection-factory="client1"
channel="in"
client-mode="true"/>
<int:channel id="cbChannel" />
<int:control-bus input-channel="cbChannel" />
<int:gateway default-request-channel="cbChannel" service-interface="org.springframework.integration.ip.tcp.ClientModeControlBusTests$ControlBus"/>
</beans>

62
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java Normal file
View File

@@ -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);
}
}

59
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java
View File

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

88
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
View File

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

91
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

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