INT-1269 Refactor TCP Adapters to Separate Out Connection Factories - Docs to Follow

This commit is contained in:
Gary Russell
2010-07-26 20:54:51 +00:00
parent bdc49cb25f
commit 46bc422174
42 changed files with 4677 additions and 4 deletions

View File

@@ -44,4 +44,6 @@ public abstract class IpHeaders {
public static final String REMOTE_PORT = TCP + "remote_port";
public static final String CONNECTION_ID = IP + "connection_id";
}

View File

@@ -93,6 +93,15 @@ public abstract class IpAdapterParserUtils {
static final String TASK_EXECUTOR = "task-executor";
static final String TCP_CONNECTION_TYPE = "type";
static final String INPUT_CONVERTER = "input-converter";
static final String OUTPUT_CONVERTER = "output-converter";
static final String SINGLE_USE = "single-use";
static final String TCP_CONNECTION_FACTORY = "connection-factory";
/**
* Adds a constructor-arg to the provided bean definition builder
@@ -129,6 +138,17 @@ public abstract class IpAdapterParserUtils {
builder.addConstructorArgValue(port);
}
/**
* @param element
* @param builder
* @param parserContext
*/
public static void addPortToConstructor(Element element,
BeanDefinitionBuilder builder, ParserContext parserContext) {
String port = IpAdapterParserUtils.getPort(element, parserContext);
builder.addConstructorArgValue(port);
}
/**
* Asserts that a protocol attribute (udp or tcp) is supplied,
* @param element

View File

@@ -31,6 +31,9 @@ public class IpNamespaceHandler extends AbstractIntegrationNamespaceHandler {
this.registerBeanDefinitionParser("outbound-channel-adapter", new IpOutboundChannelAdapterParser());
this.registerBeanDefinitionParser("inbound-gateway", new IpInboundGatewayParser());
this.registerBeanDefinitionParser("outbound-gateway", new IpOutboundGatewayParser());
this.registerBeanDefinitionParser("tcp-connection-factory", new TcpConnectionParser());
this.registerBeanDefinitionParser("tcp-inbound-channel-adapter", new TcpInboundChannelAdapterParser());
this.registerBeanDefinitionParser("tcp-outbound-channel-adapter", new TcpOutboundChannelAdapterParser());
}
}

View File

@@ -0,0 +1,101 @@
/*
* Copyright 2002-2010 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.config;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
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.util.StringUtils;
import org.w3c.dom.Element;
/**
* @author Gary Russell
* @since 2.0
*
*/
public class TcpConnectionParser extends AbstractBeanDefinitionParser {
@Override
protected AbstractBeanDefinition parseInternal(Element element,
ParserContext parserContext) {
BeanDefinitionBuilder builder = null;
String useNio = IpAdapterParserUtils.getUseNio(element);
String type = element.getAttribute(IpAdapterParserUtils.TCP_CONNECTION_TYPE);
if (!StringUtils.hasText(type)) {
parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE +
" is required for a tcp connection", element);
}
if (type.equals("client")) {
if (useNio.equals("true")) {
builder = BeanDefinitionBuilder.genericBeanDefinition(
TcpNioClientConnectionFactory.class);
IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext);
} else {
builder = BeanDefinitionBuilder.genericBeanDefinition(
TcpNetClientConnectionFactory.class);
IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext);
}
} else if (type.equals("server")) {
if (useNio.equals("true")) {
builder = BeanDefinitionBuilder.genericBeanDefinition(
TcpNioServerConnectionFactory.class);
IpAdapterParserUtils.addPortToConstructor(element, builder, parserContext);
} else {
builder = BeanDefinitionBuilder.genericBeanDefinition(
TcpNetServerConnectionFactory.class);
IpAdapterParserUtils.addPortToConstructor(element, builder, parserContext);
}
} else {
parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE +
" must be 'client' or 'server' for an IP channel adapter", element);
}
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.USING_DIRECT_BUFFERS);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_KEEP_ALIVE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_LINGER);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_TCP_NODELAY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_TRAFFIC_CLASS);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.POOL_SIZE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TASK_EXECUTOR);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.INPUT_CONVERTER);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.OUTPUT_CONVERTER);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SINGLE_USE);
return builder.getBeanDefinition();
}
}

View File

@@ -0,0 +1,45 @@
/*
* Copyright 2002-2010 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.config;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
import org.w3c.dom.Element;
/**
* Channel Adapter that receives UDP datagram packets and maps them to Messages.
*
* @author Gary Russell
* @since 2.0
*/
public class TcpInboundChannelAdapterParser extends AbstractChannelAdapterParser {
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
TcpReceivingChannelAdapter.class);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TCP_CONNECTION_FACTORY);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder,
element, "channel", "outputChannel");
return builder.getBeanDefinition();
}
}

View File

@@ -0,0 +1,42 @@
/*
* Copyright 2002-2010 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.config;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
import org.w3c.dom.Element;
/**
* @author Gary Russell
* @since 2.0
*/
public class TcpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
TcpSendingMessageHandler.class);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TCP_CONNECTION_FACTORY);
return builder.getBeanDefinition();
}
}

View File

@@ -16,6 +16,7 @@
package org.springframework.integration.ip.tcp;
import java.lang.reflect.Constructor;
import java.net.InetAddress;
import java.net.Socket;
import java.nio.channels.SocketChannel;
@@ -122,4 +123,14 @@ public class SocketIoUtils {
return writer;
}
public static String getSocketId(Socket socket) {
InetAddress inetAddress = socket.getInetAddress();
String hostName = "";
if (inetAddress != null) {
hostName = inetAddress.getHostName();
}
return hostName + ":" + socket.getPort() + ":" +
socket.hashCode();
}
}

View File

@@ -0,0 +1,89 @@
/*
* Copyright 2002-2010 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 java.net.ServerSocket;
import org.springframework.integration.core.Message;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.ConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpListener;
/**
* Tcp inbound channel adapter using a TcpConnection to
* receive data - if the connection factory is a server
* factory, this Listener owns the connections. If it is
* a client factory, the sender owns the connection.
*
* @author Gary Russell
* @since 2.0
*
*/
public class TcpReceivingChannelAdapter
extends MessageProducerSupport implements TcpListener {
protected ServerSocket serverSocket;
protected Class<NetSocketReader> customSocketReaderClass;
protected ConnectionFactory clientConnectionFactory;
protected ConnectionFactory serverConnectionFactory;
public void onMessage(Message<?> message) {
sendMessage(message);
}
@Override
protected void doStart() {
// Nothing to do; we're passive
}
@Override
protected void doStop() {
// Nothing to do; we're passive
}
/**
* Sets the client or server connection factory; for this (an inbound adapter), if
* the factory is a client connection factory, the sockets are owned by a sending
* channel adapter and this adapter is used to receive replies.
*
* @param connectionFactory the connectionFactory to set
*/
public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
if (connectionFactory instanceof AbstractClientConnectionFactory) {
this.clientConnectionFactory = connectionFactory;
} else {
this.serverConnectionFactory = connectionFactory;
}
connectionFactory.registerListener(this);
}
public boolean isListening() {
if (this.serverConnectionFactory == null) {
return false;
}
if (this.serverConnectionFactory instanceof AbstractServerConnectionFactory) {
return ((AbstractServerConnectionFactory) this.serverConnectionFactory).isListening();
}
return false;
}
}

View File

@@ -0,0 +1,157 @@
/*
* Copyright 2002-2010 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 java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.ConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpConnection;
import org.springframework.integration.ip.tcp.connection.TcpSender;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageMappingException;
import org.springframework.integration.message.MessageRejectedException;
/**
* 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.
* @author Gary Russell
* @since 2.0
*
*/
public class TcpSendingMessageHandler implements MessageHandler, TcpSender {
protected Log logger = LogFactory.getLog(this.getClass());
protected TcpConnection connection;
protected ConnectionFactory clientConnectionFactory;
protected ConnectionFactory serverConnectionFactory;
protected Map<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
protected synchronized TcpConnection getConnection() {
try {
this.connection = 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;
}
/**
* Writes the message payload to the underlying socket, using the specified
* message format.
* @see org.springframework.integration.message.MessageHandler#handleMessage(org.springframework.integration.core.Message)
*/
public void handleMessage(final Message<?> message) throws MessageRejectedException,
MessageHandlingException, MessageDeliveryException {
if (this.serverConnectionFactory != null) {
// We don't own the connection
Object connectionId = message.getHeaders().get(IpHeaders.CONNECTION_ID);
TcpConnection connection = connections.get(connectionId);
if (connection != null) {
try {
connection.send(message);
} catch (Exception e) {
logger.error("Error sending message", e);
connection.close();
}
} else {
logger.error("Unable to find incoming socket for " + message);
}
return;
}
try {
doWrite(message);
} catch (MessageMappingException e) {
// retry - socket may have closed
if (e.getCause() instanceof IOException) {
doWrite(message);
} else {
throw e;
}
}
}
/**
* Method that actually does the write.
* @param message The message to write.
*/
protected void doWrite(Message<?> message) {
try {
TcpConnection connection = getConnection();
if (connection == null) {
throw new MessageMappingException(message, "Failed to create connection");
}
connection.send(message);
} catch (Exception e) {
this.connection = null;
if (e instanceof MessageMappingException) {
throw (MessageMappingException) e;
}
throw new MessageMappingException(message, "Failed to map message", e);
}
}
/**
* 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) {
if (connectionFactory instanceof AbstractClientConnectionFactory) {
this.clientConnectionFactory = connectionFactory;
} else {
this.serverConnectionFactory = connectionFactory;
connectionFactory.registerSender(this);
}
}
public void addNewConnection(TcpConnection connection) {
connections.put(connection.getConnectionId(), connection);
}
public void removeDeadConnection(TcpConnection connection) {
connections.remove(connection.getConnectionId());
}
}

View File

@@ -0,0 +1,75 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.Executors;
import org.springframework.util.Assert;
/**
* Abstract class for client connection factories; client connection factories
* establish outgoing connections.
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractClientConnectionFactory extends AbstractConnectionFactory {
/**
* Constructs a factory that will established connections to the host and port.
* @param host The host.
* @param port The port.
*/
public AbstractClientConnectionFactory(String host, int port) {
Assert.notNull(host, "host must not be null");
this.host = host;
this.port = port;
}
/**
* Transfers attributes such as converters, singleUse etc to a new connection.
* When the connection factory has a reference to a TCPListener (to read
* responses), or for single use connections, the connection is executed.
* Single use connections need to read from the connection in order to
* close it after the socket timeout.
* @param connection The new connection.
* @param socket The new socket.
*/
protected void initializeConnection(TcpConnection connection, Socket socket) {
if (this.listener != null) {
connection.registerListener(this.listener);
}
if (this.listener != null || this.singleUse) {
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newFixedThreadPool(this.poolSize);
}
if (this.soTimeout <= 0) {
try {
socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT);
} catch (SocketException e) {
logger.error("Error setting default reply timeout", e);
}
}
}
connection.setMapper(this.mapper);
connection.setInputConverter(this.inputConverter);
connection.setOutputConverter(this.outputConverter);
connection.setSingleUse(this.singleUse);
}
}

View File

@@ -0,0 +1,313 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter;
import org.springframework.util.Assert;
/**
* Base class for all connection factories.
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractConnectionFactory
implements ConnectionFactory, Runnable, Lifecycle {
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 TcpSender sender;
protected int soTimeout;
private int soSendBufferSize;
private int soReceiveBufferSize;
private boolean soTcpNoDelay;
private int soLinger;
private boolean soKeepAlive;
private int soTrafficClass;
protected Executor taskExecutor;
protected InputStreamingConverter<?> inputConverter = new ByteArrayCrLfConverter();
protected OutputStreamingConverter<?> outputConverter = new ByteArrayCrLfConverter();
protected TcpMessageMapper mapper = new TcpMessageMapper();
protected boolean singleUse;
protected int poolSize = 5;
protected boolean active;
/**
* Sets socket attributes on the socket.
* @param socket The socket.
* @throws SocketException
*/
protected void setSocketAttributes(Socket socket) throws SocketException {
if (this.soTimeout >= 0) {
socket.setSoTimeout(this.soTimeout);
}
if (this.soSendBufferSize > 0) {
socket.setSendBufferSize(this.soSendBufferSize);
}
if (this.soReceiveBufferSize > 0) {
socket.setReceiveBufferSize(this.soReceiveBufferSize);
}
socket.setTcpNoDelay(this.soTcpNoDelay);
if (this.soLinger >= 0) {
socket.setSoLinger(true, this.soLinger);
}
if (this.soTrafficClass >= 0) {
socket.setTrafficClass(this.soTrafficClass);
}
socket.setKeepAlive(this.soKeepAlive);
}
/**
* @return the soTimeout
*/
public int getSoTimeout() {
return soTimeout;
}
/**
* @param soTimeout the soTimeout to set
*/
public void setSoTimeout(int soTimeout) {
this.soTimeout = soTimeout;
}
/**
* @return the soReceiveBufferSize
*/
public int getSoReceiveBufferSize() {
return soReceiveBufferSize;
}
/**
* @param soReceiveBufferSize the soReceiveBufferSize to set
*/
public void setSoReceiveBufferSize(int soReceiveBufferSize) {
this.soReceiveBufferSize = soReceiveBufferSize;
}
/**
* @return the soSendBufferSize
*/
public int getSoSendBufferSize() {
return soSendBufferSize;
}
/**
* @param soSendBufferSize the soSendBufferSize to set
*/
public void setSoSendBufferSize(int soSendBufferSize) {
this.soSendBufferSize = soSendBufferSize;
}
/**
* @return the soTcpNoDelay
*/
public boolean isSoTcpNoDelay() {
return soTcpNoDelay;
}
/**
* @param soTcpNoDelay the soTcpNoDelay to set
*/
public void setSoTcpNoDelay(boolean soTcpNoDelay) {
this.soTcpNoDelay = soTcpNoDelay;
}
/**
* @return the soLinger
*/
public int getSoLinger() {
return soLinger;
}
/**
* @param soLinger the soLinger to set
*/
public void setSoLinger(int soLinger) {
this.soLinger = soLinger;
}
/**
* @return the soKeepAlive
*/
public boolean isSoKeepAlive() {
return soKeepAlive;
}
/**
* @param soKeepAlive the soKeepAlive to set
*/
public void setSoKeepAlive(boolean soKeepAlive) {
this.soKeepAlive = soKeepAlive;
}
/**
* @return the soTrafficClass
*/
public int getSoTrafficClass() {
return soTrafficClass;
}
/**
* @param soTrafficClass the soTrafficClass to set
*/
public void setSoTrafficClass(int soTrafficClass) {
this.soTrafficClass = soTrafficClass;
}
/**
* @return the host
*/
public String getHost() {
return host;
}
/**
* @return the port
*/
public int getPort() {
return port;
}
/**
* Registers a TcpListener to receive messages after
* the payload has been converted from the input data.
* @param listener the TcpListener.
*/
public void registerListener(TcpListener listener) {
Assert.isNull(this.listener, this.getClass().getName() +
" may only be used by one inbound adapter");
this.listener = listener;
}
/**
* Registers a TcpSender; for server sockets, used to
* provide connection information so a sender can be used
* to reply to incoming messages.
* @param tcpSendingMessageHandler
*/
public void registerSender(TcpSender sender) {
Assert.isNull(this.sender, this.getClass().getName() +
" may only be used by one outbound adapter");
this.sender = sender;
}
/**
* @param taskExecutor the taskExecutor to set
*/
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
*
* @param converter the inputConverter to set
*/
public void setInputConverter(InputStreamingConverter<?> converter) {
this.inputConverter = converter;
}
/**
*
* @param outputConverter the outputConverter to set
*/
public void setOutputConverter(OutputStreamingConverter<?> outputConverter) {
this.outputConverter = outputConverter;
}
/**
*
* @param mapper the mapper to set; defaults to a {@link TcpMessageMapper}
*/
public void setMapper(TcpMessageMapper mapper) {
this.mapper = mapper;
}
/**
* If true, sockets created by this factory will be used once.
* @param singleUse
*/
public void setSingleUse(boolean singleUse) {
this.singleUse = singleUse;
}
public void setPoolSize(int poolSize) {
this.poolSize = poolSize;
}
/**
* Closes the server.
*/
public abstract void close();
/**
* Creates a taskExecutor (if one was not provided) and starts
* the listening process on one of its threads.
*/
public void start() {
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newFixedThreadPool(this.poolSize);
}
this.active = true;
this.taskExecutor.execute(this);
}
/**
* Stops the server.
*/
public void stop() {
this.active = false;
this.close();
}
}

View File

@@ -0,0 +1,105 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.net.SocketException;
/**
* Base class for all server connection factories. Server connection factories
* listen on a port for incoming connections and create new TcpConnection objects
* for each new connection.
* @author Gary Russell
*
*/
public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory {
protected boolean listening;
protected String localAddress;
/**
* The port on which the factory will listen.
* @param port
*/
public AbstractServerConnectionFactory(int port) {
this.port = port;
}
/**
* Not supported because the factory manages multiple connections and this
* method cannot discriminate.
*/
public TcpConnection getConnection() throws Exception {
throw new UnsupportedOperationException("Getting a connection from a server factory is not supported");
}
/**
*
* @return true if the server is listening on the port.
*/
public boolean isListening() {
return listening;
}
/**
* Transfers attributes such as converters, singleUse etc to a new connection.
* For single use sockets, enforces a socket timeout (default 10 seconds).
* @param connection The new connection.
* @param socket The new socket.
*/
protected void initializeConnection(TcpConnection connection, Socket socket) {
connection.registerListener(this.listener);
connection.registerSender(this.sender);
connection.setMapper(this.mapper);
connection.setInputConverter(this.inputConverter);
connection.setOutputConverter(this.outputConverter);
connection.setSingleUse(this.singleUse);
/*
* 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) {
try {
socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT);
} catch (SocketException e) {
logger.error("Error setting default reply timeout", e);
}
}
}
/**
*
* @return the localAddress
*/
public String getLocalAddress() {
return localAddress;
}
/**
* Used on multi-homed systems to enforce the server to listen
* on a specfic network address instead of all network adapters.
* @param localAddress the ip address of the required adapter.
*/
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
}

View File

@@ -0,0 +1,138 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
import org.springframework.integration.ip.tcp.converter.AbstractByteArrayStreamingConverter;
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());
@SuppressWarnings("rawtypes")
protected InputStreamingConverter inputConverter;
@SuppressWarnings("rawtypes")
protected OutputStreamingConverter outputConverter;
protected TcpMessageMapper mapper;
protected TcpListener listener;
protected TcpSender sender;
protected boolean singleUse;
/**
* Closes this connection.
*/
public void close() {
if (this.sender != null) {
this.sender.removeDeadConnection(this);
}
}
/**
* @return the mapper
*/
public TcpMessageMapper getMapper() {
return mapper;
}
/**
* @param mapper the mapper to set
*/
public void setMapper(TcpMessageMapper mapper) {
Assert.notNull(mapper, this.getClass().getName() + " Mapper may not be null");
this.mapper = mapper;
if (this.outputConverter != null &&
!(this.outputConverter instanceof AbstractByteArrayStreamingConverter)) {
mapper.setStringToBytes(false);
}
}
/**
* @param converter the input converter to set
*/
public void setInputConverter(InputStreamingConverter<?> inputConverter) {
this.inputConverter = inputConverter;
}
/**
* @param converter the output converter to set
*/
public void setOutputConverter(OutputStreamingConverter<?> outputConverter) {
this.outputConverter = outputConverter;
if (!(outputConverter instanceof AbstractByteArrayStreamingConverter)) {
mapper.setStringToBytes(false);
}
}
/**
* @param sender the listener to set
*/
public void registerListener(TcpListener listener) {
this.listener = listener;
}
/**
* @param sender the sender to set
*/
public void registerSender(TcpSender sender) {
this.sender = sender;
if (sender != null) {
sender.addNewConnection(this);
}
}
/**
* @return the listener
*/
public TcpListener getListener() {
return this.listener;
}
/**
* @param singleUse true if this socket is to used once and
* discarded.
*/
public void setSingleUse(boolean singleUse) {
this.singleUse = singleUse;
}
/**
*
* @return True if connection is used once.
*/
public boolean isSingleUse() {
return this.singleUse;
}
}

View File

@@ -0,0 +1,32 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
/**
* A factory used to create TcpConnection objects.
*
* @author Gary Russell
* @since 2.0
*
*/
public interface ConnectionFactory {
public TcpConnection getConnection() throws Exception;
}

View File

@@ -0,0 +1,124 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.nio.channels.SocketChannel;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
import org.springframework.integration.core.Message;
/**
* An abstraction over {@link Socket} and {@link SocketChannel} that
* sends {@link Message} objects by converting the payload
* and streaming it to the destination. Requires a {@link TcpListener}
* to receive incoming messages.
*
* @author Gary Russell
* @since 2.0
*
*/
public interface TcpConnection extends Runnable {
/**
* Closes the connection.
*/
public void close();
/**
* @return true if the connection is open.
*/
public boolean isOpen();
/**
* Converts and sends the message.
* @param message The message
* @throws Exception
*/
public void send(Message<?> message) throws Exception;
/**
* Uses the input converter to obtain the message payload
* from the connection's input stream.
* @return
* @throws Exception
*/
public Object getPayload() throws Exception;
/**
* @return the host name
*/
public String getHostName();
/**
* @return the host address
*/
public String getHostAddress();
/**
* @return the port
*/
public int getPort();
/**
* Sets the listener that will receive incoming Messages.
* @param listener
*/
public 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
*/
public void registerSender(TcpSender sender);
/**
* @return a string uniquely representing a connection.
*/
public String getConnectionId();
/**
* When true, the socket is used once and discarded.
* @param singleUse
*/
public void setSingleUse(boolean singleUse);
/**
*
* @return True if connection is used once.
*/
public boolean isSingleUse();
/**
* @param mapper the mapper
*/
public void setMapper(TcpMessageMapper mapper);
/**
* @param inputConverter the inputConverter to set
*/
public void setInputConverter(InputStreamingConverter<?> inputConverter);
/**
* @param outputConverter the outputConverter to set
*/
public void setOutputConverter(OutputStreamingConverter<?> outputConverter);
}

View File

@@ -0,0 +1,44 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import org.springframework.integration.core.Message;
/**
* Classes that implement this interface may register with a
* connection factory to receive messages retrieved from a
* {@link TcpConnection}
* @author Gary Russell
* @since 2.0
*
*/
public interface TcpListener {
/**
* Called by a TCPConnection when a new message arrives.
* @param message The message.
*/
public abstract void onMessage(Message<?> message);
/**
* Return true if the connection factory is a server
* and it is listening.
* @return
*/
public boolean isListening();
}

View File

@@ -0,0 +1,114 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.io.UnsupportedEncodingException;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.message.InboundMessageMapper;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.OutboundMessageMapper;
/**
* Maps incoming data from a {@link TcpConnection} to a {@link Message}.
* If StringToBytes is true (default),
* payloads of type String are converted to a byte[] using the supplied
* charset (UTF-8 by default).
* Inbound messages include headers representing the remote end of the
* connection as well as a connection id that can be used by a {@link TcpSender}
* to correlate which connection to send a reply.
* @author Gary Russell
* @since 2.0
*
*/
public class TcpMessageMapper implements
InboundMessageMapper<TcpConnection>,
OutboundMessageMapper<Object> {
private volatile String charset = "UTF-8";
private volatile boolean stringToBytes = true;
public Message<Object> toMessage(TcpConnection connection) throws Exception {
Message<Object> message = null;
Object payload = connection.getPayload();
if (payload != null) {
message = MessageBuilder.withPayload(payload)
.setHeader(IpHeaders.HOSTNAME, connection.getHostName())
.setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress())
.setHeader(IpHeaders.REMOTE_PORT, connection.getPort())
.setHeader(IpHeaders.CONNECTION_ID, connection.getConnectionId())
.build();
}
return message;
}
public Object fromMessage(Message<?> message) throws Exception {
if (this.stringToBytes) {
return getPayloadAsBytes(message);
}
return message.getPayload();
}
/**
* Extracts the payload as a byte array.
* @param message
* @return
*/
private byte[] getPayloadAsBytes(Message<?> message) {
byte[] bytes = null;
Object payload = message.getPayload();
if (payload instanceof byte[]) {
bytes = (byte[]) payload;
}
else if (payload instanceof String) {
try {
bytes = ((String) payload).getBytes(this.charset);
}
catch (UnsupportedEncodingException e) {
throw new MessageHandlingException(message, e);
}
}
else {
throw new MessageHandlingException(message,
"When using a byte array streaming converter, the socket mapper expects " +
"either a byte array or String payload, but received: " + payload.getClass());
}
return bytes;
}
/**
* @param charset the charset to set
*/
public void setCharset(String charset) {
this.charset = charset;
}
/**
* Sets whether outbound String payloads are to be converted
* to byte[]. Default is true.
* @param stringToBytes
*/
public void setStringToBytes(boolean stringToBytes) {
this.stringToBytes = stringToBytes;
}
}

View File

@@ -0,0 +1,74 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import javax.net.SocketFactory;
/**
* A client connection factory that creates {@link TcpNetConection}s.
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNetClientConnectionFactory extends
AbstractClientConnectionFactory {
protected TcpNetConnection theConnection;
/**
* Creates a TcpNetClientConnectionFactory for connections to the host and port.
* @param host the host
* @param port the port
*/
public TcpNetClientConnectionFactory(String host, int port) {
super(host, port);
}
/**
* Obtains a connection - if {@link #setSingleUse(boolean)} was called with
* true, a new connection is returned; otherwise a single connection is
* reused for all requests while the connection remains open.
*/
public TcpNetConnection getConnection() throws Exception {
if (this.theConnection != null && this.theConnection.isOpen()) {
return this.theConnection;
}
logger.debug("Opening new socket connection to " + this.host + ":" + this.port);
Socket socket = SocketFactory.getDefault().createSocket(this.host, this.port);
setSocketAttributes(socket);
TcpNetConnection connection = new TcpNetConnection(socket, false);
initializeConnection(connection, socket);
this.taskExecutor.execute(connection);
if (!this.singleUse) {
this.theConnection = connection;
}
return connection;
}
public void close() {
}
public void run() {
}
public boolean isRunning() {
return this.active;
}
}

View File

@@ -0,0 +1,152 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.net.SocketTimeoutException;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.tcp.SocketIoUtils;
import org.springframework.integration.ip.tcp.converter.SoftEndOfStreamException;
/**
* A TcpConnection that uses and underlying {@link Socket}.
*
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNetConnection extends AbstractTcpConnection {
private final Socket socket;
private final boolean server;
/**
* Constructs a TcpNetConnection for the socket.
* @param socket the socket
* @param server if true this connection was created as
* a result of an incoming request.
*/
public TcpNetConnection(Socket socket, boolean server) {
this.socket = socket;
this.server = server;
}
/**
* Closes this connection.
*/
public void close() {
try {
this.socket.close();
} catch (Exception e) {}
super.close();
}
public boolean isOpen() {
return !this.socket.isClosed();
}
@SuppressWarnings("unchecked")
public void send(Message<?> message) throws Exception {
Object object = mapper.fromMessage(message);
this.outputConverter.convert(object, this.socket.getOutputStream());
if (logger.isDebugEnabled())
logger.debug("Message sent " + message);
}
public String getHostAddress() {
return this.socket.getInetAddress().getHostAddress();
}
public String getHostName() {
return this.socket.getInetAddress().getHostName();
}
public Object getPayload() throws Exception {
return this.inputConverter.convert(this.socket.getInputStream());
}
public int getPort() {
return this.socket.getPort();
}
/**
* If there is no listener, and this connection is not for single use,
* this method exits. When there is a listener, the method runs in a
* loop reading input from the connections's stream, data is converted
* to an object using the {@link InputStreamingConverter} and the listener's
* {@link TcpListener#onMessage(Message)} method is called. For single use
* connections with no listener, the socket is closed after its timeout
* expires. If data is received on a single use socket with no listener,
* a warning is logged.
*/
public void run() {
if (this.listener == null && !this.singleUse) {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
Message<?> message = null;
boolean okToRun = true;
logger.debug("Reading...");
while (okToRun) {
try {
message = this.mapper.toMessage(this);
} catch (Exception e) {
this.close();
if (!(e instanceof SoftEndOfStreamException)) {
if (e instanceof SocketTimeoutException && this.singleUse) {
logger.debug("Closing single use socket after timeout");
} else {
logger.error("Read exception " +
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
}
break;
}
/*
* For single use sockets, we close after receipt if we are on the client
* side, or the server side has no outbound adapter registered
*/
if (this.singleUse && (!this.server || this.sender == null)) {
logger.debug("Closing single use socket after inbound message");
this.close();
okToRun = false;
}
if (logger.isDebugEnabled())
logger.debug("Message received " + message);
try {
if (listener == null) {
logger.warn("Unexpected message - no inbound adapter registered with connection " + message);
continue;
}
listener.onMessage(message);
} catch (Exception e) {
logger.error("Exception sending meeeage: " + message, e);
}
}
}
public String getConnectionId() {
return SocketIoUtils.getSocketId(this.socket);
}
}

View File

@@ -0,0 +1,100 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import javax.net.ServerSocketFactory;
/**
* Implements a server connection factory that produces {@link TcpNetConnection}s using
* a {@link ServerSocket}. Must have a {@link TcpListener} registered.
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNetServerConnectionFactory extends AbstractServerConnectionFactory {
protected ServerSocket serverSocket;
/**
* Listens for incoming connections on the port.
* @param port The port.
*/
public TcpNetServerConnectionFactory(int port) {
super(port);
}
/**
* If no listener registers, exits.
* Accepts incoming connections and creates TcpConnections for each new connection.
* Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the
* connection {@link TcpConnection#run()} using the task executor.
* I/O errors on the server socket/channel are logged and the factory is stopped.
*/
public void run() {
if (this.listener == null) {
logger.info("No listener bound to server connection factory; will not read; exiting...");
return;
}
try {
if (this.localAddress == null) {
this.serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(this.port, Math.abs(this.poolSize));
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(port, Math.abs(poolSize), whichNic);
}
this.listening = true;
logger.info("Listening on port " + this.port);
while (true) {
final Socket socket = serverSocket.accept();
logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress());
setSocketAttributes(socket);
TcpNetConnection connection = new TcpNetConnection(socket, true);
this.initializeConnection(connection, socket);
this.taskExecutor.execute(connection);
}
} catch (IOException e) {
this.listening = false;
if (this.active) {
logger.error("Error on ServerSocket", e);
}
this.active = false;
}
}
public boolean isRunning() {
return this.active;
}
public void close() {
if (this.serverSocket == null) {
return;
}
try {
this.serverSocket.close();
} catch (IOException e) {}
this.serverSocket = null;
}
}

View File

@@ -0,0 +1,194 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
/**
* A client connection factory that creates {@link TcpNioConnection}s.
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNioClientConnectionFactory extends
AbstractClientConnectionFactory {
protected TcpNioConnection theConnection;
protected boolean usingDirectBuffers;
private Selector selector;
protected Map<SocketChannel, TcpNioConnection> connections = new ConcurrentHashMap<SocketChannel, TcpNioConnection>();
protected BlockingQueue<SocketChannel> newChannels = new LinkedBlockingQueue<SocketChannel>();
/**
* Creates a TcpNioClientConnectionFactory for connections to the host and port.
* @param host the host
* @param port the port
*/
public TcpNioClientConnectionFactory(String host, int port) {
super(host, port);
}
/**
* Obtains a connection - if {@link #setSingleUse(boolean)} was called with
* true, a new connection is returned; otherwise a single connection is
* reused for all requests while the connection remains open.
*/
public TcpNioConnection getConnection() throws Exception {
int n = 0;
while (this.selector == null) {
Thread.sleep(100);
if (n++ > 600) {
throw new Exception("Factory failed to start");
}
}
if (this.theConnection != null && this.theConnection.isOpen()) {
return this.theConnection;
}
logger.debug("Opening new socket channel connection to " + this.host + ":" + this.port);
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.host, this.port));
setSocketAttributes(socketChannel.socket());
TcpNioConnection connection = new TcpNioConnection(socketChannel, false);
if (this.taskExecutor == null) {
connection.setTaskExecutor(Executors.newSingleThreadExecutor());
} else {
connection.setTaskExecutor(this.taskExecutor);
}
initializeConnection(connection, socketChannel.socket());
connection.setUsingDirectBuffers(this.usingDirectBuffers);
socketChannel.configureBlocking(false);
if (this.soTimeout > 0) {
connection.setLastRead(System.currentTimeMillis());
}
this.connections.put(socketChannel, connection);
newChannels.add(socketChannel);
selector.wakeup();
if (!this.singleUse) {
this.theConnection = connection;
}
return connection;
}
/**
* When set to true, connections created by this factory attempt
* to use direct buffers where possible.
* @param usingDirectBuffers
* @see ByteBuffer
*/
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
}
public void close() {
}
public void run() {
logger.debug("Read selector running for connections to " + host + ":" + port);
try {
this.selector = Selector.open();
while (this.active) {
int selectionCount = selector.select(this.soTimeout);
SocketChannel newChannel;
while ((newChannel = newChannels.poll()) != null) {
newChannel.register(this.selector, SelectionKey.OP_READ, connections.get(newChannel));
}
if (logger.isTraceEnabled())
logger.trace("Connection " + host + ":" + port + " SelectionCount: " + selectionCount);
long now = 0;
if (this.soTimeout > 0) {
Iterator<SocketChannel> it = connections.keySet().iterator();
now = System.currentTimeMillis();
while (it.hasNext()) {
SocketChannel channel = it.next();
if (!channel.isOpen()) {
logger.debug("Removing closed channel");
it.remove();
} else {
TcpNioConnection connection = this.connections.get(channel);
if (now - connection.getLastRead() > this.soTimeout) {
logger.warn("Timing out TcpNioConnection " + connection.getConnectionId());
connection.timeout();
}
}
}
}
if (selectionCount > 0) {
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
while (iterator.hasNext()) {
final SelectionKey key = iterator.next();
iterator.remove();
if (!key.isValid()) {
logger.debug("Selection key no longer valid");
}
else if (key.isReadable()) {
key.interestOps(key.interestOps() - key.readyOps());
final TcpNioConnection connection;
connection = (TcpNioConnection) key.attachment();
connection.setLastRead(System.currentTimeMillis());
this.taskExecutor.execute(new Runnable() {
public void run() {
try {
connection.readPacket();
} catch (Exception e) {
if (connection.isOpen()) {
logger.error("Exception on read " +
connection.getConnectionId() + " " +
e.getMessage());
connection.close();
} else {
logger.debug("Conection closed");
}
}
if (key.channel().isOpen()) {
key.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
}});
}
else {
logger.error("Unexpected key: " + key);
}
}
}
}
} catch (Exception e) {
logger.error("Exception in reader thread", e);
}
logger.debug("Read selector exiting for connections to " + host + ":" + port);
}
public boolean isRunning() {
return this.active;
}
}

View File

@@ -0,0 +1,353 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.tcp.SocketIoUtils;
import org.springframework.integration.ip.tcp.converter.SoftEndOfStreamException;
/**
* A TcpConnection that uses and underlying {@link SocketChannel}.
*
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNioConnection extends AbstractTcpConnection {
private final SocketChannel socketChannel;
private final boolean server;
private OutputStream channelOutputStream;
private PipedOutputStream pipedOutputStream;
private PipedInputStream pipedInputStream;
private boolean usingDirectBuffers;
private Executor taskExecutor;
private ByteBuffer rawBuffer;
private int maxMessageSize = 60 * 1024;
private boolean active = true;
private long lastRead;
/**
* Constructs a TcpNetConnection for the SocketChannel.
* @param socketChannel the socketChannel
* @param server if true this connection was created as
* a result of an incoming request.
*/
public TcpNioConnection(SocketChannel socketChannel, boolean server) throws Exception {
this.socketChannel = socketChannel;
this.server = server;
this.pipedInputStream = new PipedInputStream();
this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream);
this.channelOutputStream = new ChannelOutputStream();
}
public void close() {
doClose();
}
private void doClose() {
this.active = false;
if (pipedOutputStream != null) {
try {
pipedOutputStream.close();
} catch (IOException e) {}
}
try {
this.socketChannel.close();
} catch (Exception e) {}
super.close();
}
public boolean isOpen() {
return this.socketChannel.isOpen();
}
@SuppressWarnings("unchecked")
public void send(Message<?> message) throws Exception {
Object object = mapper.fromMessage(message);
this.outputConverter.convert(object, this.channelOutputStream);
}
public String getHostAddress() {
return this.socketChannel.socket().getInetAddress().getHostAddress();
}
public String getHostName() {
return this.socketChannel.socket().getInetAddress().getHostName();
}
public Object getPayload() throws Exception {
return this.inputConverter.convert(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.
*/
protected ByteBuffer allocate(int length) {
ByteBuffer buffer;
if (this.usingDirectBuffers) {
buffer = ByteBuffer.allocateDirect(length);
} else {
buffer = ByteBuffer.allocate(length);
}
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
* data into messages by invoking convertAndSend whenever there is
* data in the input Stream. Method exits when a message is complete
* and there is no more data; thus freeing the thread to work on other
* sockets.
*/
public void run() {
logger.debug("Nio message assembler running...");
try {
if (this.listener == null && !this.singleUse) {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
if (active) {
try {
while (pipedInputStream.available() > 0) {
convertAndSend();
}
} catch (IOException e) {
logger.error("Unexpected exception, exiting...", e);
return;
}
}
} finally {
logger.debug("Nio message assembler exiting...");
}
}
private synchronized void convertAndSend() throws IOException {
if (this.pipedInputStream.available() <= 0) {
System.err.println("NO WORK");
return;
}
Message<?> message = null;
try {
message = this.mapper.toMessage(this);
} catch (Exception e) {
this.close();
if (e instanceof SocketTimeoutException && this.singleUse) {
logger.debug("Closing single use socket after timeout");
} else {
if (!(e instanceof SoftEndOfStreamException)) {
logger.error("Read exception " +
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
}
return;
}
/*
* For single use sockets, we close after receipt if we are on the client
* side, or the server side has no outbound adapter registered
*/
if (this.singleUse && (!this.server || this.sender == null)) {
logger.debug("Closing single use socket after inbound message");
this.close();
}
try {
if (message != null) {
listener.onMessage(message);
}
} catch (Exception e) {
logger.error("Exception sending meeeage: " + message, e);
}
}
private void doRead() throws Exception {
if (rawBuffer == null) {
rawBuffer = allocate(maxMessageSize);
}
rawBuffer.clear();
int len = socketChannel.read(rawBuffer);
if (len < 0) {
this.close();
throw new IOException("Channel closed");
}
rawBuffer.flip();
if (logger.isDebugEnabled()) {
logger.debug("Read " + rawBuffer.limit() + " into raw buffer");
}
pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit());
pipedOutputStream.flush();
if (!socketChannel.isBlocking()) {
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newSingleThreadExecutor();
}
this.taskExecutor.execute(this);
}
}
/**
* Invoked by the factory when there is data to be read.
*/
public void readPacket() {
try {
doRead();
} catch (Exception e) {
logger.error("Exception on Read " +
this.getConnectionId() + " " +
e.getMessage());
this.close();
}
}
/**
* Close the socket due to timeout.
*/
void timeout() {
this.close();
}
/**
*
* @param taskExecutor the taskExecutor to set
*/
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* If true, connection will attempt to use direct buffers where
* possible.
* @param usingDirectBuffers
*/
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
}
public String getConnectionId() {
return SocketIoUtils.getSocketId(this.socketChannel.socket());
}
/**
*
* @return Time of last read.
*/
public long getLastRead() {
return lastRead;
}
/**
*
* @param lastRead The time of the last read.
*/
public void setLastRead(long lastRead) {
this.lastRead = lastRead;
}
/**
* OutputStream to wrap a SocketChannel; implements timeout on write.
*
*/
class ChannelOutputStream extends OutputStream {
private Selector selector;
private int soTimeout;
@Override
public void write(int b) throws IOException {
byte[] bytes = new byte[1];
bytes[0] = (byte) b;
ByteBuffer buffer = ByteBuffer.wrap(bytes);
doWrite(buffer);
}
@Override
public void close() throws IOException {
doClose();
}
@Override
public void flush() throws IOException {
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
ByteBuffer buffer = ByteBuffer.wrap(b, off, len);
doWrite(buffer);
}
@Override
public void write(byte[] b) throws IOException {
ByteBuffer buffer = ByteBuffer.wrap(b);
doWrite(buffer);
}
private void doWrite(ByteBuffer buffer) throws IOException {
socketChannel.write(buffer);
int remaining = buffer.remaining();
if (remaining == 0) {
return;
}
if (this.selector == null) {
this.selector = Selector.open();
this.soTimeout = socketChannel.socket().getSoTimeout();
}
socketChannel.register(selector, SelectionKey.OP_WRITE);
while (remaining > 0) {
int selectionCount = this.selector.select(this.soTimeout);
if (selectionCount == 0) {
throw new SocketTimeoutException("Timeout on write");
}
selector.selectedKeys().clear();
socketChannel.write(buffer);
remaining = buffer.remaining();
}
}
}
}

View File

@@ -0,0 +1,218 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
/**
/**
* Implements a server connection factory that produces {@link TcpNioConnection}s using
* a {@link ServerSocketChannel}. Must have a {@link TcpListener} registered.
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNioServerConnectionFactory extends AbstractServerConnectionFactory {
protected ServerSocketChannel serverChannel;
protected boolean usingDirectBuffers;
protected Map<SocketChannel, TcpNioConnection> connections = new HashMap<SocketChannel, TcpNioConnection>();
/**
* Listens for incoming connections on the port.
* @param port The port.
*/
public TcpNioServerConnectionFactory(int port) {
super(port);
}
/**
* If no listener registers, exits.
* Accepts incoming connections and creates TcpConnections for each new connection.
* Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the
* connection {@link TcpConnection#run()} using the task executor.
* I/O errors on the server socket/channel are logged and the factory is stopped.
*/
public void run() {
if (this.listener == null) {
logger.info("No listener bound to server connection factory; will not read; exiting...");
return;
}
try {
this.serverChannel = ServerSocketChannel.open();
this.listening = true;
this.serverChannel.configureBlocking(false);
if (this.localAddress == null) {
this.serverChannel.socket().bind(new InetSocketAddress(this.port),
Math.abs(this.poolSize));
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.serverChannel.socket().bind(new InetSocketAddress(whichNic, this.port),
Math.abs(this.poolSize));
}
final Selector selector = Selector.open();
this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
doSelect(this.serverChannel, selector);
} catch (IOException e) {
this.close();
this.listening = false;
if (this.active) {
logger.error("Error on ServerSocketChannel", e);
this.active = false;
}
}
}
/**
* Listens for incoming connections and for notifications that a connected
* socket is ready for reading.
* Accepts incoming connections, registers the new socket with the
* selector for reading.
* When a socket is ready for reading, unregisters the read interest and
* schedules a call to doRead which reads all available data. When the read
* is complete, the socket is again registered for read interest.
* @param server
* @param selector
* @throws IOException
* @throws ClosedChannelException
* @throws SocketException
*/
private void doSelect(ServerSocketChannel server, final Selector selector)
throws IOException, ClosedChannelException, SocketException {
while (this.active) {
int selectionCount = selector.select(this.soTimeout);
if (logger.isTraceEnabled())
logger.trace("Port " + port + " SelectionCount: " + selectionCount);
long now = 0;
if (this.soTimeout > 0) {
Iterator<SocketChannel> it = connections.keySet().iterator();
now = System.currentTimeMillis();
while (it.hasNext()) {
SocketChannel channel = it.next();
if (!channel.isOpen()) {
logger.debug("Removing closed channel");
it.remove();
} else {
TcpNioConnection connection = this.connections.get(channel);
if (now - connection.getLastRead() > this.soTimeout) {
logger.warn("Timing out TcpNioConnection " +
this.port + " : " +
connection.getConnectionId());
connection.timeout();
}
}
}
}
if (selectionCount > 0) {
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
SocketChannel channel = null;
while (iterator.hasNext()) {
final SelectionKey key = iterator.next();
iterator.remove();
if (!key.isValid()) {
logger.debug("Selection key no longer valid");
}
else if (key.isAcceptable()) {
channel = server.accept();
channel.configureBlocking(false);
Socket socket = channel.socket();
setSocketAttributes(socket);
TcpNioConnection connection = createTcpNioConnection(channel);
connection.setLastRead(now);
connections.put(channel, connection);
channel.register(selector, SelectionKey.OP_READ, connection);
}
else if (key.isReadable()) {
key.interestOps(key.interestOps() - key.readyOps());
final TcpNioConnection connection;
connection = (TcpNioConnection) key.attachment();
connection.setLastRead(System.currentTimeMillis());
this.taskExecutor.execute(new Runnable() {
public void run() {
try {
connection.readPacket();
} catch (Exception e) {
if (connection.isOpen()) {
logger.error("Exception on read " + e.getMessage());
connection.close();
} else {
logger.debug("Conection closed");
}
}
if (key.channel().isOpen()) {
key.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
}});
}
else {
logger.error("Unexpected key: " + key);
}
}
}
}
}
private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) {
try {
TcpNioConnection connection = new TcpNioConnection(socketChannel, true);
this.initializeConnection(connection, socketChannel.socket());
connection.setUsingDirectBuffers(this.usingDirectBuffers);
return connection;
} catch (Exception e) {
logger.error("Failed to establish new incoming connection", e);
return null;
}
}
public boolean isRunning() {
return this.active;
}
public void close() {
if (this.serverChannel == null) {
return;
}
try {
this.serverChannel.close();
} catch (IOException e) {}
this.serverChannel = null;
}
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
}
}

View File

@@ -0,0 +1,42 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
/**
* An interface representing a sending client of a connection
* factory.
* @author Gary Russell
* @since 2.0
*
*/
public interface TcpSender {
/**
* When we are using sockets owned by a {@link TcpListener}, this
* method is called each time a new connection is made.
* @param connection The connection.
*/
void addNewConnection(TcpConnection connection);
/**
* When we are using sockets owned by a {@link TcpListener}, this
* method is called each time a connection is closed.
* @param connection The connection.
*/
void removeDeadConnection(TcpConnection connection);
}

View File

@@ -0,0 +1,6 @@
/**
* All things related to tcp connections - client and
* server factories; listener and sender interfaces.
*/
package org.springframework.integration.ip.tcp.connection;

View File

@@ -0,0 +1,66 @@
/*
* Copyright 2002-2010 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.converter;
import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
/**
* Base class for streaming converters that convert to/from a byte array.
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractByteArrayStreamingConverter implements
InputStreamingConverter<byte[]>,
OutputStreamingConverter<byte[]> {
protected int maxMessageSize = 2048;
protected Log logger = LogFactory.getLog(this.getClass());
/**
* The maximum supported message size for this converter.
* Default 2048.
* @return The max message size.
*/
public int getMaxMessageSize() {
return maxMessageSize;
}
/**
* The maximum supported message size for this converter.
* Default 2048.
* @param maxMessageSize The max message size.
*/
public void setMaxMessageSize(int maxMessageSize) {
this.maxMessageSize = maxMessageSize;
}
protected void checkClosure(int bite) throws IOException {
if (bite < 0) {
logger.debug("Socket closed");
throw new IOException("Socket closed");
}
}
}

View File

@@ -0,0 +1,76 @@
/*
* Copyright 2002-2010 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.converter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* Converts data in an InputStream to a byte[]; data is terminated by \r\n
* (not included in resulting byte[]).
* Writes a byte[] to an OutputStream and adds \r\n.
*
* @author Gary Russell
* @since 2.0
*
*/
public class ByteArrayCrLfConverter extends AbstractByteArrayStreamingConverter {
/**
* Converts the data in the inputstream to a byte[]. Data must be terminated
* by CRLF (\r\n). Throws a {@link SoftEndOfStreamException} if the stream
* is closed immediately after the \r\n (i.e. no data is in the process of
* being read).
*/
public byte[] convert(InputStream inputStream) throws IOException {
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
int bite;
if (logger.isDebugEnabled())
logger.debug("Available to read:" + inputStream.available());
while (true) {
bite = inputStream.read();
// logger.debug("Read:" + (char) bite);
if (bite < 0 && n == 0) {
throw new SoftEndOfStreamException("Stream closed between payloads");
}
checkClosure(bite);
if (n > 0 && bite == '\n' && buffer[n-1] == '\r')
break;
buffer[n++] = (byte) bite;
if (n >= this.maxMessageSize) {
throw new IOException("CRLF not found before max message length: "
+ this.maxMessageSize);
}
};
byte[] assembledData = new byte[n-1];
System.arraycopy(buffer, 0, assembledData, 0, n-1);
return assembledData;
}
/**
* Writes the byte[] to the stream and appends \r\n.
*/
public void convert(byte[] bytes, OutputStream outputStream) throws IOException {
outputStream.write(bytes);
outputStream.write('\r');
outputStream.write('\n');
outputStream.flush();
}
}

View File

@@ -0,0 +1,113 @@
/*
* Copyright 2002-2010 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.converter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Converts data in an InputStream to a byte[]; data is preceded by
* a 4 byte binary length (network byte order,
* not included in resulting byte[]).
* Writes a byte[] to an OutputStream after a 4 byte binary length.\
* The length field contains the length of data following the length
* field.
* (network byte order).
* @author Gary Russell
*
*/
public class ByteArrayLengthHeaderConverter extends AbstractByteArrayStreamingConverter {
private Log logger = LogFactory.getLog(this.getClass());
/**
* Reads a 4 byte length from the stream and then reads that length
* from the stream and returns the data in a byte[]. Throws an
* IOException if the length field exceeds the maxMessageSize.
* Throws a {@link SoftEndOfStreamException} if the stream
* is closed between messages.
*/
public byte[] convert(InputStream inputStream) throws IOException {
byte[] lengthPart = new byte[4];
int status = read(inputStream, lengthPart, true);
if (status < 0) {
throw new SoftEndOfStreamException("Stream closed between payloads");
}
int messageLength = ByteBuffer.wrap(lengthPart).getInt();
if (logger.isDebugEnabled()) {
logger.debug("Message length is " + messageLength);
}
if (messageLength > this.maxMessageSize) {
throw new IOException("Message length " + messageLength +
" exceeds max message length: " + this.maxMessageSize);
}
byte[] messagePart = new byte[messageLength];
read(inputStream, messagePart, false);
return messagePart;
}
/**
* Writes the byte[] to the output stream, preceded by a 4 byte
* length in network byte order (big endian).
*/
public void convert(byte[] bytes, OutputStream outputStream)
throws IOException {
ByteBuffer lengthPart = ByteBuffer.allocate(4);
lengthPart.putInt(bytes.length);
outputStream.write(lengthPart.array());
outputStream.write(bytes);
outputStream.flush();
}
/**
* Reads data from the socket and puts the data in buffer. Blocks until
* buffer is full or a socket timeout occurs.
* @param buffer
* @param header true if we are reading the header
* @return < 0 if socket closed and not in the middle of a message
* @throws IOException
*/
protected int read(InputStream inputStream, byte[] buffer, boolean header)
throws IOException {
int lengthRead = 0;
int needed = buffer.length;
while (lengthRead < needed) {
int len;
len = inputStream.read(buffer, lengthRead,
needed - lengthRead);
if (len < 0 && header && lengthRead == 0) {
return len;
}
if (len < 0) {
throw new IOException("Stream closed after " + lengthRead + " of " + needed);
}
lengthRead += len;
if (logger.isDebugEnabled()) {
logger.debug("Read " + len + " bytes, buffer is now at " +
lengthRead + " of " +
needed);
}
}
return 0;
}
}

View File

@@ -0,0 +1,81 @@
/*
* Copyright 2002-2010 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.converter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import org.springframework.integration.message.MessageMappingException;
/**
* Converts data in an InputStream to a byte[]; data is prefixed by &lt;stx&gt; and
* terminated by &lt;etx&gt; (not included in resulting byte[]).
* Writes a byte[] to an OutputStream and prefixed by &lt;stx&gt; terminated by &lt;etx&gt;
*
* @author Gary Russell
* @since 2.0
*
*/
public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverter {
public static final int STX = 0x02;
public static final int ETX = 0x03;
/**
* Converts the data in the inputstream to a byte[]. Data must be prefixed
* with an ASCII STX character, and terminated with an ASCII ETX character.
* Throws a {@link SoftEndOfStreamException} if the stream
* is closed immediately before the STX (i.e. no data is in the process of
* being read).
*
*/
public byte[] convert(InputStream inputStream) throws IOException {
int bite = inputStream.read();
if (bite < 0) {
throw new SoftEndOfStreamException("Stream closed between payloads");
}
if (bite != STX)
throw new MessageMappingException("Expected STX to begin message");
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
while ((bite = inputStream.read()) != ETX) {
checkClosure(bite);
buffer[n++] = (byte) bite;
if (n >= this.maxMessageSize) {
throw new IOException("ETX not found before max message length: "
+ this.maxMessageSize);
}
}
byte[] assembledData = new byte[n];
System.arraycopy(buffer, 0, assembledData, 0, n);
return assembledData;
}
/**
* Writes the byte[] to the stream, prefixed by an ASCII STX character and
* terminated with an ASCII ETX character.
*/
public void convert(byte[] bytes, OutputStream outputStream) throws IOException {
outputStream.write(STX);
outputStream.write(bytes);
outputStream.write(ETX);
outputStream.flush();
}
}

View File

@@ -0,0 +1,49 @@
/*
* Copyright 2002-2010 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.converter;
import java.io.IOException;
/**
* Used to communicate that a stream has closed, but between logical
* messages.
*
* @author Gary Russell
* @since 2.0
*
*/
public class SoftEndOfStreamException extends IOException {
private static final long serialVersionUID = 7309907445617226978L;
public SoftEndOfStreamException() {
super();
}
public SoftEndOfStreamException(String message, Throwable cause) {
super(message, cause);
}
public SoftEndOfStreamException(String message) {
super(message);
}
public SoftEndOfStreamException(Throwable cause) {
super(cause);
}
}

View File

@@ -0,0 +1,6 @@
/**
* Byte array converters for putting some protocol on the
* wire so we can delimit incoming messages.
*/
package org.springframework.integration.ip.tcp.converter;

View File

@@ -189,6 +189,183 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="tcp-outbound-channel-adapter">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an outbound adapter. If the connection factory has a type 'client',
the factory is 'owned' by this adapter. If it has a type 'server', it is owned by an inbound channel
adapter and this adapter will attempt to correlate messages to the connection on which an original
inbound message was received.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="tcp-inbound-channel-adapter">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an inbound adapter. If the connection factory has a type 'server',
the factory is 'owned' by this adapter. If it has a type 'client', it is owned by an outbound channel
adapter and this adapter will receive any incoming messages on the connection created by the outbound
adapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="tcp-connection-factory">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="type">
<xsd:annotation>
<xsd:documentation>
Connection factories can be 'client' or 'server'. Client factories
open a connection to a server using a host and port. Server factories
listen on a port and create a separate connection for each incoming
connection request.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="client" />
<xsd:enumeration value="server" />
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="host" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The host to which a client connection factory will connect.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="port" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
For client factories, the port to which a client connection factory will connect.
For server factories, the port on which the factory will listen for incoming
connections.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="using-nio" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
If true, the factory will use java.nio.channel.SocketChannel for communication;
for a large number of connections on the server side, this can provide better
performance and may use fewer threads.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="so-keep-alive" type="xsd:string" />
<xsd:attribute name="so-linger" type="xsd:string" />
<xsd:attribute name="so-receive-buffer-size" type="xsd:string" />
<xsd:attribute name="so-send-buffer-size" type="xsd:string" />
<xsd:attribute name="so-tcp-no-delay" type="xsd:string" />
<xsd:attribute name="so-timeout" type="xsd:string" />
<xsd:attribute name="so-traffic-class" type="xsd:string" />
<xsd:attribute name="single-use" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
If true, a new connection will be created for each use. For inbound adapters
where there is no outbound adapter sharing the factory, the connection will
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="input-converter" type="xsd:string" >
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.serializer.InputStreamingConverter"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
An InputStreamingConverter that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfConverter. Input and output converters
would normally be the same but this is not required.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="output-converter" type="xsd:string" >
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.serializer.OutputStreamingConverter"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
An OutputStreamingConverter that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfConverter. Input and output converters
would normally be the same but this is not required.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="local-address" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
On a multi-homed system, specifies the ip address of the network interface used to communicate.
For inbound adapters and gateways, specifies the interface used to listen for incoming connections.
If omitted, the endpoint will listen on all available adapters. For the UDP multicast outbound adapter
specifies the interface to which multicast packets will be sent. For UDP unicast and multicast
adapters, specifies which interface to which the acknowledgment socket will be bound. Does not
apply to TCP outbound adapters and gateways.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string" />
<xsd:attribute name="pool-size" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
The number of threads that will be used for socket/channel handling. Only applies
if an external task-executor is NOT being used. When using an external task executor,
its configuration specifies the number of threads.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="ipAdapterType">
<xsd:annotation>

View File

@@ -277,5 +277,64 @@
so-timeout="226"
close="true"
/>
<ip:tcp-connection-factory
id="client1"
type="client"
host="localhost"
port="9876"
input-converter="serial"
output-converter="serial"
so-keep-alive="true"
so-linger="54"
so-receive-buffer-size="1234"
so-send-buffer-size="1235"
so-tcp-no-delay="true"
so-timeout="1236"
so-traffic-class="12"
using-nio="true"
single-use="true"
task-executor="externalTE"
pool-size="321"
/>
<ip:tcp-connection-factory
id="server1"
type="server"
port="9876"
local-address="127.0.0.1"
input-converter="serial"
output-converter="serial"
so-keep-alive="true"
so-linger="55"
so-receive-buffer-size="1234"
so-send-buffer-size="1235"
so-tcp-no-delay="true"
so-timeout="1236"
so-traffic-class="12"
using-nio="true"
single-use="true"
task-executor="externalTE"
pool-size="123"
/>
<bean id="serial" class="org.springframework.commons.serializer.JavaSerializationConverter" />
<ip:tcp-outbound-channel-adapter id="tcpNewOut1"
channel="tcpChannel"
connection-factory="client1" />
<ip:tcp-outbound-channel-adapter id="tcpNewOut2"
channel="tcpChannel"
connection-factory="server1" />
<ip:tcp-inbound-channel-adapter id="tcpNewIn1"
channel="tcpChannel"
connection-factory="client1" />
<ip:tcp-inbound-channel-adapter id="tcpNewIn2"
channel="tcpChannel"
connection-factory="server1" />
</beans>

View File

@@ -25,6 +25,7 @@ import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.context.ApplicationContext;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.ip.tcp.CustomNetSocketReader;
@@ -38,6 +39,11 @@ import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler;
import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.TcpNioSendingMessageHandler;
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
import org.springframework.integration.ip.udp.DatagramPacketMessageMapper;
import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter;
import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
@@ -49,6 +55,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@@ -125,6 +132,33 @@ public class ParserUnitTests {
@Qualifier(value="externalTE")
TaskExecutor taskExecutor;
@Autowired
@Qualifier(value="client1")
AbstractConnectionFactory client1;
@Autowired
InputStreamingConverter<byte[]> converter;
@Autowired
@Qualifier(value="server1")
AbstractConnectionFactory server1;
@Autowired
@Qualifier(value="org.springframework.integration.ip.tcp.TcpSendingMessageHandler#0")
TcpSendingMessageHandler tcpNewOut1;
@Autowired
@Qualifier(value="org.springframework.integration.ip.tcp.TcpSendingMessageHandler#1")
TcpSendingMessageHandler tcpNewOut2;
@Autowired
@Qualifier(value="tcpNewIn1")
TcpReceivingChannelAdapter tcpNewIn1;
@Autowired
@Qualifier(value="tcpNewIn2")
TcpReceivingChannelAdapter tcpNewIn2;
@Test
public void testInUdp() {
DirectFieldAccessor dfa = new DirectFieldAccessor(udpIn);
@@ -383,4 +417,66 @@ public class ParserUnitTests {
assertEquals(226, delegateDfa.getPropertyValue("soTimeout"));
assertEquals(true, dfa.getPropertyValue("close"));
}
@Test
public void testConnClient1() {
assertTrue(client1 instanceof TcpNioClientConnectionFactory);
assertEquals("localhost", client1.getHost());
assertEquals(9876, client1.getPort());
assertEquals(54, client1.getSoLinger());
assertEquals(1234, client1.getSoReceiveBufferSize());
assertEquals(1235, client1.getSoSendBufferSize());
assertEquals(1236, client1.getSoTimeout());
assertEquals(12, client1.getSoTrafficClass());
DirectFieldAccessor dfa = new DirectFieldAccessor(client1);
assertSame(converter, dfa.getPropertyValue("inputConverter"));
assertSame(converter, dfa.getPropertyValue("outputConverter"));
assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
assertEquals(true, dfa.getPropertyValue("singleUse"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
assertEquals(321, dfa.getPropertyValue("poolSize"));
}
@Test
public void testConnServer1() {
assertTrue(server1 instanceof TcpNioServerConnectionFactory);
assertEquals(9876, server1.getPort());
assertEquals(55, server1.getSoLinger());
assertEquals(1234, server1.getSoReceiveBufferSize());
assertEquals(1235, server1.getSoSendBufferSize());
assertEquals(1236, server1.getSoTimeout());
assertEquals(12, server1.getSoTrafficClass());
DirectFieldAccessor dfa = new DirectFieldAccessor(server1);
assertSame(converter, dfa.getPropertyValue("inputConverter"));
assertSame(converter, dfa.getPropertyValue("outputConverter"));
assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
assertEquals(true, dfa.getPropertyValue("singleUse"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
assertEquals(123, dfa.getPropertyValue("poolSize"));
}
@Test
public void testNewOut1() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewOut1);
assertSame(client1, dfa.getPropertyValue("clientConnectionFactory"));
}
@Test
public void testNewOut2() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewOut2);
assertSame(server1, dfa.getPropertyValue("serverConnectionFactory"));
}
@Test
public void testNewIn1() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewIn1);
assertSame(client1, dfa.getPropertyValue("clientConnectionFactory"));
}
@Test
public void testNewIn2() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewIn2);
assertSame(server1, dfa.getPropertyValue("serverConnectionFactory"));
}
}

View File

@@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-ip="http://www.springframework.org/schema/integration/ip"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
<int-ip:tcp-connection-factory id="server"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(10000)}"
/>
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="#{server.port}"
single-use="true"
so-timeout="100000"
/>
</beans>

View File

@@ -0,0 +1,81 @@
/*
* Copyright 2002-2010 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.assertNotNull;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpConnection;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 2.0
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class ConnectionToConnectionTests {
@Autowired
AbstractApplicationContext ctx;
@Autowired
private AbstractClientConnectionFactory client;
@Autowired
private AbstractServerConnectionFactory server;
private TcpReceivingChannelAdapter receiver;
@Before
public void setup() {
receiver = new TcpReceivingChannelAdapter();
server.registerListener(receiver);
ctx.start();
}
@Test
public void testConnect() throws Exception {
int n = 0;
while (!server.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
throw new Exception("Failed to listen");
}
}
TcpConnection connection = client.getConnection();
QueueChannel channel = new QueueChannel();
receiver.setOutputChannel(channel);
connection.send(MessageBuilder.withPayload("Test").build());
Message<?> m = channel.receive(10000);
assertNotNull(m);
}
}

View File

@@ -0,0 +1,58 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-ip="http://www.springframework.org/schema/integration/ip"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
<bean id="serializer" class="org.springframework.commons.serializer.JavaSerializationConverter" />
<int-ip:tcp-connection-factory id="server"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(10000)}"
input-converter="serializer"
output-converter="serializer"
using-nio="true"
single-use="true"
/>
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="#{server.port}"
single-use="true"
so-timeout="10000"
input-converter="serializer"
output-converter="serializer"
/>
<int:channel id="input" />
<int:channel id="replies">
<int:queue/>
</int:channel>
<int-ip:tcp-outbound-channel-adapter id="outboundClient"
channel="input"
connection-factory="client"/>
<int-ip:tcp-inbound-channel-adapter id="inboundClient"
channel="replies"
connection-factory="client"/>
<int-ip:tcp-inbound-channel-adapter id="inboundServer"
channel="loop"
connection-factory="server"/>
<int-ip:tcp-outbound-channel-adapter id="outboundServer"
channel="loop"
connection-factory="server"/>
<int:channel id="loop" />
</beans>

View File

@@ -0,0 +1,80 @@
/*
* Copyright 2002-2010 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.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.ip.tcp.connection.TcpListener;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 2.0
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class SharedConnectionTests {
@Autowired
AbstractApplicationContext ctx;
@Autowired
@Qualifier(value="inboundServer")
TcpListener listener;
/**
* Tests a loopback. The client-side outbound adapter sends a message over
* a connection from the client connection factory; the server side
* receives the message, puts in on a channel which is the input channel
* for the outbound adapter that's sharing the connections. The response
* comes back to an inbound adapter that is sharing the client's
* connection and we verify we get the echo back as expected.
*
* @throws Exception
*/
@Test
public void test1() throws Exception {
int n = 0;
while (!listener.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to listen");
}
}
MessageChannel input = ctx.getBean("input", MessageChannel.class);
input.send(MessageBuilder.withPayload("Test").build());
QueueChannel replies = ctx.getBean("replies", QueueChannel.class);
Message<?> message = replies.receive(10000);
assertNotNull(message);
assertEquals("Test", message.getPayload());
}
}

View File

@@ -17,13 +17,31 @@ package org.springframework.integration.ip.tcp;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter;
import org.springframework.integration.ip.util.SocketUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -212,5 +230,354 @@ public class TcpReceivingChannelAdapterTests {
new String((byte[])message.getPayload()));
adapter.stop();
}
@Test
public void newTestNet() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
int n = 0;
while (!scf.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to start listening");
}
}
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write("Test1\r\n".getBytes());
socket.getOutputStream().write("Test2\r\n".getBytes());
Message<?> message = channel.receive(10000);
assertNotNull(message);
assertEquals("Test1", new String((byte[]) message.getPayload()));
message = channel.receive(10000);
assertNotNull(message);
assertEquals("Test2", new String((byte[]) message.getPayload()));
}
@Test
public void newTestNio() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSoTimeout(5000);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
int n = 0;
while (!scf.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to start listening");
}
}
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
for (int i = 0; i < 100; i++) {
socket.getOutputStream().write(("Test" + i + "\r\n").getBytes());
// if (i % 10 == 0) {
// Thread.sleep(1000);
// }
}
for (int i = 0; i < 100; i++) {
Message<?> message = channel.receive(10000);
assertNotNull(message);
assertEquals("Test" + i, new String((byte[]) message.getPayload()));
}
}
@Test
public void newTestNetShared() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
int n = 0;
while (!scf.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to listen");
}
}
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.setSoTimeout(2000);
socket.getOutputStream().write("Test\r\n".getBytes());
socket.getOutputStream().write("Test\r\n".getBytes());
Message<?> message = channel.receive(10000);
assertNotNull(message);
handler.handleMessage(message);
message = channel.receive(10000);
assertNotNull(message);
handler.handleMessage(message);
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
assertEquals("Test\r\n", new String(b));
readFully(socket.getInputStream(), b);
assertEquals("Test\r\n", new String(b));
}
@Test
public void newTestNioShared() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
int n = 0;
while (!scf.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to listen");
}
}
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.setSoTimeout(2000);
socket.getOutputStream().write("Test\r\n".getBytes());
socket.getOutputStream().write("Test\r\n".getBytes());
Message<?> message = channel.receive(10000);
assertNotNull(message);
handler.handleMessage(message);
message = channel.receive(10000);
assertNotNull(message);
handler.handleMessage(message);
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
assertEquals("Test\r\n", new String(b));
readFully(socket.getInputStream(), b);
assertEquals("Test\r\n", new String(b));
}
@Test
public void newTestNetSingleNoOutbound() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSingleUse(true);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
int n = 0;
while (!scf.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to start listening");
}
}
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write("Test1\r\n".getBytes());
socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write("Test2\r\n".getBytes());
Message<?> message = channel.receive(10000);
assertNotNull(message);
// with single use, results may come back in a different order
Set<String> results = new HashSet<String>();
results.add(new String((byte[]) message.getPayload()));
message = channel.receive(10000);
assertNotNull(message);
results.add(new String((byte[]) message.getPayload()));
assertTrue(results.contains("Test1"));
assertTrue(results.contains("Test2"));
}
@Test
public void newTestNioSingleNoOutbound() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSingleUse(true);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
int n = 0;
while (!scf.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to start listening");
}
}
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write("Test1\r\n".getBytes());
socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write("Test2\r\n".getBytes());
Message<?> message = channel.receive(10000);
assertNotNull(message);
// with single use, results may come back in a different order
Set<String> results = new HashSet<String>();
results.add(new String((byte[]) message.getPayload()));
message = channel.receive(10000);
assertNotNull(message);
results.add(new String((byte[]) message.getPayload()));
assertTrue(results.contains("Test1"));
assertTrue(results.contains("Test2"));
}
/**
* @param is
* @param buff
*/
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {
buff[i] = (byte) is.read();
}
}
@Test
public void newTestNetSingleShared() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
int n = 0;
while (!scf.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to listen");
}
}
Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port);
socket1.setSoTimeout(2000);
socket1.getOutputStream().write("Test1\r\n".getBytes());
Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port);
socket2.setSoTimeout(2000);
socket2.getOutputStream().write("Test2\r\n".getBytes());
Message<?> message = channel.receive(10000);
assertNotNull(message);
handler.handleMessage(message);
message = channel.receive(10000);
assertNotNull(message);
handler.handleMessage(message);
byte[] b = new byte[7];
readFully(socket1.getInputStream(), b);
assertEquals("Test1\r\n", new String(b));
readFully(socket2.getInputStream(), b);
assertEquals("Test2\r\n", new String(b));
}
@Test
public void newTestNioSingleShared() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
int n = 0;
while (!scf.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to listen");
}
}
Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port);
socket1.setSoTimeout(2000);
socket1.getOutputStream().write("Test1\r\n".getBytes());
Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port);
socket2.setSoTimeout(2000);
socket2.getOutputStream().write("Test2\r\n".getBytes());
Message<?> message = channel.receive(10000);
assertNotNull(message);
handler.handleMessage(message);
message = channel.receive(10000);
assertNotNull(message);
handler.handleMessage(message);
byte[] b = new byte[7];
readFully(socket1.getInputStream(), b);
assertEquals("Test1\r\n", new String(b));
readFully(socket2.getInputStream(), b);
assertEquals("Test2\r\n", new String(b));
}
@Test
public void newTestNioSingleSharedMany() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
Executor te = Executors.newFixedThreadPool(100);
scf.setTaskExecutor(te);
scf.start();
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
int n = 0;
while (!scf.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
fail("Failed to listen");
}
}
List<Socket> sockets = new LinkedList<Socket>();
for (int i = 100; i < 200; i++) {
Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port);
socket1.setSoTimeout(2000);
socket1.getOutputStream().write(("Test" + i + "\r\n").getBytes());
sockets.add(socket1);
}
for (int i = 100; i < 200; i++) {
Message<?> message = channel.receive(10000);
assertNotNull(message);
handler.handleMessage(message);
}
byte[] b = new byte[9];
for (int i = 100; i < 200; i++) {
readFully(sockets.remove(0).getInputStream(), b);
assertEquals("Test" + i + "\r\n", new String(b));
}
}
}

View File

@@ -16,22 +16,42 @@
package org.springframework.integration.ip.tcp;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.JavaSerializationConverter;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayLengthHeaderConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter;
import org.springframework.integration.ip.util.SocketUtils;
import org.springframework.integration.message.MessageBuilder;
/**
* @author Gary Russell
* @since 2.0
*
*/
public class TcpSendingMessageHandlerTests {
@@ -197,13 +217,655 @@ public class TcpSendingMessageHandlerTests {
server.close();
}
/**
* @param is
* @param buff
*/
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {
buff[i] = (byte) is.read();
}
}
@Test
public void newTestNetCrLf() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
int i = 0;
while (true) {
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
b = ("Reply" + (++i) + "\r\n").getBytes();
socket.getOutputStream().write(b);
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply1", new String((byte[]) mOut.getPayload()));
mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply2", new String((byte[]) mOut.getPayload()));
done.set(true);
}
@Test
public void newTestNio() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
int i = 0;
while (true) {
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
b = ("Reply" + (++i) + "\r\n").getBytes();
socket.getOutputStream().write(b);
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
// ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply1", new String((byte[]) mOut.getPayload()));
mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply2", new String((byte[]) mOut.getPayload()));
done.set(true);
}
@Test
public void newTestNetStxEtx() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
int i = 0;
while (true) {
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
b = ("\u0002Reply" + (++i) + "\u0003").getBytes();
socket.getOutputStream().write(b);
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply1", new String((byte[]) mOut.getPayload()));
mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply2", new String((byte[]) mOut.getPayload()));
done.set(true);
}
@Test
public void newTestNioStxEtx() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
int i = 0;
while (true) {
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
b = ("\u0002Reply" + (++i) + "\u0003").getBytes();
socket.getOutputStream().write(b);
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply1", new String((byte[]) mOut.getPayload()));
mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply2", new String((byte[]) mOut.getPayload()));
done.set(true);
}
@Test
public void newTestNetLength() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
int i = 0;
while (true) {
byte[] b = new byte[8];
readFully(socket.getInputStream(), b);
if (!"\u0000\u0000\u0000\u0004Test".equals(new String(b))) {
throw new RuntimeException("Bad Data");
}
b = ("\u0000\u0000\u0000\u0006Reply" + (++i)).getBytes();
socket.getOutputStream().write(b);
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply1", new String((byte[]) mOut.getPayload()));
mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply2", new String((byte[]) mOut.getPayload()));
done.set(true);
}
@Test
public void newTestNioLength() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
int i = 0;
while (true) {
byte[] b = new byte[8];
readFully(socket.getInputStream(), b);
if (!"\u0000\u0000\u0000\u0004Test".equals(new String(b))) {
throw new RuntimeException("Bad Data");
}
b = ("\u0000\u0000\u0000\u0006Reply" + (++i)).getBytes();
socket.getOutputStream().write(b);
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply1", new String((byte[]) mOut.getPayload()));
mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply2", new String((byte[]) mOut.getPayload()));
done.set(true);
}
@Test
public void newTestNetSerial() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
int i = 0;
while (true) {
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
ois.readObject();
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
oos.writeObject("Reply" + (++i));
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
JavaSerializationConverter converter = new JavaSerializationConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply1", mOut.getPayload());
mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply2", mOut.getPayload());
done.set(true);
}
@Test
public void newTestNioSerial() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
int i = 0;
while (true) {
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
ois.readObject();
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
oos.writeObject("Reply" + (++i));
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
JavaSerializationConverter converter = new JavaSerializationConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply1", mOut.getPayload());
mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply2", mOut.getPayload());
done.set(true);
}
@Test
public void newTestNetSingleUseNoInbound() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
while (true) {
Socket socket = server.accept();
semaphore.release();
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
semaphore.release();
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
assertTrue(semaphore.tryAcquire(4, 10000, TimeUnit.MILLISECONDS));
done.set(true);
}
@Test
public void newTestNioSingleUseNoInbound() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
while (true) {
Socket socket = server.accept();
semaphore.release();
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
semaphore.release();
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
assertTrue(semaphore.tryAcquire(4, 10000, TimeUnit.MILLISECONDS));
done.set(true);
}
@Test
public void newTestNetSingleUseWithInbound() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
int i = 0;
while (true) {
Socket socket = server.accept();
semaphore.release();
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
b = ("Reply" + (++i) + "\r\n").getBytes();
socket.getOutputStream().write(b);
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS));
Set<String> replies = new HashSet<String>();
for (int i = 0; i < 2; i++) {
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
replies.add(new String((byte[])mOut.getPayload()));
}
assertTrue(replies.remove("Reply1"));
assertTrue(replies.remove("Reply2"));
done.set(true);
}
@Test
public void newTestNioSingleUseWithInbound() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
int i = 0;
while (true) {
Socket socket = server.accept();
semaphore.release();
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
b = ("Reply" + (++i) + "\r\n").getBytes();
socket.getOutputStream().write(b);
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS));
Set<String> replies = new HashSet<String>();
for (int i = 0; i < 2; i++) {
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
replies.add(new String((byte[])mOut.getPayload()));
}
assertTrue(replies.remove("Reply1"));
assertTrue(replies.remove("Reply2"));
done.set(true);
}
@Test
public void newTestNioSingleUseWithInboundMany() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
int i = 0;
while (true) {
Socket socket = server.accept();
semaphore.release();
byte[] b = new byte[8];
readFully(socket.getInputStream(), b);
b = ("Reply" + (i++) + "\r\n").getBytes();
socket.getOutputStream().write(b);
}
} catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(ccf);
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
assertTrue(latch.await(10, TimeUnit.SECONDS));
for (int i = 100; i < 200; i++) {
handler.handleMessage(MessageBuilder.withPayload("Test" + i).build());
}
assertTrue(semaphore.tryAcquire(10, 10000, TimeUnit.MILLISECONDS));
Set<String> replies = new HashSet<String>();
for (int i = 100; i < 200; i++) {
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
replies.add(new String((byte[])mOut.getPayload()));
}
for (int i = 0; i < 100; i++) {
assertTrue("Reply" + i + " missing", replies.remove("Reply" + i));
}
done.set(true);
}
}

View File

@@ -0,0 +1,121 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.util.SocketUtils;
import org.springframework.integration.message.MessageBuilder;
/**
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNioConnectionTests {
@Test
public void testWriteTimeout() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", port);
factory.setSoTimeout(1000);
factory.start();
final CountDownLatch latch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
server.accept();
// block so we fill the buffer
server.accept();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
try {
TcpNioConnection connection = factory.getConnection();
connection.send(MessageBuilder.withPayload(new byte[1000000]).build());
} catch (Exception e) {
assertTrue("Expected SocketTimeoutException, got " + e.getClass().getSimpleName() +
":" + e.getMessage(), e instanceof SocketTimeoutException);
}
}
@Test
public void testReadTimeout() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", port);
factory.setSoTimeout(1000);
factory.start();
final CountDownLatch latch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
// block to cause timeout on read.
server.accept();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
try {
TcpNioConnection connection = factory.getConnection();
connection.send(MessageBuilder.withPayload("Test").build());
Thread.sleep(2000);
assertTrue(!connection.isOpen());
} catch (Exception e) {
fail("Unexptected exception " + e);
}
}
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {
buff[i] = (byte) is.read();
}
}
}

View File

@@ -5,6 +5,7 @@ Bundle-ManifestVersion: 2
Import-Template:
org.apache.commons.logging;version="[1.1.1, 2.0.0)",
org.springframework.integration.*;version="[2.0.0, 2.0.1)",
org.springframework.commons.*;version="[2.0.0, 2.0.1)",
org.springframework.beans.*;version="[3.0.0, 4.0.0)",
org.springframework.context;version="[3.0.0, 4.0.0)",
org.springframework.core.*;version="[3.0.0, 4.0.0)",