Separate Multicast tests from Unicast; @Ignore Multicast tests; improve regex escaping on SI prefix.

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java

@@ -38,9 +38,9 @@ public abstract class AbstractInternetProtocolSendingMessageHandler implements M
 
 	protected int soReceiveBufferSize = -1;
 
-	protected int soSendBufferSize = -1;
+	protected volatile int soSendBufferSize = -1;
 
-	protected int soTimeout = -1;
+	protected volatile int soTimeout = -1;
 
 
 	public AbstractInternetProtocolSendingMessageHandler(String host, int port) {

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java

@@ -26,6 +26,7 @@ import org.springframework.integration.adapter.MessageMappingException;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessageHeaders;
 import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.ip.util.RegexUtils;
 import org.springframework.integration.message.InboundMessageMapper;
 import org.springframework.integration.message.MessageBuilder;
 import org.springframework.integration.message.MessageHandlingException;
@@ -65,8 +66,10 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
 	private boolean lengthCheck = false;
 
 	private static Pattern udpHeadersPattern = 
-		Pattern.compile("\\" + IpHeaders.ACK_ADDRESS + "=" + "([^;]*);\\" + 
-				MessageHeaders.ID + "=" + "([^;]*);");
+		Pattern.compile(RegexUtils.escapeRegExSpecials(IpHeaders.ACK_ADDRESS) +
+				"=" + "([^;]*);" + 
+				RegexUtils.escapeRegExSpecials(MessageHeaders.ID) + 
+				"=" + "([^;]*);");
 
 
 	public void setCharset(String charset) {
@@ -208,7 +211,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
 	}
 
 	/**
-	 * Peeks at data in he buffer to see if starts with the prefix.
+	 * Peeks at data in the buffer to see if starts with the prefix.
 	 * @param buffer
 	 * @param prefix
 	 * @return

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java

@@ -53,7 +53,7 @@ import org.springframework.util.Assert;
 public class UnicastSendingMessageHandler extends
 		AbstractInternetProtocolSendingMessageHandler implements Runnable {
 
-	private final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
+	protected final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
 
 	protected volatile DatagramSocket socket;
 
@@ -188,7 +188,7 @@ public class UnicastSendingMessageHandler extends
 			logger.debug("Sent packet for message id " + message.getHeaders().getId());
 			if (this.waitForAck) {
 				if (!countdownLatch.await(this.ackTimeout, TimeUnit.MILLISECONDS)) {
-					throw new MessagingException(message, "Failed to received UDP Ack in " + ackTimeout + " millis");
+					throw new MessagingException(message, "Failed to receive UDP Ack in " + ackTimeout + " millis");
 				}
 			}
 		}
@@ -238,6 +238,9 @@ public class UnicastSendingMessageHandler extends
 	public void run() {
 		Exception fatalException = null;
 		try {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Listening for acks on port: " + ackPort);
+			}
 			this.ackSocket = new DatagramSocket(this.ackPort);
 			if (this.soReceiveBufferSize > 0) {
 				ackSocket.setReceiveBufferSize(this.soReceiveBufferSize);

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/util/RegexUtils.java

@@ -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.util;
+
+/**
+ * Regular Expression Utilities.
+ * 
+ * @author Gary Russell
+ *
+ */
+public abstract class RegexUtils {
+
+	/**
+	 * Escapes (precedes with \) any characters in the parameter in the set<br/><br/>
+	 * <code>.$[]^*+{}()\?|</code><br/><br/>
+	 * Used to escape a string that is used as a regular expression pattern, to remove
+	 * the special meaning of these characters.
+	 * @param stringToEscape The string to escape.
+	 * @return The escaped string.
+	 */
+	public static String escapeRegExSpecials(String stringToEscape) {
+		// In the following, we look for all the specials and any we find 
+		// are escaped in the output string, allowing that string to 
+		// be used as a pattern containing the literal specials.
+		String out = stringToEscape.replaceAll(
+				"(\\.|\\$|\\[|\\]|\\^|\\*|\\+|\\{|\\}|\\(|\\)|\\\\|\\?|\\|)", 
+				"\\\\$1");
+		return out;
+	}
+
+}

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/TestIpEndToEnd.java

@@ -65,7 +65,6 @@ public class TestIpEndToEnd implements Runnable {
 
 
 	@Test
-	@Ignore
 	public void runIt() throws Exception {
 		TestIpEndToEnd launcher = new TestIpEndToEnd();
 		Thread t = new Thread(launcher);
@@ -102,24 +101,6 @@ public class TestIpEndToEnd implements Runnable {
 		}
 		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);
-		}
 	}
 
 

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/TestIpMulticastEndToEnd.java

@@ -0,0 +1,145 @@
+/*
+ * 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 TestIpMulticastEndToEnd 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 {
+		TestIpMulticastEndToEnd launcher = new TestIpMulticastEndToEnd();
+		Thread t = new Thread(launcher);
+		t.start(); // launch the receiver
+		AbstractApplicationContext applicationContext = new ClassPathXmlApplicationContext(
+				"testIp-out-multicast-context.xml",
+				TestIpMulticastEndToEnd.class);	
+		launcher.launchSender(applicationContext);
+		applicationContext.stop();
+	}
+
+
+	public void launchSender(ApplicationContext applicationContext) throws Exception {
+		ChannelResolver channelResolver = new BeanFactoryChannelResolver(applicationContext);
+		MessageChannel inputChannel = channelResolver.resolveChannelName("mcInputChannel");
+		try {
+			testingIpText = ">>>>>>> Testing IP (multicast) " + 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);
+	}
+
+
+	/**
+	 * Instantiate the receiving context
+	 */
+	public void run() {
+		AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
+				"testIp-in-multicast-context.xml",
+				TestIpMulticastEndToEnd.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 TestIpMulticastEndToEnd().runIt();
+	}
+
+}

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-common-context.xml

@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:stream="http://www.springframework.org/schema/integration/stream"
+	xmlns:ip="http://www.springframework.org/schema/integration/ip"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/integration/stream
+			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
+			http://www.springframework.org/schema/integration/ip
+			http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
+
+	<beans:bean id="testIp" class="org.springframework.integration.ip.TestIp"/>
+
+ 	<channel id="udpToStdOutChannel">
+		<interceptors>
+			<beans:ref bean="stdoutCatcher"/>
+		</interceptors>
+	</channel>
+
+	<beans:bean id="stdoutCatcher" class = "org.springframework.integration.ip.StdOutCatcher"/>
+
+	<stream:stderr-channel-adapter id="stdout" channel="udpToStdOutChannel" append-newline="true"/> 
+
+	<beans:bean id="taskScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
+		<beans:property name="daemon" value="true" />
+	</beans:bean>
+
+</beans:beans>

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-in-context.xml

@@ -25,28 +25,6 @@
 		multicast="false"
 		check-length="true" />
 
-	<ip:inbound-channel-adapter id="mcUdpReceiver"
-		channel="udpToStdOutChannel"
-		protocol="udp"
-		port="11112"
-		receive-buffer-size="500"
-		multicast="true"
-		multicast-address="225.6.7.8"
-		check-length="true" />
-
- 	<channel id="udpToStdOutChannel">
-		<interceptors>
-			<beans:ref bean="stdoutCatcher"/>
-		</interceptors>
-	</channel>
-
-	<beans:bean id="stdoutCatcher" class = "org.springframework.integration.ip.StdOutCatcher"/>
-
-	<stream:stderr-channel-adapter id="stdout" channel="udpToStdOutChannel" append-newline="true"/> 
-
-	<beans:bean id="taskScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
-		<beans:property name="daemon" value="true" />
-		<beans:property name="poolSize" value="2" />
-	</beans:bean>
-
+	<beans:import resource="testIp-common-context.xml" />
+	
 </beans:beans>

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-in-multicast-context.xml

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:stream="http://www.springframework.org/schema/integration/stream"
+	xmlns:ip="http://www.springframework.org/schema/integration/ip"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/integration/stream
+			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
+			http://www.springframework.org/schema/integration/ip
+			http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
+
+	<!--  
+		Play with the buffer size to force errors. If the checkLength property is 
+		set and this buffer is too small, we'll throw an exception.
+	-->
+	<ip:inbound-channel-adapter id="mcUdpReceiver"
+		channel="udpToStdOutChannel"
+		protocol="udp"
+		port="11112"
+		receive-buffer-size="500"
+		multicast="true"
+		multicast-address="225.6.7.8"
+		check-length="true" />
+
+	<beans:import resource="testIp-common-context.xml" />
+	
+</beans:beans>

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-out-context.xml

@@ -28,8 +28,6 @@
 	                   ref="testIp"
 	                   method="testIp"/>
 
-	<beans:bean id="testIp" class="org.springframework.integration.ip.TestIp"/>
-
 	<ip:outbound-channel-adapter id="udpSender" protocol="udp"
 		host="localhost"
 		port="11111"
@@ -37,32 +35,9 @@
 		acknowledge="true"
 		ack-host="localhost"
 		ack-port="22222"
-		ack-timeout="5000"
+		ack-timeout="10000"
 		channel="outputChannel"/>
 
-	<channel id="mcInputChannel"/>
+	<beans:import resource="testIp-common-context.xml" />
 
-	<channel id="mcOutputChannel" />
-
-	<service-activator input-channel="mcInputChannel"
-	                   output-channel="mcOutputChannel"
-	                   ref="testIp"
-	                   method="testIp"/>
-
-	<ip:outbound-channel-adapter id="mcUdpSender" protocol="udp"
-		multicast="true"
-		time-to-live="2"
-		host="225.6.7.8"
-		port="11112"
-		check-length="true"
-		acknowledge="true"
-		ack-host="localhost"
-		ack-port="22223"
-		ack-timeout="50000"
-		channel="mcOutputChannel"/>
-		
-	<beans:bean id="taskScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
-		<beans:property name="daemon" value="true" />
-	</beans:bean>
- 
 </beans:beans>

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-out-multicast-context.xml

@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:stream="http://www.springframework.org/schema/integration/stream"
+	xmlns:ip="http://www.springframework.org/schema/integration/ip"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/integration/stream
+			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
+			http://www.springframework.org/schema/integration/ip
+			http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
+
+	<stream:stdin-channel-adapter id="stdin" channel="mcOutputChannel" >
+		<poller>
+			<interval-trigger interval="100" time-unit="MILLISECONDS"/>
+		</poller>
+	</stream:stdin-channel-adapter>
+
+	<channel id="mcInputChannel"/>
+
+	<channel id="mcOutputChannel" />
+
+	<service-activator input-channel="mcInputChannel"
+	                   output-channel="mcOutputChannel"
+	                   ref="testIp"
+	                   method="testIp"/>
+
+	<ip:outbound-channel-adapter id="mcUdpSender" protocol="udp"
+		multicast="true"
+		time-to-live="2"
+		host="225.6.7.8"
+		port="11112"
+		check-length="true"
+		acknowledge="true"
+		ack-host="localhost"
+		ack-port="22223"
+		ack-timeout="10000"
+		channel="mcOutputChannel"/>
+
+	<beans:import resource="testIp-common-context.xml" />
+	 		
+</beans:beans>

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/util/RegexUtilsTest.java

@@ -0,0 +1,53 @@
+/*
+ * 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.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.springframework.integration.ip.util.RegexUtils.escapeRegExSpecials;
+
+import org.junit.Test;
+import org.springframework.integration.core.MessageHeaders;
+
+/**
+ * 
+ * @author Gary Russell
+ *
+ */
+public class RegexUtilsTest {
+	/**
+	 * Verify that we properly escape all special characters for matching regex
+	 */
+	@Test
+	public void testRegex () {
+		String s = "xxx$^[]{()}+*\\?|.xxx";
+		assertEquals("xxx\\$\\^\\[\\]\\{\\(\\)\\}\\+\\*\\\\\\?\\|\\.xxx",
+					 escapeRegExSpecials(s));
+	}
+
+	/**
+	 * And one of the ones we are actually using
+	 */
+	@Test
+	public void testSiPrefix () {
+		// protect the test in case we ever change the prefix
+		if ("$^[]{()}+*\\?|.".contains(MessageHeaders.PREFIX)) {
+			assertEquals(escapeRegExSpecials(MessageHeaders.PREFIX), "\\"
+					+ MessageHeaders.PREFIX);
+		}
+	}
+
+}