diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index 6638f4c6a9..6faf8b55f5 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -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; } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java index 4ad09a7f1c..ed860ecab6 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java @@ -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)); diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java index b11cd5222e..8204b7c5b5 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java @@ -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, diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java index 8e512a13b9..0a4eb4c391 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java @@ -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; } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java index bf7499f39c..1e231dd960 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java @@ -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; - } - } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java index f520665a12..4445cfa112 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java @@ -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(); diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java index ae2ddca46e..02043634df 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java @@ -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); diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java index abfdd9e04f..bce73cd76a 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java @@ -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); diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 75e4380186..e7652f3653 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -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 diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index dd8ef3d85d..20c995217f 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -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) { diff --git a/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index 388a49f426..4aa0e873be 100644 --- a/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -70,7 +70,6 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T - diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java index 050bfa3cc5..5fc5af8d9d 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java @@ -29,8 +29,8 @@ 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.ip.util.SocketUtils; import org.springframework.integration.message.MessageBuilder; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -106,7 +106,7 @@ public class IpChannelAdapterParserTests @SuppressWarnings("unchecked") @Test public void testTcpInbound1() { - Utils.testSendFragmented(tcp1.getPort(), true); + SocketUtils.testSendFragmented(tcp1.getPort(), true); Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("xx", new String(message.getPayload())); @@ -115,7 +115,7 @@ public class IpChannelAdapterParserTests @SuppressWarnings("unchecked") @Test public void testTcpInbound2() { - Utils.testSendFragmented(tcp2.getPort(), true); + SocketUtils.testSendFragmented(tcp2.getPort(), true); Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("xx", new String(message.getPayload())); @@ -124,7 +124,7 @@ public class IpChannelAdapterParserTests @SuppressWarnings("unchecked") @Test public void testTcpInbound3() { - Utils.testSendFragmented(tcp3.getPort(), true); + SocketUtils.testSendFragmented(tcp3.getPort(), true); Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("xx", new String(message.getPayload())); @@ -133,38 +133,38 @@ public class IpChannelAdapterParserTests @SuppressWarnings("unchecked") @Test public void testTcpInbound4() { - Utils.testSendStxEtx(tcp4.getPort(), null); + SocketUtils.testSendStxEtx(tcp4.getPort(), null); Message message = (Message) channel.receive(10000); assertNotNull(message); - assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(message.getPayload())); message = (Message) channel.receive(10000); assertNotNull(message); - assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(message.getPayload())); } @SuppressWarnings("unchecked") @Test public void testTcpInbound5() { - Utils.testSendCrLf(tcp5.getPort(), null); + SocketUtils.testSendCrLf(tcp5.getPort(), null); Message message = (Message) channel.receive(10000); assertNotNull(message); - assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(message.getPayload())); message = (Message) channel.receive(10000); assertNotNull(message); - assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(message.getPayload())); } @SuppressWarnings("unchecked") @Test public void testTcpInbound6() { - Utils.testSendStxEtx(tcp6.getPort(), null); + SocketUtils.testSendStxEtx(tcp6.getPort(), null); Message message = (Message) channel.receive(10000); assertNotNull(message); - assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String(message.getPayload())); message = (Message) channel.receive(10000); assertNotNull(message); - assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String(message.getPayload())); } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml index daa22359f3..e5121f06b5 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml @@ -10,7 +10,7 @@ http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd"> - + message = MessageBuilder.withPayload(payload).build(); + sender.handleMessage(message); + // and again + sender.handleMessage(message); + // and again + sender.handleMessage(message); + }}); + t.setDaemon(true); + t.start(); + } + for (int i = 0; i < drivers * 3 ; i++) { + Message messageOut = (Message) queue.receive(6000); + assertNotNull(messageOut); + Assert.assertEquals(payload, new String(messageOut.getPayload())); + } + adapter.stop(); + } + + @SuppressWarnings("unchecked") + @Test + public void testNio() throws Exception { + final String payload = largePayload(10000); // force fragmentation + final TcpNioReceivingChannelAdapter adapter = + new TcpNioReceivingChannelAdapter(SocketUtils.findAvailableServerSocket()); + adapter.setPoolSize(4); + int drivers = 10; + QueueChannel queue = new QueueChannel(drivers * 3); + adapter.setOutputChannel(queue); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + while (!adapter.isRunning()) { + Thread.sleep(50); // wait for server to start its listener. + } + Thread.sleep(250); // wait for listener + for (int i = 0; i < drivers; i++) { + Thread t = new Thread( new Runnable() { + public void run() { + TcpNioSendingMessageHandler sender = new TcpNioSendingMessageHandler("localhost", adapter.getPort()); + Message message = MessageBuilder.withPayload(payload).build(); + sender.handleMessage(message); + // and again + sender.handleMessage(message); + // and again + sender.handleMessage(message); + }}); + t.setDaemon(true); + t.start(); + } + for (int i = 0; i < drivers * 3 ; i++) { + Message messageOut = (Message) queue.receive(6000); + assertNotNull(messageOut); + Assert.assertEquals(payload, new String(messageOut.getPayload())); + } + adapter.stop(); + } + + /** + * @param i + * @return + */ + private String largePayload(int n) { + StringBuilder sb = new StringBuilder(n); + for (int i = 0; i < n; i++) { + sb.append('x'); + } + return sb.toString(); + } + +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java index 32e2665504..3e8b3ccd3f 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java @@ -24,6 +24,7 @@ import java.net.Socket; import javax.net.ServerSocketFactory; import org.junit.Test; +import org.springframework.integration.ip.util.SocketUtils; /** * @author Gary Russell @@ -37,21 +38,21 @@ public class NetSocketReaderTests { */ @Test public void testReadLength() throws Exception { - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); - Utils.testSendLength(port, null); + SocketUtils.testSendLength(port, null); Socket socket = server.accept(); socket.setSoTimeout(5000); NetSocketReader reader = new NetSocketReader(socket); if (reader.assembleData()) { - assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } else { fail("Failed to assemble first message"); } if (reader.assembleData()) { - assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } else { @@ -66,22 +67,22 @@ public class NetSocketReaderTests { */ @Test public void testReadStxEtx() throws Exception { - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); - Utils.testSendStxEtx(port, null); + SocketUtils.testSendStxEtx(port, null); Socket socket = server.accept(); socket.setSoTimeout(5000); NetSocketReader reader = new NetSocketReader(socket); reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX); if (reader.assembleData()) { - assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } else { fail("Failed to assemble first message"); } if (reader.assembleData()) { - assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } else { @@ -96,22 +97,22 @@ public class NetSocketReaderTests { */ @Test public void testReadCrLf() throws Exception { - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); - Utils.testSendCrLf(port, null); + SocketUtils.testSendCrLf(port, null); Socket socket = server.accept(); socket.setSoTimeout(5000); NetSocketReader reader = new NetSocketReader(socket); reader.setMessageFormat(MessageFormats.FORMAT_CRLF); if (reader.assembleData()) { - assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } else { fail("Failed to assemble first message"); } if (reader.assembleData()) { - assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } else { diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java index 4409c63547..4faea4561f 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java @@ -27,6 +27,7 @@ import javax.net.ServerSocketFactory; import javax.net.SocketFactory; import org.junit.Test; +import org.springframework.integration.ip.util.SocketUtils; /** * @author Gary Russell @@ -36,7 +37,7 @@ public class NetSocketWriterTests { @Test public void testWriteLengthHeader() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -68,7 +69,7 @@ public class NetSocketWriterTests { @Test public void testWriteStxEtx() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -100,7 +101,7 @@ public class NetSocketWriterTests { @Test public void testWriteCrLf() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java index f3c95afbdc..da2f97d16c 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java @@ -29,6 +29,7 @@ import java.util.Set; import java.util.concurrent.CountDownLatch; import org.junit.Test; +import org.springframework.integration.ip.util.SocketUtils; /** * @author Gary Russell @@ -44,13 +45,13 @@ public class NioSocketReaderTests { public void testReadLength() throws Exception { ServerSocketChannel server = ServerSocketChannel.open(); server.configureBlocking(false); - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); server.socket().bind(new InetSocketAddress(port)); final Selector selector = Selector.open(); server.register(selector, SelectionKey.OP_ACCEPT); // Fire up the sender. - Utils.testSendLength(port, latch); + SocketUtils.testSendLength(port, latch); if(selector.select(10000) <= 0) { fail("Socket failed to connect"); @@ -81,7 +82,7 @@ public class NioSocketReaderTests { if (key.isReadable()) { assertEquals(channel, key.channel()); if (reader.assembleData()) { - assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); count++; } @@ -100,13 +101,13 @@ public class NioSocketReaderTests { public void testFragmented() throws Exception { ServerSocketChannel server = ServerSocketChannel.open(); server.configureBlocking(false); - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); server.socket().bind(new InetSocketAddress(port)); final Selector selector = Selector.open(); server.register(selector, SelectionKey.OP_ACCEPT); // Fire up the sender. - Utils.testSendFragmented(port, false); + SocketUtils.testSendFragmented(port, false); if(selector.select(10000) <= 0) { fail("Socket failed to connect"); @@ -159,13 +160,13 @@ public class NioSocketReaderTests { public void testReadStxEtx() throws Exception { ServerSocketChannel server = ServerSocketChannel.open(); server.configureBlocking(false); - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); server.socket().bind(new InetSocketAddress(port)); final Selector selector = Selector.open(); server.register(selector, SelectionKey.OP_ACCEPT); // Fire up the sender. - Utils.testSendStxEtx(port, latch); + SocketUtils.testSendStxEtx(port, latch); if(selector.select(10000) <= 0) { fail("Socket failed to connect"); @@ -197,7 +198,7 @@ public class NioSocketReaderTests { if (key.isReadable()) { assertEquals(channel, key.channel()); if (reader.assembleData()) { - assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); count++; } @@ -219,13 +220,13 @@ public class NioSocketReaderTests { public void testReadCrLf() throws Exception { ServerSocketChannel server = ServerSocketChannel.open(); server.configureBlocking(false); - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); server.socket().bind(new InetSocketAddress(port)); final Selector selector = Selector.open(); server.register(selector, SelectionKey.OP_ACCEPT); // Fire up the sender. - Utils.testSendCrLf(port, latch); + SocketUtils.testSendCrLf(port, latch); if(selector.select(10000) <= 0) { fail("Socket failed to connect"); @@ -257,7 +258,7 @@ public class NioSocketReaderTests { if (key.isReadable()) { assertEquals(channel, key.channel()); if (reader.assembleData()) { - assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); count++; } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java index 67784be253..f888647dc3 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java @@ -28,6 +28,7 @@ import java.nio.channels.SocketChannel; import javax.net.ServerSocketFactory; import org.junit.Test; +import org.springframework.integration.ip.util.SocketUtils; /** * @author Gary Russell @@ -37,7 +38,7 @@ public class NioSocketWriterTests { @Test public void testWriteLengthHeader() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -69,7 +70,7 @@ public class NioSocketWriterTests { @Test public void testWriteStxEtx() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -101,7 +102,7 @@ public class NioSocketWriterTests { @Test public void testWriteCrLf() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -133,7 +134,7 @@ public class NioSocketWriterTests { @Test public void testWriteLengthHeaderDirect() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -166,7 +167,7 @@ public class NioSocketWriterTests { @Test public void testWriteStxEtxDirect() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -199,7 +200,7 @@ public class NioSocketWriterTests { @Test public void testWriteCrLfDirect() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index e333992751..622f1dae6b 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -22,6 +22,7 @@ import org.junit.Test; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter; +import org.springframework.integration.ip.util.SocketUtils; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** @@ -36,7 +37,7 @@ public class TcpReceivingChannelAdapterTests { @Test public void testNet() throws Exception { QueueChannel channel = new QueueChannel(2); - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); AbstractInternetProtocolReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); adapter.setOutputChannel(channel); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); @@ -44,15 +45,15 @@ public class TcpReceivingChannelAdapterTests { adapter.setTaskScheduler(taskScheduler); adapter.start(); Thread.sleep(2000); // wait for server to start listening - Utils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice + SocketUtils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice Thread.sleep(2000); // wait for asynch processing Message message = channel.receive(0); assertNotNull(message); - assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String((byte[])message.getPayload())); message = channel.receive(0); assertNotNull(message); - assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String((byte[])message.getPayload())); adapter.stop(); } @@ -64,7 +65,7 @@ public class TcpReceivingChannelAdapterTests { @Test public void testNetCustom() throws Exception { QueueChannel channel = new QueueChannel(2); - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); TcpNetReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); adapter.setOutputChannel(channel); adapter.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNetSocketReader"); @@ -74,15 +75,15 @@ public class TcpReceivingChannelAdapterTests { adapter.setTaskScheduler(taskScheduler); adapter.start(); Thread.sleep(2000); // wait for server to start listening - Utils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice + SocketUtils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice Thread.sleep(4000); // wait for asynch processing Message message = channel.receive(0); assertNotNull(message); - assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); message = channel.receive(0); assertNotNull(message); - assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); adapter.stop(); } @@ -94,7 +95,7 @@ public class TcpReceivingChannelAdapterTests { @Test public void testNio() throws Exception { QueueChannel channel = new QueueChannel(2); - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); TcpNioReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port); adapter.setOutputChannel(channel); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); @@ -102,15 +103,15 @@ public class TcpReceivingChannelAdapterTests { adapter.setTaskScheduler(taskScheduler); adapter.start(); Thread.sleep(2000); // wait for server to start listening - Utils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice + SocketUtils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice Thread.sleep(2000); // wait for asynch processing Message message = channel.receive(0); assertNotNull(message); - assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String((byte[])message.getPayload())); message = channel.receive(0); assertNotNull(message); - assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String((byte[])message.getPayload())); adapter.stop(); } @@ -121,7 +122,7 @@ public class TcpReceivingChannelAdapterTests { @Test public void testNioCustom() throws Exception { QueueChannel channel = new QueueChannel(2); - int port = Utils.findAvailableServerSocket(); + int port = SocketUtils.findAvailableServerSocket(); TcpNioReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port); adapter.setOutputChannel(channel); adapter.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNioSocketReader"); @@ -131,15 +132,15 @@ public class TcpReceivingChannelAdapterTests { adapter.setTaskScheduler(taskScheduler); adapter.start(); Thread.sleep(2000); // wait for server to start listening - Utils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice + SocketUtils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice Thread.sleep(4000); // wait for asynch processing Message message = channel.receive(0); assertNotNull(message); - assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); message = channel.receive(0); assertNotNull(message); - assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); adapter.stop(); } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index 3f8ea433f3..bc6335755a 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -26,6 +26,7 @@ import javax.net.ServerSocketFactory; import org.junit.Test; import org.springframework.integration.core.Message; +import org.springframework.integration.ip.util.SocketUtils; import org.springframework.integration.message.MessageBuilder; @@ -36,8 +37,8 @@ import org.springframework.integration.message.MessageBuilder; public class TcpSendingMessageHandlerTests { @Test - public void testNetBlocking() throws Exception { - final int port = Utils.findAvailableServerSocket(); + public void testNet() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -45,38 +46,6 @@ public class TcpSendingMessageHandlerTests { try { TcpNetSendingMessageHandler handler = new TcpNetSendingMessageHandler("localhost", port); handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); - handler.setBlockingWrite(true); - Message message = MessageBuilder.withPayload(testString).build(); - handler.handleMessage(message); - Thread.sleep(1000000000L); - } catch (Exception e) { - e.printStackTrace(); - } - } - }); - t.setDaemon(true); - t.start(); - Socket socket = server.accept(); - InputStream is = socket.getInputStream(); - byte[] buff = new byte[testString.length() + 2]; - readFully(is, buff); - assertEquals(MessageFormats.STX, buff[0]); - assertEquals(testString, new String(buff, 1, testString.length())); - assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); - server.close(); - } - - @Test - public void testNetNonBlocking() throws Exception { - final int port = Utils.findAvailableServerSocket(); - final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); - Thread t = new Thread(new Runnable() { - public void run() { - try { - TcpNetSendingMessageHandler handler = new TcpNetSendingMessageHandler("localhost", port); - handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); - handler.setBlockingWrite(false); Message message = MessageBuilder.withPayload(testString).build(); handler.handleMessage(message); Thread.sleep(1000000000L); @@ -99,7 +68,7 @@ public class TcpSendingMessageHandlerTests { @Test public void testNetCustom() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -107,7 +76,6 @@ public class TcpSendingMessageHandlerTests { try { TcpNetSendingMessageHandler handler = new TcpNetSendingMessageHandler("localhost", port); handler.setMessageFormat(MessageFormats.FORMAT_CUSTOM); - handler.setBlockingWrite(true); handler.setCustomSocketWriterClassName("org.springframework.integration.ip.tcp.CustomNetSocketWriter"); Message message = MessageBuilder.withPayload(testString).build(); handler.handleMessage(message); @@ -127,11 +95,10 @@ public class TcpSendingMessageHandlerTests { new String(buff)); server.close(); } - @Test - public void testNioBlocking() throws Exception { - final int port = Utils.findAvailableServerSocket(); + public void testNio() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -139,7 +106,6 @@ public class TcpSendingMessageHandlerTests { try { TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); - handler.setBlockingWrite(true); Message message = MessageBuilder.withPayload(testString).build(); handler.handleMessage(message); Thread.sleep(1000000000L); @@ -161,8 +127,8 @@ public class TcpSendingMessageHandlerTests { } @Test - public void testNioNonBlocking() throws Exception { - final int port = Utils.findAvailableServerSocket(); + public void testNioDirect() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -170,70 +136,6 @@ public class TcpSendingMessageHandlerTests { try { TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); - handler.setBlockingWrite(false); - Message message = MessageBuilder.withPayload(testString).build(); - handler.handleMessage(message); - Thread.sleep(1000000000L); - } catch (Exception e) { - e.printStackTrace(); - } - } - }); - t.setDaemon(true); - t.start(); - Socket socket = server.accept(); - InputStream is = socket.getInputStream(); - byte[] buff = new byte[testString.length() + 2]; - readFully(is, buff); - assertEquals(MessageFormats.STX, buff[0]); - assertEquals(testString, new String(buff, 1, testString.length())); - assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); - server.close(); - } - - @Test - public void testNioBlockingDirect() throws Exception { - final int port = Utils.findAvailableServerSocket(); - final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); - Thread t = new Thread(new Runnable() { - public void run() { - try { - TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); - handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); - handler.setBlockingWrite(true); - handler.setUsingDirectBuffers(true); - Message message = MessageBuilder.withPayload(testString).build(); - handler.handleMessage(message); - Thread.sleep(1000000000L); - } catch (Exception e) { - e.printStackTrace(); - } - } - }); - t.setDaemon(true); - t.start(); - Socket socket = server.accept(); - InputStream is = socket.getInputStream(); - byte[] buff = new byte[testString.length() + 2]; - readFully(is, buff); - assertEquals(MessageFormats.STX, buff[0]); - assertEquals(testString, new String(buff, 1, testString.length())); - assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); - server.close(); - } - - @Test - public void testNioNonBlockingDirect() throws Exception { - final int port = Utils.findAvailableServerSocket(); - final String testString = "abcdef"; - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); - Thread t = new Thread(new Runnable() { - public void run() { - try { - TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); - handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); - handler.setBlockingWrite(false); handler.setUsingDirectBuffers(true); Message message = MessageBuilder.withPayload(testString).build(); handler.handleMessage(message); @@ -257,7 +159,7 @@ public class TcpSendingMessageHandlerTests { @Test public void testNioCustom() throws Exception { - final int port = Utils.findAvailableServerSocket(); + final int port = SocketUtils.findAvailableServerSocket(); final String testString = "abcdef"; ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Thread t = new Thread(new Runnable() { @@ -265,7 +167,6 @@ public class TcpSendingMessageHandlerTests { try { TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); handler.setMessageFormat(MessageFormats.FORMAT_CUSTOM); - handler.setBlockingWrite(true); handler.setCustomSocketWriteriClassName("org.springframework.integration.ip.tcp.CustomNioSocketWriter"); Message message = MessageBuilder.withPayload(testString).build(); handler.handleMessage(message); diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java new file mode 100644 index 0000000000..d33c8d72e4 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java @@ -0,0 +1,196 @@ +/* + * 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.udp; + +import static org.junit.Assert.assertNotNull; +import junit.framework.Assert; + +import org.junit.Test; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.util.SocketUtils; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + + +/** + * + * For both .net. and .nio. adapters, creates a single server and 10 clients + * and sends 3 messages from each client to the associated server. + * Ensures that all messages are correctly assembled and received ok. + * Since udp is inherently unreliable, we have to single thread our requests + * through a blocking queue, to get a reliable test case. Otherwise collisions + * will cause messages to be lost. + * Even with this restriction, we are still testing the receiving adapter's + * ability to handle multiple requests from multiple clients. + * + * @author Gary Russell + * + */ +public class MultiClientTests { + + @SuppressWarnings("unchecked") + @Test + public void testNoAck() throws Exception { + final String payload = largePayload(1000); + final UnicastReceivingChannelAdapter adapter = + new UnicastReceivingChannelAdapter(SocketUtils.findAvailableUdpSocket()); + int drivers = 10; + adapter.setPoolSize(drivers); + QueueChannel queue = new QueueChannel(drivers * 3); + adapter.setOutputChannel(queue); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + final QueueChannel queueIn = new QueueChannel(1000); + while (!adapter.isRunning()) { + Thread.sleep(50); // wait for server to start listening + } + Thread.sleep(250); // wait for listener + for (int i = 0; i < drivers; i++) { + Thread t = new Thread( new Runnable() { + public void run() { + UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler( + "localhost", adapter.getPort()); + while (true) { + Message message = queueIn.receive(); + sender.handleMessage(message); + } + }}); + t.setDaemon(true); + t.start(); + } + for (int i = 0; i < drivers * 3 ; i++) { + queueIn.send(MessageBuilder.withPayload(payload).build()); + } + for (int i = 0; i < drivers * 3 ; i++) { + Message messageOut = (Message) queue.receive(10000); + assertNotNull(messageOut); + Assert.assertEquals(payload, new String(messageOut.getPayload())); + } + adapter.stop(); + } + + @SuppressWarnings("unchecked") + @Test + public void testAck() throws Exception { + Thread.sleep(1000); + final String payload = largePayload(1000); + final UnicastReceivingChannelAdapter adapter = + new UnicastReceivingChannelAdapter(SocketUtils.findAvailableUdpSocket(), false); + int drivers = 5; + adapter.setPoolSize(drivers); + QueueChannel queue = new QueueChannel(drivers * 3); + adapter.setOutputChannel(queue); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + final QueueChannel queueIn = new QueueChannel(1000); + while (!adapter.isRunning()) { + Thread.sleep(50); // wait for server to start listening + } + Thread.sleep(250); // wait for listener + for (int i = 0; i < drivers; i++) { + final int j = i; + Thread t = new Thread( new Runnable() { + public void run() { + UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler( + "localhost", adapter.getPort(), + false, true, "localhost", + SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1000), + 10000); + while (true) { + Message message = queueIn.receive(); + sender.handleMessage(message); + } + }}); + t.setDaemon(true); + t.start(); + } + for (int i = 0; i < drivers * 3 ; i++) { + queueIn.send(MessageBuilder.withPayload(payload).build()); + } + for (int i = 0; i < drivers * 3 ; i++) { + Message messageOut = (Message) queue.receive(20000); + assertNotNull(messageOut); + Assert.assertEquals(payload, new String(messageOut.getPayload())); + } + adapter.stop(); + } + + @SuppressWarnings("unchecked") + @Test + public void testAckWithLength() throws Exception { + Thread.sleep(1000); + final String payload = largePayload(1000); + final UnicastReceivingChannelAdapter adapter = + new UnicastReceivingChannelAdapter(SocketUtils.findAvailableUdpSocket(), true); + int drivers = 10; + adapter.setPoolSize(drivers); + QueueChannel queue = new QueueChannel(drivers * 3); + adapter.setOutputChannel(queue); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + final QueueChannel queueIn = new QueueChannel(1000); + while (!adapter.isRunning()) { + Thread.sleep(50); // wait for server to start listening + } + Thread.sleep(250); // wait for listener + for (int i = 0; i < drivers; i++) { + final int j = i; + Thread t = new Thread( new Runnable() { + public void run() { + UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler( + "localhost", adapter.getPort(), + true, true, "localhost", + SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1000), + 10000); + while (true) { + Message message = queueIn.receive(); + sender.handleMessage(message); + } + }}); + t.setDaemon(true); + t.start(); + } + for (int i = 0; i < drivers * 3 ; i++) { + queueIn.send(MessageBuilder.withPayload(payload).build()); + } + for (int i = 0; i < drivers * 3 ; i++) { + Message messageOut = (Message) queue.receive(10000); + assertNotNull(messageOut); + Assert.assertEquals(payload, new String(messageOut.getPayload())); + } + adapter.stop(); + } + + /** + * @param i + * @return + */ + private String largePayload(int n) { + StringBuilder sb = new StringBuilder(n); + for (int i = 0; i < n; i++) { + sb.append('x'); + } + return sb.toString(); + } + +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java similarity index 90% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java index 582faf67ca..c6e1c80872 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java @@ -13,9 +13,10 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.ip.tcp; +package org.springframework.integration.ip.util; import java.io.OutputStream; +import java.net.DatagramSocket; import java.net.InetAddress; import java.net.ServerSocket; import java.net.Socket; @@ -33,11 +34,11 @@ import org.apache.commons.logging.LogFactory; * @author Gary Russell * */ -public class Utils { +public class SocketUtils { public static final String TEST_STRING = "TestMessage"; - private static final Log logger = LogFactory.getLog(Utils.class); + private static final Log logger = LogFactory.getLog(SocketUtils.class); /** * Sends a message in two chunks with a preceding length. Two such messages are sent. @@ -104,7 +105,7 @@ public class Utils { private static void writeByte(OutputStream os, int b, boolean noDelay) throws Exception { os.write(b); - logger.debug("Wrote 0x%x\n" + Integer.toHexString(b)); + logger.debug("Wrote 0x" + Integer.toHexString(b)); if (noDelay) { return; } @@ -191,4 +192,21 @@ public class Utils { public static int findAvailableServerSocket() { return findAvailableServerSocket(5678); } + + public static int findAvailableUdpSocket(int seed) { + for (int i = seed; i < seed+200; i++) { + try { + DatagramSocket sock = new DatagramSocket(i); + sock.close(); + Thread.sleep(100); + return i; + } catch (Exception e) { } + } + throw new RuntimeException("Cannot find a free server socket"); + } + + public static int findAvailableUdpSocket() { + return findAvailableUdpSocket(9876); + } + }