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

INT-1008 Simple TCP Inbound Gateway; incl tests and namespace parser tests

Browse Source
This commit is contained in:
Gary Russell
2010-04-17 19:52:03 +00:00
parent 1387dbfe83
commit c697c1ea45
10 changed files with 211 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java
View File

@@ -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;

31
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java
View File

@@ -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<NetSocketWriter> 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<NetSocketWriter>) 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<NetSocketWriter> 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 {

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

@@ -154,6 +154,9 @@ public class TcpNetReceivingChannelAdapter extends
if (customSocketReaderClassName != null) {
this.customSocketReader = (Class<NetSocketReader>) Class
.forName(customSocketReaderClassName);
if (!(NetSocketReader.class.isAssignableFrom(this.customSocketReader))) {
throw new IllegalArgumentException("Custom socket reader must be of type NetSocketReader");
}
}
}

9
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java
View File

@@ -77,8 +77,13 @@ public class TcpNetSendingMessageHandler extends
@SuppressWarnings("unchecked")
public void setCustomSocketWriterClassName(
String customSocketWriterClassName) throws ClassNotFoundException {
this.customSocketWriter = (Class<NetSocketWriter>) Class
.forName(customSocketWriterClassName);
if (customSocketWriterClassName != null) {
this.customSocketWriter = (Class<NetSocketWriter>) Class
.forName(customSocketWriterClassName);
if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriter))) {
throw new IllegalArgumentException("Custom socket writer must be of type NetSocketWriter");
}
}
}
}

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

@@ -220,6 +220,9 @@ public class TcpNioReceivingChannelAdapter extends
throws ClassNotFoundException {
this.customSocketReader = (Class<NioSocketReader>) Class
.forName(customSocketReaderClassName);
if (!(NioSocketReader.class.isAssignableFrom(this.customSocketReader))) {
throw new IllegalArgumentException("Custom socket reader must be of type NioSocketReader");
}
}
}

9
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java
View File

@@ -85,8 +85,13 @@ public class TcpNioSendingMessageHandler extends
@SuppressWarnings("unchecked")
public void setCustomSocketWriterClassName(
String customSocketWriterClassName) throws ClassNotFoundException {
this.customSocketWriter = (Class<NioSocketWriter>) Class
.forName(customSocketWriterClassName);
if (customSocketWriterClassName != null) {
this.customSocketWriter = (Class<NioSocketWriter>) Class
.forName(customSocketWriterClassName);
if (!(NioSocketWriter.class.isAssignableFrom(this.customSocketWriter))) {
throw new IllegalArgumentException("Custom socket writer must be of type NioSocketWriter");
}
}
}
/**

17
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -11,6 +11,7 @@
<int:channel id="udpChannel" />
<int:channel id="tcpChannel" />
<int:channel id="replyChannel" />
<ip:inbound-channel-adapter id="testInUdp"
channel="udpChannel"
@@ -173,6 +174,20 @@
so-traffic-class="27"
/>
<ip:inbound-gateway id="simpleInGateway"
request-channel="tcpChannel"
reply-channel="replyChannel"
custom-socket-reader-class-name="org.springframework.integration.ip.tcp.CustomNetSocketReader"
custom-socket-writer-class-name="org.springframework.integration.ip.tcp.CustomNetSocketWriter"
message-format="crlf"
pool-size="23"
port="#{tcpIpUtils.findAvailableServerSocket(6500)}"
receive-buffer-size="123"
so-keep-alive="true"
so-receive-buffer-size="124"
so-send-buffer-size="125"
so-timeout="126"
/>
</beans>

22
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

@@ -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"));
}
}

27
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml
View File

@@ -12,25 +12,36 @@
<beans:bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
<ip:inbound-gateway id="tcpGateway"
<ip:inbound-gateway id="gatewayCrLf"
port="#{tcpIpUtils.findAvailableServerSocket(5200)}"
request-channel="toSA"
reply-channel="fromSA"
message-format="crlf" />
<ip:inbound-gateway id="gatewayStxEtx"
port="#{tcpIpUtils.findAvailableServerSocket(5300)}"
request-channel="toSA"
message-format="stx-etx" />
<ip:inbound-gateway id="gatewayLength"
port="#{tcpIpUtils.findAvailableServerSocket(5400)}"
request-channel="toSA"
message-format="length-header" />
<ip:inbound-gateway id="gatewayCustom"
port="#{tcpIpUtils.findAvailableServerSocket(5500)}"
request-channel="toSA"
message-format="custom"
custom-socket-reader-class-name="org.springframework.integration.ip.tcp.CustomNetSocketReader"
custom-socket-writer-class-name="org.springframework.integration.ip.tcp.CustomNetSocketWriter"/>
<channel id="toSA" />
<channel id="fromSA">
<queue capacity="1"/>
</channel>
<service-activator id="SA"
input-channel="toSA"
output-channel="fromSA"
ref="service"
method="test"
/>
<beans:bean id="service" class="org.springframework.integration.ip.tcp.TestService" />
</beans:beans>

104
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java
View File

@@ -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());
}
}