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

INT-781 Outbound TCP Adapters Namespace & Tests (Work in progress)

Browse Source
This commit is contained in:
Gary Russell
2010-02-26 22:40:19 +00:00
parent 6e2295b625
commit 81584599ae
12 changed files with 221 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java
View File

@@ -82,4 +82,12 @@ public abstract class AbstractInternetProtocolSendingMessageHandler implements M
this.soSendBufferSize = size;
}
/**
* @return the port
*/
public int getPort() {
return port;
}
}

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

@@ -77,8 +77,15 @@ public abstract class IpAdapterParserUtils {
static final String CUSTOM_SOCKET_READER_CLASS_NAME =
"custom-socket-reader-class-name";
// static final String
static final String CUSTOM_SOCKET_WRITER_CLASS_NAME =
"custom-socket-writer-class-name";
static final String SO_LINGER = "so-linger";
static final String SO_TCP_NODELAY = "so-tcp-nodelay";
static final String SO_TRAFFIC_CLASS = "so-traffic-class";
/**
* Adds a constructor-arg to the bean definition with the value

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

@@ -16,17 +16,19 @@
package org.springframework.integration.ip.config;
import org.w3c.dom.Element;
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.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler;
import org.springframework.integration.ip.tcp.TcpNioSendingMessageHandler;
import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
import org.springframework.integration.ip.udp.UnicastSendingMessageHandler;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* @author Gary Russell
@@ -38,25 +40,21 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
String protocol = IpAdapterParserUtils.getProtocol(element);
BeanDefinitionBuilder builder = null;
if (protocol.equals("tcp")) {
throw new BeanCreationException("tcp not yet supported");
builder = parseTcp(element);
}
else if (protocol.equals("udp")) {
String multicast = IpAdapterParserUtils.getMulticast(element);
if (multicast.equals("true")) {
builder = BeanDefinitionBuilder
.genericBeanDefinition(MulticastSendingMessageHandler.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.MIN_ACKS_SUCCESS,
"minAcksForSuccess");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.TIME_TO_LIVE,
"timeToLive");
}
else {
builder = BeanDefinitionBuilder
.genericBeanDefinition(UnicastSendingMessageHandler.class);
}
builder = parseUdp(element);
}
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
return builder.getBeanDefinition();
}
/**
* @param element
* @param builder
*/
private void addHostAndPortToConstructor(Element element,
BeanDefinitionBuilder builder) {
String host = element.getAttribute(IpAdapterParserUtils.HOST);
if (!StringUtils.hasText(host)) {
throw new BeanCreationException(IpAdapterParserUtils.HOST
@@ -65,6 +63,30 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
builder.addConstructorArgValue(host);
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("true")) {
builder = BeanDefinitionBuilder
.genericBeanDefinition(MulticastSendingMessageHandler.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.MIN_ACKS_SUCCESS,
"minAcksForSuccess");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.TIME_TO_LIVE,
"timeToLive");
}
else {
builder = BeanDefinitionBuilder
.genericBeanDefinition(UnicastSendingMessageHandler.class);
}
addHostAndPortToConstructor(element, builder);
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.CHECK_LENGTH, true);
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
@@ -91,10 +113,43 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
+ " must be supplied");
}
}
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
return builder.getBeanDefinition();
return builder;
}
/**
* @param element
* @return
*/
private BeanDefinitionBuilder parseTcp(Element element) {
BeanDefinitionBuilder builder;
String useNio = IpAdapterParserUtils.getUseNio(element);
if (useNio.equals("false")) {
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNetSendingMessageHandler.class);
}
else {
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNioSendingMessageHandler.class);
}
addHostAndPortToConstructor(element, 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);
return builder;
}
}

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

@@ -57,7 +57,7 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
*/
public void run() {
if (logger.isDebugEnabled()) {
logger.debug(this.getClass().getSimpleName() + " running...");
logger.debug(this.getClass().getSimpleName() + " running on port: " + port);
}
if (this.active && this.threadPoolTaskScheduler == null) {
this.threadPoolTaskScheduler = new ThreadPoolTaskScheduler();

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

@@ -50,7 +50,7 @@ public abstract class AbstractTcpSendingMessageHandler extends
protected boolean soKeepAlive = false;
protected int messageFormat;
protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
protected boolean blockingWrite = false;
@@ -111,7 +111,7 @@ public abstract class AbstractTcpSendingMessageHandler extends
.newSingleThreadExecutor(new ThreadFactory() {
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("UDP-Ack-Handler");
thread.setName("Tcp-NonBlocking-Handler-port-" + port);
thread.setDaemon(true);
return thread;
}
@@ -131,9 +131,16 @@ public abstract class AbstractTcpSendingMessageHandler extends
protected void doWrite(Message<?> message) {
try {
byte[] bytes = mapper.fromMessage(message);
this.getWriter().write(bytes);
SocketWriter writer = this.getWriter();
if (writer == null) {
throw new MessageMappingException("Failed to create SocketWriter");
}
writer.write(bytes);
} catch (Exception e) {
writer = null;
if (e instanceof MessageMappingException) {
throw (MessageMappingException) e;
}
throw new MessageMappingException("Failed to map message", e);
}
}

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

@@ -230,6 +230,10 @@ public class NioSocketReader extends AbstractSocketReader {
protected void readChannel(ByteBuffer buffer) throws IOException {
try {
int len = channel.read(buffer);
if (len < 0) {
logger.debug("Socket closed");
throw new IOException("Socket closed");
}
if (logger.isDebugEnabled()) {
logger.debug("Read " + len + " bytes, buffer is now at " +
buffer.position() + " of " +

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

@@ -67,7 +67,7 @@ public class TcpNioReceivingChannelAdapter extends
try {
serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.socket().bind(new InetSocketAddress(port));
serverChannel.socket().bind(new InetSocketAddress(port), 10);
final Selector selector = Selector.open();
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
doSelect(serverChannel, selector);
@@ -107,7 +107,7 @@ public class TcpNioReceivingChannelAdapter extends
while (active) {
int selectionCount = selector.select();
if (logger.isDebugEnabled())
logger.debug("SelectionCount: " + selectionCount);
logger.debug("Port " + port + " SelectionCount: " + selectionCount);
if (selectionCount > 0) {
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();

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

@@ -66,6 +66,11 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T
<xsd:attribute name="ack-port" type="xsd:string" />
<xsd:attribute name="ack-timeout" type="xsd:string" />
<xsd:attribute name="time-to-live" type="xsd:string" />
<xsd:attribute name="custom-socket-writer-class-name" type="xsd:string" />
<xsd:attribute name="so-linger" type="xsd:string" />
<xsd:attribute name="so-tcp-nodelay" type="xsd:boolean" />
<xsd:attribute name="so-traffic-class" type="xsd:string" />
<xsd:attribute name="blocking-write" type="xsd:boolean" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

68
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java
View File

@@ -18,17 +18,29 @@ package org.springframework.integration.ip.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter;
import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler;
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.Utils;
import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -37,7 +49,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @author Gary Russell
*
*/
@ContextConfiguration(locations="inboundAdapters.xml")
@ContextConfiguration(locations={"inboundAdapters.xml"
,"outboundAdapters.xml"
})
@RunWith(SpringJUnit4ClassRunner.class)
public class IpChannelAdapterParserTests {
@@ -72,11 +86,15 @@ public class IpChannelAdapterParserTests {
@Qualifier(value="udp1")
UnicastReceivingChannelAdapter udp1;
@Autowired
@Qualifier(value="org.springframework.integration.ip.tcp.TcpNioSendingMessageHandler#0")
TcpNioSendingMessageHandler tcpOut1;
@SuppressWarnings("unchecked")
@Test
public void testTcpInbound1() {
Utils.testSendFragmented(tcp1.getPort());
Message<byte[]> message = (Message<byte[]>) channel.receive();
Message<byte[]> message = (Message<byte[]>) channel.receive(10000);
assertNotNull(message);
assertEquals("xx", new String(message.getPayload()));
}
@@ -85,7 +103,7 @@ public class IpChannelAdapterParserTests {
@Test
public void testTcpInbound2() {
Utils.testSendFragmented(tcp2.getPort());
Message<byte[]> message = (Message<byte[]>) channel.receive();
Message<byte[]> message = (Message<byte[]>) channel.receive(10000);
assertNotNull(message);
assertEquals("xx", new String(message.getPayload()));
}
@@ -94,7 +112,7 @@ public class IpChannelAdapterParserTests {
@Test
public void testTcpInbound3() {
Utils.testSendFragmented(tcp3.getPort());
Message<byte[]> message = (Message<byte[]>) channel.receive();
Message<byte[]> message = (Message<byte[]>) channel.receive(10000);
assertNotNull(message);
assertEquals("xx", new String(message.getPayload()));
}
@@ -103,10 +121,10 @@ public class IpChannelAdapterParserTests {
@Test
public void testTcpInbound4() {
Utils.testSendStxEtx(tcp4.getPort(), null);
Message<byte[]> message = (Message<byte[]>) channel.receive();
Message<byte[]> message = (Message<byte[]>) channel.receive(10000);
assertNotNull(message);
assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload()));
message = (Message<byte[]>) channel.receive();
message = (Message<byte[]>) channel.receive(10000);
assertNotNull(message);
assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload()));
}
@@ -115,10 +133,10 @@ public class IpChannelAdapterParserTests {
@Test
public void testTcpInbound5() {
Utils.testSendCrLf(tcp5.getPort(), null);
Message<byte[]> message = (Message<byte[]>) channel.receive();
Message<byte[]> message = (Message<byte[]>) channel.receive(10000);
assertNotNull(message);
assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload()));
message = (Message<byte[]>) channel.receive();
message = (Message<byte[]>) channel.receive(10000);
assertNotNull(message);
assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload()));
}
@@ -127,11 +145,11 @@ public class IpChannelAdapterParserTests {
@Test
public void testTcpInbound6() {
Utils.testSendStxEtx(tcp6.getPort(), null);
Message<byte[]> message = (Message<byte[]>) channel.receive();
Message<byte[]> message = (Message<byte[]>) channel.receive(10000);
assertNotNull(message);
assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003",
new String(message.getPayload()));
message = (Message<byte[]>) channel.receive();
message = (Message<byte[]>) channel.receive(10000);
assertNotNull(message);
assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003",
new String(message.getPayload()));
@@ -141,4 +159,34 @@ public class IpChannelAdapterParserTests {
public void testUdpInbound1() {
assertNotNull(udp1);
}
@SuppressWarnings("unchecked")
@Test
public void testTcpOutbound1() {
setPort(tcpOut1, tcp1);
Message<String> message = MessageBuilder.withPayload("TESTING").build();
tcpOut1.handleMessage(message);
Message<byte[]> mOut = (Message<byte[]>) channel.receive(10000);
assertNotNull(mOut);
assertEquals("TESTING", new String(mOut.getPayload()));
}
private void setPort(AbstractInternetProtocolSendingMessageHandler tcpSMA,
AbstractInternetProtocolReceivingChannelAdapter tcpRCA) {
try {
int port = tcpRCA.getPort();
Field portField = tcpSMA.getClass().getSuperclass().getSuperclass().getDeclaredField("port");
portField.setAccessible(true);
assertEquals(9999, portField.getInt(tcpSMA));
portField.setInt(tcpSMA, port);
InetSocketAddress address = new InetSocketAddress("localhost", port);
Field addressField = tcpSMA.getClass().getSuperclass().getSuperclass().getDeclaredField("destinationAddress");
addressField.setAccessible(true);
addressField.set(tcpSMA, address);
} catch (Exception e) {
fail("Couldn't fix port:" + e);
}
}
}

18
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml
View File

@@ -9,12 +9,14 @@
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">
<beans:bean id="tcpIpUtils" class="org.springframework.integration.ip.tcp.Utils" />
<!-- nio without direct buffers -->
<ip:inbound-channel-adapter id="tcp1"
channel="channel"
protocol="tcp"
port="9876"
port="#{tcpIpUtils.findAvailableServerSocket(5000)}"
message-format="length-header"
using-nio="true"
using-direct-buffers="false"
@@ -27,7 +29,7 @@
<ip:inbound-channel-adapter id="tcp2"
channel="channel"
protocol="tcp"
port="9877"
port="#{tcpIpUtils.findAvailableServerSocket(6000)}"
message-format="length-header"
using-nio="true"
using-direct-buffers="true"
@@ -40,7 +42,7 @@
<ip:inbound-channel-adapter id="tcp3"
channel="channel"
protocol="tcp"
port="9878"
port="#{tcpIpUtils.findAvailableServerSocket(7000)}"
message-format="length-header"
using-nio="false"
pool-size="2"
@@ -52,7 +54,7 @@
<ip:inbound-channel-adapter id="tcp4"
channel="channel"
protocol="tcp"
port="9879"
port="#{tcpIpUtils.findAvailableServerSocket(8000)}"
message-format="stx-etx"
using-nio="false"
pool-size="2"
@@ -64,7 +66,7 @@
<ip:inbound-channel-adapter id="tcp5"
channel="channel"
protocol="tcp"
port="9880"
port="#{tcpIpUtils.findAvailableServerSocket(9000)}"
message-format="crlf"
using-nio="false"
pool-size="2"
@@ -76,7 +78,7 @@
<ip:inbound-channel-adapter id="tcp6"
channel="channel"
protocol="tcp"
port="9881"
port="#{tcpIpUtils.findAvailableServerSocket(10000)}"
message-format="custom"
custom-socket-reader-class-name="org.springframework.integration.ip.tcp.CustomNetSocketReader"
using-nio="false"
@@ -88,7 +90,7 @@
<ip:inbound-channel-adapter id="udp1"
channel="channel"
protocol="udp"
port="9976"
port="#{tcpIpUtils.findAvailableServerSocket(11000)}"
receive-buffer-size="500"
multicast="false"
check-length="true" />

33
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/outboundAdapters.xml Normal file
View File

@@ -0,0 +1,33 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:ip="http://www.springframework.org/schema/integration/ip"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
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">
<channel id="inChannel"/>
<!-- nio without direct buffers -->
<ip:outbound-channel-adapter id="tcpOut1"
channel="inChannel"
protocol="tcp"
host="localhost"
port="9999"
message-format="length-header"
using-nio="true"
using-direct-buffers="false"
so-keep-alive="true"
so-timeout="100000"
/>
<beans:bean id="taskScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
<beans:property name="daemon" value="true" />
<beans:property name="poolSize" value="20" />
</beans:bean>
</beans:beans>

9
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java
View File

@@ -78,6 +78,7 @@ public class Utils {
Thread thread = new Thread(new Runnable() {
public void run() {
try {
System.out.println("Connecting to " + port);
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
OutputStream os = socket.getOutputStream();
writeByte(os, 0);
@@ -168,8 +169,8 @@ public class Utils {
thread.start();
}
public static int findAvailableServerSocket() {
for (int i = 5678; i < 5878; i++) {
public static int findAvailableServerSocket(int seed) {
for (int i = seed; i < seed+200; i++) {
try {
ServerSocket sock = ServerSocketFactory.getDefault().createServerSocket(i);
sock.close();
@@ -178,4 +179,8 @@ public class Utils {
}
throw new RuntimeException("Cannot find a free server socket");
}
public static int findAvailableServerSocket() {
return findAvailableServerSocket(5678);
}
}