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

Merge pull request #106 from garyrussell/INT-2162

Browse Source 
INT-2162 Add apply-sequence to TCP Connection Factory
  INT-2154 Refactor TCP ConnectionFactorys to facilitate subclassing
  INT-2050 Code Polishing
This commit is contained in:
Mark Fisher
2011-10-04 14:41:18 -04:00
parent 03d6105f51 7981855c78
commit 28dd2571c0
46 changed files with 1487 additions and 758 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -652,7 +652,7 @@
Two sequential
messages arriving on the <emphasis>same</emphasis> socket <emphasis>might</emphasis>
be processed by different threads. This means that the order in which the messages are
sent to the channel is indeterminate; the strict ordering of the messages on the
sent to the channel is indeterminate; the strict ordering of the messages arriving on the
socket is not maintained.
</para>
<para>
@@ -660,6 +660,14 @@
is required, consider setting <classname>using-nio</classname> to false
and using async handoff.
</para>
<para>
Alternatively, you may choose to insert a resequencer downstream of the inbound endpoint to
return the messages to their proper sequence. Set <emphasis>apply-sequence</emphasis>
to true on the connection factory, and messages arriving on a TCP connection will
have <emphasis>sequenceNumber</emphasis> and <emphasis>correlationId</emphasis> headers
set. The resequencer uses these headers to return the messages to their proper
sequence.
</para>
</section>
<section id="ip-endpoint-reference">
<title>IP Configuration Attributes</title>
@@ -727,7 +735,7 @@
<entry>true, false</entry>
<entry>Whether or not connection uses NIO. Refer to the java.nio
package for more information.
See <xref linkend="note_nio" />.
See <xref linkend="note_nio" />.
Default false.</entry>
</row>
<row>
@@ -739,6 +747,18 @@
Refer to <classname>java.nio.ByteBuffer</classname> documentation for
more information. Must be false if using-nio is false. </entry>
</row>
<row>
<entry>apply-sequence</entry>
<entry>Y</entry>
<entry>Y</entry>
<entry>true, false</entry>
<entry>When using NIO, it may be necessary to resequence messages. When this
attribute is set to true, <emphasis>correlationId</emphasis> and
<emphasis>sequenceNumber</emphasis> headers will be added to
received messages.
See <xref linkend="note_nio" />.
Default false.</entry>
</row>
<row>
<entry>so-timeout</entry>
<entry>Y</entry>

3
spring-integration-ip/.springBeans
View File

@@ -1,13 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.5.1.201011101000-RELEASE]]></pluginVersion>
<pluginVersion><![CDATA[2.7.2.201109122348-RELEASE]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[false]]></enableImports>
<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>
</configs>
<configSets>
</configSets>

67
spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java
View File

@@ -35,23 +35,23 @@ import org.springframework.util.Assert;
public abstract class AbstractInternetProtocolReceivingChannelAdapter
extends MessageProducerSupport implements Runnable, CommonSocketOptions {
protected final int port;
private final int port;
protected volatile int soTimeout = 0;
private volatile int soTimeout = 0;
protected volatile int soReceiveBufferSize = -1;
private volatile int soReceiveBufferSize = -1;
protected volatile int receiveBufferSize = 2048;
private volatile int receiveBufferSize = 2048;
protected volatile boolean active;
private volatile boolean active;
protected volatile boolean listening;
private volatile boolean listening;
protected volatile String localAddress;
private volatile String localAddress;
protected volatile Executor taskExecutor;
private volatile Executor taskExecutor;
protected volatile int poolSize = 5;
private volatile int poolSize = 5;
public AbstractInternetProtocolReceivingChannelAdapter(int port) {
@@ -66,30 +66,39 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
return port;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.SocketOptions#setSoTimeout(int)
*/
public void setSoTimeout(int soTimeout) {
this.soTimeout = soTimeout;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.SocketOptions#setSoReceiveBufferSize(int)
/**
* @return the soTimeout
*/
public int getSoTimeout() {
return soTimeout;
}
public void setSoReceiveBufferSize(int soReceiveBufferSize) {
this.soReceiveBufferSize = soReceiveBufferSize;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.CommonSocketOptions#setSoSendBufferSize(int)
/**
* @return the soReceiveBufferSize
*/
public void setSoSendBufferSize(int soSendBufferSize) {
public int getSoReceiveBufferSize() {
return soReceiveBufferSize;
}
public void setReceiveBufferSize(int receiveBufferSize) {
this.receiveBufferSize = receiveBufferSize;
}
/**
* @return the receiveBufferSize
*/
public int getReceiveBufferSize() {
return receiveBufferSize;
}
@Override
protected void doStart() {
TaskScheduler taskScheduler = this.getTaskScheduler();
@@ -117,9 +126,6 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
}
}
/* (non-Javadoc)
* @see org.springframework.integration.endpoint.AbstractEndpoint#doStop()
*/
@Override
protected void doStop() {
this.active = false;
@@ -129,6 +135,13 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
return listening;
}
/**
* @param listening the listening to set
*/
public void setListening(boolean listening) {
this.listening = listening;
}
public String getLocalAddress() {
return localAddress;
}
@@ -145,4 +158,18 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
this.taskExecutor = taskExecutor;
}
/**
* @return the taskExecutor
*/
public Executor getTaskExecutor() {
return taskExecutor;
}
/**
* @return the active
*/
public boolean isActive() {
return active;
}
}

42
spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java
View File

@@ -36,15 +36,15 @@ public abstract class AbstractInternetProtocolSendingMessageHandler extends Abst
protected final Log logger = LogFactory.getLog(getClass());
protected final SocketAddress destinationAddress;
private final SocketAddress destinationAddress;
protected final String host;
private final String host;
protected final int port;
private final int port;
protected volatile int soSendBufferSize = -1;
private volatile int soSendBufferSize = -1;
protected volatile int soTimeout = -1;
private volatile int soTimeout = -1;
public AbstractInternetProtocolSendingMessageHandler(String host, int port) {
Assert.notNull(host, "host must not be null");
@@ -80,6 +80,14 @@ public abstract class AbstractInternetProtocolSendingMessageHandler extends Abst
this.soSendBufferSize = size;
}
/**
* @return the host
*/
public String getHost() {
return host;
}
/**
* @return the port
*/
@@ -87,4 +95,28 @@ public abstract class AbstractInternetProtocolSendingMessageHandler extends Abst
return port;
}
/**
* @return the destinationAddress
*/
public SocketAddress getDestinationAddress() {
return destinationAddress;
}
/**
* @return the soTimeout
*/
public int getSoTimeout() {
return soTimeout;
}
/**
* @return the soSendBufferSize
*/
public int getSoSendBufferSize() {
return soSendBufferSize;
}
}

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -29,19 +29,19 @@ public interface CommonSocketOptions {
* @see Socket#setSoTimeout(int)
* @see DatagramSocket#setSoTimeout(int)
*/
public void setSoTimeout(int soTimeout);
void setSoTimeout(int soTimeout);
/**
* @see Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
*/
public void setSoReceiveBufferSize(int soReceiveBufferSize);
void setSoReceiveBufferSize(int soReceiveBufferSize);
/**
* @see Socket#setSendBufferSize(int)
* @see DatagramSocket#setSendBufferSize(int)
*/
public void setSoSendBufferSize(int soSendBufferSize);
void setSoSendBufferSize(int soSendBufferSize);
/**
* On a multi-homed system, specifies the ip address of the network interface used to communicate.
@@ -53,6 +53,6 @@ public interface CommonSocketOptions {
*
* @param localAddress
*/
public void setLocalAddress(String localAddress);
void setLocalAddress(String localAddress);
}

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java
View File

@@ -42,6 +42,11 @@ public abstract class IpHeaders {
public static final String CONNECTION_ID = IP + "connection_id";
/**
* Use apply-sequence and sequenceNumber instead
* @deprecated
*/
@Deprecated
public static final String CONNECTION_SEQ = IP + "connection_seq";
}

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

@@ -107,6 +107,11 @@ public abstract class IpAdapterParserUtils {
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";
/**
* Adds a constructor-arg to the provided bean definition builder

90
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java
View File

@@ -17,6 +17,7 @@ package org.springframework.integration.ip.config;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.serializer.Deserializer;
@@ -24,13 +25,11 @@ import org.springframework.core.serializer.Serializer;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain;
import org.springframework.integration.ip.tcp.connection.TcpListener;
import org.springframework.integration.ip.tcp.connection.TcpMessageMapper;
import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpSender;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
/**
@@ -42,58 +41,56 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer
*
*/
public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<AbstractConnectionFactory>
implements SmartLifecycle {
implements SmartLifecycle, BeanNameAware {
private AbstractConnectionFactory connectionFactory;
private String type;
private volatile AbstractConnectionFactory connectionFactory;
protected String host;
protected int port;
protected TcpListener listener;
private volatile String type;
protected TcpSender sender;
private volatile String host;
protected int soTimeout;
private volatile int port;
private int soSendBufferSize;
private volatile int soTimeout;
private int soReceiveBufferSize;
private boolean soTcpNoDelay;
private volatile int soSendBufferSize;
private int soLinger = -1; // don't set by default
private volatile int soReceiveBufferSize;
private boolean soKeepAlive;
private volatile boolean soTcpNoDelay;
private int soTrafficClass = -1; // don't set by default
private Executor taskExecutor;
protected Deserializer<?> deserializer = new ByteArrayCrLfSerializer();
protected Serializer<?> serializer = new ByteArrayCrLfSerializer();
protected TcpMessageMapper mapper = new TcpMessageMapper();
private volatile int soLinger = -1; // don't set by default
protected boolean singleUse;
private volatile boolean soKeepAlive;
protected int poolSize = 5;
private volatile int soTrafficClass = -1; // don't set by default
protected volatile boolean active;
private volatile Executor taskExecutor;
protected TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
private boolean lookupHost = true;
private String localAddress;
private volatile Deserializer<?> deserializer = new ByteArrayCrLfSerializer();
private volatile Serializer<?> serializer = new ByteArrayCrLfSerializer();
private volatile TcpMessageMapper mapper = new TcpMessageMapper();
private volatile boolean singleUse;
private volatile int poolSize = 5;
private volatile TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
private volatile boolean lookupHost = true;
private volatile String localAddress;
private volatile boolean usingNio;
private volatile boolean usingDirectBuffers;
private volatile String beanName;
private volatile boolean applySequence;
private boolean usingNio;
private boolean usingDirectBuffers;
@Override
public Class<?> getObjectType() {
return this.connectionFactory != null ? this.connectionFactory.getClass()
@@ -136,6 +133,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
factory.setDeserializer(this.deserializer);
factory.setInterceptorFactoryChain(this.interceptorFactoryChain);
factory.setLookupHost(this.lookupHost);
this.mapper.setApplySequence(this.applySequence);
factory.setMapper(this.mapper);
factory.setPoolSize(this.poolSize);
factory.setSerializer(this.serializer);
@@ -148,6 +146,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
factory.setSoTimeout(this.soTimeout);
factory.setSoTrafficClass(this.soTrafficClass);
factory.setTaskExecutor(this.taskExecutor);
factory.setBeanName(this.beanName);
}
private void setServerAttributes(AbstractServerConnectionFactory factory) {
@@ -363,5 +362,16 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
return this.connectionFactory.isRunning();
}
public void setBeanName(String name) {
this.beanName = name;
}
/**
* @param applySequence the applySequence to set
*/
public void setApplySequence(boolean applySequence) {
this.applySequence = applySequence;
}
}

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java
View File

@@ -81,6 +81,8 @@ public class TcpConnectionParser extends AbstractBeanDefinitionParser {
IpAdapterParserUtils.INTERCEPTOR_FACTORY_CHAIN);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.LOOKUP_HOST);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.APPLY_SEQUENCE);
return builder.getBeanDefinition();
}

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

@@ -42,6 +42,10 @@ public class TcpInboundChannelAdapterParser extends AbstractChannelAdapterParser
element, "channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder,
element, "error-channel", "errorChannel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.AUTO_STARTUP);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.PHASE);
return builder.getBeanDefinition();
}

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

@@ -36,6 +36,10 @@ public class TcpOutboundChannelAdapterParser extends AbstractOutboundChannelAdap
".TcpSendingMessageHandler");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TCP_CONNECTION_FACTORY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.AUTO_STARTUP);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.PHASE);
return builder.getBeanDefinition();
}

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java
View File

@@ -49,6 +49,10 @@ public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser {
IpAdapterParserUtils.REQUEST_TIMEOUT);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.REPLY_TIMEOUT);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.AUTO_STARTUP);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.PHASE);
return builder;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* 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.
@@ -40,7 +40,7 @@ import org.springframework.integration.ip.tcp.connection.TcpSender;
*/
public class TcpInboundGateway extends MessagingGatewaySupport implements TcpListener, TcpSender {
protected AbstractServerConnectionFactory connectionFactory;
private AbstractServerConnectionFactory connectionFactory;
private Map<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
@@ -94,4 +94,11 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements TcpLis
public String getComponentType(){
return "ip:tcp-inbound-gateway";
}
/**
* @return the connectionFactory
*/
protected AbstractServerConnectionFactory getConnectionFactory() {
return connectionFactory;
}
}

98
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -22,6 +22,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageTimeoutException;
@@ -35,28 +36,34 @@ import org.springframework.integration.ip.tcp.connection.TcpListener;
import org.springframework.integration.ip.tcp.connection.TcpSender;
import org.springframework.util.Assert;
/**
/**
* TCP outbound gateway that uses a client connection factory. If the factory is configured
* for single-use connections, each request is sent on a new connection; if the factory does not use
* single use connections, each request is blocked until the previous response is received
* (or times out). Asynchronous requests/responses over the same connection are not
* (or times out). Asynchronous requests/responses over the same connection are not
* supported - use a pair of outbound/inbound adapters for that use case.
*
* <p/>
* {@link SmartLifecycle} methods delegate to the underlying {@link AbstractConnectionFactory}
*
*
* @author Gary Russell
* @since 2.0
*/
public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler implements TcpSender, TcpListener {
public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler implements TcpSender, TcpListener, SmartLifecycle {
private volatile AbstractConnectionFactory connectionFactory;
protected AbstractConnectionFactory connectionFactory;
private Map<String, AsyncReply> pendingReplies = new ConcurrentHashMap<String, AsyncReply>();
private Semaphore semaphore = new Semaphore(1, true);
private long replyTimeout = 10000;
private long requestTimeout = 10000;
private volatile long replyTimeout = 10000;
private volatile long requestTimeout = 10000;
private volatile boolean autoStartup = true;
private volatile int phase;
/**
* @param requestTimeout the requestTimeout to set
@@ -98,7 +105,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
connection.send(requestMessage);
Message<?> replyMessage = reply.getReply();
if (replyMessage == null) {
throw new MessageTimeoutException(requestMessage, "Timed out waiting for response");
throw new MessageTimeoutException(requestMessage, "Timed out waiting for response");
}
if (logger.isDebugEnabled()) {
logger.debug("Respose " + replyMessage);
@@ -137,12 +144,8 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
return false;
}
public boolean isListening() {
return false;
}
public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
Assert.isTrue(connectionFactory instanceof AbstractClientConnectionFactory,
Assert.isTrue(connectionFactory instanceof AbstractClientConnectionFactory,
this.getClass().getName() + " requires a client connection factory");
this.connectionFactory = connectionFactory;
connectionFactory.registerListener(this);
@@ -157,10 +160,59 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
// do nothing - no asynchronous multiplexing supported
}
/**
* Specify the Spring Integration reply channel. If this property is not
* set the gateway will check for a 'replyChannel' header on the request.
*/
public void setReplyChannel(MessageChannel replyChannel) {
this.setOutputChannel(replyChannel);
}
public String getComponentType(){
return "ip:tcp-outbound-gateway";
}
public void start() {
this.connectionFactory.start();
}
public void stop() {
this.connectionFactory.stop();
}
public boolean isRunning() {
return this.connectionFactory.isRunning();
}
public int getPhase() {
return this.phase;
}
public boolean isAutoStartup() {
return this.autoStartup;
}
public void stop(Runnable callback) {
this.connectionFactory.stop(callback);
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
public void setPhase(int phase) {
this.phase = phase;
}
/**
* @return the connectionFactory
*/
protected AbstractConnectionFactory getConnectionFactory() {
return connectionFactory;
}
/**
* Class used to coordinate the asynchronous reply to its request.
*
*
* @author Gary Russell
* @since 2.0
*/
@@ -197,14 +249,4 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
}
/**
* Specify the Spring Integration reply channel. If this property is not
* set the gateway will check for a 'replyChannel' header on the request.
*/
public void setReplyChannel(MessageChannel replyChannel) {
this.setOutputChannel(replyChannel);
}
public String getComponentType(){
return "ip:tcp-outbound-gateway";
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* 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.
@@ -15,8 +15,6 @@
*/
package org.springframework.integration.ip.tcp;
import java.net.ServerSocket;
import org.springframework.integration.Message;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
@@ -38,11 +36,9 @@ import org.springframework.integration.ip.tcp.connection.TcpListener;
public class TcpReceivingChannelAdapter
extends MessageProducerSupport implements TcpListener {
protected ServerSocket serverSocket;
private ConnectionFactory clientConnectionFactory;
protected ConnectionFactory clientConnectionFactory;
protected ConnectionFactory serverConnectionFactory;
private ConnectionFactory serverConnectionFactory;
public boolean onMessage(Message<?> message) {
sendMessage(message);
@@ -51,12 +47,22 @@ public class TcpReceivingChannelAdapter
@Override
protected void doStart() {
// Nothing to do; we're passive
if (this.serverConnectionFactory != null) {
this.serverConnectionFactory.start();
}
if (this.clientConnectionFactory != null) {
this.clientConnectionFactory.start();
}
}
@Override
protected void doStop() {
// Nothing to do; we're passive
if (this.clientConnectionFactory != null) {
this.clientConnectionFactory.stop();
}
if (this.serverConnectionFactory != null) {
this.serverConnectionFactory.stop();
}
}
/**
@@ -88,4 +94,18 @@ public class TcpReceivingChannelAdapter
public String getComponentType(){
return "ip:tcp-inbound-channel-adapter";
}
/**
* @return the clientConnectionFactory
*/
protected ConnectionFactory getClientConnectionFactory() {
return clientConnectionFactory;
}
/**
* @return the serverConnectionFactory
*/
protected ConnectionFactory getServerConnectionFactory() {
return serverConnectionFactory;
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* 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.
@@ -21,6 +21,7 @@ import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageHandlingException;
@@ -35,7 +36,7 @@ import org.springframework.integration.ip.tcp.connection.TcpSender;
import org.springframework.integration.mapping.MessageMappingException;
/**
* Tcp outbound channel adapter using a TcpConnection to
* Tcp outbound channel adapter using a TcpConnection to
* send data - if the connection factory is a server
* factory, the TcpListener owns the connections. If it is
* a client factory, this object owns the connection.
@@ -43,39 +44,36 @@ import org.springframework.integration.mapping.MessageMappingException;
* @since 2.0
*
*/
public class TcpSendingMessageHandler extends AbstractMessageHandler implements TcpSender {
protected Log logger = LogFactory.getLog(this.getClass());
protected TcpConnection connection;
protected ConnectionFactory clientConnectionFactory;
protected ConnectionFactory serverConnectionFactory;
protected Map<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
protected synchronized TcpConnection getConnection() {
public class TcpSendingMessageHandler extends AbstractMessageHandler implements TcpSender, SmartLifecycle {
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile ConnectionFactory clientConnectionFactory;
private volatile ConnectionFactory serverConnectionFactory;
private Map<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
private volatile boolean autoStartup;
private volatile int phase;
protected TcpConnection getConnection() {
TcpConnection connection = null;
if (this.clientConnectionFactory == null) {
return null;
}
try {
this.connection = clientConnectionFactory.getConnection();
connection = this.clientConnectionFactory.getConnection();
} catch (Exception e) {
logger.error("Error creating SocketWriter", e);
}
return this.connection;
}
/**
* Close the underlying socket and prepare to establish a new socket on
* the next write.
*/
protected void close() {
this.connection.close();
this.connection = null;
return connection;
}
/**
* Writes the message payload to the underlying socket, using the specified
* message format.
* message format.
* @see org.springframework.integration.core.MessageHandler#handleMessage(org.springframework.integration.Message)
*/
public void handleMessageInternal(final Message<?> message) throws MessageRejectedException,
@@ -96,7 +94,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
}
return;
}
// we own the connection
try {
doWrite(message);
@@ -116,8 +114,9 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
* @param message The message to write.
*/
protected void doWrite(Message<?> message) {
TcpConnection connection = null;
try {
TcpConnection connection = getConnection();
connection = getConnection();
if (connection == null) {
throw new MessageMappingException(message, "Failed to create connection");
}
@@ -126,11 +125,10 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
}
connection.send(message);
} catch (Exception e) {
String connectionId = null;
if (this.connection != null) {
connectionId = this.connection.getConnectionId();
String connectionId = null;
if (connection != null) {
connectionId = connection.getConnectionId();
}
this.connection = null;
if (e instanceof MessageMappingException) {
throw (MessageMappingException) e;
}
@@ -142,7 +140,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
* Sets the client or server connection factory; for this (an outbound adapter), if
* the factory is a server connection factory, the sockets are owned by a receiving
* channel adapter and this adapter is used to send replies.
*
*
* @param connectionFactory the connectionFactory to set
*/
public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
@@ -157,11 +155,83 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
public void addNewConnection(TcpConnection connection) {
connections.put(connection.getConnectionId(), connection);
}
public void removeDeadConnection(TcpConnection connection) {
connections.remove(connection.getConnectionId());
}
public String getComponentType(){
return "ip:tcp-outbound-channel-adapter";
}
public void start() {
if (this.clientConnectionFactory != null) {
this.clientConnectionFactory.start();
}
if (this.serverConnectionFactory != null) {
this.serverConnectionFactory.start();
}
}
public void stop() {
if (this.clientConnectionFactory != null) {
this.clientConnectionFactory.stop();
}
if (this.serverConnectionFactory != null) {
this.serverConnectionFactory.stop();
}
}
public boolean isRunning() {
boolean cfRunning = this.clientConnectionFactory != null ? this.clientConnectionFactory.isRunning() : false;
boolean sfRunning = this.serverConnectionFactory != null ? this.serverConnectionFactory.isRunning() : false;
return cfRunning | sfRunning;
}
public int getPhase() {
return this.phase;
}
public boolean isAutoStartup() {
return this.autoStartup;
}
public void stop(Runnable callback) {
if (this.clientConnectionFactory != null) {
this.clientConnectionFactory.stop(callback);
}
if (this.serverConnectionFactory != null) {
this.serverConnectionFactory.stop(callback);
}
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
public void setPhase(int phase) {
this.phase = phase;
}
/**
* @return the clientConnectionFactory
*/
protected ConnectionFactory getClientConnectionFactory() {
return clientConnectionFactory;
}
/**
* @return the serverConnectionFactory
*/
protected ConnectionFactory getServerConnectionFactory() {
return serverConnectionFactory;
}
/**
* @return the connections
*/
protected Map<String, TcpConnection> getConnections() {
return connections;
}
}

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

@@ -19,8 +19,6 @@ package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.net.SocketException;
import org.springframework.util.Assert;
/**
* Abstract class for client connection factories; client connection factories
* establish outgoing connections.
@@ -30,7 +28,7 @@ import org.springframework.util.Assert;
*/
public abstract class AbstractClientConnectionFactory extends AbstractConnectionFactory {
protected TcpConnection theConnection;
private TcpConnection theConnection;
/**
* Constructs a factory that will established connections to the host and port.
@@ -38,9 +36,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
* @param port The port.
*/
public AbstractClientConnectionFactory(String host, int port) {
Assert.notNull(host, "host must not be null");
this.host = host;
this.port = port;
super(host, port);
}
/**
@@ -53,11 +49,12 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
* @param socket The new socket.
*/
protected void initializeConnection(TcpConnection connection, Socket socket) {
if (this.listener != null) {
connection.registerListener(this.listener);
TcpListener listener = this.getListener();
if (listener != null) {
connection.registerListener(listener);
}
if (this.listener != null || this.singleUse) {
if (this.soTimeout <= 0) {
if (listener != null || this.isSingleUse()) {
if (this.getSoTimeout() <= 0) {
try {
socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT);
} catch (SocketException e) {
@@ -65,10 +62,24 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
}
}
}
connection.setMapper(this.mapper);
connection.setDeserializer(this.deserializer);
connection.setSerializer(this.serializer);
connection.setSingleUse(this.singleUse);
connection.setMapper(this.getMapper());
connection.setDeserializer(this.getDeserializer());
connection.setSerializer(this.getSerializer());
connection.setSingleUse(this.isSingleUse());
}
/**
* @param theConnection the theConnection to set
*/
protected void setTheConnection(TcpConnection theConnection) {
this.theConnection = theConnection;
}
/**
* @return the theConnection
*/
protected TcpConnection getTheConnection() {
return theConnection;
}
}

195
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -36,73 +36,85 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.MessagingException;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.util.Assert;
/**
* Base class for all connection factories.
*
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractConnectionFactory
implements ConnectionFactory, Runnable, SmartLifecycle {
public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
implements ConnectionFactory, Runnable, SmartLifecycle, BeanNameAware {
protected Log logger = LogFactory.getLog(this.getClass());
protected final static int DEFAULT_REPLY_TIMEOUT = 10000;
protected String host;
protected int port;
protected TcpListener listener;
protected final Log logger = LogFactory.getLog(this.getClass());
protected TcpSender sender;
protected static final int DEFAULT_REPLY_TIMEOUT = 10000;
protected int soTimeout;
private volatile String host;
private int soSendBufferSize;
private volatile int port;
private int soReceiveBufferSize;
private boolean soTcpNoDelay;
private volatile TcpListener listener;
private int soLinger = -1; // don't set by default
private volatile TcpSender sender;
private boolean soKeepAlive;
private volatile int soTimeout;
private int soTrafficClass = -1; // don't set by default
private Executor taskExecutor;
private boolean privateExecutor;
private volatile int soSendBufferSize;
protected Deserializer<?> deserializer = new ByteArrayCrLfSerializer();
protected Serializer<?> serializer = new ByteArrayCrLfSerializer();
protected TcpMessageMapper mapper = new TcpMessageMapper();
private volatile int soReceiveBufferSize;
protected boolean singleUse;
private volatile boolean soTcpNoDelay;
protected int poolSize = 5;
private volatile int soLinger = -1; // don't set by default
protected volatile boolean active;
private volatile boolean soKeepAlive;
protected TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
private boolean lookupHost = true;
private List<TcpConnection> connections = new LinkedList<TcpConnection>();
private volatile int soTrafficClass = -1; // don't set by default
private volatile Executor taskExecutor;
private volatile boolean privateExecutor;
private volatile Deserializer<?> deserializer = new ByteArrayCrLfSerializer();
private volatile Serializer<?> serializer = new ByteArrayCrLfSerializer();
private volatile TcpMessageMapper mapper = new TcpMessageMapper();
private volatile boolean singleUse;
private volatile int poolSize = 5;
private volatile boolean active;
private volatile TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
private volatile boolean lookupHost = true;
private volatile List<TcpConnection> connections = new LinkedList<TcpConnection>();
protected final Object lifecycleMonitor = new Object();
public AbstractConnectionFactory(int port) {
this.port = port;
}
public AbstractConnectionFactory(String host, int port) {
Assert.notNull(host, "host must not be null");
this.host = host;
this.port = port;
}
/**
* Sets socket attributes on the socket.
* @param socket The socket.
@@ -240,6 +252,48 @@ public abstract class AbstractConnectionFactory
return port;
}
/**
* @return the listener
*/
public TcpListener getListener() {
return listener;
}
/**
* @return the sender
*/
public TcpSender getSender() {
return sender;
}
/**
* @return the serializer
*/
public Serializer<?> getSerializer() {
return serializer;
}
/**
* @return the deserializer
*/
public Deserializer<?> getDeserializer() {
return deserializer;
}
/**
* @return the mapper
*/
public TcpMessageMapper getMapper() {
return mapper;
}
/**
* @return the poolSize
*/
public int getPoolSize() {
return poolSize;
}
/**
* Registers a TcpListener to receive messages after
* the payload has been converted from the input data.
@@ -252,7 +306,7 @@ public abstract class AbstractConnectionFactory
}
/**
* Registers a TcpSender; for server sockets, used to
* Registers a TcpSender; for server sockets, used to
* provide connection information so a sender can be used
* to reply to incoming messages.
* @param sender The sender
@@ -271,7 +325,7 @@ public abstract class AbstractConnectionFactory
}
/**
*
*
* @param deserializer the deserializer to set
*/
public void setDeserializer(Deserializer<?> deserializer) {
@@ -279,7 +333,7 @@ public abstract class AbstractConnectionFactory
}
/**
*
*
* @param serializer the serializer to set
*/
public void setSerializer(Serializer<?> serializer) {
@@ -287,7 +341,7 @@ public abstract class AbstractConnectionFactory
}
/**
*
*
* @param mapper the mapper to set; defaults to a {@link TcpMessageMapper}
*/
public void setMapper(TcpMessageMapper mapper) {
@@ -309,7 +363,7 @@ public abstract class AbstractConnectionFactory
this.singleUse = singleUse;
}
public void setPoolSize(int poolSize) {
this.poolSize = poolSize;
}
@@ -349,10 +403,13 @@ public abstract class AbstractConnectionFactory
this.getTaskExecutor().execute(this);
}
}
if (logger.isInfoEnabled()) {
logger.info("started " + this);
}
}
/**
* Creates a taskExecutor (if one was not provided).
* Creates a taskExecutor (if one was not provided).
*/
protected Executor getTaskExecutor() {
synchronized (this.lifecycleMonitor) {
@@ -388,7 +445,7 @@ public abstract class AbstractConnectionFactory
try {
if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
logger.debug("Forcing executor shutdown");
executorService.shutdownNow();
executorService.shutdownNow();
if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
logger.debug("Executor failed to shutdown");
}
@@ -402,6 +459,9 @@ public abstract class AbstractConnectionFactory
}
}
}
if (logger.isInfoEnabled()) {
logger.info("stopped " + this);
}
}
protected TcpConnection wrapConnection(TcpConnection connection) throws Exception {
@@ -409,7 +469,7 @@ public abstract class AbstractConnectionFactory
if (this.interceptorFactoryChain == null) {
return connection;
}
TcpConnectionInterceptorFactory[] interceptorFactories =
TcpConnectionInterceptorFactory[] interceptorFactories =
this.interceptorFactoryChain.getInterceptorFactories();
if (interceptorFactories == null) {
return connection;
@@ -433,9 +493,9 @@ public abstract class AbstractConnectionFactory
}
/**
*
*
* Times out any expired connections then, if selectionCount > 0, processes the selected keys.
*
*
* @param selectionCount
* @param selector
* @param connections
@@ -483,7 +543,7 @@ public abstract class AbstractConnectionFactory
else if (key.isReadable()) {
try {
key.interestOps(key.interestOps() - key.readyOps());
final TcpNioConnection connection;
final TcpNioConnection connection;
connection = (TcpNioConnection) key.attachment();
connection.setLastRead(System.currentTimeMillis());
this.taskExecutor.execute(new Runnable() {
@@ -535,13 +595,17 @@ public abstract class AbstractConnectionFactory
protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException {
throw new UnsupportedOperationException("Nio server factory must override this method");
}
public int getPhase() {
return 0;
}
/**
* We are controlled by the startup options of
* the bound endpoint.
*/
public boolean isAutoStartup() {
return true;
return false;
}
public void stop(Runnable callback) {
@@ -558,7 +622,7 @@ public abstract class AbstractConnectionFactory
this.connections.add(connection);
}
}
protected void harvestClosedConnections() {
synchronized (this.connections) {
Iterator<TcpConnection> iterator = this.connections.iterator();
@@ -570,4 +634,29 @@ public abstract class AbstractConnectionFactory
}
}
}
public boolean isRunning() {
return this.active;
}
/**
* @return the active
*/
protected boolean isActive() {
return active;
}
/**
* @param active the active to set
*/
protected void setActive(boolean active) {
this.active = active;
}
protected void checkActive() throws IOException {
if (!this.isActive()) {
throw new IOException(this + " connection factory has not been started");
}
}
}

33
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -29,9 +29,9 @@ import java.net.SocketException;
*/
public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory {
protected boolean listening;
private boolean listening;
protected String localAddress;
private String localAddress;
/**
@@ -39,7 +39,7 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
* @param port
*/
public AbstractServerConnectionFactory(int port) {
this.port = port;
super(port);
}
@@ -51,6 +51,14 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
throw new UnsupportedOperationException("Getting a connection from a server factory is not supported");
}
/**
* @param listening the listening to set
*/
protected void setListening(boolean listening) {
this.listening = listening;
}
/**
*
* @return true if the server is listening on the port.
@@ -66,20 +74,21 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
* @param socket The new socket.
*/
protected void initializeConnection(TcpConnection connection, Socket socket) {
if (this.listener != null) {
connection.registerListener(this.listener);
TcpListener listener = this.getListener();
if (listener != null) {
connection.registerListener(listener);
}
connection.registerSender(this.sender);
connection.setMapper(this.mapper);
connection.setDeserializer(this.deserializer);
connection.setSerializer(this.serializer);
connection.setSingleUse(this.singleUse);
connection.registerSender(this.getSender());
connection.setMapper(this.getMapper());
connection.setDeserializer(this.getDeserializer());
connection.setSerializer(this.getSerializer());
connection.setSingleUse(this.isSingleUse());
/*
* If we have a collaborating outbound channel adapter and we are configured
* for single use; need to enforce a timeout on the socket so we will close
* it some period after the response was sent (timeout on the next read).
*/
if (this.singleUse && this.soTimeout <= 0 && this.listener != null) {
if (this.isSingleUse() && this.getSoTimeout() <= 0 && listener != null) {
try {
socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT);
} catch (SocketException e) {

94
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp.connection;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.logging.Log;
@@ -33,68 +34,62 @@ import org.springframework.util.Assert;
* Base class for TcpConnections. TcpConnections are established by
* client connection factories (outgoing) or server connection factories
* (incoming).
*
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractTcpConnection implements TcpConnection {
protected Log logger = LogFactory.getLog(this.getClass());
protected final Log logger = LogFactory.getLog(this.getClass());
@SuppressWarnings("rawtypes")
protected Deserializer deserializer;
private volatile Deserializer deserializer;
@SuppressWarnings("rawtypes")
protected Serializer serializer;
protected TcpMessageMapper mapper;
protected TcpListener listener;
private TcpListener actualListener;
private volatile Serializer serializer;
protected TcpSender sender;
private volatile TcpMessageMapper mapper;
protected boolean singleUse;
private volatile TcpListener listener;
protected final boolean server;
private volatile TcpListener actualListener;
protected String connectionId;
private AtomicLong sequence = new AtomicLong();
private int soLinger = -1;
private volatile TcpSender sender;
private String hostName = "unknown";
private volatile boolean singleUse;
private String hostAddress = "unknown";
private int port;
private final boolean lookupHost;
private final boolean server;
private volatile String connectionId;
private final AtomicLong sequence = new AtomicLong();
private volatile int soLinger = -1;
private volatile String hostName = "unknown";
private volatile String hostAddress = "unknown";
private volatile int port;
private int hashCode;
public AbstractTcpConnection(Socket socket, boolean server, boolean lookupHost) {
this.server = server;
this.lookupHost = lookupHost;
this.hashCode = socket.hashCode();
InetAddress inetAddress = socket.getInetAddress();
if (inetAddress != null) {
this.hostAddress = inetAddress.getHostAddress();
if (this.lookupHost) {
if (lookupHost) {
this.hostName = inetAddress.getHostName();
} else {
this.hostName = this.hostAddress;
}
}
this.connectionId = this.hostName + ":" + this.port + ":" + this.hashCode;
this.connectionId = this.hostName + ":" + this.port + ":" + UUID.randomUUID().toString();
try {
this.soLinger = socket.getSoLinger();
} catch (SocketException e) { }
}
public void afterSend(Message<?> message) throws Exception {
if (logger.isDebugEnabled())
logger.debug("Message sent " + message);
@@ -119,7 +114,7 @@ public abstract class AbstractTcpConnection implements TcpConnection {
}
/**
* If we have been intercepted, propagate the close from the outermost interceptor;
* If we have been intercepted, propagate the close from the outermost interceptor;
* otherwise, just call close().
*/
protected void closeConnection() {
@@ -147,14 +142,14 @@ public abstract class AbstractTcpConnection implements TcpConnection {
public void setMapper(TcpMessageMapper mapper) {
Assert.notNull(mapper, this.getClass().getName() + " Mapper may not be null");
this.mapper = mapper;
if (this.serializer != null &&
if (this.serializer != null &&
!(this.serializer instanceof AbstractByteArraySerializer)) {
mapper.setStringToBytes(false);
}
}
/**
*
*
* @return the deserializer
*/
public Deserializer<?> getDeserializer() {
@@ -169,7 +164,7 @@ public abstract class AbstractTcpConnection implements TcpConnection {
}
/**
*
*
* @return the serializer
*/
public Serializer<?> getSerializer() {
@@ -177,7 +172,7 @@ public abstract class AbstractTcpConnection implements TcpConnection {
}
/**
* @param serializer the serializer to set
* @param serializer the serializer to set
*/
public void setSerializer(Serializer<?> serializer) {
this.serializer = serializer;
@@ -202,7 +197,7 @@ public abstract class AbstractTcpConnection implements TcpConnection {
this.actualListener = outerInterceptor.getListener();
}
}
/**
* @param sender the sender to set
*/
@@ -219,9 +214,16 @@ public abstract class AbstractTcpConnection implements TcpConnection {
public TcpListener getListener() {
return this.listener;
}
/**
* @param singleUse true if this socket is to used once and
* @return the sender
*/
public TcpSender getSender() {
return sender;
}
/**
* @param singleUse true if this socket is to used once and
* discarded.
*/
public void setSingleUse(boolean singleUse) {
@@ -229,7 +231,7 @@ public abstract class AbstractTcpConnection implements TcpConnection {
}
/**
*
*
* @return True if connection is used once.
*/
public boolean isSingleUse() {
@@ -240,8 +242,8 @@ public abstract class AbstractTcpConnection implements TcpConnection {
return server;
}
public long getConnectionSeq() {
return sequence.incrementAndGet();
public long incrementAndGetConnectionSequence() {
return this.sequence.incrementAndGet();
}
public String getHostAddress() {
@@ -255,5 +257,5 @@ public abstract class AbstractTcpConnection implements TcpConnection {
public String getConnectionId() {
return this.connectionId;
}
}

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java
View File

@@ -157,8 +157,8 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI
}
}
public long getConnectionSeq() {
return this.theConnection.getConnectionSeq();
public long incrementAndGetConnectionSequence() {
return this.theConnection.incrementAndGetConnectionSequence();
}
TcpSender getSender() {
@@ -176,5 +176,5 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI
this.realSender = sender != null;
return this.realSender;
}
}

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -16,6 +16,8 @@
package org.springframework.integration.ip.tcp.connection;
import org.springframework.context.SmartLifecycle;
/**
@@ -25,8 +27,8 @@ package org.springframework.integration.ip.tcp.connection;
* @since 2.0
*
*/
public interface ConnectionFactory {
public interface ConnectionFactory extends SmartLifecycle {
public TcpConnection getConnection() throws Exception;
TcpConnection getConnection() throws Exception;
}

62
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -26,9 +26,9 @@ import org.springframework.integration.Message;
/**
* An abstraction over {@link Socket} and {@link SocketChannel} that
* sends {@link Message} objects by serializing the payload
* and streaming it to the destination. Requires a {@link TcpListener}
* and streaming it to the destination. Requires a {@link TcpListener}
* to receive incoming messages.
*
*
* @author Gary Russell
* @since 2.0
*
@@ -38,115 +38,115 @@ public interface TcpConnection extends Runnable {
/**
* Closes the connection.
*/
public void close();
void close();
/**
* @return true if the connection is open.
*/
public boolean isOpen();
boolean isOpen();
/**
* Converts and sends the message.
* @param message The message
* @throws Exception
* @throws Exception
*/
public void send(Message<?> message) throws Exception;
void send(Message<?> message) throws Exception;
/**
* Uses the deserializer to obtain the message payload
* from the connection's input stream.
* @return The payload
* @throws Exception
* @throws Exception
*/
public Object getPayload() throws Exception;
Object getPayload() throws Exception;
/**
* @return the host name
*/
public String getHostName();
String getHostName();
/**
* @return the host address
*/
public String getHostAddress();
String getHostAddress();
/**
* @return the port
*/
public int getPort();
int getPort();
/**
* Sets the listener that will receive incoming Messages.
* Sets the listener that will receive incoming Messages.
* @param listener The listener
*/
public void registerListener(TcpListener listener);
void registerListener(TcpListener listener);
/**
* Registers a sender. Used on server side sockets so a
* sender can determine which connection to send a reply
* to.
* @param sender the sender
* @param sender the sender
*/
public void registerSender(TcpSender sender);
void registerSender(TcpSender sender);
/**
* @return a string uniquely representing a connection.
*/
public String getConnectionId();
String getConnectionId();
/**
* When true, the socket is used once and discarded.
* @param singleUse the singleUse
*/
public void setSingleUse(boolean singleUse);
void setSingleUse(boolean singleUse);
/**
*
*
* @return True if connection is used once.
*/
public boolean isSingleUse();
boolean isSingleUse();
/**
*
*
* @return True if connection is used once.
*/
public boolean isServer();
boolean isServer();
/**
* @param mapper the mapper
*/
public void setMapper(TcpMessageMapper mapper);
void setMapper(TcpMessageMapper mapper);
/**
*
*
* @return the deserializer
*/
public Deserializer<?> getDeserializer();
Deserializer<?> getDeserializer();
/**
* @param deserializer the deserializer to set
*/
public void setDeserializer(Deserializer<?> deserializer);
void setDeserializer(Deserializer<?> deserializer);
/**
*
*
* @return the serializer
*/
public Serializer<?> getSerializer();
Serializer<?> getSerializer();
/**
* @param serializer the serializer to set
*/
public void setSerializer(Serializer<?> serializer);
void setSerializer(Serializer<?> serializer);
/**
* @return this connection's listener
*/
public TcpListener getListener();
TcpListener getListener();
/**
* @return the next sequence number for a message received on this socket
*/
public long getConnectionSeq();
long incrementAndGetConnectionSequence();
}

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -22,6 +22,6 @@ package org.springframework.integration.ip.tcp.connection;
*/
public interface TcpConnectionInterceptor extends TcpConnection, TcpListener, TcpSender {
public void setTheConnection(TcpConnection connection);
void setTheConnection(TcpConnection connection);
}

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -31,6 +31,6 @@ public interface TcpConnectionInterceptorFactory {
*
* @return the TcpInterceptor
*/
public abstract TcpConnectionInterceptor getInterceptor();
abstract TcpConnectionInterceptor getInterceptor();
}

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -33,6 +33,6 @@ public interface TcpListener {
* @param message The message.
* @return true if the message was intercepted
*/
public abstract boolean onMessage(Message<?> message);
abstract boolean onMessage(Message<?> message);
}

40
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java
View File

@@ -41,20 +41,35 @@ public class TcpMessageMapper implements
OutboundMessageMapper<Object> {
private volatile String charset = "UTF-8";
private volatile boolean stringToBytes = true;
private volatile boolean applySequence = false;
@SuppressWarnings("deprecation")
public Message<Object> toMessage(TcpConnection connection) throws Exception {
Message<Object> message = null;
Object payload = connection.getPayload();
if (payload != null) {
message = MessageBuilder.withPayload(payload)
.setHeader(IpHeaders.HOSTNAME, connection.getHostName())
.setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress())
.setHeader(IpHeaders.REMOTE_PORT, connection.getPort())
.setHeader(IpHeaders.CONNECTION_ID, connection.getConnectionId())
.setHeader(IpHeaders.CONNECTION_SEQ, connection.getConnectionSeq())
.build();
String connectionId = connection.getConnectionId();
if (this.applySequence) {
message = MessageBuilder.withPayload(payload)
.setHeader(IpHeaders.HOSTNAME, connection.getHostName())
.setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress())
.setHeader(IpHeaders.REMOTE_PORT, connection.getPort())
.setHeader(IpHeaders.CONNECTION_ID, connectionId)
.setCorrelationId(connectionId)
.setSequenceNumber((int) connection.incrementAndGetConnectionSequence())
.build();
} else {
message = MessageBuilder.withPayload(payload)
.setHeader(IpHeaders.HOSTNAME, connection.getHostName())
.setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress())
.setHeader(IpHeaders.REMOTE_PORT, connection.getPort())
.setHeader(IpHeaders.CONNECTION_ID, connectionId)
.setHeader(IpHeaders.CONNECTION_SEQ, connection.incrementAndGetConnectionSequence())
.build();
}
}
return message;
@@ -112,4 +127,11 @@ public class TcpMessageMapper implements
this.stringToBytes = stringToBytes;
}
/**
* @param applySequence the applySequence to set
*/
public void setApplySequence(boolean applySequence) {
this.applySequence = applySequence;
}
}

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

@@ -45,18 +45,22 @@ public class TcpNetClientConnectionFactory extends
* reused for all requests while the connection remains open.
*/
public TcpConnection getConnection() throws Exception {
if (this.theConnection != null && this.theConnection.isOpen()) {
return this.theConnection;
this.checkActive();
TcpConnection theConnection = this.getTheConnection();
if (theConnection != null && theConnection.isOpen()) {
return theConnection;
}
logger.debug("Opening new socket connection to " + this.host + ":" + this.port);
Socket socket = createSocket(this.host, this.port);
if (logger.isDebugEnabled()) {
logger.debug("Opening new socket connection to " + this.getHost() + ":" + this.getPort());
}
Socket socket = createSocket(this.getHost(), this.getPort());
setSocketAttributes(socket);
TcpConnection connection = new TcpNetConnection(socket, false, this.isLookupHost());
connection = wrapConnection(connection);
initializeConnection(connection, socket);
this.getTaskExecutor().execute(connection);
if (!this.singleUse) {
this.theConnection = connection;
if (!this.isSingleUse()) {
this.setTheConnection(connection);
}
this.harvestClosedConnections();
return connection;
@@ -81,8 +85,4 @@ public class TcpNetClientConnectionFactory extends
public void run() {
}
public boolean isRunning() {
return this.active;
}
}

25
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.net.SocketTimeoutException;
import org.springframework.core.serializer.Serializer;
import org.springframework.core.serializer.Deserializer;
import org.springframework.integration.Message;
import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
@@ -64,13 +65,13 @@ public class TcpNetConnection extends AbstractTcpConnection {
@SuppressWarnings("unchecked")
public synchronized void send(Message<?> message) throws Exception {
Object object = mapper.fromMessage(message);
this.serializer.serialize(object, this.socket.getOutputStream());
Object object = this.getMapper().fromMessage(message);
((Serializer<Object>) this.getSerializer()).serialize(object, this.socket.getOutputStream());
this.afterSend(message);
}
public Object getPayload() throws Exception {
return this.deserializer.deserialize(this.socket.getInputStream());
return this.getDeserializer().deserialize(this.socket.getInputStream());
}
public int getPort() {
@@ -88,7 +89,9 @@ public class TcpNetConnection extends AbstractTcpConnection {
* a warning is logged.
*/
public void run() {
if (this.listener == null && !this.singleUse) {
boolean singleUse = this.isSingleUse();
TcpListener listener = this.getListener();
if (listener == null && !singleUse) {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
@@ -98,11 +101,11 @@ public class TcpNetConnection extends AbstractTcpConnection {
boolean intercepted = false;
while (okToRun) {
try {
message = this.mapper.toMessage(this);
message = this.getMapper().toMessage(this);
} catch (Exception e) {
this.closeConnection();
if (!(e instanceof SoftEndOfStreamException)) {
if (e instanceof SocketTimeoutException && this.singleUse) {
if (e instanceof SocketTimeoutException && singleUse) {
logger.debug("Closing single use socket after timeout");
} else {
if (this.noReadErrorOnClose) {
@@ -135,8 +138,8 @@ public class TcpNetConnection extends AbstractTcpConnection {
intercepted = listener.onMessage(message);
} catch (Exception e) {
if (e instanceof NoListenerException) {
if (this.singleUse) {
logger.debug("Closing single use socket after inbound message " + this.connectionId);
if (singleUse) {
logger.debug("Closing single use socket after inbound message " + this.getConnectionId());
this.closeConnection();
okToRun = false;
} else {
@@ -151,8 +154,8 @@ public class TcpNetConnection extends AbstractTcpConnection {
* side, and the data was not intercepted,
* or the server side has no outbound adapter registered
*/
if (this.singleUse && ((!this.server && !intercepted) || (this.server && this.sender == null))) {
logger.debug("Closing single use socket after inbound message " + this.connectionId);
if (singleUse && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) {
logger.debug("Closing single use socket after inbound message " + this.getConnectionId());
this.closeConnection();
okToRun = false;
}

38
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java
View File

@@ -33,7 +33,7 @@ import javax.net.ServerSocketFactory;
*/
public class TcpNetServerConnectionFactory extends AbstractServerConnectionFactory {
protected ServerSocket serverSocket;
private ServerSocket serverSocket;
/**
* Listens for incoming connections on the port.
@@ -52,20 +52,20 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
*/
public void run() {
ServerSocket theServerSocket = null;
if (this.listener == null) {
if (this.getListener() == null) {
logger.info("No listener bound to server connection factory; will not read; exiting...");
return;
}
try {
if (this.localAddress == null) {
theServerSocket = createServerSocket(this.port, this.poolSize, null);
if (this.getLocalAddress() == null) {
theServerSocket = createServerSocket(this.getPort(), this.getPoolSize(), null);
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
theServerSocket = createServerSocket(this.port, this.poolSize, whichNic);
InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
theServerSocket = createServerSocket(this.getPort(), this.getPoolSize(), whichNic);
}
this.serverSocket = theServerSocket;
this.listening = true;
logger.info("Listening on port " + this.port);
this.setListening(true);
logger.info("Listening on port " + this.getPort());
while (true) {
final Socket socket = serverSocket.accept();
logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress());
@@ -77,14 +77,14 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
this.harvestClosedConnections();
}
} catch (Exception e) {
this.listening = false;
this.setListening(false);
// don't log an error if we had a good socket once and now it's closed
if (e instanceof SocketException && theServerSocket != null) {
logger.warn("Server Socket closed");
} else if (this.active) {
} else if (this.isActive()) {
logger.error("Error on ServerSocket", e);
}
this.active = false;
this.setActive(false);
}
}
@@ -102,17 +102,13 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
protected ServerSocket createServerSocket(int port, int backlog, InetAddress whichNic) throws IOException {
if (whichNic == null) {
return ServerSocketFactory.getDefault().createServerSocket(port,
Math.abs(poolSize));
Math.abs(backlog));
} else {
return ServerSocketFactory.getDefault().createServerSocket(port,
Math.abs(poolSize), whichNic);
Math.abs(backlog), whichNic);
}
}
public boolean isRunning() {
return this.active;
}
public void close() {
if (this.serverSocket == null) {
return;
@@ -122,6 +118,12 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
} catch (IOException e) {}
this.serverSocket = null;
}
/**
* @return the serverSocket
*/
protected ServerSocket getServerSocket() {
return serverSocket;
}
}

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

@@ -36,13 +36,13 @@ import java.util.concurrent.LinkedBlockingQueue;
public class TcpNioClientConnectionFactory extends
AbstractClientConnectionFactory {
protected boolean usingDirectBuffers;
private boolean usingDirectBuffers;
private Selector selector;
protected Map<SocketChannel, TcpNioConnection> connections = new ConcurrentHashMap<SocketChannel, TcpNioConnection>();
private Map<SocketChannel, TcpNioConnection> connections = new ConcurrentHashMap<SocketChannel, TcpNioConnection>();
protected BlockingQueue<SocketChannel> newChannels = new LinkedBlockingQueue<SocketChannel>();
private BlockingQueue<SocketChannel> newChannels = new LinkedBlockingQueue<SocketChannel>();
/**
@@ -60,6 +60,7 @@ public class TcpNioClientConnectionFactory extends
* reused for all requests while the connection remains open.
*/
public TcpConnection getConnection() throws Exception {
this.checkActive();
int n = 0;
while (this.selector == null) {
try {
@@ -71,11 +72,13 @@ public class TcpNioClientConnectionFactory extends
throw new Exception("Factory failed to start");
}
}
if (this.theConnection != null && this.theConnection.isOpen()) {
return this.theConnection;
if (this.getTheConnection() != null && this.getTheConnection().isOpen()) {
return this.getTheConnection();
}
logger.debug("Opening new socket channel connection to " + this.host + ":" + this.port);
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.host, this.port));
if (logger.isDebugEnabled()) {
logger.debug("Opening new socket channel connection to " + this.getHost() + ":" + this.getPort());
}
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.getHost(), this.getPort()));
setSocketAttributes(socketChannel.socket());
TcpNioConnection connection = new TcpNioConnection(socketChannel, false, this.isLookupHost());
connection.setUsingDirectBuffers(this.usingDirectBuffers);
@@ -83,14 +86,14 @@ public class TcpNioClientConnectionFactory extends
TcpConnection wrappedConnection = wrapConnection(connection);
initializeConnection(wrappedConnection, socketChannel.socket());
socketChannel.configureBlocking(false);
if (this.soTimeout > 0) {
if (this.getSoTimeout() > 0) {
connection.setLastRead(System.currentTimeMillis());
}
this.connections.put(socketChannel, connection);
newChannels.add(socketChannel);
selector.wakeup();
if (!this.singleUse) {
this.theConnection = wrappedConnection;
if (!this.isSingleUse()) {
this.setTheConnection(wrappedConnection);
}
return wrappedConnection;
}
@@ -112,12 +115,14 @@ public class TcpNioClientConnectionFactory extends
}
public void run() {
logger.debug("Read selector running for connections to " + host + ":" + port);
if (logger.isDebugEnabled()) {
logger.debug("Read selector running for connections to " + this.getHost() + ":" + this.getPort());
}
try {
this.selector = Selector.open();
while (this.active) {
while (this.isActive()) {
SocketChannel newChannel;
int selectionCount = selector.select(this.soTimeout);
int selectionCount = selector.select(this.getSoTimeout());
while ((newChannel = newChannels.poll()) != null) {
newChannel.register(this.selector, SelectionKey.OP_READ, connections.get(newChannel));
}
@@ -125,13 +130,32 @@ public class TcpNioClientConnectionFactory extends
}
} catch (Exception e) {
logger.error("Exception in read selector thread", e);
this.active = false;
this.setActive(false);
}
if (logger.isDebugEnabled()) {
logger.debug("Read selector exiting for connections to " + this.getHost() + ":" + this.getPort());
}
logger.debug("Read selector exiting for connections to " + host + ":" + port);
}
public boolean isRunning() {
return this.active;
/**
* @return the usingDirectBuffers
*/
protected boolean isUsingDirectBuffers() {
return usingDirectBuffers;
}
/**
* @return the connections
*/
protected Map<SocketChannel, TcpNioConnection> getConnections() {
return connections;
}
/**
* @return the newChannels
*/
protected BlockingQueue<SocketChannel> getNewChannels() {
return newChannels;
}
}

75
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -30,6 +30,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.Message;
import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
@@ -43,27 +44,27 @@ import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamExceptio
public class TcpNioConnection extends AbstractTcpConnection {
private final SocketChannel socketChannel;
private volatile OutputStream channelOutputStream;
private volatile PipedOutputStream pipedOutputStream;
private volatile PipedInputStream pipedInputStream;
private volatile boolean usingDirectBuffers;
private volatile Executor taskExecutor;
private volatile ByteBuffer rawBuffer;
private volatile int maxMessageSize = 60 * 1024;
private volatile long lastRead;
private AtomicInteger executionControl = new AtomicInteger();
private volatile boolean writingToPipe;
/**
* Constructs a TcpNetConnection for the SocketChannel.
* @param socketChannel the socketChannel
@@ -77,7 +78,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream);
this.channelOutputStream = new ChannelOutputStream();
}
public void close() {
doClose();
}
@@ -100,21 +101,21 @@ public class TcpNioConnection extends AbstractTcpConnection {
@SuppressWarnings("unchecked")
public void send(Message<?> message) throws Exception {
synchronized(mapper) {
Object object = mapper.fromMessage(message);
this.serializer.serialize(object, this.channelOutputStream);
synchronized(this.getMapper()) {
Object object = this.getMapper().fromMessage(message);
((Serializer<Object>) this.getSerializer()).serialize(object, this.channelOutputStream);
this.afterSend(message);
}
}
public Object getPayload() throws Exception {
return this.deserializer.deserialize(pipedInputStream);
return this.getDeserializer().deserialize(pipedInputStream);
}
public int getPort() {
return this.socketChannel.socket().getPort();
}
/**
* Allocates a ByteBuffer of the requested length using normal or
* direct buffers, depending on the usingDirectBuffers field.
@@ -128,7 +129,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
return buffer;
}
/**
* If there is no listener, and this connection is not for single use,
* this method exits. When there is a listener, this method assembles
@@ -142,8 +143,8 @@ public class TcpNioConnection extends AbstractTcpConnection {
logger.trace(this.getConnectionId() + " Nio message assembler running...");
}
try {
if (this.listener == null && !this.singleUse) {
logger.debug("TcpListener exiting - no listener and not single use");
if (this.getListener() == null && !this.isSingleUse()) {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
try {
@@ -206,12 +207,12 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
Message<?> message = null;
try {
message = this.mapper.toMessage(this);
message = this.getMapper().toMessage(this);
} catch (Exception e) {
this.closeConnection();
if (e instanceof SocketTimeoutException && this.singleUse) {
if (e instanceof SocketTimeoutException && this.isSingleUse()) {
if (logger.isDebugEnabled()) {
logger.debug("Closing single use socket after timeout " + this.connectionId);
logger.debug("Closing single use socket after timeout " + this.getConnectionId());
}
} else {
if (!(e instanceof SoftEndOfStreamException)) {
@@ -219,7 +220,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
}
return null;
}
}
return message;
}
@@ -227,13 +228,13 @@ public class TcpNioConnection extends AbstractTcpConnection {
boolean intercepted = false;
try {
if (message != null) {
intercepted = listener.onMessage(message);
intercepted = getListener().onMessage(message);
}
} catch (Exception e) {
if (e instanceof NoListenerException) {
if (this.singleUse) {
if (this.isSingleUse()) {
if (logger.isDebugEnabled()) {
logger.debug("Closing single use channel after inbound message " + this.connectionId);
logger.debug("Closing single use channel after inbound message " + this.getConnectionId());
}
this.closeConnection();
}
@@ -246,12 +247,14 @@ public class TcpNioConnection extends AbstractTcpConnection {
* side, and the data was not intercepted,
* or the server side has no outbound adapter registered
*/
if (this.singleUse && ((!this.server && !intercepted) || (this.server && this.sender == null))) {
logger.debug("Closing single use cbannel after inbound message " + this.connectionId);
if (this.isSingleUse() && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) {
if (logger.isDebugEnabled()) {
logger.debug("Closing single use cbannel after inbound message " + this.getConnectionId());
}
this.closeConnection();
}
}
private void doRead() throws Exception {
if (this.rawBuffer == null) {
this.rawBuffer = allocate(maxMessageSize);
@@ -311,14 +314,14 @@ public class TcpNioConnection extends AbstractTcpConnection {
this.closeConnection();
}
}
/**
* Close the socket due to timeout.
*/
void timeout() {
this.closeConnection();
}
/**
*
* @param taskExecutor the taskExecutor to set
@@ -326,7 +329,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* If true, connection will attempt to use direct buffers where
* possible.
@@ -359,9 +362,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
class ChannelOutputStream extends OutputStream {
private Selector selector;
private int soTimeout;
@Override
public void write(int b) throws IOException {
byte[] bytes = new byte[1];
@@ -390,7 +393,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
ByteBuffer buffer = ByteBuffer.wrap(b);
doWrite(buffer);
}
private synchronized void doWrite(ByteBuffer buffer) throws IOException {
socketChannel.write(buffer);
int remaining = buffer.remaining();
@@ -414,5 +417,5 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
}
}

60
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java
View File

@@ -39,11 +39,11 @@ import java.util.Map;
*/
public class TcpNioServerConnectionFactory extends AbstractServerConnectionFactory {
protected ServerSocketChannel serverChannel;
private ServerSocketChannel serverChannel;
protected boolean usingDirectBuffers;
private boolean usingDirectBuffers;
protected Map<SocketChannel, TcpNioConnection> connections = new HashMap<SocketChannel, TcpNioConnection>();
private Map<SocketChannel, TcpNioConnection> connections = new HashMap<SocketChannel, TcpNioConnection>();
private Selector selector;
@@ -63,34 +63,35 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
* I/O errors on the server socket/channel are logged and the factory is stopped.
*/
public void run() {
if (this.listener == null) {
if (this.getListener() == null) {
logger.info("No listener bound to server connection factory; will not read; exiting...");
return;
}
try {
this.serverChannel = ServerSocketChannel.open();
logger.info("Listening on port " + this.port);
int port = this.getPort();
logger.info("Listening on port " + port);
this.serverChannel.configureBlocking(false);
if (this.localAddress == null) {
this.serverChannel.socket().bind(new InetSocketAddress(this.port),
Math.abs(this.poolSize));
if (this.getLocalAddress() == null) {
this.serverChannel.socket().bind(new InetSocketAddress(port),
Math.abs(this.getPoolSize()));
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.serverChannel.socket().bind(new InetSocketAddress(whichNic, this.port),
Math.abs(this.poolSize));
InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
this.serverChannel.socket().bind(new InetSocketAddress(whichNic, port),
Math.abs(this.getPoolSize()));
}
final Selector selector = Selector.open();
this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
this.listening = true;
this.setListening(true);
this.selector = selector;
doSelect(this.serverChannel, selector);
} catch (IOException e) {
this.close();
this.listening = false;
if (this.active) {
this.setListening(false);
if (this.isActive()) {
logger.error("Error on ServerSocketChannel", e);
this.active = false;
this.setActive(false);
}
}
}
@@ -111,8 +112,8 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
*/
private void doSelect(ServerSocketChannel server, final Selector selector)
throws IOException, ClosedChannelException, SocketException {
while (this.active) {
int selectionCount = selector.select(this.soTimeout);
while (this.isActive()) {
int selectionCount = selector.select(this.getSoTimeout());
this.processNioSelections(selectionCount, selector, server, this.connections);
}
}
@@ -155,10 +156,6 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
}
}
public boolean isRunning() {
return this.active;
}
public void close() {
if (this.selector != null) {
this.selector.wakeup();
@@ -175,6 +172,27 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
}
/**
* @return the serverChannel
*/
protected ServerSocketChannel getServerChannel() {
return serverChannel;
}
/**
* @return the usingDirectBuffers
*/
protected boolean isUsingDirectBuffers() {
return usingDirectBuffers;
}
/**
* @return the connections
*/
protected Map<SocketChannel, TcpNioConnection> getConnections() {
return connections;
}
}

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java
View File

@@ -37,7 +37,7 @@ public abstract class AbstractByteArraySerializer implements
protected int maxMessageSize = 2048;
protected Log logger = LogFactory.getLog(this.getClass());
protected final Log logger = LogFactory.getLog(this.getClass());
/**
* The maximum supported message size for this serializer.

20
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* 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.
@@ -32,7 +32,7 @@ import org.springframework.integration.MessagingException;
*/
public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAdapter {
protected String group;
private String group;
/**
@@ -61,25 +61,23 @@ public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAda
@Override
protected synchronized DatagramSocket getSocket() {
if (this.socket == null) {
if (this.getTheSocket() == null) {
try {
MulticastSocket socket = new MulticastSocket(this.port);
MulticastSocket socket = new MulticastSocket(this.getPort());
String localAddress = this.getLocalAddress();
if (localAddress != null) {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
InetAddress whichNic = InetAddress.getByName(localAddress);
socket.setInterface(whichNic);
}
socket.setSoTimeout(this.soTimeout);
if (this.soReceiveBufferSize > 0) {
socket.setReceiveBufferSize(this.soReceiveBufferSize);
}
this.setSocketAttributes(socket);
socket.joinGroup(InetAddress.getByName(this.group));
this.socket = socket;
this.setSocket(socket);
}
catch (IOException e) {
throw new MessagingException("failed to create DatagramSocket", e);
}
}
return this.socket;
return super.getSocket();
}
}

26
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -39,9 +39,9 @@ import org.springframework.integration.core.MessageHandler;
*/
public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler {
protected int timeToLive = -1;
private int timeToLive = -1;
protected String localAddress;
private String localAddress;
/**
* Constructs a MulticastSendingMessageHandler to send data to the multicast address/port.
@@ -99,20 +99,20 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
@Override
protected synchronized DatagramSocket getSocket() throws IOException {
if (this.socket == null) {
if (this.getTheSocket() == null) {
MulticastSocket socket;
if (acknowledge) {
if (this.isAcknowledge()) {
if (logger.isDebugEnabled()) {
logger.debug("Listening for acks on port: " + ackPort);
logger.debug("Listening for acks on port: " + this.getAckPort());
}
if (localAddress == null) {
socket = new MulticastSocket(this.ackPort);
socket = new MulticastSocket(this.getAckPort());
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
socket = new MulticastSocket(new InetSocketAddress(whichNic, this.ackPort));
socket = new MulticastSocket(new InetSocketAddress(whichNic, this.getAckPort()));
}
if (this.soReceiveBufferSize > 0) {
socket.setReceiveBufferSize(this.soReceiveBufferSize);
if (this.getSoReceiveBufferSize() > 0) {
socket.setReceiveBufferSize(this.getSoReceiveBufferSize());
}
} else {
socket = new MulticastSocket();
@@ -126,9 +126,9 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
NetworkInterface intfce = NetworkInterface.getByInetAddress(whichNic);
socket.setNetworkInterface(intfce);
}
this.socket = socket;
this.setSocket(socket);
}
return this.socket;
return this.getSocket();
}
@@ -137,7 +137,7 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
* @param minAcksForSuccess
*/
public void setMinAcksForSuccess(int minAcksForSuccess) {
this.ackCounter = minAcksForSuccess;
this.setAckCounter(minAcksForSuccess);
}
/**

74
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
View File

@@ -33,20 +33,20 @@ import org.springframework.integration.ip.AbstractInternetProtocolReceivingChann
import org.springframework.integration.ip.IpHeaders;
/**
* A channel adapter to receive incoming UDP packets. Packets can optionally be preceded by a
* A channel adapter to receive incoming UDP packets. Packets can optionally be preceded by a
* 4 byte length field, used to validate that all data was received. Packets may also contain
* information indicating an acknowledgment needs to be sent.
*
*
* @author Gary Russell
* @since 2.0
*/
public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolReceivingChannelAdapter {
protected volatile DatagramSocket socket;
private volatile DatagramSocket socket;
protected final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
private final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
protected volatile int soSendBufferSize = -1;
private volatile int soSendBufferSize = -1;
private static Pattern addressPattern = Pattern.compile("([^:]*):([0-9]*)");
@@ -75,15 +75,15 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
public void run() {
if (logger.isDebugEnabled()) {
logger.debug("UDP Receiver running on port:" + port);
logger.debug("UDP Receiver running on port:" + this.getPort());
}
checkTaskExecutor("UDP-Incoming-Msg-Handler");
listening = true;
this.setListening(true);
// Do as little as possible here so we can loop around and catch the next packet.
// Just schedule the packet for processing.
while (this.active) {
while (this.isActive()) {
try {
asyncSendMessage(receive());
}
@@ -91,7 +91,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
// continue
}
catch (SocketException e) {
doStop();
doStop();
}
catch (Exception e) {
if (e instanceof MessagingException) {
@@ -100,7 +100,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
throw new MessagingException("failed to receive DatagramPacket", e);
}
}
listening = false;
this.setListening(false);
}
protected void sendAck(Message<byte[]> message) {
@@ -133,7 +133,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
}
protected boolean asyncSendMessage(final DatagramPacket packet) {
this.taskExecutor.execute(new Runnable(){
this.getTaskExecutor().execute(new Runnable(){
public void run() {
Message<byte[]> message = null;
try {
@@ -156,26 +156,36 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
protected DatagramPacket receive() throws Exception {
DatagramSocket socket = this.getSocket();
final byte[] buffer = new byte[this.receiveBufferSize];
final byte[] buffer = new byte[this.getReceiveBufferSize()];
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
socket.receive(packet);
return packet;
}
/**
* @param socket the socket to set
*/
public void setSocket(DatagramSocket socket) {
this.socket = socket;
}
protected DatagramSocket getTheSocket() {
return this.socket;
}
protected synchronized DatagramSocket getSocket() {
if (this.socket == null) {
try {
DatagramSocket socket = null;
String localAddress = this.getLocalAddress();
if (localAddress == null) {
this.socket = new DatagramSocket(this.port);
socket = new DatagramSocket(this.getPort());
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.socket = new DatagramSocket(this.port, whichNic);
}
this.socket.setSoTimeout(this.soTimeout);
if (this.soReceiveBufferSize > 0) {
this.socket.setReceiveBufferSize(this.soReceiveBufferSize);
InetAddress whichNic = InetAddress.getByName(localAddress);
socket = new DatagramSocket(this.getPort(), whichNic);
}
setSocketAttributes(socket);
this.socket = socket;
}
catch (IOException e) {
throw new MessagingException("failed to create DatagramSocket", e);
@@ -184,12 +194,28 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
return this.socket;
}
/**
* Sets timeout and receive buffer size
*
* @param socket
* @throws SocketException
*/
protected void setSocketAttributes(DatagramSocket socket)
throws SocketException {
socket.setSoTimeout(this.getSoTimeout());
int soReceiveBufferSize = this.getSoReceiveBufferSize();
if (soReceiveBufferSize > 0) {
socket.setReceiveBufferSize(soReceiveBufferSize);
}
}
@Override
protected void doStop() {
super.doStop();
try {
this.socket.close();
DatagramSocket socket = this.socket;
this.socket = null;
socket.close();
}
catch (Exception e) {
// ignore
@@ -199,11 +225,11 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
public void setSoSendBufferSize(int soSendBufferSize) {
this.soSendBufferSize = soSendBufferSize;
}
public void setLookupHost(boolean lookupHost) {
this.mapper.setLookupHost(lookupHost);
}
public String getComponentType(){
return "ip:udp-inbound-channel-adapter";
}

92
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2001-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.
@@ -20,7 +20,6 @@ import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.util.Collections;
import java.util.HashMap;
@@ -37,12 +36,11 @@ import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler;
import org.springframework.util.Assert;
/**
* A {@link MessageHandler} implementation that maps a Message into
* A {@link org.springframework.integration.core.MessageHandler} implementation that maps a Message into
* a UDP datagram packet and sends that to the specified host and port.
*
* Messages can be basic, with no support for reliability, can be prefixed
@@ -55,39 +53,39 @@ import org.springframework.util.Assert;
public class UnicastSendingMessageHandler extends
AbstractInternetProtocolSendingMessageHandler implements Runnable{
protected final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
private final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
protected volatile DatagramSocket socket;
private volatile DatagramSocket socket;
/**
* If true adds headers to instruct receiving adapter to return an ack.
*/
protected volatile boolean waitForAck = false;
protected volatile boolean acknowledge = false;
private volatile boolean waitForAck = false;
protected volatile int ackPort;
private volatile boolean acknowledge = false;
protected volatile int ackTimeout = 5000;
private volatile int ackPort;
protected volatile int ackCounter = 1;
private volatile int ackTimeout = 5000;
protected volatile Map<String, CountDownLatch> ackControl = Collections
private volatile int ackCounter = 1;
private volatile Map<String, CountDownLatch> ackControl = Collections
.synchronizedMap(new HashMap<String, CountDownLatch>());
protected volatile Exception fatalException;
private volatile Exception fatalException;
protected int soReceiveBufferSize = -1;
private volatile int soReceiveBufferSize = -1;
protected String localAddress;
private CountDownLatch ackLatch;
private volatile String localAddress;
private boolean ackThreadRunning;
private volatile CountDownLatch ackLatch;
private volatile boolean ackThreadRunning;
private volatile Executor taskExecutor;
protected volatile Executor taskExecutor;
/**
* Basic constructor; no reliability; no acknowledgment.
* @param host Destination host.
@@ -153,7 +151,7 @@ public class UnicastSendingMessageHandler extends
ackTimeout);
}
protected void setReliabilityAttributes(boolean lengthCheck,
protected final void setReliabilityAttributes(boolean lengthCheck,
boolean acknowledge, String ackHost, int ackPort, int ackTimeout) {
this.mapper.setLengthCheck(lengthCheck);
this.waitForAck = acknowledge;
@@ -248,10 +246,18 @@ public class UnicastSendingMessageHandler extends
protected void send(DatagramPacket packet) throws Exception {
DatagramSocket socket = this.getSocket();
packet.setSocketAddress(this.destinationAddress);
packet.setSocketAddress(this.getDestinationAddress());
socket.send(packet);
}
protected void setSocket(DatagramSocket socket) {
this.socket = socket;
}
protected DatagramSocket getTheSocket() {
return this.socket;
}
protected synchronized DatagramSocket getSocket() throws IOException {
if (this.socket == null) {
if (acknowledge) {
@@ -276,11 +282,11 @@ public class UnicastSendingMessageHandler extends
}
protected void setSocketAttributes(DatagramSocket socket) throws SocketException {
if (this.soTimeout >= 0) {
socket.setSoTimeout(this.soTimeout);
if (this.getSoTimeout() >= 0) {
socket.setSoTimeout(this.getSoTimeout());
}
if (this.soSendBufferSize > 0) {
socket.setSendBufferSize(this.soSendBufferSize);
if (this.getSoSendBufferSize() > 0) {
socket.setSendBufferSize(this.getSoSendBufferSize());
}
}
@@ -334,7 +340,7 @@ public class UnicastSendingMessageHandler extends
}
/**
* @see Socket#setReceiveBufferSize(int)
* @see java.net.Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
*/
public void setSoReceiveBufferSize(int size) {
@@ -348,8 +354,36 @@ public class UnicastSendingMessageHandler extends
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* @param ackCounter the ackCounter to set
*/
public void setAckCounter(int ackCounter) {
this.ackCounter = ackCounter;
}
public String getComponentType(){
return "ip:udp-outbound-channel-adapter";
}
/**
* @return the acknowledge
*/
public boolean isAcknowledge() {
return acknowledge;
}
/**
* @return the ackPort
*/
public int getAckPort() {
return ackPort;
}
/**
* @return the soReceiveBufferSize
*/
public int getSoReceiveBufferSize() {
return soReceiveBufferSize;
}
}

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

@@ -65,7 +65,7 @@ its configuration specifies the number of threads.
<xsd:attribute name="lookup-host" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
Whether or not to do a DNS reverse-lookup on the remote ip address to insert the host name into the
Whether or not to do a DNS reverse-lookup on the remote ip address to insert the host name into the
message headers (ip_hostName). Default "true".
</xsd:documentation>
</xsd:annotation>
@@ -115,183 +115,199 @@ task executors such as a WorkManagerTaskExecutor.
<xsd:element name="tcp-inbound-channel-adapter">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an inbound adapter. If the connection factory has a type 'server',
the factory is 'owned' by this adapter. If it has a type 'client', it is owned by an outbound channel
adapter and this adapter will receive any incoming messages on the connection created by the outbound
adapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
If a (synchronous) downstream exception is thrown and an "error-channel" is specified,
the MessagingException will be sent to this channel. Otherwise, any such exception
will simply be logged by the channel adapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:complexContent>
<xsd:extension base="smartLifeCycleType">
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an inbound adapter. If the connection factory has a type 'server',
the factory is 'owned' by this adapter. If it has a type 'client', it is owned by an outbound channel
adapter and this adapter will receive any incoming messages on the connection created by the outbound
adapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
If a (synchronous) downstream exception is thrown and an "error-channel" is specified,
the MessagingException will be sent to this channel. Otherwise, any such exception
will simply be logged by the channel adapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="tcp-outbound-channel-adapter">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an outbound adapter. If the connection factory has a type 'client',
the factory is 'owned' by this adapter. If it has a type 'server', it is owned by an inbound channel
adapter and this adapter will attempt to correlate messages to the connection on which an original
inbound message was received.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:complexContent>
<xsd:extension base="smartLifeCycleType">
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an outbound adapter. If the connection factory has a type 'client',
the factory is 'owned' by this adapter. If it has a type 'server', it is owned by an inbound channel
adapter and this adapter will attempt to correlate messages to the connection on which an original
inbound message was received.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="tcp-inbound-gateway">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an inbound adapter. The connection factory must be of type 'server'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-timeout" type="xsd:string"/>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
If a (synchronous) downstream exception is thrown and an "error-channel" is specified,
the MessagingException will be sent to this channel and the ultimate response
of the error flow will be returned as a response by the gateway. If no
"error-channel" is specified, any such exception
will simply be logged by the gateway. In such a situation, no response is sent
to the client.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:complexContent>
<xsd:extension base="smartLifeCycleType">
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an inbound adapter. The connection factory must be of type 'server'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-timeout" type="xsd:string"/>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
If a (synchronous) downstream exception is thrown and an "error-channel" is specified,
the MessagingException will be sent to this channel and the ultimate response
of the error flow will be returned as a response by the gateway. If no
"error-channel" is specified, any such exception
will simply be logged by the gateway. In such a situation, no response is sent
to the client.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="tcp-outbound-gateway">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an outbound adapter. The connection factory must be of 'client'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-timeout" type="xsd:string"/>
<xsd:attribute name="reply-timeout" type="xsd:string"/>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:complexContent>
<xsd:extension base="smartLifeCycleType">
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an outbound adapter. The connection factory must be of 'client'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-timeout" type="xsd:string"/>
<xsd:attribute name="reply-timeout" type="xsd:string"/>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
@@ -317,7 +333,7 @@ connection request.
<xsd:attribute name="host" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The host to which a client connection factory will connect.
The host to which a client connection factory will connect.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -326,14 +342,14 @@ The host to which a client connection factory will connect.
<xsd:documentation>
For client factories, the port to which a client connection factory will connect.
For server factories, the port on which the factory will listen for incoming
connections.
connections.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="using-nio" type="xsd:string" default="false">
<xsd:annotation>
<xsd:documentation>
If true, the factory will use java.nio.channel.SocketChannel for communication;
If true, the factory will use java.nio.channel.SocketChannel for communication;
for a large number of connections on the server side, this can provide better
performance and may use fewer threads.
</xsd:documentation>
@@ -350,10 +366,10 @@ performance and may use fewer threads.
<xsd:annotation>
<xsd:documentation>
If true, instructs the factory to use direct buffers if possible; only applies if
using-nio is true. Refer to ByteBuffer javadocs for more information.
using-nio is true. Refer to ByteBuffer javadocs for more information.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:attribute>
<xsd:attribute name="single-use" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
@@ -363,7 +379,7 @@ be closed after a message is received. For outbound adapters where there is
no inbound adapter sharing the factory, or for inbound adapters where an
outbound adapter shares the factory, the connection will be closed after
so-timeout milliseconds. For outbound adapters where an inbound adapter shares
the factory, the connection will be closed after a response is received.
the factory, the connection will be closed after a response is received.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -377,7 +393,7 @@ the factory, the connection will be closed after a response is received.
<xsd:documentation>
A Serializer that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfSerializer. Serializer and Deserializer
would normally be the same but this is not required.
would normally be the same but this is not required.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -391,7 +407,7 @@ would normally be the same but this is not required.
<xsd:documentation>
A Deserializer that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfSerializer. Serializer and Deserializer
would normally be the same but this is not required.
would normally be the same but this is not required.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -420,7 +436,7 @@ its configuration specifies the number of threads.
<xsd:attribute name="lookup-host" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
Whether or not to do a DNS reverse-lookup on the remote ip address to insert the host name into the
Whether or not to do a DNS reverse-lookup on the remote ip address to insert the host name into the
message headers (ip_connectionId, ip_hostName). Default "true".
</xsd:documentation>
</xsd:annotation>
@@ -434,6 +450,14 @@ message headers (ip_connectionId, ip_hostName). Default "true".
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="apply-sequence" type="xsd:string" default="false">
<xsd:annotation>
<xsd:documentation>
When set to "true", adds sequenceNumber and correlationId headers to messages originating from
connections created by this factory. Facilitates resequencing if necessary. Default "false".
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
@@ -480,4 +504,23 @@ apply to TCP outbound adapters and gateways.
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="smartLifeCycleType">
<xsd:attribute name="auto-startup" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Boolean value indicating whether this endpoint should start automatically.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="phase" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
The lifecycle phase within which this endpoint should start and stop.
The lower the value the earlier this endpoint will start and the later it will stop. The
default is 0. Values can be negative. See SmartLifeCycle.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:schema>

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

@@ -12,15 +12,16 @@
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketTestUtils" />
<int:channel id="udpChannel" />
<int:channel id="tcpChannel" />
<int:channel id="replyChannel" />
<task:executor id="externalTE" pool-size="10"/>
<ip:udp-inbound-channel-adapter id="testInUdp"
channel="udpChannel"
check-length="true"
@@ -48,22 +49,25 @@
so-receive-buffer-size="30"
so-send-buffer-size="31"
so-timeout="32"
local-address="127.0.0.1"
local-address="127.0.0.1"
/>
<ip:tcp-connection-factory id="cfS1"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(5200)}"
lookup-host="false"
apply-sequence="true"
/>
<ip:tcp-inbound-channel-adapter id="testInTcp"
channel="tcpChannel"
error-channel="errorChannel"
connection-factory="cfS1"
auto-startup="false"
phase="124"
/>
<ip:udp-outbound-channel-adapter id="testOutUdp"
<ip:udp-outbound-channel-adapter id="testOutUdp"
ack-host="somehost"
ack-port="#{tcpIpUtils.findAvailableUdpSocket(5300)}"
ack-timeout="51"
@@ -76,12 +80,12 @@
so-receive-buffer-size="52"
so-send-buffer-size="53"
so-timeout="54"
local-address="127.0.0.1"
local-address="127.0.0.1"
task-executor="externalTE"
order="23"
order="23"
/>
<ip:udp-outbound-channel-adapter id="testOutUdpiMulticast"
<ip:udp-outbound-channel-adapter id="testOutUdpiMulticast"
ack-host="somehost"
ack-port="#{tcpIpUtils.findAvailableUdpSocket(5500)}"
ack-timeout="51"
@@ -104,45 +108,50 @@
port="#{tcpIpUtils.findAvailableServerSocket(5700)}"
host="localhost"
lookup-host="false"
apply-sequence="false"
/>
<ip:tcp-outbound-channel-adapter id="testOutTcpNio"
<ip:tcp-outbound-channel-adapter id="testOutTcpNio"
channel="tcpChannel"
connection-factory="cfC1"
order="35"
auto-startup="false"
phase="125"
/>
<ip:tcp-connection-factory id="cfS2"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(5800)}"
/>
<ip:tcp-inbound-gateway id="inGateway1"
request-channel="tcpChannel"
reply-channel="replyChannel"
error-channel="errorChannel"
connection-factory="cfS2"
reply-timeout="456"
/>
auto-startup="false"
phase="126"
/>
<ip:tcp-connection-factory id="cfS3"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(5850)}"
/>
<ip:tcp-inbound-gateway id="inGateway2"
request-channel="tcpChannel"
reply-channel="replyChannel"
connection-factory="cfS3"
reply-timeout="456"
/>
/>
<ip:tcp-connection-factory id="cfC2"
type="client"
port="#{tcpIpUtils.findAvailableServerSocket(5900)}"
host="localhost"
/>
<ip:tcp-outbound-gateway id="outGateway"
request-channel="tcpChannel"
reply-channel="replyChannel"
@@ -150,7 +159,9 @@
request-timeout="234"
reply-timeout="567"
order="24"
/>
auto-startup="false"
phase="127"
/>
<ip:tcp-connection-factory
id="client1"
@@ -169,15 +180,15 @@
using-nio="#{props['use.nio']}"
single-use="true"
task-executor="externalTE"
pool-size="321"
pool-size="321"
using-direct-buffers="true"
interceptor-factory-chain="interceptors"
/>
<util:properties id="props">
<prop key="use.nio">true</prop>
</util:properties>
<ip:tcp-connection-factory
id="server1"
type="server"
@@ -199,7 +210,7 @@
using-direct-buffers="true"
interceptor-factory-chain="interceptors"
/>
<ip:tcp-connection-factory
id="client2"
type="client"
@@ -217,10 +228,10 @@
using-nio="false"
single-use="true"
task-executor="externalTE"
pool-size="321"
pool-size="321"
interceptor-factory-chain="interceptors"
/>
<ip:tcp-connection-factory
id="server2"
type="server"
@@ -241,32 +252,29 @@
pool-size="123"
interceptor-factory-chain="interceptors"
/>
<bean id="interceptors" class="org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain" />
<bean id="defaultSerializer" class="org.springframework.core.serializer.DefaultSerializer" />
<bean id="defaultDeserializer" class="org.springframework.core.serializer.DefaultDeserializer" />
<ip:tcp-outbound-channel-adapter id="tcpNewOut1"
channel="tcpChannel"
connection-factory="client1"
connection-factory="client1"
order="25"/>
<ip:tcp-outbound-channel-adapter id="tcpNewOut2"
channel="tcpChannel"
connection-factory="server1"
connection-factory="server1"
order="15"/>
<ip:tcp-inbound-channel-adapter id="tcpNewIn1"
channel="tcpChannel"
connection-factory="client1" />
<ip:tcp-inbound-channel-adapter id="tcpNewIn2"
channel="tcpChannel"
connection-factory="server1" />
</beans>
</beans>

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

@@ -40,6 +40,7 @@ import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.ip.tcp.TcpInboundGateway;
import org.springframework.integration.ip.tcp.TcpOutboundGateway;
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
@@ -68,7 +69,7 @@ public class ParserUnitTests {
@Autowired
ApplicationContext ctx;
@Autowired
@Qualifier(value="testInUdp")
UnicastReceivingChannelAdapter udpIn;
@@ -80,7 +81,7 @@ public class ParserUnitTests {
@Autowired
@Qualifier(value="testInTcp")
TcpReceivingChannelAdapter tcpIn;
@Autowired
@Qualifier(value="org.springframework.integration.ip.udp.UnicastSendingMessageHandler#0")
UnicastSendingMessageHandler udpOut;
@@ -93,6 +94,9 @@ public class ParserUnitTests {
@Qualifier(value="org.springframework.integration.ip.tcp.TcpSendingMessageHandler#0")
TcpSendingMessageHandler tcpOut;
@Autowired
EventDrivenConsumer testOutTcpNio;
@Autowired
@Qualifier(value="inGateway1")
TcpInboundGateway tcpInboundGateway1;
@@ -104,11 +108,14 @@ public class ParserUnitTests {
@Autowired
@Qualifier(value="org.springframework.integration.ip.tcp.TcpOutboundGateway#0")
TcpOutboundGateway tcpOutboundGateway;
@Autowired
EventDrivenConsumer outGateway;
@Autowired
@Qualifier(value="externalTE")
TaskExecutor taskExecutor;
@Autowired
AbstractConnectionFactory client1;
@@ -158,7 +165,7 @@ public class ParserUnitTests {
@Autowired
private MessageChannel errorChannel;
@Autowired
private DirectChannel udpChannel;
@@ -183,7 +190,7 @@ public class ParserUnitTests {
DirectFieldAccessor mapperAccessor = new DirectFieldAccessor(mapper);
assertFalse((Boolean)mapperAccessor.getPropertyValue("lookupHost"));
}
@Test
public void testInUdpMulticast() {
DirectFieldAccessor dfa = new DirectFieldAccessor(udpInMulticast);
@@ -195,13 +202,13 @@ public class ParserUnitTests {
assertEquals(31, dfa.getPropertyValue("soSendBufferSize"));
assertEquals(32, dfa.getPropertyValue("soTimeout"));
assertEquals("127.0.0.1", dfa.getPropertyValue("localAddress"));
assertNotSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
assertNotSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
assertNull(dfa.getPropertyValue("errorChannel"));
DatagramPacketMessageMapper mapper = (DatagramPacketMessageMapper) dfa.getPropertyValue("mapper");
DirectFieldAccessor mapperAccessor = new DirectFieldAccessor(mapper);
assertTrue((Boolean)mapperAccessor.getPropertyValue("lookupHost"));
}
@Test
public void testInTcp() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpIn);
@@ -210,8 +217,12 @@ public class ParserUnitTests {
assertEquals("ip:tcp-inbound-channel-adapter", tcpIn.getComponentType());
assertEquals(errorChannel, dfa.getPropertyValue("errorChannel"));
assertFalse(cfS1.isLookupHost());
assertFalse(tcpIn.isAutoStartup());
assertEquals(124, tcpIn.getPhase());
assertTrue((Boolean) TestUtils.getPropertyValue(
TestUtils.getPropertyValue(cfS1, "mapper"), "applySequence"));
}
@Test
public void testOutUdp() {
DirectFieldAccessor dfa = new DirectFieldAccessor(udpOut);
@@ -235,7 +246,7 @@ public class ParserUnitTests {
assertEquals("testOutUdp",udpOut.getComponentName());
assertEquals("ip:udp-outbound-channel-adapter", udpOut.getComponentType());
}
@Test
public void testOutUdpMulticast() {
DirectFieldAccessor dfa = new DirectFieldAccessor(udpOutMulticast);
@@ -254,9 +265,9 @@ public class ParserUnitTests {
assertEquals(53, dfa.getPropertyValue("soSendBufferSize"));
assertEquals(54, dfa.getPropertyValue("soTimeout"));
assertEquals(55, dfa.getPropertyValue("timeToLive"));
assertEquals(12, dfa.getPropertyValue("order"));
assertEquals(12, dfa.getPropertyValue("order"));
}
@Test
public void testUdpOrder() {
@SuppressWarnings("unchecked")
@@ -277,6 +288,10 @@ public class ParserUnitTests {
assertEquals("ip:tcp-outbound-channel-adapter", tcpOut.getComponentType());
assertFalse(cfC1.isLookupHost());
assertEquals(35, dfa.getPropertyValue("order"));
assertFalse(tcpOut.isAutoStartup());
assertEquals(125, tcpOut.getPhase());
assertFalse((Boolean) TestUtils.getPropertyValue(
TestUtils.getPropertyValue(cfC1, "mapper"), "applySequence"));
}
@Test
@@ -288,6 +303,10 @@ public class ParserUnitTests {
assertEquals("ip:tcp-inbound-gateway", tcpInboundGateway1.getComponentType());
assertEquals(errorChannel, dfa.getPropertyValue("errorChannel"));
assertTrue(cfS2.isLookupHost());
assertFalse(tcpInboundGateway1.isAutoStartup());
assertEquals(126, tcpInboundGateway1.getPhase());
assertFalse((Boolean) TestUtils.getPropertyValue(
TestUtils.getPropertyValue(cfS2, "mapper"), "applySequence"));
}
@Test
@@ -309,7 +328,9 @@ public class ParserUnitTests {
assertEquals("outGateway",tcpOutboundGateway.getComponentName());
assertEquals("ip:tcp-outbound-gateway", tcpOutboundGateway.getComponentType());
assertTrue(cfC2.isLookupHost());
assertEquals(24, dfa.getPropertyValue("order"));
assertEquals(24, dfa.getPropertyValue("order"));
assertFalse(tcpOutboundGateway.isAutoStartup());
assertEquals(127, tcpOutboundGateway.getPhase());
}
@Test
@@ -350,7 +371,7 @@ public class ParserUnitTests {
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
assertEquals(123, dfa.getPropertyValue("poolSize"));
assertEquals(true, dfa.getPropertyValue("usingDirectBuffers"));
assertNotNull(dfa.getPropertyValue("interceptorFactoryChain"));
assertNotNull(dfa.getPropertyValue("interceptorFactoryChain"));
}
@Test
@@ -389,7 +410,7 @@ public class ParserUnitTests {
assertEquals(true, dfa.getPropertyValue("singleUse"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
assertEquals(123, dfa.getPropertyValue("poolSize"));
assertNotNull(dfa.getPropertyValue("interceptorFactoryChain"));
assertNotNull(dfa.getPropertyValue("interceptorFactoryChain"));
}
@Test
@@ -398,36 +419,38 @@ public class ParserUnitTests {
assertSame(client1, dfa.getPropertyValue("clientConnectionFactory"));
assertEquals(25, dfa.getPropertyValue("order"));
}
@Test
public void testNewOut2() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewOut2);
assertSame(server1, dfa.getPropertyValue("serverConnectionFactory"));
assertEquals(15, dfa.getPropertyValue("order"));
assertEquals(15, dfa.getPropertyValue("order"));
}
@Test
public void testNewIn1() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewIn1);
assertSame(client1, dfa.getPropertyValue("clientConnectionFactory"));
assertNull(dfa.getPropertyValue("errorChannel"));
}
@Test
public void testNewIn2() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewIn2);
assertSame(server1, dfa.getPropertyValue("serverConnectionFactory"));
}
@Test
public void testtCPOrder() {
this.outGateway.start();
this.testOutTcpNio.start();
@SuppressWarnings("unchecked")
Set<MessageHandler> handlers = (Set<MessageHandler>) TestUtils
.getPropertyValue(
TestUtils.getPropertyValue(this.tcpChannel, "dispatcher"),
"handlers");
Iterator<MessageHandler> iterator = handlers.iterator();
assertSame(this.tcpNewOut2, iterator.next()); //15
assertSame(this.tcpNewOut2, iterator.next()); //15
assertSame(this.tcpOutboundGateway, iterator.next()); //24
assertSame(this.tcpNewOut1, iterator.next()); //25
assertSame(this.tcpOut, iterator.next()); //35

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

@@ -0,0 +1,28 @@
<?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: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" />
<ip:tcp-connection-factory id="cfS1"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(15200)}"
lookup-host="false"
/>
<ip:tcp-inbound-channel-adapter id="tcpNetIn"
channel="tcpChannel1"
error-channel="errorChannel"
connection-factory="cfS1"
auto-startup="false"
/>
<int:channel id="tcpChannel1">
<int:queue/>
</int:channel>
</beans>

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

@@ -0,0 +1,76 @@
/*
* 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.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.fail;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
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 AutoStartTests {
@Autowired
AbstractServerConnectionFactory cfS1;
@Autowired
TcpReceivingChannelAdapter tcpNetIn;
@Test
public void testNetIn() throws Exception {
assertFalse(cfS1.isAutoStartup());
DirectFieldAccessor dfa = new DirectFieldAccessor(cfS1);
assertNull(dfa.getPropertyValue("serverSocket"));
startAndStop();
assertNull(dfa.getPropertyValue("serverSocket"));
startAndStop();
assertNull(dfa.getPropertyValue("serverSocket"));
}
/**
* @throws InterruptedException
*/
private void startAndStop() throws InterruptedException {
tcpNetIn.start();
int n = 0;
while (!cfS1.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to start listening");
}
}
tcpNetIn.stop();
while (cfS1.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to stop listening");
}
}
}
}

3
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java
View File

@@ -86,6 +86,7 @@ public class ConnectionToConnectionTests {
throw new Exception("Failed to listen");
}
}
client.start();
for (int i = 0; i < 100; i++) {
TcpConnection connection = client.getConnection();
connection.send(MessageBuilder.withPayload("Test").build());
@@ -105,6 +106,7 @@ public class ConnectionToConnectionTests {
ByteArrayRawSerializer serializer = new ByteArrayRawSerializer();
client.setSerializer(serializer);
server.setDeserializer(serializer);
client.start();
TcpConnection connection = client.getConnection();
connection.send(MessageBuilder.withPayload("Test").build());
Message<?> message = serverSideChannel.receive(10000);
@@ -119,6 +121,7 @@ public class ConnectionToConnectionTests {
@Test
public void testLookup() throws Exception {
client.start();
TcpConnection connection = client.getConnection();
assertFalse(connection.getConnectionId().contains("localhost"));
connection.close();

3
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java
View File

@@ -133,6 +133,7 @@ public class TcpConfigOutboundGatewayTests {
@Test
public void testOutboundStxEtx() throws Exception {
TcpOutboundGateway gateway = new TcpOutboundGateway();
stxEtxClient.start();
gateway.setConnectionFactory(stxEtxClient);
waitListening(inboundGatewayStxEtx);
Message<String> message = MessageBuilder.withPayload("test").build();
@@ -144,6 +145,7 @@ public class TcpConfigOutboundGatewayTests {
@Test
public void testOutboundSerialized() throws Exception {
TcpOutboundGateway gateway = new TcpOutboundGateway();
javaSerialClient.start();
gateway.setConnectionFactory(javaSerialClient);
waitListening(inboundGatewaySerialized);
Message<String> message = MessageBuilder.withPayload("test").build();
@@ -155,6 +157,7 @@ public class TcpConfigOutboundGatewayTests {
@Test
public void testOutboundLength() throws Exception {
TcpOutboundGateway gateway = new TcpOutboundGateway();
lengthHeaderClient.start();
gateway.setConnectionFactory(lengthHeaderClient);
waitListening(inboundGatewayLength);
Message<String> message = MessageBuilder.withPayload("test").build();

77
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.ip.tcp.connection;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -26,8 +27,6 @@ import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.connection.TcpConnection;
import org.springframework.integration.ip.tcp.connection.TcpMessageMapper;
import org.springframework.integration.support.MessageBuilder;
/**
@@ -41,10 +40,6 @@ public class TcpMessageMapperTests {
*/
private static final String TEST_PAYLOAD = "abcdefghijkl";
/**
* Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#toMessage(org.springframework.integration.ip.tcp.SocketReader)}.
* Tests segmented reads into the payload and verifies reassembly.
*/
@Test
public void testToMessage() throws Exception {
@@ -64,10 +59,7 @@ public class TcpMessageMapperTests {
.getHeaders().get(IpHeaders.REMOTE_PORT));
}
/**
* Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#toMessage(org.springframework.integration.ip.tcp.SocketReader)}.
* Tests segmented reads into the payload and verifies reassembly.
*/
@SuppressWarnings("deprecation")
@Test
public void testToMessageSequence() throws Exception {
@@ -119,10 +111,67 @@ public class TcpMessageMapperTests {
.getHeaders().get(IpHeaders.CONNECTION_SEQ));
}
/**
* Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#fromMessage(org.springframework.integration.Message)}.
* @throws Exception
*/
@SuppressWarnings("deprecation")
@Test
public void testToMessageSequenceNew() throws Exception {
TcpMessageMapper mapper = new TcpMessageMapper();
mapper.setApplySequence(true);
Socket socket = SocketFactory.getDefault().createSocket();
TcpConnection connection = new AbstractTcpConnection(socket, false, false) {
public void run() {
}
public void send(Message<?> message) throws Exception {
}
public boolean isOpen() {
return false;
}
public int getPort() {
return 1234;
}
public Object getPayload() throws Exception {
return TEST_PAYLOAD.getBytes();
}
public String getHostName() {
return "MyHost";
}
public String getHostAddress() {
return "1.1.1.1";
}
public String getConnectionId() {
return "anId";
}
};
Message<Object> message = mapper.toMessage(connection);
assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload()));
assertEquals("MyHost", message
.getHeaders().get(IpHeaders.HOSTNAME));
assertEquals("1.1.1.1", message
.getHeaders().get(IpHeaders.IP_ADDRESS));
assertEquals(1234, message
.getHeaders().get(IpHeaders.REMOTE_PORT));
assertNull(message
.getHeaders().get(IpHeaders.CONNECTION_SEQ));
assertEquals(Integer.valueOf(1), message
.getHeaders().getSequenceNumber());
assertEquals(message.getHeaders().get(IpHeaders.CONNECTION_ID), message
.getHeaders().getCorrelationId());
message = mapper.toMessage(connection);
assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload()));
assertEquals("MyHost", message
.getHeaders().get(IpHeaders.HOSTNAME));
assertEquals("1.1.1.1", message
.getHeaders().get(IpHeaders.IP_ADDRESS));
assertEquals(1234, message
.getHeaders().get(IpHeaders.REMOTE_PORT));
assertNull(message
.getHeaders().get(IpHeaders.CONNECTION_SEQ));
assertEquals(Integer.valueOf(2), message
.getHeaders().getSequenceNumber());
assertEquals(message.getHeaders().get(IpHeaders.CONNECTION_ID), message
.getHeaders().getCorrelationId());
}
@Test
public void testFromMessageBytes() throws Exception {
String s = "test";