diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java index cdb144142d..3c30003a1e 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java @@ -49,6 +49,13 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter this.port = port; } + /** + * + * @return The port on which this receiver is listening. + */ + public int getPort() { + return port; + } /* (non-Javadoc) * @see org.springframework.integration.ip.SocketOptions#setSoTimeout(int) @@ -83,4 +90,12 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter taskScheduler.schedule(this, new Date()); } + /* (non-Javadoc) + * @see org.springframework.integration.endpoint.AbstractEndpoint#doStop() + */ + @Override + protected void doStop() { + this.active = false; + } + } 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 f9d2e72055..afa5734521 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 @@ -21,6 +21,7 @@ import org.w3c.dom.Element; import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.ip.tcp.MessageFormats; import org.springframework.util.StringUtils; /** @@ -45,10 +46,12 @@ public abstract class IpAdapterParserUtils { static final String SO_TIMEOUT = "so-timeout"; - static final String SO_RECEIVE_BUFFER_SIZE = "so-receive-bufffer-size"; + static final String SO_RECEIVE_BUFFER_SIZE = "so-receive-buffer-size"; static final String SO_SEND_BUFFER_SIZE = "so-send-buffer-size"; + static final String SO_KEEP_ALIVE = "so-keep-alive"; + static final String RECEIVE_BUFFER_SIZE = "receive-buffer-size"; static final String POOL_SIZE = "pool-size"; @@ -64,6 +67,17 @@ public abstract class IpAdapterParserUtils { static final String MIN_ACKS_SUCCESS = "min-acks-for-success"; static final String TIME_TO_LIVE = "time-to-live"; + + static final String USING_NIO = "using-nio"; + + static final String USING_DIRECT_BUFFERS = "using-direct-buffers"; + + static final String MESSAGE_FORMAT = "message-format"; + + static final String CUSTOM_SOCKET_READER_CLASS_NAME = + "custom-socket-reader-class-name"; + +// static final String /** @@ -76,7 +90,7 @@ public abstract class IpAdapterParserUtils { * @param attributeName the name of the attribute whose value will be * used to populate the property */ - public static void addConstuctirValueIfAttributeDefined(BeanDefinitionBuilder builder, + public static void addConstuctorValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName, boolean trueFalse) { String attributeValue = element.getAttribute(attributeName); if (StringUtils.hasText(attributeValue)) { @@ -131,6 +145,45 @@ public abstract class IpAdapterParserUtils { return multicast; } + /** + * Gets the use-nio attribute, if present; if not returns 'false'. + * @param element + * @return The value of the attribute or false. + */ + static String getUseNio(Element element) { + String useNio = element.getAttribute(IpAdapterParserUtils.USING_NIO); + if (!StringUtils.hasText(useNio)) { + useNio = "false"; + } + return useNio; + } + + /** + * Gets the message-format attribute, if present; if not returns + * {@link MessageFormats#FORMAT_LENGTH_HEADER}. + * @param element + * @return The value of the attribute or false. + */ + static Integer getMessageFormat(Element element) { + String useNio = element.getAttribute(IpAdapterParserUtils.MESSAGE_FORMAT); + if (!StringUtils.hasText(useNio)) { + return MessageFormats.FORMAT_LENGTH_HEADER; + } + if (useNio.equals("length-header")) { + return MessageFormats.FORMAT_LENGTH_HEADER; + } + if (useNio.equals("stx-etx")) { + return MessageFormats.FORMAT_STX_ETX; + } + if (useNio.equals("crlf")) { + return MessageFormats.FORMAT_CRLF; + } + if (useNio.equals("custom")) { + return MessageFormats.FORMAT_CUSTOM; + } + return MessageFormats.FORMAT_LENGTH_HEADER; + } + /** * Sets the common port attributes on the bean being built (timeout, receive buffer size, * send buffer size). 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 0f3f1316e1..4ad09a7f1c 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 @@ -22,8 +22,11 @@ import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.core.Conventions; import org.springframework.integration.config.xml.AbstractChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter; import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter; import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter; import org.springframework.util.StringUtils; @@ -38,33 +41,12 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { String protocol = IpAdapterParserUtils.getProtocol(element); - String multicast = IpAdapterParserUtils.getMulticast(element); BeanDefinitionBuilder builder = null; if (protocol.equals("tcp")) { - throw new BeanCreationException("tcp not yet supported"); + builder = parseTcp(element); + } else if (protocol.equals("udp")) { + builder = parseUdp(element); } - else if (protocol.equals("udp")) { - if (multicast.equals("false")) { - builder = BeanDefinitionBuilder - .genericBeanDefinition(UnicastReceivingChannelAdapter.class); - } - else { - builder = BeanDefinitionBuilder - .genericBeanDefinition(MulticastReceivingChannelAdapter.class); - String mcAddress = element - .getAttribute(IpAdapterParserUtils.MULTICAST_ADDRESS); - if (!StringUtils.hasText(mcAddress)) { - throw new BeanCreationException( - IpAdapterParserUtils.MULTICAST_ADDRESS - + " is required for a multicast UDP/IP channel adapter"); - } - builder.addConstructorArgValue(mcAddress); - } - } - String port = IpAdapterParserUtils.getPort(element); - builder.addConstructorArgValue(port); - IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder, - element, IpAdapterParserUtils.CHECK_LENGTH, true); IpAdapterParserUtils.addCommonSocketOptions(builder, element); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.RECEIVE_BUFFER_SIZE); @@ -75,4 +57,71 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser return builder.getBeanDefinition(); } + /** + * @param element + * @param builder + */ + private void addPortToConstructor(Element element, + BeanDefinitionBuilder builder) { + String port = IpAdapterParserUtils.getPort(element); + builder.addConstructorArgValue(port); + } + + /** + * @param element + * @return + */ + private BeanDefinitionBuilder parseUdp(Element element) { + BeanDefinitionBuilder builder; + String multicast = IpAdapterParserUtils.getMulticast(element); + if (multicast.equals("false")) { + builder = BeanDefinitionBuilder + .genericBeanDefinition(UnicastReceivingChannelAdapter.class); + } + else { + builder = BeanDefinitionBuilder + .genericBeanDefinition(MulticastReceivingChannelAdapter.class); + String mcAddress = element + .getAttribute(IpAdapterParserUtils.MULTICAST_ADDRESS); + if (!StringUtils.hasText(mcAddress)) { + throw new BeanCreationException( + IpAdapterParserUtils.MULTICAST_ADDRESS + + " is required for a multicast UDP/IP channel adapter"); + } + builder.addConstructorArgValue(mcAddress); + } + addPortToConstructor(element, builder); + IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, + element, IpAdapterParserUtils.CHECK_LENGTH, true); + return builder; + } + + /** + * @param element + * @return + */ + private BeanDefinitionBuilder parseTcp(Element element) { + BeanDefinitionBuilder builder; + String useNio = IpAdapterParserUtils.getUseNio(element); + if (useNio.equals("false")) { + builder = BeanDefinitionBuilder + .genericBeanDefinition(TcpNetReceivingChannelAdapter.class); + } + else { + builder = BeanDefinitionBuilder + .genericBeanDefinition(TcpNioReceivingChannelAdapter.class); + } + addPortToConstructor(element, builder); + builder.addPropertyValue( + Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT), + IpAdapterParserUtils.getMessageFormat(element)); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.CUSTOM_SOCKET_READER_CLASS_NAME); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.USING_DIRECT_BUFFERS); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SO_KEEP_ALIVE); + return builder; + } + } 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 7a4e420d42..a2383bd33e 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 @@ -65,15 +65,15 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt builder.addConstructorArgValue(host); String port = IpAdapterParserUtils.getPort(element); builder.addConstructorArgValue(port); - IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder, + IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.CHECK_LENGTH, true); - IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder, + IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK, true); - IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder, + IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK_HOST, false); - IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder, + IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK_PORT, false); - IpAdapterParserUtils.addConstuctirValueIfAttributeDefined(builder, + IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.ACK_TIMEOUT, false); String ack = element.getAttribute(IpAdapterParserUtils.ACK); if (ack.equals("true")) { 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 42c3a9f352..ae07a830f5 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 @@ -55,15 +55,6 @@ public abstract class AbstractTcpReceivingChannelAdapter extends super(port); } - /* (non-Javadoc) - * @see org.springframework.integration.endpoint.AbstractEndpoint#doStop() - */ - @Override - protected void doStop() { - // TODO Auto-generated method stub - - } - /** * Creates the ThreadPoolTaskScheduler, if necessary, and calls * {@link #server()}. @@ -129,7 +120,7 @@ public abstract class AbstractTcpReceivingChannelAdapter extends * @throws ClassNotFoundException */ @SuppressWarnings("unchecked") - public void setCustomSocketReader(String customSocketReaderClassName) throws ClassNotFoundException { + public void setCustomSocketReaderClassName(String customSocketReaderClassName) throws ClassNotFoundException { this.customSocketReader = (Class) Class.forName(customSocketReaderClassName); } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java index bcd1141191..ea56b58ab5 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java @@ -94,7 +94,14 @@ public class NioSocketReader extends AbstractSocketReader { return false; } } - assembledData = dataPart.array(); + if (usingDirectBuffers) { + byte[] assembledData = new byte[dataPart.capacity()]; + dataPart.flip(); + dataPart.get(assembledData); + this.assembledData = assembledData; + } else { + assembledData = dataPart.array(); + } lengthPart = dataPart = null; return true; } catch (Exception e) { 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 ed178798c0..416cdb29b8 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 @@ -35,6 +35,7 @@ import org.springframework.integration.core.Message; public class TcpNetReceivingChannelAdapter extends AbstractTcpReceivingChannelAdapter { + protected ServerSocket serverSocket; /** * Constructs a TcpNetReceivingChannelAdapter that listens on the port. * @param port The port. @@ -52,11 +53,11 @@ public class TcpNetReceivingChannelAdapter extends */ @Override protected void server() { - while (true) { + while (active) { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + serverSocket = ServerSocketFactory.getDefault().createServerSocket(port); while (true) { - final Socket socket = server.accept(); + final Socket socket = serverSocket.accept(); setSocketOptions(socket); this.threadPoolTaskScheduler.execute(new Runnable() { public void run() { @@ -64,6 +65,15 @@ public class TcpNetReceivingChannelAdapter extends }}); } } catch (IOException e) { + if (!active) { + if (serverSocket != null) { + try { + serverSocket.close(); + } catch (IOException e1) {} + } + serverSocket = null; + return; + } // TODO Auto-generated catch block e.printStackTrace(); } @@ -116,5 +126,15 @@ public class TcpNetReceivingChannelAdapter extends } } } + @Override + protected void doStop() { + super.doStop(); + try { + this.serverSocket.close(); + } + catch (Exception e) { + // ignore + } + } } 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 f263d690e8..ef56331970 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 @@ -41,6 +41,8 @@ import org.springframework.integration.core.Message; public class TcpNioReceivingChannelAdapter extends AbstractTcpReceivingChannelAdapter { + protected ServerSocketChannel serverChannel; + /** * Constructs a TcpNioReceivingChannelAdapter to listen on the port. * @param port The port. @@ -58,14 +60,24 @@ public class TcpNioReceivingChannelAdapter extends @Override protected void server() { try { - final ServerSocketChannel server = ServerSocketChannel.open(); - server.configureBlocking(false); - server.socket().bind(new InetSocketAddress(port)); + serverChannel = ServerSocketChannel.open(); + serverChannel.configureBlocking(false); + serverChannel.socket().bind(new InetSocketAddress(port)); final Selector selector = Selector.open(); - server.register(selector, SelectionKey.OP_ACCEPT); - doSelect(server, selector); + serverChannel.register(selector, SelectionKey.OP_ACCEPT); + doSelect(serverChannel, selector); } catch (IOException e) { + if (!active) { + try { + serverChannel.close(); + } catch (IOException e1) { + // TODO Auto-generated catch block + e1.printStackTrace(); + } + serverChannel = null; + return; + } // TODO Auto-generated catch block e.printStackTrace(); } @@ -87,7 +99,7 @@ public class TcpNioReceivingChannelAdapter extends */ private void doSelect(ServerSocketChannel server, final Selector selector) throws IOException, ClosedChannelException, SocketException { - while (true) { + while (active) { int selectionCount = selector.select(); if (logger.isDebugEnabled()) logger.debug("SelectionCount: " + selectionCount); @@ -184,5 +196,16 @@ public class TcpNioReceivingChannelAdapter extends } } + @Override + protected void doStop() { + super.doStop(); + try { + this.serverChannel.close(); + } + catch (Exception e) { + // ignore + } + } + } \ No newline at end of file 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 bbfbd18437..75e4380186 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 @@ -196,7 +196,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece @Override protected void doStop() { - this.active = false; + super.doStop(); try { this.socket.close(); } 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 5aa81d2c66..28343ce950 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 @@ -38,6 +38,14 @@ + + + +If message-format = 'custom' you must provide a sub class of the appropriate type to implement +the custom message format. See java docs for TcpNetReceivingChannelAdapter and TcpNioReceivingChannelAdapter. + + + @@ -85,6 +93,20 @@ + + + + + + + + + + + + + + \ No newline at end of file diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/StdOutCatcher.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/StdOutCatcher.java deleted file mode 100644 index ac90b80829..0000000000 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/StdOutCatcher.java +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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; - -import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter; -import org.springframework.integration.core.Message; -import org.springframework.integration.core.MessageChannel; - -/** - * Grabs a copy of the last message sent out, if its payload is a byte[] - * - * @author Gary Russell - * @since 2.0 - */ -public class StdOutCatcher extends ChannelInterceptorAdapter{ - - private String content; - - public String getContent() { - return content; - } - - @Override - public Message preSend(Message message, MessageChannel channel) { - if (message.getPayload() instanceof byte[]) { - content = new String((byte[]) message.getPayload()); - } - return message; - } - -} 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 new file mode 100644 index 0000000000..6cb80aaf6a --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java @@ -0,0 +1,144 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.config; + + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.Utils; +import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + + +/** + * @author Gary Russell + * + */ +@ContextConfiguration(locations="inboundAdapters.xml") +@RunWith(SpringJUnit4ClassRunner.class) +public class IpChannelAdapterParserTests { + + @Autowired + QueueChannel channel; + + @Autowired + @Qualifier(value="tcp1") + TcpNioReceivingChannelAdapter tcp1; + + @Autowired + @Qualifier(value="tcp2") + TcpNioReceivingChannelAdapter tcp2; + + @Autowired + @Qualifier(value="tcp3") + TcpNetReceivingChannelAdapter tcp3; + + @Autowired + @Qualifier(value="tcp4") + TcpNetReceivingChannelAdapter tcp4; + + @Autowired + @Qualifier(value="tcp5") + TcpNetReceivingChannelAdapter tcp5; + + @Autowired + @Qualifier(value="tcp6") + TcpNetReceivingChannelAdapter tcp6; + + @Autowired + @Qualifier(value="udp1") + UnicastReceivingChannelAdapter udp1; + + @SuppressWarnings("unchecked") + @Test + public void testTcpInbound1() { + Utils.testSendFragmented(tcp1.getPort()); + Message message = (Message) channel.receive(); + assertNotNull(message); + assertEquals("xx", new String(message.getPayload())); + } + + @SuppressWarnings("unchecked") + @Test + public void testTcpInbound2() { + Utils.testSendFragmented(tcp2.getPort()); + Message message = (Message) channel.receive(); + assertNotNull(message); + assertEquals("xx", new String(message.getPayload())); + } + + @SuppressWarnings("unchecked") + @Test + public void testTcpInbound3() { + Utils.testSendFragmented(tcp3.getPort()); + Message message = (Message) channel.receive(); + assertNotNull(message); + assertEquals("xx", new String(message.getPayload())); + } + + @SuppressWarnings("unchecked") + @Test + public void testTcpInbound4() { + Utils.testSendStxEtx(tcp4.getPort(), null); + Message message = (Message) channel.receive(); + assertNotNull(message); + assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); + message = (Message) channel.receive(); + assertNotNull(message); + assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); + } + + @SuppressWarnings("unchecked") + @Test + public void testTcpInbound5() { + Utils.testSendCrLf(tcp5.getPort(), null); + Message message = (Message) channel.receive(); + assertNotNull(message); + assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); + message = (Message) channel.receive(); + assertNotNull(message); + assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); + } + + @SuppressWarnings("unchecked") + @Test + public void testTcpInbound6() { + Utils.testSendStxEtx(tcp6.getPort(), null); + Message message = (Message) channel.receive(); + assertNotNull(message); + assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + new String(message.getPayload())); + message = (Message) channel.receive(); + assertNotNull(message); + assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + new String(message.getPayload())); + } + + @Test + public void testUdpInbound1() { + assertNotNull(udp1); + } +} 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 new file mode 100644 index 0000000000..c18000503e --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml @@ -0,0 +1,100 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 9833b62263..43920a9783 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 @@ -65,7 +65,7 @@ public class TcpReceivingChannelAdapterTests { int port = 12346; AbstractTcpReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); adapter.setOutputChannel(channel); - adapter.setCustomSocketReader("org.springframework.integration.ip.tcp.CustomNetSocketReader"); + adapter.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNetSocketReader"); adapter.setMessageFormat(MessageFormats.FORMAT_CUSTOM); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.initialize(); @@ -120,7 +120,7 @@ public class TcpReceivingChannelAdapterTests { int port = 12356; TcpNioReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port); adapter.setOutputChannel(channel); - adapter.setCustomSocketReader("org.springframework.integration.ip.tcp.CustomNioSocketReader"); + adapter.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNioSocketReader"); adapter.setMessageFormat(MessageFormats.FORMAT_CUSTOM); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.initialize(); diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java index 36788beca8..ce73eea32e 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java @@ -25,14 +25,14 @@ import java.util.concurrent.TimeUnit; import org.junit.Ignore; import org.junit.Test; - import org.springframework.context.ApplicationContext; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.ChannelResolver; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.ip.StdOutCatcher; import org.springframework.integration.message.StringMessage; /** @@ -52,7 +52,7 @@ public class UdpMulticastEndToEndTests implements Runnable { private String testingIpText; - private String stdOutput; + private Message finalMessage; private CountDownLatch sentFirst = new CountDownLatch(1); @@ -104,13 +104,14 @@ public class UdpMulticastEndToEndTests implements Runnable { doneProcessing.countDown(); } assertTrue(firstReceived.await(2, TimeUnit.SECONDS)); - assertEquals(testingIpText, stdOutput); + assertEquals(testingIpText, new String(finalMessage.getPayload())); } /** * Instantiate the receiving context */ + @SuppressWarnings("unchecked") public void run() { AbstractApplicationContext ctx = new ClassPathXmlApplicationContext( "testIp-in-multicast-context.xml", @@ -118,14 +119,12 @@ public class UdpMulticastEndToEndTests implements Runnable { while (okToRun) { try { sentFirst.await(); - // wait another second to allow for the asynch handoffs - Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } - StdOutCatcher out = ctx.getBean(StdOutCatcher.class); - stdOutput = out.getContent(); + QueueChannel channel = ctx.getBean("udpOutChannel", QueueChannel.class); + finalMessage = (Message) channel.receive(); firstReceived.countDown(); try { doneProcessing.await(); diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java index e49dfd2aaf..6866b63a53 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java @@ -29,8 +29,9 @@ import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.ChannelResolver; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.ip.StdOutCatcher; import org.springframework.integration.message.StringMessage; /** @@ -49,7 +50,7 @@ public class UdpUnicastEndToEndTests implements Runnable { private String testingIpText; - private String stdOutput; + private Message finalMessage; private CountDownLatch sentFirst = new CountDownLatch(1); @@ -98,26 +99,25 @@ public class UdpUnicastEndToEndTests implements Runnable { doneProcessing.countDown(); } assertTrue(firstReceived.await(2, TimeUnit.SECONDS)); - assertEquals(testingIpText, stdOutput); + assertEquals(testingIpText, new String(finalMessage.getPayload())); } /** * Instantiate the receiving context */ + @SuppressWarnings("unchecked") public void run() { AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("testIp-in-context.xml", UdpUnicastEndToEndTests.class); while (okToRun) { try { sentFirst.await(); - // wait another second to allow for the asynch handoffs - Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } - StdOutCatcher out = ctx.getBean(StdOutCatcher.class); - stdOutput = out.getContent(); + QueueChannel channel = ctx.getBean("udpOutChannel", QueueChannel.class); + finalMessage = (Message) channel.receive(); firstReceived.countDown(); try { doneProcessing.await(); diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-common-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-common-context.xml index ea1a5ab78e..21cbc40a4b 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-common-context.xml +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-common-context.xml @@ -15,16 +15,10 @@ - - - - + + - - - - diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml index 569ef5fadd..19f20c6764 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml @@ -18,7 +18,7 @@ set and this buffer is too small, we'll throw an exception. -->