INT-4341: RedisQueueIn: support receiveTimeout=0

JIRA: https://jira.spring.io/browse/INT-4341
Fixes spring-projects/spring-integration#2228

Add to the `RedisQueueInboundGateway` and `RedisQueueMessageDrivenEndpoint`support
for the `receiveTimeout` of 0 and block indefinitely

Added some simple tests to check if the receive timeout can be set to 0

**Cherry-pick to 4.3.x**

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -46,6 +46,8 @@ import org.springframework.util.Assert;
  * @author David Liu
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Matthias Jeschke
+ *
  * @since 4.1
  */
 @ManagedResource
@@ -126,7 +128,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
 	 * @param receiveTimeout Must be non-negative. Specified in milliseconds.
 	 */
 	public void setReceiveTimeout(long receiveTimeout) {
-		Assert.isTrue(receiveTimeout > 0, "'receiveTimeout' must be > 0.");
+		Assert.isTrue(receiveTimeout >= 0, "'receiveTimeout' must be >= 0.");
 		this.receiveTimeout = receiveTimeout;
 	}
 

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2016 the original author or authors.
+ * Copyright 2013-2017 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.
@@ -49,6 +49,8 @@ import org.springframework.util.Assert;
  * @author Artem Bilan
  * @author Gary Russell
  * @author Rainer Frey
+ * @author Matthias Jeschke
+ *
  * @since 3.0
  */
 @ManagedResource
@@ -132,7 +134,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 	 * @param receiveTimeout Must be non-negative. Specified in milliseconds.
 	 */
 	public void setReceiveTimeout(long receiveTimeout) {
-		Assert.isTrue(receiveTimeout > 0, "'receiveTimeout' must be > 0.");
+		Assert.isTrue(receiveTimeout >= 0, "'receiveTimeout' must be >= 0.");
 		this.receiveTimeout = receiveTimeout;
 	}
 

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml

@@ -11,7 +11,8 @@
 
 	<bean id="redisConnectionFactory"
 		  class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
-		<property name="port" value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
+		<property name="port"
+				  value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
 	</bean>
 
 	<bean id="customRedisConnectionFactory" parent="redisConnectionFactory"/>
@@ -21,22 +22,29 @@
 	<int:channel id="sendChannel"/>
 
 	<int-redis:queue-inbound-channel-adapter id="customAdapter"
-											  queue="si.test.Int3017.Inbound2"
-											  channel="sendChannel"
-											  connection-factory="customRedisConnectionFactory"
-											  expect-message="true"
-											  serializer="serializer"
-											  error-channel="errorChannel"
-											  receive-timeout="2000"
-											  recovery-interval="3000"
-											  task-executor="executor"
-											  auto-startup="false"
-											  phase="100"
-											  right-pop="false"/>
+											 queue="si.test.Int3017.Inbound2"
+											 channel="sendChannel"
+											 connection-factory="customRedisConnectionFactory"
+											 expect-message="true"
+											 serializer="serializer"
+											 error-channel="errorChannel"
+											 receive-timeout="2000"
+											 recovery-interval="3000"
+											 task-executor="executor"
+											 auto-startup="false"
+											 phase="100"
+											 right-pop="false"/>
+
+	<int-redis:queue-inbound-channel-adapter id="zeroReceiveTimeoutAdapter"
+											 queue="si.test.Int3017.Inbound2"
+											 channel="sendChannel"
+											 connection-factory="customRedisConnectionFactory"
+											 receive-timeout="0"/>
 
 	<bean id="executor" class="org.springframework.integration.util.ErrorHandlingTaskExecutor">
 		<constructor-arg ref="threadPoolTaskExecutor"/>
-		<constructor-arg value="#{T(org.springframework.scheduling.support.TaskUtils).LOG_AND_SUPPRESS_ERROR_HANDLER}"/>
+		<constructor-arg
+				value="#{T(org.springframework.scheduling.support.TaskUtils).LOG_AND_SUPPRESS_ERROR_HANDLER}"/>
 	</bean>
 
 	<task:executor id="threadPoolTaskExecutor" pool-size="5"/>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2016 the original author or authors.
+ * Copyright 2013-2017 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.
@@ -41,10 +41,13 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+
 /**
  * @author Artem Bilan
  * @author Gary Russell
  * @author Rainer Frey
+ * @author Matthias Jeschke
+ *
  * @since 3.0
  */
 @ContextConfiguration
@@ -72,6 +75,10 @@ public class RedisQueueInboundChannelAdapterParserTests {
 	@Qualifier("customAdapter")
 	private RedisQueueMessageDrivenEndpoint customAdapter;
 
+	@Autowired
+	@Qualifier("zeroReceiveTimeoutAdapter")
+	private RedisQueueMessageDrivenEndpoint zeroReceiveTimeoutAdapter;
+
 	@Autowired
 	@Qualifier("errorChannel")
 	private MessageChannel errorChannel;
@@ -88,11 +95,13 @@ public class RedisQueueInboundChannelAdapterParserTests {
 	@Autowired
 	private RedisSerializer<?> serializer;
 
+
 	@Test
 	public void testInt3017DefaultConfig() {
 		assertSame(this.connectionFactory,
 				TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.ops.template.connectionFactory"));
-		assertEquals("si.test.Int3017.Inbound1", TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.key"));
+		assertEquals("si.test.Int3017.Inbound1",
+				TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.key"));
 		assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "expectMessage", Boolean.class));
 		assertEquals(1000L, TestUtils.getPropertyValue(this.defaultAdapter, "receiveTimeout"));
 		assertEquals(5000L, TestUtils.getPropertyValue(this.defaultAdapter, "recoveryInterval"));
@@ -107,11 +116,13 @@ public class RedisQueueInboundChannelAdapterParserTests {
 		assertTrue(TestUtils.getPropertyValue(this.defaultAdapter, "rightPop", Boolean.class));
 	}
 
+
 	@Test
 	public void testInt3017CustomConfig() {
 		assertSame(this.customRedisConnectionFactory,
 				TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.ops.template.connectionFactory"));
-		assertEquals("si.test.Int3017.Inbound2", TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.key"));
+		assertEquals("si.test.Int3017.Inbound2",
+				TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.key"));
 		assertTrue(TestUtils.getPropertyValue(this.customAdapter, "expectMessage", Boolean.class));
 		assertEquals(2000L, TestUtils.getPropertyValue(this.customAdapter, "receiveTimeout"));
 		assertEquals(3000L, TestUtils.getPropertyValue(this.customAdapter, "recoveryInterval"));
@@ -124,4 +135,10 @@ public class RedisQueueInboundChannelAdapterParserTests {
 		assertFalse(TestUtils.getPropertyValue(this.customAdapter, "rightPop", Boolean.class));
 	}
 
+
+	@Test
+	public void testInt4341ZeroReceiveTimeoutConfig() {
+		assertEquals(0L, TestUtils.getPropertyValue(this.zeroReceiveTimeoutAdapter, "receiveTimeout"));
+	}
+
 }

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests-context.xml

@@ -1,26 +1,26 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:int="http://www.springframework.org/schema/integration"
-	xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
-	xmlns:task="http://www.springframework.org/schema/task"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
+	   xmlns:task="http://www.springframework.org/schema/task"
+	   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/task http://www.springframework.org/schema/task/spring-task.xsd
 		http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd">
 
 	<int-redis:queue-inbound-gateway id="inboundGateway"
-		request-channel="requestChannel"
-		connection-factory="redisConnectionFactory"
-		reply-channel="receiveChannel"
-		request-timeout="3000"
-		reply-timeout="2000"
-		queue="si.test.queue"
-		task-executor="executor"
-		serializer="serializer"
-		auto-startup="false"
-		extract-payload="false"
-		phase="3"/>
+									 request-channel="requestChannel"
+									 connection-factory="redisConnectionFactory"
+									 reply-channel="receiveChannel"
+									 request-timeout="3000"
+									 reply-timeout="2000"
+									 queue="si.test.queue"
+									 task-executor="executor"
+									 serializer="serializer"
+									 auto-startup="false"
+									 extract-payload="false"
+									 phase="3"/>
 
 	<int:channel id="si.test.queue">
 		<int:queue/>
@@ -35,7 +35,8 @@
 	</int:channel>
 
 	<bean id="redisConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
-		<constructor-arg value="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
+		<constructor-arg
+				value="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
 		</constructor-arg>
 	</bean>
 
@@ -43,4 +44,9 @@
 
 	<task:executor id="executor" pool-size="10"/>
 
+	<int-redis:queue-inbound-gateway id="zeroReceiveTimeoutGateway"
+									 request-channel="requestChannel"
+									 receive-timeout="0"
+									 queue="si.test.queue"/>
+
 </beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2015 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -35,9 +35,12 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+
 /**
  * @author David Liu
  * @author Artem Bilan
+ * @author Matthias Jeschke
+ *
  * since 4.1
  */
 @ContextConfiguration
@@ -49,6 +52,10 @@ public class RedisQueueInboundGatewayParserTests {
 	@Qualifier("inboundGateway")
 	private RedisQueueInboundGateway defaultGateway;
 
+	@Autowired
+	@Qualifier("zeroReceiveTimeoutGateway")
+	private RedisQueueInboundGateway zeroReceiveTimeoutGateway;
+
 	@Autowired
 	@Qualifier("receiveChannel")
 	private MessageChannel receiveChannel;
@@ -60,6 +67,7 @@ public class RedisQueueInboundGatewayParserTests {
 	@Autowired
 	private RedisSerializer<?> serializer;
 
+
 	@Test
 	public void testDefaultConfig() throws Exception {
 		assertFalse(TestUtils.getPropertyValue(this.defaultGateway, "extractPayload", Boolean.class));
@@ -73,4 +81,9 @@ public class RedisQueueInboundGatewayParserTests {
 		assertEquals(3, TestUtils.getPropertyValue(this.defaultGateway, "phase"));
 	}
 
+	@Test
+	public void testZeroReceiveTimeoutConfig() throws Exception {
+		assertEquals(0L, TestUtils.getPropertyValue(this.zeroReceiveTimeoutGateway, "receiveTimeout"));
+	}
+
 }