diff --git a/docs/src/reference/docbook/ip.xml b/docs/src/reference/docbook/ip.xml index b661c1259f..b7f357fef9 100644 --- a/docs/src/reference/docbook/ip.xml +++ b/docs/src/reference/docbook/ip.xml @@ -652,7 +652,7 @@ Two sequential messages arriving on the same socket might 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. @@ -660,6 +660,14 @@ is required, consider setting using-nio to false and using async handoff. + + Alternatively, you may choose to insert a resequencer downstream of the inbound endpoint to + return the messages to their proper sequence. Set apply-sequence + to true on the connection factory, and messages arriving on a TCP connection will + have sequenceNumber and correlationId headers + set. The resequencer uses these headers to return the messages to their proper + sequence. +
IP Configuration Attributes @@ -727,7 +735,7 @@ true, false Whether or not connection uses NIO. Refer to the java.nio package for more information. - See . + See . Default false. @@ -739,6 +747,18 @@ Refer to java.nio.ByteBuffer documentation for more information. Must be false if using-nio is false. + + apply-sequence + Y + Y + true, false + When using NIO, it may be necessary to resequence messages. When this + attribute is set to true, correlationId and + sequenceNumber headers will be added to + received messages. + See . + Default false. + so-timeout Y diff --git a/spring-integration-ip/.springBeans b/spring-integration-ip/.springBeans index 4d34282f88..253efca1a7 100644 --- a/spring-integration-ip/.springBeans +++ b/spring-integration-ip/.springBeans @@ -1,13 +1,14 @@ 1 - + src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml + src/test/java/org/springframework/integration/ip/tcp/AutoStartTests-context.xml diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java index 108bdc9b04..23391ef289 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java @@ -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; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java index 16189c5e8e..abe6438315 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java @@ -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; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java index 1201ce3b36..bdbd844bdf 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/CommonSocketOptions.java @@ -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); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java index fc191d09ce..f628ca0de8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java @@ -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"; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index 2ba9fc45a4..a5cee11ea0 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -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 diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java index ab2238400b..0e85635220 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java @@ -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 - 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 connections = new ConcurrentHashMap(); @@ -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; + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index 0f8f978856..194a0ebb63 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -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. - * + *

+ * {@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 pendingReplies = new ConcurrentHashMap(); - + 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"; - } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java index ed342ff01d..9cb093a452 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -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; + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index 59884c6c71..83dd53b3b7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -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 connections = new ConcurrentHashMap(); - - 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 connections = new ConcurrentHashMap(); + + 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 getConnections() { + return connections; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index 6253e3706b..259d47bb36 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -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; } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 5b5566f2b6..181e6477ac 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -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 connections = new LinkedList(); + 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 connections = new LinkedList(); 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 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"); + } + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index cbb9724e07..8bb5e3726d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -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) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java index 5860a2dc22..c2fc6cdee1 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java @@ -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; } - + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java index b4ef4b2a3a..7a47ea3547 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java @@ -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; } - + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java index ea5fa3a148..8b30feeb41 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java @@ -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; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java index 3fbfe27708..ebc9f79a67 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java @@ -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(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java index 3a7e9caeab..4db2d00083 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java @@ -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); } \ No newline at end of file diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java index 6a69543134..84edf4167b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java @@ -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(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java index c93b493a6f..aa617331ab 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java @@ -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); } \ No newline at end of file diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java index 56489a706a..a905506d1d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java @@ -41,20 +41,35 @@ public class TcpMessageMapper implements OutboundMessageMapper { private volatile String charset = "UTF-8"; - + private volatile boolean stringToBytes = true; - + + private volatile boolean applySequence = false; + + @SuppressWarnings("deprecation") public Message toMessage(TcpConnection connection) throws Exception { Message 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; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java index d4f714ef02..46e84ee55a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java @@ -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; - } - } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index d44c64b5ac..05b5a8c43f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 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) 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; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index b3ae401739..108e02b539 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java @@ -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; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index e782e41e15..1676e54735 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -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 connections = new ConcurrentHashMap(); + private Map connections = new ConcurrentHashMap(); - protected BlockingQueue newChannels = new LinkedBlockingQueue(); + private BlockingQueue newChannels = new LinkedBlockingQueue(); /** @@ -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 getConnections() { + return connections; + } + + /** + * @return the newChannels + */ + protected BlockingQueue getNewChannels() { + return newChannels; } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index ab72f628b4..f3d4701cc5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -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) 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 { } } - + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index 9221c112f9..24c549bc60 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -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 connections = new HashMap(); + private Map connections = new HashMap(); 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 getConnections() { + return connections; + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java index 3bdd8d41fd..f630962a79 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java @@ -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. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java index 6cbdb248ec..75f5660692 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java @@ -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(); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java index 4ad27021e6..24a7f471ea 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java @@ -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); } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 1b49345761..52c3c258f0 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -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 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 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"; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index e544ebfd6d..0c0e981d83 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -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 ackControl = Collections + private volatile int ackCounter = 1; + + private volatile Map ackControl = Collections .synchronizedMap(new HashMap()); - 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; + } } diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd index 05bdba4c82..104296f0eb 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.1.xsd @@ -65,7 +65,7 @@ its configuration specifies the number of threads. -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". @@ -115,183 +115,199 @@ task executors such as a WorkManagerTaskExecutor. - - - - - - - - - -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. - - - - - - - - - - - - - - - - - - - - - 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. - - - + + + + + + + + + + + + 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. + + + + + + + + + + + + + + + + + + + + + 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. + + + + + - - - - - - - - - -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. - - - - - - - - - - - - - - - - Specifies the order for invocation when this endpoint is connected as a - subscriber to a SubscribableChannel. - - - + + + + + + + + + + + + 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. + + + + + + + + + + + + + + + + Specifies the order for invocation when this endpoint is connected as a + subscriber to a SubscribableChannel. + + + + + - - - - - - - - - -A connection factory is needed by an inbound adapter. The connection factory must be of type 'server'. - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - 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. - - - + + + + + + + + + + + + A connection factory is needed by an inbound adapter. The connection factory must be of type 'server'. + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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. + + + + + - - - - - - - - - -A connection factory is needed by an outbound adapter. The connection factory must be of 'client'. - - - - - - - - - - - - - - - - - - - - - - - - - - - Specifies the order for invocation when this endpoint is connected as a - subscriber to a SubscribableChannel. - - - + + + + + + + + + + + + A connection factory is needed by an outbound adapter. The connection factory must be of 'client'. + + + + + + + + + + + + + + + + + + + + + + + + + + + Specifies the order for invocation when this endpoint is connected as a + subscriber to a SubscribableChannel. + + + + + @@ -317,7 +333,7 @@ connection request. -The host to which a client connection factory will connect. +The host to which a client connection factory will connect. @@ -326,14 +342,14 @@ The host to which a client connection factory will connect. 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. -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. @@ -350,10 +366,10 @@ performance and may use fewer threads. 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. - + @@ -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. @@ -377,7 +393,7 @@ the factory, the connection will be closed after a response is received. 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. @@ -391,7 +407,7 @@ would normally be the same but this is not required. 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. @@ -420,7 +436,7 @@ its configuration specifies the number of threads. -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". @@ -434,6 +450,14 @@ message headers (ip_connectionId, ip_hostName). Default "true". + + + +When set to "true", adds sequenceNumber and correlationId headers to messages originating from +connections created by this factory. Facilitates resequencing if necessary. Default "false". + + + @@ -480,4 +504,23 @@ apply to TCP outbound adapters and gateways. + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 75fb5d5479..962f9db509 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -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"> - + + - + - + - + - - - - - - + + auto-startup="false" + phase="126" + /> - + + /> - + + auto-startup="false" + phase="127" + /> - + true - + - + - + - + - + - + - + - + - + - - - - + \ No newline at end of file diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index 2d138181ac..07d701316b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -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 handlers = (Set) TestUtils .getPropertyValue( TestUtils.getPropertyValue(this.tcpChannel, "dispatcher"), "handlers"); Iterator 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 diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests-context.xml new file mode 100644 index 0000000000..21c2b67343 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests-context.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests.java new file mode 100644 index 0000000000..935ed65048 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests.java @@ -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"); + } + } + } +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java index cf4cc87451..469ac4452d 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java @@ -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(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java index 45ef0b7ebd..329a12213b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java @@ -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 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 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 message = MessageBuilder.withPayload("test").build(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java index 29cde829e5..7cc0f98cbf 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java @@ -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 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";