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

Inbound tcp channel adapters namespace support and tests.

Browse Source
This commit is contained in:
Gary Russell
2010-02-22 00:50:45 +00:00
parent d2d1b8f0dd
commit 512859f3c7
19 changed files with 496 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java
View File

@@ -49,6 +49,13 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
this.port = port;
}
/**
*
* @return The port on which this receiver is listening.
*/
public int getPort() {
return port;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.SocketOptions#setSoTimeout(int)
@@ -83,4 +90,12 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
taskScheduler.schedule(this, new Date());
}
/* (non-Javadoc)
* @see org.springframework.integration.endpoint.AbstractEndpoint#doStop()
*/
@Override
protected void doStop() {
this.active = false;
}
}

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

@@ -21,6 +21,7 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.MessageFormats;
import org.springframework.util.StringUtils;
/**
@@ -45,10 +46,12 @@ public abstract class IpAdapterParserUtils {
static final String SO_TIMEOUT = "so-timeout";
static final String SO_RECEIVE_BUFFER_SIZE = "so-receive-bufffer-size";
static final String SO_RECEIVE_BUFFER_SIZE = "so-receive-buffer-size";
static final String SO_SEND_BUFFER_SIZE = "so-send-buffer-size";
static final String SO_KEEP_ALIVE = "so-keep-alive";
static final String RECEIVE_BUFFER_SIZE = "receive-buffer-size";
static final String POOL_SIZE = "pool-size";
@@ -64,6 +67,17 @@ public abstract class IpAdapterParserUtils {
static final String MIN_ACKS_SUCCESS = "min-acks-for-success";
static final String TIME_TO_LIVE = "time-to-live";
static final String USING_NIO = "using-nio";
static final String USING_DIRECT_BUFFERS = "using-direct-buffers";
static final String MESSAGE_FORMAT = "message-format";
static final String CUSTOM_SOCKET_READER_CLASS_NAME =
"custom-socket-reader-class-name";
// static final String
/**
@@ -76,7 +90,7 @@ public abstract class IpAdapterParserUtils {
* @param attributeName the name of the attribute whose value will be
* used to populate the property
*/
public static void addConstuctirValueIfAttributeDefined(BeanDefinitionBuilder builder,
public static void addConstuctorValueIfAttributeDefined(BeanDefinitionBuilder builder,
Element element, String attributeName, boolean trueFalse) {
String attributeValue = element.getAttribute(attributeName);
if (StringUtils.hasText(attributeValue)) {
@@ -131,6 +145,45 @@ public abstract class IpAdapterParserUtils {
return multicast;
}
/**
* Gets the use-nio attribute, if present; if not returns 'false'.
* @param element
* @return The value of the attribute or false.
*/
static String getUseNio(Element element) {
String useNio = element.getAttribute(IpAdapterParserUtils.USING_NIO);
if (!StringUtils.hasText(useNio)) {
useNio = "false";
}
return useNio;
}
/**
* Gets the message-format attribute, if present; if not returns
* {@link MessageFormats#FORMAT_LENGTH_HEADER}.
* @param element
* @return The value of the attribute or false.
*/
static Integer getMessageFormat(Element element) {
String useNio = element.getAttribute(IpAdapterParserUtils.MESSAGE_FORMAT);
if (!StringUtils.hasText(useNio)) {
return MessageFormats.FORMAT_LENGTH_HEADER;
}
if (useNio.equals("length-header")) {
return MessageFormats.FORMAT_LENGTH_HEADER;
}
if (useNio.equals("stx-etx")) {
return MessageFormats.FORMAT_STX_ETX;
}
if (useNio.equals("crlf")) {
return MessageFormats.FORMAT_CRLF;
}
if (useNio.equals("custom")) {
return MessageFormats.FORMAT_CUSTOM;
}
return MessageFormats.FORMAT_LENGTH_HEADER;
}
/**
* Sets the common port attributes on the bean being built (timeout, receive buffer size,
* send buffer size).

97
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java
View File

@@ -22,8 +22,11 @@ import org.springframework.beans.factory.BeanCreationException;
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.AbstractChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter;
import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter;
import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
import org.springframework.util.StringUtils;
@@ -38,33 +41,12 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
String protocol = IpAdapterParserUtils.getProtocol(element);
String multicast = IpAdapterParserUtils.getMulticast(element);
BeanDefinitionBuilder builder = null;
if (protocol.equals("tcp")) {
throw new BeanCreationException("tcp not yet supported");
builder = parseTcp(element);
} else if (protocol.equals("udp")) {
builder = parseUdp(element);
}
else if (protocol.equals("udp")) {
if (multicast.equals("false")) {
builder = BeanDefinitionBuilder
.genericBeanDefinition(UnicastReceivingChannelAdapter.class);
}
else {
builder = BeanDefinitionBuilder
.genericBeanDefinition(MulticastReceivingChannelAdapter.class);
String mcAddress = element
.getAttribute(IpAdapterParserUtils.MULTICAST_ADDRESS);
if (!StringUtils.hasText(mcAddress)) {
throw new BeanCreationException(
IpAdapterParserUtils.MULTICAST_ADDRESS
+ " is required for a multicast UDP/IP channel adapter");
}
builder.addConstructorArgValue(mcAddress);
}
}
String port = IpAdapterParserUtils.getPort(element);
builder.addConstructorArgValue(port);
IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.CHECK_LENGTH, true);
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
@@ -75,4 +57,71 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser
return builder.getBeanDefinition();
}
/**
* @param element
* @param builder
*/
private void addPortToConstructor(Element element,
BeanDefinitionBuilder builder) {
String port = IpAdapterParserUtils.getPort(element);
builder.addConstructorArgValue(port);
}
/**
* @param element
* @return
*/
private BeanDefinitionBuilder parseUdp(Element element) {
BeanDefinitionBuilder builder;
String multicast = IpAdapterParserUtils.getMulticast(element);
if (multicast.equals("false")) {
builder = BeanDefinitionBuilder
.genericBeanDefinition(UnicastReceivingChannelAdapter.class);
}
else {
builder = BeanDefinitionBuilder
.genericBeanDefinition(MulticastReceivingChannelAdapter.class);
String mcAddress = element
.getAttribute(IpAdapterParserUtils.MULTICAST_ADDRESS);
if (!StringUtils.hasText(mcAddress)) {
throw new BeanCreationException(
IpAdapterParserUtils.MULTICAST_ADDRESS
+ " is required for a multicast UDP/IP channel adapter");
}
builder.addConstructorArgValue(mcAddress);
}
addPortToConstructor(element, builder);
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.CHECK_LENGTH, true);
return builder;
}
/**
* @param element
* @return
*/
private BeanDefinitionBuilder parseTcp(Element element) {
BeanDefinitionBuilder builder;
String useNio = IpAdapterParserUtils.getUseNio(element);
if (useNio.equals("false")) {
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNetReceivingChannelAdapter.class);
}
else {
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNioReceivingChannelAdapter.class);
}
addPortToConstructor(element, builder);
builder.addPropertyValue(
Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT),
IpAdapterParserUtils.getMessageFormat(element));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.CUSTOM_SOCKET_READER_CLASS_NAME);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.USING_DIRECT_BUFFERS);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_KEEP_ALIVE);
return builder;
}
}

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

@@ -65,15 +65,15 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
builder.addConstructorArgValue(host);
String port = IpAdapterParserUtils.getPort(element);
builder.addConstructorArgValue(port);
IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.CHECK_LENGTH, true);
IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.ACK, true);
IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.ACK_HOST, false);
IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.ACK_PORT, false);
IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder,
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.ACK_TIMEOUT, false);
String ack = element.getAttribute(IpAdapterParserUtils.ACK);
if (ack.equals("true")) {

11
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java
View File

@@ -55,15 +55,6 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
super(port);
}
/* (non-Javadoc)
* @see org.springframework.integration.endpoint.AbstractEndpoint#doStop()
*/
@Override
protected void doStop() {
// TODO Auto-generated method stub
}
/**
* Creates the ThreadPoolTaskScheduler, if necessary, and calls
* {@link #server()}.
@@ -129,7 +120,7 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketReader(String customSocketReaderClassName) throws ClassNotFoundException {
public void setCustomSocketReaderClassName(String customSocketReaderClassName) throws ClassNotFoundException {
this.customSocketReader = (Class<SocketReader>) Class.forName(customSocketReaderClassName);
}

9
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java
View File

@@ -94,7 +94,14 @@ public class NioSocketReader extends AbstractSocketReader {
return false;
}
}
assembledData = dataPart.array();
if (usingDirectBuffers) {
byte[] assembledData = new byte[dataPart.capacity()];
dataPart.flip();
dataPart.get(assembledData);
this.assembledData = assembledData;
} else {
assembledData = dataPart.array();
}
lengthPart = dataPart = null;
return true;
} catch (Exception e) {

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

@@ -35,6 +35,7 @@ import org.springframework.integration.core.Message;
public class TcpNetReceivingChannelAdapter extends
AbstractTcpReceivingChannelAdapter {
protected ServerSocket serverSocket;
/**
* Constructs a TcpNetReceivingChannelAdapter that listens on the port.
* @param port The port.
@@ -52,11 +53,11 @@ public class TcpNetReceivingChannelAdapter extends
*/
@Override
protected void server() {
while (true) {
while (active) {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
serverSocket = ServerSocketFactory.getDefault().createServerSocket(port);
while (true) {
final Socket socket = server.accept();
final Socket socket = serverSocket.accept();
setSocketOptions(socket);
this.threadPoolTaskScheduler.execute(new Runnable() {
public void run() {
@@ -64,6 +65,15 @@ public class TcpNetReceivingChannelAdapter extends
}});
}
} catch (IOException e) {
if (!active) {
if (serverSocket != null) {
try {
serverSocket.close();
} catch (IOException e1) {}
}
serverSocket = null;
return;
}
// TODO Auto-generated catch block
e.printStackTrace();
}
@@ -116,5 +126,15 @@ public class TcpNetReceivingChannelAdapter extends
}
}
}
@Override
protected void doStop() {
super.doStop();
try {
this.serverSocket.close();
}
catch (Exception e) {
// ignore
}
}
}

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

@@ -41,6 +41,8 @@ import org.springframework.integration.core.Message;
public class TcpNioReceivingChannelAdapter extends
AbstractTcpReceivingChannelAdapter {
protected ServerSocketChannel serverChannel;
/**
* Constructs a TcpNioReceivingChannelAdapter to listen on the port.
* @param port The port.
@@ -58,14 +60,24 @@ public class TcpNioReceivingChannelAdapter extends
@Override
protected void server() {
try {
final ServerSocketChannel server = ServerSocketChannel.open();
server.configureBlocking(false);
server.socket().bind(new InetSocketAddress(port));
serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.socket().bind(new InetSocketAddress(port));
final Selector selector = Selector.open();
server.register(selector, SelectionKey.OP_ACCEPT);
doSelect(server, selector);
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
doSelect(serverChannel, selector);
} catch (IOException e) {
if (!active) {
try {
serverChannel.close();
} catch (IOException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
serverChannel = null;
return;
}
// TODO Auto-generated catch block
e.printStackTrace();
}
@@ -87,7 +99,7 @@ public class TcpNioReceivingChannelAdapter extends
*/
private void doSelect(ServerSocketChannel server, final Selector selector)
throws IOException, ClosedChannelException, SocketException {
while (true) {
while (active) {
int selectionCount = selector.select();
if (logger.isDebugEnabled())
logger.debug("SelectionCount: " + selectionCount);
@@ -184,5 +196,16 @@ public class TcpNioReceivingChannelAdapter extends
}
}
@Override
protected void doStop() {
super.doStop();
try {
this.serverChannel.close();
}
catch (Exception e) {
// ignore
}
}
}

2
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
View File

@@ -196,7 +196,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
@Override
protected void doStop() {
this.active = false;
super.doStop();
try {
this.socket.close();
}

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

@@ -38,6 +38,14 @@
<xsd:attribute name="pool-size" type="xsd:string" />
<xsd:attribute name="receive-buffer-size" type="xsd:string" />
<xsd:attribute name="multicast-address" 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:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -85,6 +93,20 @@
<xsd:attribute name="so-timeout" type="xsd:string" />
<xsd:attribute name="check-length" type="xsd:string" />
<xsd:attribute name="multicast" type="xsd:string" />
<xsd:attribute name="using-nio" type="xsd:boolean" />
<xsd:attribute name="message-format">
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="length-header" />
<xsd:enumeration value="stx-etx" />
<xsd:enumeration value="crlf" />
<xsd:enumeration value="custom" />
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="using-direct-buffers" type="xsd:boolean" />
<xsd:attribute name="so-keep-alive" type="xsd:boolean" />
</xsd:complexType>
</xsd:schema>