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

INT-4341: RedisQueueIn: support receiveTimeout=0

Browse Source 
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**

# Conflicts:
#	spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java
#	spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java
#	spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml
#	spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java
This commit is contained in:
Matthias Jeschke
2017-09-06 19:48:59 +02:00
committed by Artem Bilan
parent d3928b0a06
commit 684dd5a609
4 changed files with 38 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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.
@@ -45,6 +45,8 @@ import org.springframework.util.Assert;
* @author David Liu
* @author Artem Bilan
* @author Gary Russell
* @author Matthias Jeschke
*
* @since 4.1
*/
@ManagedResource
@@ -125,7 +127,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;
}

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

@@ -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.
@@ -48,6 +48,8 @@ import org.springframework.util.Assert;
* @author Artem Bilan
* @author Gary Russell
* @author Rainer Frey
* @author Matthias Jeschke
*
* @since 3.0
*/
@ManagedResource
@@ -131,7 +133,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;
}

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

@@ -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"/>
@@ -33,9 +34,16 @@
auto-startup="false"
phase="100"/>
<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"/>

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

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2015 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,9 +41,12 @@ 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 Matthias Jeschke
*
* @since 3.0
*/
@ContextConfiguration
@@ -71,6 +74,10 @@ public class RedisQueueInboundChannelAdapterParserTests {
@Qualifier("customAdapter")
private RedisQueueMessageDrivenEndpoint customAdapter;
@Autowired
@Qualifier("zeroReceiveTimeoutAdapter")
private RedisQueueMessageDrivenEndpoint zeroReceiveTimeoutAdapter;
@Autowired
@Qualifier("errorChannel")
private MessageChannel errorChannel;
@@ -87,11 +94,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"));
@@ -105,11 +114,13 @@ public class RedisQueueInboundChannelAdapterParserTests {
assertSame(this.defaultAdapterChannel, TestUtils.getPropertyValue(this.defaultAdapter, "outputChannel"));
}
@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"));
@@ -121,4 +132,10 @@ public class RedisQueueInboundChannelAdapterParserTests {
assertSame(this.sendChannel, TestUtils.getPropertyValue(this.customAdapter, "outputChannel"));
}
@Test
public void testInt4341ZeroReceiveTimeoutConfig() {
assertEquals(0L, TestUtils.getPropertyValue(this.zeroReceiveTimeoutAdapter, "receiveTimeout"));
}
}