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

INT-781 Namespace cleanup; multi-client/multi-message tests (UDP and TCP)

Browse Source
This commit is contained in:
Gary Russell
2010-03-02 23:25:06 +00:00
parent 687e75f841
commit a732f3c255
23 changed files with 487 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,8 +18,8 @@ package org.springframework.integration.ip.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.MessageFormats;
import org.springframework.util.StringUtils;
@@ -86,8 +86,6 @@ public abstract class IpAdapterParserUtils {
static final String SO_TRAFFIC_CLASS = "so-traffic-class";
static final String BLOCKING_WRITE = "blocking-write";
/**
* Adds a constructor-arg to the bean definition with the value
@@ -110,18 +108,19 @@ public abstract class IpAdapterParserUtils {
/**
* Asserts that a protocol attribute (udp or tcp) is supplied,
* @param element
* @param parserContext
* @return The value of the attribute.
* @throws BeanCreationException if attribute not provided or invalid.
*/
static String getProtocol(Element element) {
static String getProtocol(Element element, ParserContext parserContext) {
String protocol = element.getAttribute(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE);
if (!StringUtils.hasText(protocol)) {
throw new BeanCreationException(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE +
" is required for an IP channel adapter");
parserContext.getReaderContext().error(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE +
" is required for an IP channel adapter", element);
}
if (!protocol.equals("tcp") && !protocol.equals("udp")) {
throw new BeanCreationException(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE +
" must be 'tcp' or 'udp' for an IP channel adapter");
parserContext.getReaderContext().error(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE +
" must be 'tcp' or 'udp' for an IP channel adapter", element);
}
return protocol;
}
@@ -129,14 +128,15 @@ public abstract class IpAdapterParserUtils {
/**
* Asserts that a port attribute is supplied.
* @param element
* @param parserContext
* @return The value of the attribute.
* @throws BeanCreationException if attribute is not provided.
*/
static String getPort(Element element) {
static String getPort(Element element, ParserContext parserContext) {
String port = element.getAttribute(IpAdapterParserUtils.PORT);
if (!StringUtils.hasText(port)) {
throw new BeanCreationException(IpAdapterParserUtils.PORT +
" is required for IP channel adapters");
parserContext.getReaderContext().error(IpAdapterParserUtils.PORT +
" is required for IP channel adapters", element);
}
return port;
}

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

@@ -18,7 +18,6 @@ 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;
@@ -40,13 +39,14 @@ import org.springframework.util.StringUtils;
public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser {
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
String protocol = IpAdapterParserUtils.getProtocol(element);
String protocol = IpAdapterParserUtils.getProtocol(element, parserContext);
BeanDefinitionBuilder builder = null;
if (protocol.equals("tcp")) {
builder = parseTcp(element);
builder = parseTcp(element, parserContext);
} else if (protocol.equals("udp")) {
builder = parseUdp(element);
builder = parseUdp(element, parserContext);
}
parserContext.extractSource(element);
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
@@ -60,18 +60,20 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser
/**
* @param element
* @param builder
* @param parserContext
*/
private void addPortToConstructor(Element element,
BeanDefinitionBuilder builder) {
String port = IpAdapterParserUtils.getPort(element);
BeanDefinitionBuilder builder, ParserContext parserContext) {
String port = IpAdapterParserUtils.getPort(element, parserContext);
builder.addConstructorArgValue(port);
}
/**
* @param element
* @param parserContext
* @return
*/
private BeanDefinitionBuilder parseUdp(Element element) {
private BeanDefinitionBuilder parseUdp(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder;
String multicast = IpAdapterParserUtils.getMulticast(element);
if (multicast.equals("false")) {
@@ -84,13 +86,14 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser
String mcAddress = element
.getAttribute(IpAdapterParserUtils.MULTICAST_ADDRESS);
if (!StringUtils.hasText(mcAddress)) {
throw new BeanCreationException(
parserContext.getReaderContext().error(
IpAdapterParserUtils.MULTICAST_ADDRESS
+ " is required for a multicast UDP/IP channel adapter");
+ " is required for a multicast UDP/IP channel adapter",
element);
}
builder.addConstructorArgValue(mcAddress);
}
addPortToConstructor(element, builder);
addPortToConstructor(element, builder, parserContext);
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.CHECK_LENGTH, true);
return builder;
@@ -98,9 +101,10 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser
/**
* @param element
* @param parserContext
* @return
*/
private BeanDefinitionBuilder parseTcp(Element element) {
private BeanDefinitionBuilder parseTcp(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder;
String useNio = IpAdapterParserUtils.getUseNio(element);
if (useNio.equals("false")) {
@@ -111,7 +115,7 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNioReceivingChannelAdapter.class);
}
addPortToConstructor(element, builder);
addPortToConstructor(element, builder, parserContext);
builder.addPropertyValue(
Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT),
IpAdapterParserUtils.getMessageFormat(element));

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

@@ -16,7 +16,6 @@
package org.springframework.integration.ip.config;
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;
@@ -37,13 +36,13 @@ import org.w3c.dom.Element;
public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
String protocol = IpAdapterParserUtils.getProtocol(element);
String protocol = IpAdapterParserUtils.getProtocol(element, parserContext);
BeanDefinitionBuilder builder = null;
if (protocol.equals("tcp")) {
builder = parseTcp(element);
builder = parseTcp(element, parserContext);
}
else if (protocol.equals("udp")) {
builder = parseUdp(element);
builder = parseUdp(element, parserContext);
}
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
return builder.getBeanDefinition();
@@ -52,24 +51,26 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
/**
* @param element
* @param builder
* @param parserContext
*/
private void addHostAndPortToConstructor(Element element,
BeanDefinitionBuilder builder) {
BeanDefinitionBuilder builder, ParserContext parserContext) {
String host = element.getAttribute(IpAdapterParserUtils.HOST);
if (!StringUtils.hasText(host)) {
throw new BeanCreationException(IpAdapterParserUtils.HOST
+ " is required for IP outbound channel adapters");
parserContext.getReaderContext().error(IpAdapterParserUtils.HOST
+ " is required for IP outbound channel adapters", element);
}
builder.addConstructorArgValue(host);
String port = IpAdapterParserUtils.getPort(element);
String port = IpAdapterParserUtils.getPort(element, parserContext);
builder.addConstructorArgValue(port);
}
/**
* @param element
* @param parserContext
* @return
*/
private BeanDefinitionBuilder parseUdp(Element element) {
private BeanDefinitionBuilder parseUdp(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder;
String multicast = IpAdapterParserUtils.getMulticast(element);
if (multicast.equals("true")) {
@@ -86,7 +87,7 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
builder = BeanDefinitionBuilder
.genericBeanDefinition(UnicastSendingMessageHandler.class);
}
addHostAndPortToConstructor(element, builder);
addHostAndPortToConstructor(element, builder, parserContext);
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
element, IpAdapterParserUtils.CHECK_LENGTH, true);
IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder,
@@ -105,12 +106,12 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
.getAttribute(IpAdapterParserUtils.ACK_PORT))
|| !StringUtils.hasText(element
.getAttribute(IpAdapterParserUtils.ACK_TIMEOUT))) {
throw new BeanCreationException("When "
parserContext.getReaderContext().error("When "
+ IpAdapterParserUtils.ACK + " is true, "
+ IpAdapterParserUtils.ACK_HOST + ", "
+ IpAdapterParserUtils.ACK_PORT + ", and "
+ IpAdapterParserUtils.ACK_TIMEOUT
+ " must be supplied");
+ " must be supplied", element);
}
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
@@ -120,9 +121,10 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
/**
* @param element
* @param parserContext
* @return
*/
private BeanDefinitionBuilder parseTcp(Element element) {
private BeanDefinitionBuilder parseTcp(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder;
String useNio = IpAdapterParserUtils.getUseNio(element);
if (useNio.equals("false")) {
@@ -133,12 +135,10 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNioSendingMessageHandler.class);
}
addHostAndPortToConstructor(element, builder);
addHostAndPortToConstructor(element, builder, parserContext);
builder.addPropertyValue(
Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT),
IpAdapterParserUtils.getMessageFormat(element));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.BLOCKING_WRITE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.CUSTOM_SOCKET_WRITER_CLASS_NAME);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,

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

@@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -62,9 +63,10 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
if (this.active && this.threadPoolTaskScheduler == null) {
this.threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
this.threadPoolTaskScheduler.setThreadFactory(new ThreadFactory() {
private AtomicInteger n = new AtomicInteger();
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("TCP-Incoming-Msg-Handler");
thread.setName("TCP-Incoming-Msg-Handler-" + n.getAndIncrement());
thread.setDaemon(true);
return thread;
}

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

@@ -52,8 +52,6 @@ public abstract class AbstractTcpSendingMessageHandler extends
protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
protected boolean blockingWrite = false;
/**
* Constructs a message handler that sends messages to the specified
* host and port.
@@ -94,34 +92,12 @@ public abstract class AbstractTcpSendingMessageHandler extends
/**
* Writes the message payload to the underlying socket, using the specified
* message format. If blockingWrite is true, the write to the socket
* will occur on the caller's thread. Otherwise, the method will return
* immediately and the write will occur on a separate thread.
*
* message format.
* @see org.springframework.integration.message.MessageHandler#handleMessage(org.springframework.integration.core.Message)
*/
public void handleMessage(final Message<?> message) throws MessageRejectedException,
MessageHandlingException, MessageDeliveryException {
if (blockingWrite) {
doWrite(message);
return;
}
if (this.executorService == null) {
this.executorService = Executors
.newSingleThreadExecutor(new ThreadFactory() {
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("Tcp-NonBlocking-Handler-port-" + port);
thread.setDaemon(true);
return thread;
}
});
}
executorService.execute(new Runnable() {
public void run() {
doWrite(message);
}
});
doWrite(message);
}
/**
@@ -186,12 +162,4 @@ public abstract class AbstractTcpSendingMessageHandler extends
this.messageFormat = messageFormat;
}
/**
* If true, socket writes will occur on the caller's thread.
* @param blockingWrite the blockingWrite to set
*/
public void setBlockingWrite(boolean blockingWrite) {
this.blockingWrite = blockingWrite;
}
}

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

@@ -124,6 +124,8 @@ public class NioSocketWriter extends AbstractSocketWriter {
}
if (lengthPart == null) {
lengthPart = ByteBuffer.allocate(4);
} else {
lengthPart.clear();
}
lengthPart.putInt(bytes.length);
lengthPart.flip();

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

@@ -58,7 +58,8 @@ public class TcpNetReceivingChannelAdapter extends
protected void server() {
while (active) {
try {
serverSocket = ServerSocketFactory.getDefault().createServerSocket(port);
serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(port, Math.abs(poolSize));
while (true) {
final Socket socket = serverSocket.accept();
setSocketOptions(socket);

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

@@ -67,7 +67,8 @@ public class TcpNioReceivingChannelAdapter extends
try {
serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.socket().bind(new InetSocketAddress(port), 10);
serverChannel.socket().bind(new InetSocketAddress(port),
Math.abs(poolSize));
final Selector selector = Selector.open();
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
doSelect(serverChannel, selector);

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

@@ -82,7 +82,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
public void run() {
if (logger.isDebugEnabled()) {
logger.debug("UDP Receiver running...");
logger.debug("UDP Receiver running on port:" + port);
}
if (this.active && this.threadPoolTaskScheduler == null) {
this.threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
@@ -199,6 +199,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
super.doStop();
try {
this.socket.close();
socket = null;
}
catch (Exception e) {
// ignore

10
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -28,6 +28,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
@@ -155,9 +156,10 @@ public class UnicastSendingMessageHandler extends
Assert.hasLength(ackHost);
this.executorService = Executors
.newSingleThreadExecutor(new ThreadFactory() {
private AtomicInteger n = new AtomicInteger();
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("UDP-Ack-Handler");
thread.setName("UDP-Ack-Handler-" + n.getAndIncrement());
thread.setDaemon(true);
return thread;
}
@@ -256,10 +258,8 @@ public class UnicastSendingMessageHandler extends
}
}
catch (IOException e) {
if (this.ackSocket != null) {
logger.error("Error on UDP Acknowledge thread");
fatalException = e;
}
logger.error("Error on UDP Acknowledge thread" + e.getMessage());
fatalException = e;
}
finally {
if (this.ackSocket != null) {

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

@@ -70,7 +70,6 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T
<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" default="false" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>