committing tests
This commit is contained in:
Mark Fisher
@@ -0,0 +1,79 @@
|
||||
/*
|
||||
* 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 static org.junit.Assert.assertEquals;
|
||||
|
||||
import java.net.DatagramPacket;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.core.Message;
|
||||
import org.springframework.integration.ip.udp.DatagramPacketMessageMapper;
|
||||
import org.springframework.integration.message.MessageBuilder;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @since 2.0
|
||||
*/
|
||||
public class DatagramPacketMessageMapperTests {
|
||||
|
||||
@Test
|
||||
public void testFromToMessage() throws Exception {
|
||||
test(false, false);
|
||||
test(true, false);
|
||||
test(false, true);
|
||||
test(true, true);
|
||||
}
|
||||
|
||||
private void test(boolean ack, boolean lengthCheck) throws Exception {
|
||||
Message<byte[]> message = MessageBuilder.withPayload("ABCD".getBytes()).build();
|
||||
DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
|
||||
mapper.setAckAddress("localhost:11111");
|
||||
mapper.setAcknowledge(ack);
|
||||
mapper.setLengthCheck(lengthCheck);
|
||||
DatagramPacket packet = mapper.fromMessage(message);
|
||||
packet.setSocketAddress(new InetSocketAddress("localhost", 22222));
|
||||
Message<byte[]> messageOut = mapper.toMessage(packet);
|
||||
assertEquals(new String(message.getPayload()), new String(messageOut.getPayload()));
|
||||
if (ack) {
|
||||
assertEquals(message.getHeaders().getId().toString(),
|
||||
messageOut.getHeaders().getId().toString());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@Ignore
|
||||
public void testTruncation() throws Exception {
|
||||
Message<byte[]> message = MessageBuilder.withPayload("ABCD".getBytes()).build();
|
||||
DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
|
||||
mapper.setAckAddress("localhost:11111");
|
||||
mapper.setAcknowledge(false);
|
||||
mapper.setLengthCheck(true);
|
||||
DatagramPacket packet = mapper.fromMessage(message);
|
||||
// Force a truncation failure
|
||||
ByteBuffer bb = ByteBuffer.wrap(packet.getData());
|
||||
bb.putInt(99999);
|
||||
packet.setSocketAddress(new InetSocketAddress("localhost", 22222));
|
||||
Message<byte[]> messageOut = mapper.toMessage(packet);
|
||||
assertEquals(new String(message.getPayload()), new String(messageOut.getPayload()));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,231 @@
|
||||
/*
|
||||
* 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 static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.net.DatagramPacket;
|
||||
import java.net.DatagramSocket;
|
||||
import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.MulticastSocket;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.core.Message;
|
||||
import org.springframework.integration.ip.udp.DatagramPacketMessageMapper;
|
||||
import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
|
||||
import org.springframework.integration.ip.udp.UnicastSendingMessageHandler;
|
||||
import org.springframework.integration.message.MessageBuilder;
|
||||
|
||||
/**
|
||||
* @author Mark Fisher
|
||||
* @author Gary Russell
|
||||
* @since 2.0
|
||||
*/
|
||||
public class DatagramPacketSendingHandlerTests {
|
||||
|
||||
@Test
|
||||
public void verifySend() throws Exception {
|
||||
final int testPort = 27816;
|
||||
byte[] buffer = new byte[8];
|
||||
final DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length);
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
public void run() {
|
||||
try {
|
||||
DatagramSocket socket = new DatagramSocket(testPort);
|
||||
socket.receive(receivedPacket);
|
||||
latch.countDown();
|
||||
socket.close();
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
});
|
||||
Thread.sleep(1000);
|
||||
UnicastSendingMessageHandler handler =
|
||||
new UnicastSendingMessageHandler("localhost", testPort);
|
||||
String payload = "foo";
|
||||
handler.handleMessage(MessageBuilder.withPayload(payload).build());
|
||||
assertTrue(latch.await(3000, TimeUnit.MILLISECONDS));
|
||||
byte[] src = receivedPacket.getData();
|
||||
int length = receivedPacket.getLength();
|
||||
int offset = receivedPacket.getOffset();
|
||||
byte[] dest = new byte[length];
|
||||
System.arraycopy(src, offset, dest, 0, length);
|
||||
assertEquals(payload, new String(dest));
|
||||
handler.shutDown();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void verifySendWithAck() throws Exception {
|
||||
final int testPort = 27816;
|
||||
final int ackPort = 17816;
|
||||
byte[] buffer = new byte[1000];
|
||||
final DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length);
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
UnicastSendingMessageHandler handler =
|
||||
new UnicastSendingMessageHandler("localhost", testPort, true,
|
||||
true, "localhost", ackPort, 5000);
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
public void run() {
|
||||
try {
|
||||
DatagramSocket socket = new DatagramSocket(testPort);
|
||||
socket.receive(receivedPacket);
|
||||
socket.close();
|
||||
DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
|
||||
mapper.setAcknowledge(true);
|
||||
mapper.setLengthCheck(true);
|
||||
Message<byte[]> message = mapper.toMessage(receivedPacket);
|
||||
Object id = message.getHeaders().getId();
|
||||
byte[] ack = id.toString().getBytes();
|
||||
DatagramPacket ackPack = new DatagramPacket(ack, ack.length,
|
||||
new InetSocketAddress("localHost", ackPort));
|
||||
DatagramSocket out = new DatagramSocket();
|
||||
out.send(ackPack);
|
||||
out.close();
|
||||
latch.countDown();
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
});
|
||||
Thread.sleep(1000);
|
||||
String payload = "foobar";
|
||||
handler.handleMessage(MessageBuilder.withPayload(payload).build());
|
||||
assertTrue(latch.await(3000, TimeUnit.MILLISECONDS));
|
||||
byte[] src = receivedPacket.getData();
|
||||
int length = receivedPacket.getLength();
|
||||
int offset = receivedPacket.getOffset();
|
||||
byte[] dest = new byte[6];
|
||||
System.arraycopy(src, offset+length-6, dest, 0, 6);
|
||||
assertEquals(payload, new String(dest));
|
||||
handler.shutDown();
|
||||
}
|
||||
|
||||
@Test
|
||||
@Ignore
|
||||
public void verifySendMulticast() throws Exception {
|
||||
final int testPort = 27816;
|
||||
final String multicastAddress = "225.6.7.8";
|
||||
final String payload = "foo";
|
||||
final CountDownLatch latch = new CountDownLatch(2);
|
||||
Runnable catcher = new Runnable() {
|
||||
public void run() {
|
||||
try {
|
||||
byte[] buffer = new byte[8];
|
||||
DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length);
|
||||
MulticastSocket socket = new MulticastSocket(testPort);
|
||||
InetAddress group = InetAddress.getByName(multicastAddress);
|
||||
socket.joinGroup(group);
|
||||
LogFactory.getLog(getClass())
|
||||
.debug(Thread.currentThread().getName() + " waiting for packet");
|
||||
socket.receive(receivedPacket);
|
||||
socket.close();
|
||||
byte[] src = receivedPacket.getData();
|
||||
int length = receivedPacket.getLength();
|
||||
int offset = receivedPacket.getOffset();
|
||||
byte[] dest = new byte[length];
|
||||
System.arraycopy(src, offset, dest, 0, length);
|
||||
assertEquals(payload, new String(dest));
|
||||
LogFactory.getLog(getClass())
|
||||
.debug(Thread.currentThread().getName() + " received packet");
|
||||
latch.countDown();
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
};
|
||||
Executor executor = Executors.newFixedThreadPool(2);
|
||||
executor.execute(catcher);
|
||||
executor.execute(catcher);
|
||||
Thread.sleep(1000);
|
||||
MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler(multicastAddress, testPort);
|
||||
handler.handleMessage(MessageBuilder.withPayload(payload).build());
|
||||
assertTrue(latch.await(3000, TimeUnit.MILLISECONDS));
|
||||
handler.shutDown();
|
||||
}
|
||||
|
||||
@Test
|
||||
@Ignore
|
||||
public void verifySendMulticastWithAcks() throws Exception {
|
||||
final int testPort = 27816;
|
||||
final int ackPort = 17817;
|
||||
final String multicastAddress = "225.6.7.8";
|
||||
final String payload = "foobar";
|
||||
final CountDownLatch latch = new CountDownLatch(2);
|
||||
Runnable catcher = new Runnable() {
|
||||
public void run() {
|
||||
try {
|
||||
byte[] buffer = new byte[1000];
|
||||
DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length);
|
||||
MulticastSocket socket = new MulticastSocket(testPort);
|
||||
InetAddress group = InetAddress.getByName(multicastAddress);
|
||||
socket.joinGroup(group);
|
||||
LogFactory.getLog(getClass()).debug(Thread.currentThread().getName() + " waiting for packet");
|
||||
socket.receive(receivedPacket);
|
||||
socket.close();
|
||||
byte[] src = receivedPacket.getData();
|
||||
int length = receivedPacket.getLength();
|
||||
int offset = receivedPacket.getOffset();
|
||||
byte[] dest = new byte[6];
|
||||
System.arraycopy(src, offset+length-6, dest, 0, 6);
|
||||
assertEquals(payload, new String(dest));
|
||||
LogFactory.getLog(getClass()).debug(Thread.currentThread().getName() + " received packet");
|
||||
DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
|
||||
mapper.setAcknowledge(true);
|
||||
mapper.setLengthCheck(true);
|
||||
Message<byte[]> message = mapper.toMessage(receivedPacket);
|
||||
Object id = message.getHeaders().getId();
|
||||
byte[] ack = id.toString().getBytes();
|
||||
DatagramPacket ackPack = new DatagramPacket(ack, ack.length,
|
||||
new InetSocketAddress("localHost", ackPort));
|
||||
DatagramSocket out = new DatagramSocket();
|
||||
out.send(ackPack);
|
||||
out.close();
|
||||
latch.countDown();
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
};
|
||||
Executor executor = Executors.newFixedThreadPool(2);
|
||||
executor.execute(catcher);
|
||||
executor.execute(catcher);
|
||||
Thread.sleep(1000);
|
||||
MulticastSendingMessageHandler handler =
|
||||
new MulticastSendingMessageHandler(multicastAddress, testPort, true,
|
||||
true, "localhost", ackPort, 500000);;
|
||||
handler.setMinAcksForSuccess(2);
|
||||
handler.handleMessage(MessageBuilder.withPayload(payload).build());
|
||||
assertTrue(latch.await(3000, TimeUnit.MILLISECONDS));
|
||||
handler.shutDown();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
/*
|
||||
* 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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
/*
|
||||
* 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;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @since 2.0
|
||||
*/
|
||||
public class TestIp {
|
||||
|
||||
public String testIp(String input) {
|
||||
return input;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
/*
|
||||
* 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 static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
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.core.MessageChannel;
|
||||
import org.springframework.integration.message.StringMessage;
|
||||
|
||||
/**
|
||||
* Sends and receives a simple message through to the Udp channel adapters.
|
||||
* If run as a JUnit just sends one message and terminates (see console).
|
||||
* TODO: Use a custom output stream and catch output to verify.
|
||||
*
|
||||
* If run from main(),
|
||||
* hangs around for a couple of minutes to allow console interaction (enter a message on the
|
||||
* console and you should see it go through the outbound context, over UDP, and
|
||||
* received in the other context (and written back to the console).
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @since 2.0
|
||||
*/
|
||||
public class TestIpEndToEnd implements Runnable {
|
||||
|
||||
private String testingIpText;
|
||||
|
||||
private String stdOutput;
|
||||
|
||||
private CountDownLatch sentFirst = new CountDownLatch(1);
|
||||
|
||||
private CountDownLatch firstReceived = new CountDownLatch(1);
|
||||
|
||||
private CountDownLatch doneProcessing = new CountDownLatch(1);
|
||||
|
||||
private boolean okToRun = true;
|
||||
|
||||
private static long hangAroundFor = 0;
|
||||
|
||||
|
||||
@Test
|
||||
@Ignore
|
||||
public void runIt() throws Exception {
|
||||
TestIpEndToEnd launcher = new TestIpEndToEnd();
|
||||
Thread t = new Thread(launcher);
|
||||
t.start(); // launch the receiver
|
||||
AbstractApplicationContext applicationContext = new ClassPathXmlApplicationContext("testIp-out-context.xml", TestIpEndToEnd.class);
|
||||
launcher.launchSender(applicationContext);
|
||||
applicationContext.stop();
|
||||
}
|
||||
|
||||
|
||||
public void launchSender(ApplicationContext applicationContext) throws Exception {
|
||||
ChannelResolver channelResolver = new BeanFactoryChannelResolver(applicationContext);
|
||||
MessageChannel inputChannel = channelResolver.resolveChannelName("inputChannel");
|
||||
try {
|
||||
testingIpText = ">>>>>>> Testing IP " + new Date();
|
||||
inputChannel.send(new StringMessage(testingIpText));
|
||||
sentFirst.countDown();
|
||||
try {
|
||||
Thread.sleep(hangAroundFor); // give some time for console interaction
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
finally {
|
||||
if (hangAroundFor == 0) {
|
||||
sentFirst = new CountDownLatch(1);
|
||||
}
|
||||
else {
|
||||
okToRun = false;
|
||||
}
|
||||
// tell the receiver to we're done
|
||||
doneProcessing.countDown();
|
||||
}
|
||||
assertTrue(firstReceived.await(2, TimeUnit.SECONDS));
|
||||
assertEquals(testingIpText, stdOutput);
|
||||
if (hangAroundFor == 0) {
|
||||
// If we're running in JUnit mode, now try the multicast version
|
||||
firstReceived = new CountDownLatch(1);
|
||||
doneProcessing = new CountDownLatch(1);
|
||||
inputChannel = channelResolver.resolveChannelName("mcInputChannel");
|
||||
try {
|
||||
testingIpText = ">>>>>>> Testing IP (multicast) " + new Date();
|
||||
inputChannel.send(new StringMessage(testingIpText));
|
||||
sentFirst.countDown();
|
||||
}
|
||||
finally {
|
||||
okToRun = false;
|
||||
// tell the receiver to shutdown
|
||||
doneProcessing.countDown();
|
||||
}
|
||||
assertTrue(firstReceived.await(2, TimeUnit.SECONDS));
|
||||
assertEquals(testingIpText, stdOutput);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Instantiate the receiving context
|
||||
*/
|
||||
public void run() {
|
||||
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("testIp-in-context.xml", TestIpEndToEnd.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();
|
||||
firstReceived.countDown();
|
||||
try {
|
||||
doneProcessing.await();
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
ctx.stop();
|
||||
}
|
||||
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
hangAroundFor = 120000;
|
||||
new TestIpEndToEnd().runIt();
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user