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

INT-781 More namespace work (Still W.i.P)

Browse Source
This commit is contained in:
Gary Russell
2010-02-27 17:23:18 +00:00
parent 44448ea94f
commit 1323aa6368
11 changed files with 214 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -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";
/**

2
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java
View File

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

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

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

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

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

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

@@ -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<NetSocketWriter>) Class
.forName(customSocketWriterClassName);

2
org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd
View File

@@ -70,7 +70,7 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T
<xsd:attribute name="so-linger" type="xsd:string" />
<xsd:attribute name="so-tcp-nodelay" type="xsd:boolean" />
<xsd:attribute name="so-traffic-class" type="xsd:string" />
<xsd:attribute name="blocking-write" type="xsd:boolean" />
<xsd:attribute name="blocking-write" type="xsd:boolean" default="false" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

135
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java
View File

@@ -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<byte[]> message = (Message<byte[]>) 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<byte[]> message = (Message<byte[]>) 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<byte[]> message = (Message<byte[]>) 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<String> message = MessageBuilder.withPayload("TESTING").build();
tcpOut1.handleMessage(message);
Message<byte[]> mOut = (Message<byte[]>) 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<String> message = MessageBuilder.withPayload("TESTING").build();
tcpOut2.handleMessage(message);
Message<byte[]> mOut = (Message<byte[]>) channel.receive(10000);
assertNotNull(mOut);
assertEquals("TESTING", new String(mOut.getPayload()));
}
@SuppressWarnings("unchecked")
@Test
public void testTcpOutbound3() {
Message<String> message = MessageBuilder.withPayload("TESTING").build();
tcpOut3.handleMessage(message);
Message<byte[]> mOut = (Message<byte[]>) channel.receive(10000);
assertNotNull(mOut);
assertEquals("TESTING", new String(mOut.getPayload()));
}
@SuppressWarnings("unchecked")
@Test
public void testTcpOutbound4() {
Message<String> message = MessageBuilder.withPayload("TESTING").build();
tcpOut4.handleMessage(message);
Message<byte[]> mOut = (Message<byte[]>) channel.receive(10000);
assertNotNull(mOut);
assertEquals("TESTING", new String(mOut.getPayload()));
}
@SuppressWarnings("unchecked")
@Test
public void testTcpOutbound5() {
Message<String> message = MessageBuilder.withPayload("TESTING").build();
tcpOut5.handleMessage(message);
Message<byte[]> mOut = (Message<byte[]>) channel.receive(10000);
assertNotNull(mOut);
assertEquals("TESTING", new String(mOut.getPayload()));
}
@SuppressWarnings("unchecked")
@Test
public void testTcpOutbound6() {
Message<String> message = MessageBuilder.withPayload("TESTING").build();
tcpOut6.handleMessage(message);
Message<byte[]> mOut = (Message<byte[]>) channel.receive(10000);
assertNotNull(mOut);
// custom format pads to 24 bytes
assertEquals("TESTING ", new String(mOut.getPayload()));
}
@SuppressWarnings("unchecked")
@Test
public void testTcpOutbound6a() {
Message<String> message = MessageBuilder.withPayload(
"abcdefghijklmnopqrdtuvwxyz").build();
tcpOut6.handleMessage(message);
Message<byte[]> mOut = (Message<byte[]>) 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);
// }
// }
}

65
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/outboundAdapters.xml
View File

@@ -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"
/>
<!-- nio with direct buffers -->
<ip:outbound-channel-adapter id="tcpOut2"
channel="inChannel"
protocol="tcp"
host="localhost"
port="#{tcp2.port}"
message-format="length-header"
using-nio="true"
using-direct-buffers="true"
so-keep-alive="true"
so-timeout="100000"
/>
<!-- net -->
<ip:outbound-channel-adapter id="tcpOut3"
channel="inChannel"
protocol="tcp"
host="localhost"
port="#{tcp3.port}"
message-format="length-header"
using-nio="false"
so-keep-alive="true"
so-timeout="100000"
/>
<!-- net stxetx -->
<ip:outbound-channel-adapter id="tcpOut4"
channel="inChannel"
protocol="tcp"
host="localhost"
port="#{tcp4.port}"
message-format="stx-etx"
using-nio="false"
so-keep-alive="true"
so-timeout="100000"
/>
<!-- net crlf -->
<ip:outbound-channel-adapter id="tcpOut5"
channel="inChannel"
protocol="tcp"
host="localhost"
port="#{tcp5.port}"
message-format="crlf"
using-nio="false"
so-keep-alive="true"
so-timeout="100000"
/>
<!-- net custom -->
<ip:outbound-channel-adapter id="tcpOut6"
channel="inChannel"
protocol="tcp"
host="localhost"
port="#{tcp6.port}"
blocking-write="true"
message-format="custom"
custom-socket-writer-class-name="org.springframework.integration.ip.tcp.CustomNetSocketWriter"
using-nio="false"
so-keep-alive="true"
so-timeout="100000"
/>
<beans:bean id="taskScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
<beans:property name="daemon" value="true" />
<beans:property name="poolSize" value="20" />

2
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java
View File

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

2
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

@@ -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<String> message = MessageBuilder.withPayload(testString).build();
handler.handleMessage(message);
Thread.sleep(1000000000L);

48
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java
View File

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