diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java index 282aef8ac8..0726489d1f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java @@ -34,6 +34,8 @@ public abstract class IpHeaders { public static final String IP_ADDRESS = IP + "address"; + public static final String PORT = IP + "port"; + public static final String ACK_ADDRESS = IP + "ackTo"; public static final String ACK_ID = IP + "ackId"; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java index ff72e6b783..2dad44d150 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java @@ -41,16 +41,16 @@ import org.springframework.util.Assert; * may be either a byte array or a String. The default charset for converting * a String to a byte array is UTF-8, but that may be changed by invoking the * {@link #setCharset(String)} method. - * + * * By default, the UDP messages will be unreliable (truncation may occur on * the receiving end; packets may be lost). - * + * * Reliability can be enhanced by one or both of the following techniques: * - * + * * @author Mark Fisher * @author Gary Russell * @author Dave Syer @@ -63,15 +63,15 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper message = MessageBuilder.withPayload("ABCD".getBytes()).build(); + DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); + DatagramPacket packet = mapper.fromMessage(message); + packet.setSocketAddress(new InetSocketAddress("localhost", port)); + final DatagramSocket socket = new DatagramSocket(SocketUtils.findAvailableUdpSocket()); + socket.send(packet); + final AtomicReference theAnswer = new AtomicReference(); + final CountDownLatch receiverReadyLatch = new CountDownLatch(1); + final CountDownLatch replyReceivedLatch = new CountDownLatch(1); + //main thread sends the reply using the headers, this thread will receive it + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + DatagramPacket answer = new DatagramPacket(new byte[2000], 2000); + try { + receiverReadyLatch.countDown(); + socket.receive(answer); + theAnswer.set(answer); + replyReceivedLatch.countDown(); + } catch (IOException e) { + e.printStackTrace(); + } + } + }); + Message receivedMessage = (Message) channel.receive(2000); + assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); + String replyString = "reply:" + System.currentTimeMillis(); + byte[] replyBytes = replyString.getBytes(); + DatagramPacket reply = new DatagramPacket(replyBytes, replyBytes.length); + reply.setSocketAddress(new InetSocketAddress( + (String) receivedMessage.getHeaders().get(IpHeaders.IP_ADDRESS), + (Integer) receivedMessage.getHeaders().get(IpHeaders.PORT))); + assertTrue(receiverReadyLatch.await(10, TimeUnit.SECONDS)); + new DatagramSocket().send(reply); + assertTrue(replyReceivedLatch.await(10, TimeUnit.SECONDS)); + DatagramPacket answerPacket = theAnswer.get(); + assertNotNull(answerPacket); + assertEquals(replyString, new String(answerPacket.getData(), 0, answerPacket.getLength())); + } + @SuppressWarnings("unchecked") @Test public void testUnicastSender() throws Exception {