From c697c1ea45a1713a695114477cc005ed70ae0f5a Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sat, 17 Apr 2010 19:52:03 +0000 Subject: [PATCH] INT-1008 Simple TCP Inbound Gateway; incl tests and namespace parser tests --- .../integration/ip/tcp/NetSocketReader.java | 8 ++ .../ip/tcp/SimpleTcpNetInboundGateway.java | 31 +++++- .../ip/tcp/TcpNetReceivingChannelAdapter.java | 3 + .../ip/tcp/TcpNetSendingMessageHandler.java | 9 +- .../ip/tcp/TcpNioReceivingChannelAdapter.java | 3 + .../ip/tcp/TcpNioSendingMessageHandler.java | 9 +- .../ip/config/ParserUnitTests-context.xml | 17 ++- .../ip/config/ParserUnitTests.java | 22 ++++ ...impleTcpNetInboundGatewayTests-context.xml | 27 +++-- .../tcp/SimpleTcpNetInboundGatewayTests.java | 104 +++++++++++++++++- 10 files changed, 211 insertions(+), 22 deletions(-) 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 481c2f01d9..7ee7b45d08 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 @@ -80,6 +80,10 @@ public class NetSocketReader extends AbstractSocketReader { int n = 0; int bite; while ((bite = inputStream.read()) != ETX) { + if (bite < 0) { + logger.debug("Socket closed"); + throw new IOException("Socket Closed"); + } buffer[n++] = (byte) bite; if (n >= maxMessageSize) { throw new IOException("ETX not found before max message length: " @@ -102,6 +106,10 @@ public class NetSocketReader extends AbstractSocketReader { int bite; while (true) { bite = inputStream.read(); + if (bite < 0) { + logger.debug("Socket closed"); + throw new IOException("Socket Closed"); + } if (n > 0 && bite == '\n' && buffer[n-1] == '\r') break; buffer[n++] = (byte) bite; diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java index f8e8663b49..65edbadcf5 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java @@ -15,9 +15,11 @@ */ package org.springframework.integration.ip.tcp; +import java.lang.reflect.Constructor; import java.net.Socket; import java.net.SocketException; +import org.springframework.beans.BeanUtils; import org.springframework.integration.adapter.MessageMappingException; import org.springframework.integration.core.Message; import org.springframework.integration.gateway.AbstractMessagingGateway; @@ -59,7 +61,7 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { private String customSocketReaderClassName; - private String customSocketWriterClassName; + private Class customSocketWriter; /* (non-Javadoc) @@ -190,10 +192,19 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { } /** - * @param customSocketWriterClassName the customSocketWriterClassName to set + * @param customSocketWriter the customSocketWriter to set + * @throws ClassNotFoundException */ - public void setCustomSocketWriterClassName(String customSocketWriterClassName) { - this.customSocketWriterClassName = customSocketWriterClassName; + @SuppressWarnings("unchecked") + public void setCustomSocketWriterClassName( + String customSocketWriterClassName) throws ClassNotFoundException { + if (customSocketWriterClassName != null) { + this.customSocketWriter = (Class) Class + .forName(customSocketWriterClassName); + if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriter))) { + throw new IllegalArgumentException("Custom socket writer must be of type NetSocketWriter"); + } + } } private class WriteCapableTcpNetReceivingChannelAdapter extends TcpNetReceivingChannelAdapter { @@ -211,7 +222,17 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { @Override protected void processMessage(NetSocketReader reader) { Socket socket = reader.getSocket(); - NetSocketWriter writer = new NetSocketWriter(socket); + NetSocketWriter writer = null; + try { + if (messageFormat == MessageFormats.FORMAT_CUSTOM){ + Constructor ctor = customSocketWriter.getConstructor(Socket.class); + writer = BeanUtils.instantiateClass(ctor, socket); + } else { + writer = new NetSocketWriter(socket); + } + } catch (Exception e) { + throw new MessageMappingException("Error creating SocketWriter", e); + } writer.setMessageFormat(messageFormat); Message message = sendAndReceiveMessage(reader); try { 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 2f5d01bca6..df0f9fafb4 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 @@ -154,6 +154,9 @@ public class TcpNetReceivingChannelAdapter extends if (customSocketReaderClassName != null) { this.customSocketReader = (Class) Class .forName(customSocketReaderClassName); + if (!(NetSocketReader.class.isAssignableFrom(this.customSocketReader))) { + throw new IllegalArgumentException("Custom socket reader must be of type NetSocketReader"); + } } } 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 77ba56ad3f..c7d2b280d4 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 @@ -77,8 +77,13 @@ public class TcpNetSendingMessageHandler extends @SuppressWarnings("unchecked") public void setCustomSocketWriterClassName( String customSocketWriterClassName) throws ClassNotFoundException { - this.customSocketWriter = (Class) Class - .forName(customSocketWriterClassName); + if (customSocketWriterClassName != null) { + this.customSocketWriter = (Class) Class + .forName(customSocketWriterClassName); + if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriter))) { + throw new IllegalArgumentException("Custom socket writer must be of type NetSocketWriter"); + } + } } } 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 456a784a22..6275b0f733 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 @@ -220,6 +220,9 @@ public class TcpNioReceivingChannelAdapter extends throws ClassNotFoundException { this.customSocketReader = (Class) Class .forName(customSocketReaderClassName); + if (!(NioSocketReader.class.isAssignableFrom(this.customSocketReader))) { + throw new IllegalArgumentException("Custom socket reader must be of type NioSocketReader"); + } } } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java index 6b1471caaf..b9576155db 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java @@ -85,8 +85,13 @@ public class TcpNioSendingMessageHandler extends @SuppressWarnings("unchecked") public void setCustomSocketWriterClassName( String customSocketWriterClassName) throws ClassNotFoundException { - this.customSocketWriter = (Class) Class - .forName(customSocketWriterClassName); + if (customSocketWriterClassName != null) { + this.customSocketWriter = (Class) Class + .forName(customSocketWriterClassName); + if (!(NioSocketWriter.class.isAssignableFrom(this.customSocketWriter))) { + throw new IllegalArgumentException("Custom socket writer must be of type NioSocketWriter"); + } + } } /** diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 89652ddda9..88e9f191f6 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -11,6 +11,7 @@ + - + diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index 3fce61e723..340b6fba64 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -28,6 +28,7 @@ import org.springframework.integration.ip.tcp.CustomNetSocketWriter; import org.springframework.integration.ip.tcp.CustomNioSocketReader; import org.springframework.integration.ip.tcp.CustomNioSocketWriter; import org.springframework.integration.ip.tcp.MessageFormats; +import org.springframework.integration.ip.tcp.SimpleTcpNetInboundGateway; import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter; import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler; import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter; @@ -89,6 +90,8 @@ public class ParserUnitTests { @Qualifier(value="org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler#0") TcpNetSendingMessageHandler tcpOutNet; + @Autowired + SimpleTcpNetInboundGateway simpleTcpNetInboundGateway; @Test public void testInUdp() { @@ -239,4 +242,23 @@ public class ParserUnitTests { assertEquals(54, dfa.getPropertyValue("soTimeout")); } + + @Test + public void testInGateway() { + DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetInboundGateway); + assertTrue(simpleTcpNetInboundGateway.getPort() >= 6500); + assertEquals(MessageFormats.FORMAT_CRLF, dfa.getPropertyValue("messageFormat")); + TcpNetReceivingChannelAdapter delegate = (TcpNetReceivingChannelAdapter) dfa + .getPropertyValue("delegate"); + DirectFieldAccessor delegateDfa = new DirectFieldAccessor(delegate); + assertEquals(CustomNetSocketReader.class, delegateDfa.getPropertyValue("customSocketReader")); + assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriter")); + assertEquals(true, dfa.getPropertyValue("soKeepAlive")); + assertEquals(123, dfa.getPropertyValue("receiveBufferSize")); + assertEquals(124, dfa.getPropertyValue("soReceiveBufferSize")); + assertEquals(125, dfa.getPropertyValue("soSendBufferSize")); + assertEquals(126, dfa.getPropertyValue("soTimeout")); + assertEquals(23, dfa.getPropertyValue("poolSize")); + + } } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml index e0256be8be..4af4c674e6 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml @@ -12,25 +12,36 @@ - + + + + + + - - - - - + diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java index 6c23943255..0dbad881ff 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java @@ -21,9 +21,11 @@ import java.net.Socket; import javax.net.SocketFactory; +import org.junit.Ignore; 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.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -35,13 +37,29 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @RunWith(SpringJUnit4ClassRunner.class) public class SimpleTcpNetInboundGatewayTests { + private static int startup = 2000; + @Autowired - SimpleTcpNetInboundGateway gateway; + @Qualifier(value="gatewayCrLf") + SimpleTcpNetInboundGateway gatewayCrLf; + + @Autowired + @Qualifier(value="gatewayStxEtx") + SimpleTcpNetInboundGateway gatewayStxEtx; + + @Autowired + @Qualifier(value="gatewayLength") + SimpleTcpNetInboundGateway gatewayLength; + + @Autowired + @Qualifier(value="gatewayCustom") + SimpleTcpNetInboundGateway gatewayCustom; @Test - public void test1() throws Exception { - Thread.sleep(2000); - Socket socket = SocketFactory.getDefault().createSocket("localhost", gateway.getPort()); + public void testCrLf() throws Exception { + Thread.sleep(startup); + startup = 0; + Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayCrLf.getPort()); String greetings = "Hello World!"; socket.getOutputStream().write((greetings + "\r\n").getBytes()); StringBuilder sb = new StringBuilder(); @@ -55,4 +73,82 @@ public class SimpleTcpNetInboundGatewayTests { } assertEquals("echo:" + greetings + "\r\n", sb.toString()); } + + @Test + public void testStxEtx() throws Exception { + Thread.sleep(startup); + startup = 0; + Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayStxEtx.getPort()); + String greetings = "Hello World!"; + socket.getOutputStream().write(MessageFormats.STX); + socket.getOutputStream().write((greetings).getBytes()); + socket.getOutputStream().write(MessageFormats.ETX); + StringBuilder sb = new StringBuilder(); + int c; + while (true) { + c = socket.getInputStream().read(); + if (c == MessageFormats.STX) { + continue; + } + if (c == MessageFormats.ETX) { + break; + } + sb.append((char) c); + } + assertEquals("echo:" + greetings, sb.toString()); + } + + @Test + public void testLength() throws Exception { + Thread.sleep(startup); + startup = 0; + Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayLength.getPort()); + String greetings = "Hello World!"; + byte[] header = new byte[4]; + header[3] = (byte) greetings.length(); + socket.getOutputStream().write(header); + socket.getOutputStream().write((greetings).getBytes()); + StringBuilder sb = new StringBuilder(); + int c; + int n = 0; + int size = 0; + while (true) { + c = socket.getInputStream().read(); + if (n++ < 3) { + continue; + } + if (n == 4) { + size = c; + continue; + } + sb.append((char) c); + if (n - 4 >= size) { + break; + } + } + assertEquals("echo:" + greetings, sb.toString()); + } + + @Test + public void testCustom() throws Exception { + Thread.sleep(startup); + startup = 0; + Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayCustom.getPort()); + String greetings = "Hello World!"; + String pad = " "; + socket.getOutputStream().write((greetings).getBytes()); + socket.getOutputStream().write(pad.getBytes()); // will be truncated + StringBuilder sb = new StringBuilder(); + int c; + int n = 0; + int size = 24; // custom format is fixed 24 bytes + while (true) { + c = socket.getInputStream().read(); + sb.append((char) c); + if (++n >= size) { + break; + } + } + assertEquals("echo:" + greetings, sb.toString().trim()); + } }