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

INT-1008 Simple TCP Outbound Gateway

Browse Source
This commit is contained in:
Gary Russell
2010-05-01 23:02:22 +00:00
parent 31970fcf11
commit c9bed59341
22 changed files with 774 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -20,6 +20,7 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
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.MessageFormats;
import org.springframework.util.StringUtils;
@@ -105,6 +106,23 @@ public abstract class IpAdapterParserUtils {
}
}
/**
* @param element
* @param builder
* @param parserContext
*/
public static void addHostAndPortToConstructor(Element element,
BeanDefinitionBuilder builder, ParserContext parserContext) {
String host = element.getAttribute(IpAdapterParserUtils.HOST);
if (!StringUtils.hasText(host)) {
parserContext.getReaderContext().error(IpAdapterParserUtils.HOST
+ " is required for IP outbound channel adapters", element);
}
builder.addConstructorArgValue(host);
String port = IpAdapterParserUtils.getPort(element, parserContext);
builder.addConstructorArgValue(port);
}
/**
* Asserts that a protocol attribute (udp or tcp) is supplied,
* @param element
@@ -193,6 +211,29 @@ public abstract class IpAdapterParserUtils {
return MessageFormats.FORMAT_LENGTH_HEADER;
}
/**
* @param element
* @param builder
*/
public static void addOutboundTcpAttributes(Element element,
BeanDefinitionBuilder builder) {
builder.addPropertyValue(
Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT),
IpAdapterParserUtils.getMessageFormat(element));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.CUSTOM_SOCKET_WRITER_CLASS_NAME);
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);
}
/**
* Sets the common port attributes on the bean being built (timeout, receive buffer size,
* send buffer size).

1
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java
View File

@@ -30,6 +30,7 @@ public class IpNamespaceHandler extends AbstractIntegrationNamespaceHandler {
this.registerBeanDefinitionParser("inbound-channel-adapter", new IpInboundChannelAdapterParser());
this.registerBeanDefinitionParser("outbound-channel-adapter", new IpOutboundChannelAdapterParser());
this.registerBeanDefinitionParser("inbound-gateway", new IpInboundGatewayParser());
this.registerBeanDefinitionParser("outbound-gateway", new IpOutboundGatewayParser());
}
}

38
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java
View File

@@ -19,7 +19,6 @@ 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.core.Conventions;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler;
@@ -48,23 +47,6 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
return builder.getBeanDefinition();
}
/**
* @param element
* @param builder
* @param parserContext
*/
private void addHostAndPortToConstructor(Element element,
BeanDefinitionBuilder builder, ParserContext parserContext) {
String host = element.getAttribute(IpAdapterParserUtils.HOST);
if (!StringUtils.hasText(host)) {
parserContext.getReaderContext().error(IpAdapterParserUtils.HOST
+ " is required for IP outbound channel adapters", element);
}
builder.addConstructorArgValue(host);
String port = IpAdapterParserUtils.getPort(element, parserContext);
builder.addConstructorArgValue(port);
}
/**
* @param element
* @param parserContext
@@ -87,7 +69,7 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
builder = BeanDefinitionBuilder
.genericBeanDefinition(UnicastSendingMessageHandler.class);
}
addHostAndPortToConstructor(element, builder, parserContext);
IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext);
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.CHECK_LENGTH, true);
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
@@ -135,22 +117,8 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNioSendingMessageHandler.class);
}
addHostAndPortToConstructor(element, builder, parserContext);
builder.addPropertyValue(
Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT),
IpAdapterParserUtils.getMessageFormat(element));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.CUSTOM_SOCKET_WRITER_CLASS_NAME);
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);
IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext);
IpAdapterParserUtils.addOutboundTcpAttributes(element, builder);
return builder;
}

52
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundGatewayParser.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* 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.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
/**
* Parser for the <outbound-gateway> element of the integration 'jms' namespace.
*
* @author Gary Russell
*/
public class IpOutboundGatewayParser extends AbstractConsumerEndpointParser {
@Override
protected String getInputChannelAttributeName() {
return "request-channel";
}
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway");
IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext);
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
IpAdapterParserUtils.addOutboundTcpAttributes(element, builder);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.CUSTOM_SOCKET_READER_CLASS_NAME);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel");
return builder;
}
}

2
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java
View File

@@ -111,7 +111,7 @@ public abstract class AbstractTcpSendingMessageHandler extends
}
writer.write(bytes);
} catch (Exception e) {
writer = null;
this.writer = null;
if (e instanceof MessageMappingException) {
throw (MessageMappingException) e;
}

16
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java
View File

@@ -59,9 +59,9 @@ public class NetSocketReader extends AbstractSocketReader {
if (logger.isDebugEnabled()) {
logger.debug("Message length is " + messageLength);
}
if (messageLength > maxMessageSize) {
if (messageLength > this.maxMessageSize) {
throw new IOException("Message length " + messageLength +
" exceeds max message length: " + maxMessageSize);
" exceeds max message length: " + this.maxMessageSize);
}
byte[] messagePart = new byte[messageLength];
read(messagePart);
@@ -77,7 +77,7 @@ public class NetSocketReader extends AbstractSocketReader {
InputStream inputStream = socket.getInputStream();
if (inputStream.read() != STX)
throw new MessageMappingException("Expected STX to begin message");
byte[] buffer = new byte[maxMessageSize];
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
int bite;
while ((bite = inputStream.read()) != ETX) {
@@ -86,9 +86,9 @@ public class NetSocketReader extends AbstractSocketReader {
throw new IOException("Socket Closed");
}
buffer[n++] = (byte) bite;
if (n >= maxMessageSize) {
if (n >= this.maxMessageSize) {
throw new IOException("ETX not found before max message length: "
+ maxMessageSize);
+ this.maxMessageSize);
}
}
assembledData = new byte[n];
@@ -102,7 +102,7 @@ public class NetSocketReader extends AbstractSocketReader {
@Override
protected boolean assembleDataCrLfFormat() throws IOException {
InputStream inputStream = socket.getInputStream();
byte[] buffer = new byte[maxMessageSize];
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
int bite;
while (true) {
@@ -114,9 +114,9 @@ public class NetSocketReader extends AbstractSocketReader {
if (n > 0 && bite == '\n' && buffer[n-1] == '\r')
break;
buffer[n++] = (byte) bite;
if (n >= maxMessageSize) {
if (n >= this.maxMessageSize) {
throw new IOException("CRLF not found before max message length: "
+ maxMessageSize);
+ this.maxMessageSize);
}
};
assembledData = new byte[n-1];

103
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java
View File

@@ -15,13 +15,12 @@
*/
package org.springframework.integration.ip.tcp;
import java.lang.reflect.Constructor;
import java.net.Socket;
import java.net.SocketException;
import org.springframework.beans.BeanUtils;
import org.springframework.integration.core.Message;
import org.springframework.integration.gateway.AbstractMessagingGateway;
import org.springframework.integration.ip.util.SocketIoUtils;
import org.springframework.integration.message.MessageMappingException;
/**
@@ -39,88 +38,71 @@ import org.springframework.integration.message.MessageMappingException;
*/
public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
private SocketMessageMapper mapper = new SocketMessageMapper();
protected SocketMessageMapper mapper = new SocketMessageMapper();
private WriteCapableTcpNetReceivingChannelAdapter delegate;
protected WriteCapableTcpNetReceivingChannelAdapter delegate;
private int port;
protected int port;
private int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
private int poolSize = 2;
protected int poolSize = 2;
private int receiveBufferSize = 2048;
protected int receiveBufferSize = 2048;
private boolean soKeepAlive;
protected boolean soKeepAlive;
private int soReceiveBufferSize = -1;
protected int soReceiveBufferSize = -1;
private int soSendBufferSize = -1;
protected int soSendBufferSize = -1;
private int soTimeout = 0;
protected int soTimeout = 0;
private String customSocketReaderClassName;
protected String customSocketReaderClassName;
private Class<NetSocketWriter> customSocketWriter;
protected Class<NetSocketWriter> customSocketWriterClass;
/* (non-Javadoc)
* @see org.springframework.integration.gateway.AbstractMessagingGateway#doStart()
*/
@Override
protected void doStart() {
super.doStart();
delegate.start();
this.delegate.start();
}
/* (non-Javadoc)
* @see org.springframework.integration.gateway.AbstractMessagingGateway#doStop()
*/
@Override
protected void doStop() {
super.doStop();
delegate.stop();
this.delegate.stop();
}
/* (non-Javadoc)
* @see org.springframework.integration.gateway.AbstractMessagingGateway#onInit()
*/
@Override
protected void onInit() throws Exception {
delegate = new WriteCapableTcpNetReceivingChannelAdapter(port);
delegate.setMessageFormat(messageFormat);
delegate.setPoolSize(poolSize);
delegate.setReceiveBufferSize(receiveBufferSize);
delegate.setSoKeepAlive(soKeepAlive);
delegate.setSoReceiveBufferSize(soReceiveBufferSize);
delegate.setSoSendBufferSize(soSendBufferSize);
delegate.setSoTimeout(soTimeout);
delegate.setTaskScheduler(getTaskScheduler());
delegate.setCustomSocketReaderClassName(customSocketReaderClassName);
this.delegate = new WriteCapableTcpNetReceivingChannelAdapter(port);
this.delegate.setMessageFormat(messageFormat);
this.delegate.setPoolSize(poolSize);
this.delegate.setReceiveBufferSize(receiveBufferSize);
this.delegate.setSoKeepAlive(soKeepAlive);
this.delegate.setSoReceiveBufferSize(soReceiveBufferSize);
this.delegate.setSoSendBufferSize(soSendBufferSize);
this.delegate.setSoTimeout(soTimeout);
this.delegate.setTaskScheduler(getTaskScheduler());
this.delegate.setCustomSocketReaderClassName(customSocketReaderClassName);
super.onInit();
}
/* (non-Javadoc)
* @see org.springframework.integration.gateway.AbstractMessagingGateway#fromMessage(org.springframework.integration.core.Message)
*/
@Override
protected Object fromMessage(Message<?> message) {
throw new MessageMappingException("Cannot map a message to an object in this gateway");
}
/* (non-Javadoc)
* @see org.springframework.integration.gateway.AbstractMessagingGateway#toMessage(java.lang.Object)
*/
@Override
protected Message<?> toMessage(Object object) {
try {
return mapper.toMessage((SocketReader) object);
return this.mapper.toMessage((SocketReader) object);
} catch (Exception e) {
throw new MessageMappingException("Failed to map message", e);
}
}
/**
* @param port the port to set
*/
@@ -160,7 +142,7 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
* @return the port
*/
public int getPort() {
return port;
return this.port;
}
/**
@@ -192,16 +174,16 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
}
/**
* @param customSocketWriter the customSocketWriter to set
* @param customSocketWriterClassName the customSocketWriterClassName to set
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketWriterClassName(
String customSocketWriterClassName) throws ClassNotFoundException {
if (customSocketWriterClassName != null) {
this.customSocketWriter = (Class<NetSocketWriter>) Class
this.customSocketWriterClass = (Class<NetSocketWriter>) Class
.forName(customSocketWriterClassName);
if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriter))) {
if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriterClass))) {
throw new IllegalArgumentException("Custom socket writer must be of type NetSocketWriter");
}
}
@@ -216,43 +198,24 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
super(port);
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter#processMessage(org.springframework.integration.core.Message)
*/
@Override
protected void processMessage(NetSocketReader reader) {
Socket socket = reader.getSocket();
NetSocketWriter writer = null;
try {
if (messageFormat == MessageFormats.FORMAT_CUSTOM){
Constructor<NetSocketWriter> ctor = customSocketWriter.getConstructor(Socket.class);
writer = BeanUtils.instantiateClass(ctor, socket);
} else {
writer = new NetSocketWriter(socket);
}
} catch (Exception e) {
throw new MessageMappingException("Error creating SocketWriter", e);
}
writer.setMessageFormat(messageFormat);
Message<?> message = sendAndReceiveMessage(reader);
NetSocketWriter writer = SocketIoUtils.createNetWriter(this.messageFormat,
customSocketWriterClass, socket);
try {
writer.write(mapper.fromMessage(message));
writer.write(this.mapper.fromMessage(message));
} catch (Exception e) {
throw new MessageMappingException("Failed to map and send response", e);
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter#doStart()
*/
@Override
protected void doStart() {
; super.doStart();
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractTcpReceivingChannelAdapter#setSocketOptions(java.net.Socket)
*/
@Override
protected void setSocketOptions(Socket socket) throws SocketException {
super.setSocketOptions(socket);

210
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java Normal file
View File

@@ -0,0 +1,210 @@
/*
* 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.Socket;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.ip.util.SocketIoUtils;
/**
* Simple TCP outbound gateway; delegates write to a {@link TcpNetSendingMessageHandler}
* then blocks on read of same socket. Uses {@link java.net.Socket} and the client
* thread is dedicated to a request/response pair. No multiplexing of requests
* over the outbound socket are supported. This class is thread safe in that
* if multiple clients attempt to send a message, they will be blocked until
* any existing request/response is processed.
*
* @author Gary Russell
* @since 2.0
*/
public class SimpleTcpNetOutboundGateway extends
AbstractReplyProducingMessageHandler {
protected TcpNetSendingMessageHandler handler;
protected Class<NetSocketReader> customSocketReaderClass;
protected int messageFormat;
protected int maxMessageSize = 2048;
protected int soReceiveBufferSize = -1;
protected NetSocketReader reader;
/**
* Constructs a SimpleTcpNetOutboundGateway that sends data to the
* specified host and port, and waits for a response.
*
* @param host The host.
* @param port The port.
*/
public SimpleTcpNetOutboundGateway(String host, int port) {
handler = new TcpNetSendingMessageHandler(host, port);
}
/**
* Synchronized to prevent multiplexing requests over the same socket.
*/
@Override
protected synchronized Object handleRequestMessage(Message<?> requestMessage) {
this.handler.handleMessage(requestMessage);
if (this.reader == null) {
Socket socket = this.handler.getSocket();
this.reader = SocketIoUtils.createNetReader(this.messageFormat,
this.customSocketReaderClass, socket, this.maxMessageSize,
this.soReceiveBufferSize);
}
try {
this.reader.assembleData(); // Net... always returns true
byte[] bytes = this.reader.getAssembledData();
return bytes;
} catch (Exception e) {
this.reader = null;
throw new MessagingException(requestMessage, e);
}
}
/**
* @param obj
* @return
* @see java.lang.Object#equals(java.lang.Object)
*/
public boolean equals(Object obj) {
return handler.equals(obj);
}
/**
* @return
* @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#getPort()
*/
public int getPort() {
return handler.getPort();
}
/**
* @return
* @see java.lang.Object#hashCode()
*/
public int hashCode() {
return handler.hashCode();
}
/**
* @param customSocketWriterClassName
* @throws ClassNotFoundException
* @see org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler#setCustomSocketWriterClassName(java.lang.String)
*/
public void setCustomSocketWriterClassName(
String customSocketWriterClassName) throws ClassNotFoundException {
handler.setCustomSocketWriterClassName(customSocketWriterClassName);
}
/**
* @param messageFormat
* @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setMessageFormat(int)
*/
public void setMessageFormat(int messageFormat) {
handler.setMessageFormat(messageFormat);
this.messageFormat = messageFormat;
}
/**
* @param soKeepAlive
* @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setSoKeepAlive(boolean)
*/
public void setSoKeepAlive(boolean soKeepAlive) {
handler.setSoKeepAlive(soKeepAlive);
}
/**
* @param soLinger
* @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setSoLinger(int)
*/
public void setSoLinger(int soLinger) {
handler.setSoLinger(soLinger);
}
/**
* @param size
* @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#setSoReceiveBufferSize(int)
*/
public void setSoReceiveBufferSize(int size) {
this.soReceiveBufferSize = size;
}
/**
* @param size
* @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#setSoSendBufferSize(int)
*/
public void setSoSendBufferSize(int size) {
handler.setSoSendBufferSize(size);
}
/**
* @param soTcpNoDelay
* @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setSoTcpNoDelay(boolean)
*/
public void setSoTcpNoDelay(boolean soTcpNoDelay) {
handler.setSoTcpNoDelay(soTcpNoDelay);
}
/**
* @param timeout
* @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#setSoTimeout(int)
*/
public void setSoTimeout(int timeout) {
handler.setSoTimeout(timeout);
}
/**
* @param soTrafficClass
* @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setSoTrafficClass(int)
*/
public void setSoTrafficClass(int soTrafficClass) {
handler.setSoTrafficClass(soTrafficClass);
}
/**
* @param customSocketReaderClass the customSocketReader to set
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketReaderClassName(
String customSocketReaderClassName) throws ClassNotFoundException {
if (customSocketReaderClassName != null) {
this.customSocketReaderClass = (Class<NetSocketReader>) Class
.forName(customSocketReaderClassName);
if (!(NetSocketReader.class.isAssignableFrom(this.customSocketReaderClass))) {
throw new IllegalArgumentException("Custom socket reader must be of type NetSocketReader");
}
}
}
/**
* Specify the Spring Integration reply channel. If this property is not
* set the gateway will check for a 'replyChannel' header on the request.
*/
public void setReplyChannel(MessageChannel replyChannel) {
this.setOutputChannel(replyChannel);
}
}

5
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java
View File

@@ -35,4 +35,9 @@ public interface SocketWriter {
*/
void write(byte[] bytes) throws IOException;
/**
* @param messageFormat the messageFormat to set
*/
public void setMessageFormat(int messageFormat);
}

30
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java
View File

@@ -16,15 +16,13 @@
package org.springframework.integration.ip.tcp;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.net.ServerSocket;
import java.net.Socket;
import javax.net.ServerSocketFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageMappingException;
import org.springframework.integration.ip.util.SocketIoUtils;
/**
* Tcp Receiving Channel adapter that uses a {@link java.net.Socket}. Each
@@ -38,7 +36,7 @@ public class TcpNetReceivingChannelAdapter extends
AbstractTcpReceivingChannelAdapter {
protected ServerSocket serverSocket;
protected Class<NetSocketReader> customSocketReader;
protected Class<NetSocketReader> customSocketReaderClass;
/**
* Constructs a TcpNetReceivingChannelAdapter that listens on the port.
* @param port The port.
@@ -93,21 +91,9 @@ public class TcpNetReceivingChannelAdapter extends
* @param socket
*/
protected void handleSocket(Socket socket) {
NetSocketReader reader = null;
if (messageFormat == MessageFormats.FORMAT_CUSTOM) {
try {
Constructor<NetSocketReader> ctor =
customSocketReader.getConstructor(Socket.class);
reader = BeanUtils.instantiateClass(ctor, socket);
} catch (Exception e) {
throw new MessageMappingException("Failed to instantiate custom reader", e);
}
}
else {
reader = new NetSocketReader(socket);
}
reader.setMessageFormat(messageFormat);
reader.setMaxMessageSize(receiveBufferSize);
NetSocketReader reader = SocketIoUtils.createNetReader(messageFormat,
customSocketReaderClass, socket, this.receiveBufferSize,
this.soReceiveBufferSize);
while (true) {
try {
if (reader.assembleData()) {
@@ -145,16 +131,16 @@ public class TcpNetReceivingChannelAdapter extends
}
/**
* @param customSocketReader the customSocketReader to set
* @param customSocketReaderClass the customSocketReader to set
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketReaderClassName(
String customSocketReaderClassName) throws ClassNotFoundException {
if (customSocketReaderClassName != null) {
this.customSocketReader = (Class<NetSocketReader>) Class
this.customSocketReaderClass = (Class<NetSocketReader>) Class
.forName(customSocketReaderClassName);
if (!(NetSocketReader.class.isAssignableFrom(this.customSocketReader))) {
if (!(NetSocketReader.class.isAssignableFrom(this.customSocketReaderClass))) {
throw new IllegalArgumentException("Custom socket reader must be of type NetSocketReader");
}
}

31
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java
View File

@@ -15,12 +15,11 @@
*/
package org.springframework.integration.ip.tcp;
import java.lang.reflect.Constructor;
import java.net.Socket;
import javax.net.SocketFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.integration.ip.util.SocketIoUtils;
/**
@@ -31,7 +30,7 @@ import org.springframework.beans.BeanUtils;
public class TcpNetSendingMessageHandler extends
AbstractTcpSendingMessageHandler {
protected Class<NetSocketWriter> customSocketWriter;
protected Class<NetSocketWriter> customSocketWriterClass;
/**
* Constructs a TcpNetSendingMessageHandler that sends data to the
@@ -46,22 +45,22 @@ public class TcpNetSendingMessageHandler extends
protected volatile Socket socket;
/**
* if
* @return the socket
*/
protected Socket getSocket() {
return socket;
}
/**
* @return the writer
*/
protected synchronized SocketWriter getWriter() {
if (writer == null) {
if (this.writer == null) {
try {
this.socket = SocketFactory.getDefault().createSocket(this.host, this.port);
this.setSocketAttributes(socket);
NetSocketWriter writer;
if (messageFormat == MessageFormats.FORMAT_CUSTOM){
Constructor<NetSocketWriter> ctor = customSocketWriter.getConstructor(Socket.class);
writer = BeanUtils.instantiateClass(ctor, socket);
} else {
writer = new NetSocketWriter(socket);
}
writer.setMessageFormat(messageFormat);
NetSocketWriter writer = SocketIoUtils.createNetWriter(messageFormat,
customSocketWriterClass, socket);
this.writer = writer;
} catch (Exception e) {
logger.error("Error creating SocketWriter", e);
@@ -71,16 +70,16 @@ public class TcpNetSendingMessageHandler extends
}
/**
* @param customSocketWriter the customSocketWriter to set
* @param customSocketWriterClassName the customSocketWriterClassName to set
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketWriterClassName(
String customSocketWriterClassName) throws ClassNotFoundException {
if (customSocketWriterClassName != null) {
this.customSocketWriter = (Class<NetSocketWriter>) Class
this.customSocketWriterClass = (Class<NetSocketWriter>) Class
.forName(customSocketWriterClassName);
if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriter))) {
if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriterClass))) {
throw new IllegalArgumentException("Custom socket writer must be of type NetSocketWriter");
}
}

29
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java
View File

@@ -17,7 +17,6 @@ package org.springframework.integration.ip.tcp;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
@@ -29,8 +28,8 @@ import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import org.springframework.beans.BeanUtils;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.util.SocketIoUtils;
/**
* Tcp Receiving Channel adapter that uses a {@link java.nio.channels.SocketChannel}.
@@ -46,7 +45,7 @@ public class TcpNioReceivingChannelAdapter extends
protected ServerSocketChannel serverChannel;
protected boolean usingDirectBuffers;
protected Class<NioSocketReader> customSocketReader;
protected Class<NioSocketReader> customSocketReaderClass;
/**
* Constructs a TcpNioReceivingChannelAdapter to listen on the port.
@@ -152,22 +151,10 @@ public class TcpNioReceivingChannelAdapter extends
* @return The NioSocketReader.
*/
private NioSocketReader createSocketReader(final SelectionKey key) {
NioSocketReader reader = null;
SocketChannel channel = (SocketChannel) key.channel();
if (messageFormat == MessageFormats.FORMAT_CUSTOM) {
try {
Constructor<NioSocketReader> ctor = customSocketReader
.getConstructor(SocketChannel.class);
reader = BeanUtils.instantiateClass(ctor, channel);
} catch (Exception e) {
logger.error("Error creating SocketReader", e);
}
} else {
reader = new NioSocketReader(channel);
}
reader.setUsingDirectBuffers(usingDirectBuffers);
reader.setMessageFormat(messageFormat);
reader.setMaxMessageSize(receiveBufferSize);
NioSocketReader reader = SocketIoUtils.createNioReader(messageFormat,
this.customSocketReaderClass, channel, this.receiveBufferSize,
this.receiveBufferSize, this.usingDirectBuffers);
return reader;
}
@@ -212,15 +199,15 @@ public class TcpNioReceivingChannelAdapter extends
}
/**
* @param customSocketReader the customSocketReader to set
* @param customSocketReaderClassName the customSocketReaderClassName to set
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketReaderClassName(String customSocketReaderClassName)
throws ClassNotFoundException {
this.customSocketReader = (Class<NioSocketReader>) Class
this.customSocketReaderClass = (Class<NioSocketReader>) Class
.forName(customSocketReaderClassName);
if (!(NioSocketReader.class.isAssignableFrom(this.customSocketReader))) {
if (!(NioSocketReader.class.isAssignableFrom(this.customSocketReaderClass))) {
throw new IllegalArgumentException("Custom socket reader must be of type NioSocketReader");
}
}

28
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java
View File

@@ -15,10 +15,9 @@
*/
package org.springframework.integration.ip.tcp;
import java.lang.reflect.Constructor;
import java.nio.channels.SocketChannel;
import org.springframework.beans.BeanUtils;
import org.springframework.integration.ip.util.SocketIoUtils;
/**
@@ -32,7 +31,7 @@ public class TcpNioSendingMessageHandler extends
protected boolean usingDirectBuffers;
protected Class<NioSocketWriter> customSocketWriter;
protected Class<NioSocketWriter> customSocketWriterClass;
protected int buffsPerConnection = 5;
@@ -48,20 +47,13 @@ public class TcpNioSendingMessageHandler extends
* @return the socket
*/
protected synchronized SocketWriter getWriter() {
if (socketChannel == null) {
if (this.socketChannel == null) {
try {
socketChannel = SocketChannel.open(this.destinationAddress);
this.socketChannel = SocketChannel.open(this.destinationAddress);
this.setSocketAttributes(socketChannel.socket());
NioSocketWriter writer;
if (messageFormat == MessageFormats.FORMAT_CUSTOM){
Constructor<NioSocketWriter> ctor = customSocketWriter
.getConstructor(SocketChannel.class, int.class, int.class);
writer = BeanUtils.instantiateClass(ctor, socketChannel, buffsPerConnection, soSendBufferSize);
} else {
writer = new NioSocketWriter(socketChannel, buffsPerConnection, soSendBufferSize);
}
writer.setMessageFormat(messageFormat);
writer.setUsingDirectBuffers(usingDirectBuffers);
NioSocketWriter writer = SocketIoUtils.createNioWriter(messageFormat,
customSocketWriterClass, socketChannel,
buffsPerConnection, soSendBufferSize, usingDirectBuffers);
this.writer = writer;
} catch (Exception e) {
logger.error("Error creating SocketWriter", e);
@@ -79,16 +71,16 @@ public class TcpNioSendingMessageHandler extends
}
/**
* @param customSocketWriter the customSocketWriter to set
* @param customSocketWriterClassName the customSocketWriterClassName to set
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketWriterClassName(
String customSocketWriterClassName) throws ClassNotFoundException {
if (customSocketWriterClassName != null) {
this.customSocketWriter = (Class<NioSocketWriter>) Class
this.customSocketWriterClass = (Class<NioSocketWriter>) Class
.forName(customSocketWriterClassName);
if (!(NioSocketWriter.class.isAssignableFrom(this.customSocketWriter))) {
if (!(NioSocketWriter.class.isAssignableFrom(this.customSocketWriterClass))) {
throw new IllegalArgumentException("Custom socket writer must be of type NioSocketWriter");
}
}

162
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/util/SocketIoUtils.java Normal file
View File

@@ -0,0 +1,162 @@
/*
* 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.util;
import java.lang.reflect.Constructor;
import java.net.Socket;
import java.nio.channels.SocketChannel;
import org.springframework.beans.BeanUtils;
import org.springframework.integration.ip.tcp.MessageFormats;
import org.springframework.integration.ip.tcp.NetSocketReader;
import org.springframework.integration.ip.tcp.NetSocketWriter;
import org.springframework.integration.ip.tcp.NioSocketReader;
import org.springframework.integration.ip.tcp.NioSocketWriter;
import org.springframework.integration.message.MessageMappingException;
/**
* @author Gary Russell
*
*/
public class SocketIoUtils {
/**
* @param messageFormat
* @param customSocketReaderClass
* @param socket
* @param receiveBufferSize
* @param soReceiveBufferSize
* @return
*/
public static NetSocketReader createNetReader(int messageFormat,
Class<NetSocketReader> customSocketReaderClass,
Socket socket,
int receiveBufferSize,
int soReceiveBufferSize) {
NetSocketReader reader;
if (messageFormat == MessageFormats.FORMAT_CUSTOM) {
try {
Constructor<NetSocketReader> ctor =
customSocketReaderClass.getConstructor(Socket.class);
reader = BeanUtils.instantiateClass(ctor, socket);
if (soReceiveBufferSize > 0) {
socket.setReceiveBufferSize(soReceiveBufferSize);
}
} catch (Exception e) {
throw new MessageMappingException("Failed to instantiate custom reader", e);
}
}
else {
reader = new NetSocketReader(socket);
}
reader.setMessageFormat(messageFormat);
reader.setMaxMessageSize(receiveBufferSize);
return reader;
}
/**
* @param messageFormat
* @param socket
* @param customSocketWriterClass
* @return
*/
public static NetSocketWriter createNetWriter(int messageFormat,
Class<NetSocketWriter> customSocketWriterClass, Socket socket) {
NetSocketWriter writer;
if (messageFormat == MessageFormats.FORMAT_CUSTOM){
try {
Constructor<NetSocketWriter> ctor = customSocketWriterClass.getConstructor(Socket.class);
writer = BeanUtils.instantiateClass(ctor, socket);
} catch (Exception e) {
throw new MessageMappingException("Failed to instantiate custom writer", e);
}
} else {
writer = new NetSocketWriter(socket);
}
writer.setMessageFormat(messageFormat);
return writer;
}
/**
* @param messageFormat
* @param customSocketReaderClass
* @param socket
* @param receiveBufferSize
* @param usingDirectBuffers
* @param soReceiveBufferSize
* @return
*/
public static NioSocketReader createNioReader(int messageFormat,
Class<NioSocketReader> customSocketReaderClass,
SocketChannel channel,
int receiveBufferSize,
int soReceiveBufferSize,
boolean usingDirectBuffers ) {
NioSocketReader reader;
if (messageFormat == MessageFormats.FORMAT_CUSTOM) {
try {
Constructor<NioSocketReader> ctor =
customSocketReaderClass.getConstructor(SocketChannel.class);
reader = BeanUtils.instantiateClass(ctor, channel);
if (soReceiveBufferSize > 0) {
channel.socket().setReceiveBufferSize(soReceiveBufferSize);
}
} catch (Exception e) {
throw new MessageMappingException("Failed to instantiate custom reader", e);
}
}
else {
reader = new NioSocketReader(channel);
}
reader.setMessageFormat(messageFormat);
reader.setMaxMessageSize(receiveBufferSize);
reader.setUsingDirectBuffers(usingDirectBuffers);
return reader;
}
/**
* @param messageFormat
* @param socket
* @param customSocketWriter
* @param maxBuffers
* @param sendBufferSize
* @param usingDirectBuffers
* @return
*/
public static NioSocketWriter createNioWriter(int messageFormat,
Class<NioSocketWriter> customSocketWriterClass,
SocketChannel channel,
int maxBuffers,
int sendBufferSize,
boolean usingDirectBuffers) {
NioSocketWriter writer;
if (messageFormat == MessageFormats.FORMAT_CUSTOM){
try {
Constructor<NioSocketWriter> ctor = customSocketWriterClass
.getConstructor(SocketChannel.class, int.class, int.class);
writer = BeanUtils.instantiateClass(ctor, channel, maxBuffers, sendBufferSize);
} catch (Exception e) {
throw new MessageMappingException("Failed to instantiate custom writer", e);
}
} else {
writer = new NioSocketWriter(channel, maxBuffers, sendBufferSize);
}
writer.setMessageFormat(messageFormat);
writer.setUsingDirectBuffers(usingDirectBuffers);
return writer;
}
}

41
org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd
View File

@@ -79,14 +79,44 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
<xsd:element name="inbound-gateway">
<xsd:annotation>
<xsd:documentation>
Defines an inbound Gateway for receiving and replying to incoming IP packets.
Defines an inbound Gateway for receiving and replying to incoming tcp messages.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="gatewayType">
<xsd:attribute name="pool-size" type="xsd:string" />
<xsd:attribute name="receive-buffer-size" type="xsd:string" />
<xsd:attribute name="custom-socket-reader-class-name" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
If message-format = 'custom' you must provide a sub class of the appropriate type to implement
the custom message format. See java docs for TcpNetReceivingChannelAdapter and TcpNioReceivingChannelAdapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="custom-socket-writer-class-name" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
If message-format = 'custom' you must provide a sub class of the appropriate type to implement
the custom message format. See java docs for TcpNetSendingChannelAdapter and TcpNioSendingChannelAdapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-gateway">
<xsd:annotation>
<xsd:documentation>
Defines an outbound Gateway for sending and receiving responses over TCP.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="gatewayType">
<xsd:attribute name="host" type="xsd:string" />
<xsd:attribute name="custom-socket-reader-class-name" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
@@ -125,7 +155,7 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="protocol">
<xsd:attribute name="protocol" use="required">
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="tcp" />
@@ -133,6 +163,8 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="using-nio" type="xsd:string" />
<xsd:attribute name="using-direct-buffers" type="xsd:string" />
<xsd:attribute name="check-length" type="xsd:string" />
<xsd:attribute name="multicast" type="xsd:string" />
</xsd:extension>
@@ -169,6 +201,7 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
<xsd:attribute name="request-timeout" type="xsd:string"/>
<xsd:attribute name="reply-timeout" type="xsd:string"/>
<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
<xsd:attribute name="receive-buffer-size" type="xsd:string" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -179,7 +212,6 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
<xsd:attribute name="so-receive-buffer-size" type="xsd:string" />
<xsd:attribute name="so-send-buffer-size" type="xsd:string" />
<xsd:attribute name="so-timeout" type="xsd:string" />
<xsd:attribute name="using-nio" type="xsd:string" />
<xsd:attribute name="message-format">
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
@@ -190,7 +222,6 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="using-direct-buffers" type="xsd:string" />
<xsd:attribute name="so-keep-alive" type="xsd:string" />
</xsd:complexType>

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

@@ -189,5 +189,21 @@
so-timeout="126"
/>
<ip:outbound-gateway id="simpleOutGateway"
request-channel="tcpChannel"
reply-channel="replyChannel"
custom-socket-reader-class-name="org.springframework.integration.ip.tcp.CustomNetSocketReader"
custom-socket-writer-class-name="org.springframework.integration.ip.tcp.CustomNetSocketWriter"
message-format="crlf"
host="localhost"
port="#{tcpIpUtils.findAvailableServerSocket(6500)}"
receive-buffer-size="223"
so-keep-alive="true"
so-receive-buffer-size="224"
so-send-buffer-size="225"
so-timeout="226"
/>
</beans>

51
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

@@ -23,12 +23,15 @@ 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.context.ApplicationContext;
import org.springframework.integration.ip.tcp.CustomNetSocketReader;
import org.springframework.integration.ip.tcp.CustomNetSocketWriter;
import org.springframework.integration.ip.tcp.CustomNioSocketReader;
import org.springframework.integration.ip.tcp.CustomNioSocketWriter;
import org.springframework.integration.ip.tcp.MessageFormats;
import org.springframework.integration.ip.tcp.SimpleTcpNetInboundGateway;
import org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway;
import org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGatewayTests;
import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler;
import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter;
@@ -50,6 +53,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
public class ParserUnitTests {
@Autowired
ApplicationContext ctx;
@Autowired
@Qualifier(value="testInUdp")
UnicastReceivingChannelAdapter udpIn;
@@ -92,6 +98,10 @@ public class ParserUnitTests {
@Autowired
SimpleTcpNetInboundGateway simpleTcpNetInboundGateway;
@Autowired
@Qualifier(value="org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway#0")
SimpleTcpNetOutboundGateway simpleTcpNetOutboundGateway;
@Test
public void testInUdp() {
@@ -120,7 +130,7 @@ public class ParserUnitTests {
public void testInTcpNio() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInNio);
assertTrue(tcpInNio.getPort() >= 5200);
assertEquals(CustomNioSocketReader.class, dfa.getPropertyValue("customSocketReader"));
assertEquals(CustomNioSocketReader.class, dfa.getPropertyValue("customSocketReaderClass"));
assertEquals(false, dfa.getPropertyValue("usingDirectBuffers"));
assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat"));
assertEquals(27, dfa.getPropertyValue("poolSize"));
@@ -134,7 +144,7 @@ public class ParserUnitTests {
public void testInTcpNioDirect() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInNioDirect);
assertTrue(tcpInNioDirect.getPort() >= 5300);
assertEquals(CustomNioSocketReader.class, dfa.getPropertyValue("customSocketReader"));
assertEquals(CustomNioSocketReader.class, dfa.getPropertyValue("customSocketReaderClass"));
assertEquals(true, dfa.getPropertyValue("usingDirectBuffers"));
assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat"));
assertEquals(27, dfa.getPropertyValue("poolSize"));
@@ -148,7 +158,7 @@ public class ParserUnitTests {
public void testInTcpNet() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInNet);
assertTrue(tcpInNet.getPort() >= 5400);
assertEquals(CustomNetSocketReader.class, dfa.getPropertyValue("customSocketReader"));
assertEquals(CustomNetSocketReader.class, dfa.getPropertyValue("customSocketReaderClass"));
assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat"));
assertEquals(27, dfa.getPropertyValue("poolSize"));
assertEquals(true, dfa.getPropertyValue("soKeepAlive"));
@@ -201,7 +211,7 @@ public class ParserUnitTests {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpOutNio);
assertTrue(tcpOutNio.getPort() >= 6200);
assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat"));
assertEquals(CustomNioSocketWriter.class, dfa.getPropertyValue("customSocketWriter"));
assertEquals(CustomNioSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass"));
assertEquals(true, dfa.getPropertyValue("soKeepAlive"));
assertEquals(3, dfa.getPropertyValue("soLinger"));
assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
@@ -217,7 +227,7 @@ public class ParserUnitTests {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpOutNioDirect);
assertTrue(tcpOutNioDirect.getPort() >= 6300);
assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat"));
assertEquals(CustomNioSocketWriter.class, dfa.getPropertyValue("customSocketWriter"));
assertEquals(CustomNioSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass"));
assertEquals(true, dfa.getPropertyValue("soKeepAlive"));
assertEquals(3, dfa.getPropertyValue("soLinger"));
assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
@@ -233,7 +243,7 @@ public class ParserUnitTests {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpOutNet);
assertTrue(tcpOutNet.getPort() >= 6400);
assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat"));
assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriter"));
assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass"));
assertEquals(true, dfa.getPropertyValue("soKeepAlive"));
assertEquals(3, dfa.getPropertyValue("soLinger"));
assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
@@ -251,8 +261,8 @@ public class ParserUnitTests {
TcpNetReceivingChannelAdapter delegate = (TcpNetReceivingChannelAdapter) dfa
.getPropertyValue("delegate");
DirectFieldAccessor delegateDfa = new DirectFieldAccessor(delegate);
assertEquals(CustomNetSocketReader.class, delegateDfa.getPropertyValue("customSocketReader"));
assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriter"));
assertEquals(CustomNetSocketReader.class, delegateDfa.getPropertyValue("customSocketReaderClass"));
assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass"));
assertEquals(true, dfa.getPropertyValue("soKeepAlive"));
assertEquals(123, dfa.getPropertyValue("receiveBufferSize"));
assertEquals(124, dfa.getPropertyValue("soReceiveBufferSize"));
@@ -260,5 +270,28 @@ public class ParserUnitTests {
assertEquals(126, dfa.getPropertyValue("soTimeout"));
assertEquals(23, dfa.getPropertyValue("poolSize"));
}
}
@Test
public void testOutGateway() {
DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetOutboundGateway);
assertTrue(simpleTcpNetOutboundGateway.getPort() >= 6500);
assertEquals(MessageFormats.FORMAT_CRLF, dfa.getPropertyValue("messageFormat"));
TcpNetSendingMessageHandler handler = (TcpNetSendingMessageHandler) dfa
.getPropertyValue("handler");
DirectFieldAccessor delegateDfa = new DirectFieldAccessor(handler);
assertEquals(CustomNetSocketReader.class, dfa.getPropertyValue("customSocketReaderClass"));
assertEquals(CustomNetSocketWriter.class, delegateDfa.getPropertyValue("customSocketWriterClass"));
assertEquals(true, delegateDfa.getPropertyValue("soKeepAlive"));
assertEquals(224, dfa.getPropertyValue("soReceiveBufferSize"));
assertEquals(225, delegateDfa.getPropertyValue("soSendBufferSize"));
assertEquals(226, delegateDfa.getPropertyValue("soTimeout"));
}
@Test
public void test() {
String[] beanDefinitionNames = ctx.getBeanDefinitionNames();
for (String x : beanDefinitionNames)
System.out.println(x);
}
}

18
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests-context.xml Normal file
View File

@@ -0,0 +1,18 @@
<?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"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<import resource="SimpleTcpNetInboundGatewayTests-context.xml" />
<bean id="tcpOutGateway" class="org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway">
<constructor-arg value="localhost"/>
<constructor-arg value="#{gatewayCrLf.port}"/>
<!-- <property name="" value=""/>-->
</bean>
</beans>

94
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests.java Normal file
View File

@@ -0,0 +1,94 @@
/*
* 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 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.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.junit.Assert.assertEquals;
/**
* @author Gary Russell
*
*/
@ContextConfiguration(locations="SimpleTcpNetInboundGatewayTests-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class SimpleTcpNetOutboundGatewayTests {
@Autowired
@Qualifier("gatewayCrLf")
private SimpleTcpNetInboundGateway inboundGatewayCrLf;
@Autowired
@Qualifier("gatewayStxEtx")
private SimpleTcpNetInboundGateway inboundGatewayStxEtx;
@Autowired
@Qualifier("gatewayLength")
private SimpleTcpNetInboundGateway inboundGatewayLength;
@Autowired
@Qualifier("gatewayCustom")
private SimpleTcpNetInboundGateway inboundGatewayCustom;
@Test
public void testOutboundCrLf() throws Exception {
SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway
("localhost", inboundGatewayCrLf.getPort());
gateway.setMessageFormat(MessageFormats.FORMAT_CRLF);
Message<String> message = MessageBuilder.withPayload("test").build();
byte[] bytes = (byte[]) gateway.handleRequestMessage(message);
assertEquals("echo:test", new String(bytes));
}
@Test
public void testOutboundStxEtx() throws Exception {
SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway
("localhost", inboundGatewayStxEtx.getPort());
gateway.setMessageFormat(MessageFormats.FORMAT_STX_ETX);
Message<String> message = MessageBuilder.withPayload("test").build();
byte[] bytes = (byte[]) gateway.handleRequestMessage(message);
assertEquals("echo:test", new String(bytes));
}
@Test
public void testOutboundLength() throws Exception {
SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway
("localhost", inboundGatewayLength.getPort());
gateway.setMessageFormat(MessageFormats.FORMAT_LENGTH_HEADER);
Message<String> message = MessageBuilder.withPayload("test").build();
byte[] bytes = (byte[]) gateway.handleRequestMessage(message);
assertEquals("echo:test", new String(bytes));
}
@Test
public void testOutboundCustom() throws Exception {
SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway
("localhost", inboundGatewayCustom.getPort());
gateway.setMessageFormat(MessageFormats.FORMAT_CUSTOM);
gateway.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNetSocketReader");
gateway.setCustomSocketWriterClassName("org.springframework.integration.ip.tcp.CustomNetSocketWriter");
Message<String> message = MessageBuilder.withPayload("test").build();
byte[] bytes = (byte[]) gateway.handleRequestMessage(message);
assertEquals("echo:test", new String(bytes).trim());
}
}

2
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java
View File

@@ -104,7 +104,7 @@ public class UdpUnicastEndToEndTests implements Runnable {
// tell the receiver to we're done
doneProcessing.countDown();
}
assertTrue(firstReceived.await(2, TimeUnit.SECONDS));
assertTrue(firstReceived.await(5, TimeUnit.SECONDS));
assertEquals(testingIpText, new String(finalMessage.getPayload()));
}