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 7476ac003d..6638f4c6a9 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 @@ -85,6 +85,8 @@ public abstract class IpAdapterParserUtils { static final String SO_TCP_NODELAY = "so-tcp-nodelay"; static final String SO_TRAFFIC_CLASS = "so-traffic-class"; + + static final String BLOCKING_WRITE = "blocking-write"; /** 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 5173e1e0b5..b11cd5222e 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 @@ -137,6 +137,8 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt 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/NetSocketReader.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java index 17c9f707aa..241d74a309 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java @@ -140,7 +140,8 @@ public class NetSocketReader extends AbstractSocketReader { len = socket.getInputStream().read(buffer, lengthRead, needed - lengthRead); if (len < 0) { - throw new IOException("EOF"); + logger.debug("Socket closed"); + throw new IOException("Socket Closed"); } lengthRead += len; if (logger.isDebugEnabled()) { 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 b770b17968..45a9993419 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 @@ -264,6 +264,10 @@ public class NioSocketReader extends AbstractSocketReader { if (len == 0) { return false; } + if (len < 0) { + logger.debug("Socket closed"); + throw new IOException("Socket closed"); + } rawBuffer.flip(); if (logger.isDebugEnabled()) { logger.debug("Read " + rawBuffer.limit() + " into raw buffer"); @@ -307,14 +311,14 @@ public class NioSocketReader extends AbstractSocketReader { } /** - * @return the useDirectBuffers + * @return the usingeDirectBuffers */ public boolean isUsingDirectBuffers() { return usingDirectBuffers; } /** - * @param useDirectBuffers the useDirectBuffers to set + * @param usingDirectBuffers the usingDirectBuffers to set */ public void setUsingDirectBuffers(boolean usingDirectBuffers) { this.usingDirectBuffers = usingDirectBuffers; diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java index 83a5868e61..0c1e723fd2 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java @@ -76,7 +76,7 @@ public class TcpNetSendingMessageHandler extends * @throws ClassNotFoundException */ @SuppressWarnings("unchecked") - public void setCustomSocketWriteriClassName( + public void setCustomSocketWriterClassName( String customSocketWriterClassName) throws ClassNotFoundException { this.customSocketWriter = (Class) Class .forName(customSocketWriterClassName); 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 7c3824eed4..388a49f426 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,7 @@ 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 4463172da2..050bfa3cc5 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 @@ -18,22 +18,13 @@ package org.springframework.integration.ip.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.fail; - -import java.lang.reflect.Field; -import java.net.InetSocketAddress; import org.junit.Test; import org.junit.runner.RunWith; -import org.springframework.beans.BeansException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; -import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter; -import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler; import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter; import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler; import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter; @@ -53,7 +44,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; ,"outboundAdapters.xml" }) @RunWith(SpringJUnit4ClassRunner.class) -public class IpChannelAdapterParserTests { +public class IpChannelAdapterParserTests +// implements ApplicationContextAware + { @Autowired QueueChannel channel; @@ -90,10 +83,30 @@ public class IpChannelAdapterParserTests { @Qualifier(value="org.springframework.integration.ip.tcp.TcpNioSendingMessageHandler#0") TcpNioSendingMessageHandler tcpOut1; + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.TcpNioSendingMessageHandler#1") + TcpNioSendingMessageHandler tcpOut2; + + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler#0") + TcpNetSendingMessageHandler tcpOut3; + + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler#1") + TcpNetSendingMessageHandler tcpOut4; + + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler#2") + TcpNetSendingMessageHandler tcpOut5; + + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler#3") + TcpNetSendingMessageHandler tcpOut6; + @SuppressWarnings("unchecked") @Test public void testTcpInbound1() { - Utils.testSendFragmented(tcp1.getPort()); + Utils.testSendFragmented(tcp1.getPort(), true); Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("xx", new String(message.getPayload())); @@ -102,7 +115,7 @@ public class IpChannelAdapterParserTests { @SuppressWarnings("unchecked") @Test public void testTcpInbound2() { - Utils.testSendFragmented(tcp2.getPort()); + Utils.testSendFragmented(tcp2.getPort(), true); Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("xx", new String(message.getPayload())); @@ -111,7 +124,7 @@ public class IpChannelAdapterParserTests { @SuppressWarnings("unchecked") @Test public void testTcpInbound3() { - Utils.testSendFragmented(tcp3.getPort()); + Utils.testSendFragmented(tcp3.getPort(), true); Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("xx", new String(message.getPayload())); @@ -163,7 +176,6 @@ public class IpChannelAdapterParserTests { @SuppressWarnings("unchecked") @Test public void testTcpOutbound1() { - setPort(tcpOut1, tcp1); Message message = MessageBuilder.withPayload("TESTING").build(); tcpOut1.handleMessage(message); Message mOut = (Message) channel.receive(10000); @@ -172,21 +184,86 @@ public class IpChannelAdapterParserTests { } - private void setPort(AbstractInternetProtocolSendingMessageHandler tcpSMA, - AbstractInternetProtocolReceivingChannelAdapter tcpRCA) { - try { - int port = tcpRCA.getPort(); - Field portField = tcpSMA.getClass().getSuperclass().getSuperclass().getDeclaredField("port"); - portField.setAccessible(true); - assertEquals(9999, portField.getInt(tcpSMA)); - portField.setInt(tcpSMA, port); - InetSocketAddress address = new InetSocketAddress("localhost", port); - Field addressField = tcpSMA.getClass().getSuperclass().getSuperclass().getDeclaredField("destinationAddress"); - addressField.setAccessible(true); - addressField.set(tcpSMA, address); - } catch (Exception e) { - fail("Couldn't fix port:" + e); - } + @SuppressWarnings("unchecked") + @Test + public void testTcpOutbound2() { + Message message = MessageBuilder.withPayload("TESTING").build(); + tcpOut2.handleMessage(message); + Message mOut = (Message) channel.receive(10000); + assertNotNull(mOut); + assertEquals("TESTING", new String(mOut.getPayload())); + } + @SuppressWarnings("unchecked") + @Test + public void testTcpOutbound3() { + Message message = MessageBuilder.withPayload("TESTING").build(); + tcpOut3.handleMessage(message); + Message mOut = (Message) channel.receive(10000); + assertNotNull(mOut); + assertEquals("TESTING", new String(mOut.getPayload())); + + } + + @SuppressWarnings("unchecked") + @Test + public void testTcpOutbound4() { + Message message = MessageBuilder.withPayload("TESTING").build(); + tcpOut4.handleMessage(message); + Message mOut = (Message) channel.receive(10000); + assertNotNull(mOut); + assertEquals("TESTING", new String(mOut.getPayload())); + + } + + @SuppressWarnings("unchecked") + @Test + public void testTcpOutbound5() { + Message message = MessageBuilder.withPayload("TESTING").build(); + tcpOut5.handleMessage(message); + Message mOut = (Message) channel.receive(10000); + assertNotNull(mOut); + assertEquals("TESTING", new String(mOut.getPayload())); + + } + + @SuppressWarnings("unchecked") + @Test + public void testTcpOutbound6() { + Message message = MessageBuilder.withPayload("TESTING").build(); + tcpOut6.handleMessage(message); + Message mOut = (Message) channel.receive(10000); + assertNotNull(mOut); + // custom format pads to 24 bytes + assertEquals("TESTING ", new String(mOut.getPayload())); + + } + + @SuppressWarnings("unchecked") + @Test + public void testTcpOutbound6a() { + Message message = MessageBuilder.withPayload( + "abcdefghijklmnopqrdtuvwxyz").build(); + tcpOut6.handleMessage(message); + Message mOut = (Message) channel.receive(10000); + assertNotNull(mOut); + // custom format truncates to 24 bytes + assertEquals("abcdefghijklmnopqrdtuvwx", new String(mOut.getPayload())); + + } + +// /* (non-Javadoc) +// * @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext) +// */ +// @Override +// public void setApplicationContext(ApplicationContext applicationContext) +// throws BeansException { +// String[] names = +// applicationContext.getBeanNamesForType(TcpNetSendingMessageHandler.class); +// for (String n : names) { +// System.out.println(n); +// } +// } + } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/outboundAdapters.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/outboundAdapters.xml index 6e333284f7..15a4c5f24a 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/outboundAdapters.xml +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/outboundAdapters.xml @@ -17,7 +17,7 @@ channel="inChannel" protocol="tcp" host="localhost" - port="9999" + port="#{tcp1.port}" message-format="length-header" using-nio="true" using-direct-buffers="false" @@ -25,6 +25,69 @@ so-timeout="100000" /> + + + + + + + + + + + + + + + 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 7631a30bdc..f3c95afbdc 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 @@ -106,7 +106,7 @@ public class NioSocketReaderTests { server.register(selector, SelectionKey.OP_ACCEPT); // Fire up the sender. - Utils.testSendFragmented(port); + Utils.testSendFragmented(port, false); if(selector.select(10000) <= 0) { fail("Socket failed to connect"); 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 8b3dd5e1af..3f8ea433f3 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 @@ -108,7 +108,7 @@ public class TcpSendingMessageHandlerTests { TcpNetSendingMessageHandler handler = new TcpNetSendingMessageHandler("localhost", port); handler.setMessageFormat(MessageFormats.FORMAT_CUSTOM); handler.setBlockingWrite(true); - handler.setCustomSocketWriteriClassName("org.springframework.integration.ip.tcp.CustomNetSocketWriter"); + handler.setCustomSocketWriterClassName("org.springframework.integration.ip.tcp.CustomNetSocketWriter"); Message message = MessageBuilder.withPayload(testString).build(); handler.handleMessage(message); Thread.sleep(1000000000L); 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/tcp/Utils.java index 90a1f3ad5a..582faf67ca 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/tcp/Utils.java @@ -24,6 +24,9 @@ import java.util.concurrent.CountDownLatch; import javax.net.ServerSocketFactory; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + /** * TCP/IP Test utilities. * @@ -34,6 +37,8 @@ public class Utils { public static final String TEST_STRING = "TestMessage"; + private static final Log logger = LogFactory.getLog(Utils.class); + /** * Sends a message in two chunks with a preceding length. Two such messages are sent. * @param latch If not null, await until counted down before sending second chunk. @@ -48,14 +53,14 @@ public class Utils { ByteBuffer.wrap(len).putInt(TEST_STRING.length() * 2); socket.getOutputStream().write(len); socket.getOutputStream().write(TEST_STRING.getBytes()); - System.out.println(i + " Wrote first part"); + logger.debug(i + " Wrote first part"); if (latch != null) { latch.await(); } Thread.sleep(500); // send the second chunk socket.getOutputStream().write(TEST_STRING.getBytes()); - System.out.println(i + " Wrote second part"); + logger.debug(i + " Wrote second part"); } Thread.sleep(1000000000L); // wait forever, but we're a daemon } catch (Exception e) { @@ -74,19 +79,19 @@ public class Utils { * @param b * @throws Exception */ - public static void testSendFragmented(final int port) { + public static void testSendFragmented(final int port, final boolean noDelay) { Thread thread = new Thread(new Runnable() { public void run() { try { - System.out.println("Connecting to " + port); + logger.debug("Connecting to " + port); Socket socket = new Socket(InetAddress.getByName("localhost"), port); OutputStream os = socket.getOutputStream(); - writeByte(os, 0); - writeByte(os, 0); - writeByte(os, 0); - writeByte(os, 2); - writeByte(os, 'x'); - writeByte(os, 'x'); + writeByte(os, 0, noDelay); + writeByte(os, 0, noDelay); + writeByte(os, 0, noDelay); + writeByte(os, 2, noDelay); + writeByte(os, 'x', noDelay); + writeByte(os, 'x', noDelay); Thread.sleep(1000000000L); // wait forever, but we're a daemon } catch (Exception e) { e.printStackTrace(); @@ -97,9 +102,12 @@ public class Utils { thread.start(); } - private static void writeByte(OutputStream os, int b) throws Exception { + private static void writeByte(OutputStream os, int b, boolean noDelay) throws Exception { os.write(b); - System.out.printf("Wrote 0x%x\n", b); + logger.debug("Wrote 0x%x\n" + Integer.toHexString(b)); + if (noDelay) { + return; + } Thread.sleep(500); } @@ -114,17 +122,17 @@ public class Utils { Socket socket = new Socket(InetAddress.getByName("localhost"), port); OutputStream outputStream = socket.getOutputStream(); for (int i = 0; i < 2; i++) { - writeByte(outputStream, 0x02); + writeByte(outputStream, 0x02, true); outputStream.write(TEST_STRING.getBytes()); - System.out.println(i + " Wrote first part"); + logger.debug(i + " Wrote first part"); if (latch != null) { latch.await(); } Thread.sleep(500); // send the second chunk outputStream.write(TEST_STRING.getBytes()); - System.out.println(i + " Wrote second part"); - writeByte(outputStream, 0x03); + logger.debug(i + " Wrote second part"); + writeByte(outputStream, 0x03, true); } Thread.sleep(1000000000L); // wait forever, but we're a daemon } catch (Exception e) { @@ -148,16 +156,16 @@ public class Utils { OutputStream outputStream = socket.getOutputStream(); for (int i = 0; i < 2; i++) { outputStream.write(TEST_STRING.getBytes()); - System.out.println(i + " Wrote first part"); + logger.debug(i + " Wrote first part"); if (latch != null) { latch.await(); } Thread.sleep(500); // send the second chunk outputStream.write(TEST_STRING.getBytes()); - System.out.println(i + " Wrote second part"); - writeByte(outputStream, '\r'); - writeByte(outputStream, '\n'); + logger.debug(i + " Wrote second part"); + writeByte(outputStream, '\r', true); + writeByte(outputStream, '\n', true); } Thread.sleep(1000000000L); // wait forever, but we're a daemon } catch (Exception e) {