From b9c8a8edf89fd127f98ef1538aa2ffa35c894fd6 Mon Sep 17 00:00:00 2001 From: Matthias Jeschke Date: Wed, 6 Sep 2017 19:48:59 +0200 Subject: [PATCH] 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** --- .../inbound/RedisQueueInboundGateway.java | 6 ++- .../RedisQueueMessageDrivenEndpoint.java | 6 ++- ...boundChannelAdapterParserTests-context.xml | 36 ++++++++++------- ...QueueInboundChannelAdapterParserTests.java | 23 +++++++++-- ...QueueInboundGatewayParserTests-context.xml | 40 +++++++++++-------- .../RedisQueueInboundGatewayParserTests.java | 15 ++++++- 6 files changed, 87 insertions(+), 39 deletions(-) diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java index 21859925e2..94b5d92fc4 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java +++ b/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; } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java index 825b09a356..ae5be9fb65 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java +++ b/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; } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml index b149ac3e6b..1d82b8173a 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml @@ -11,7 +11,8 @@ - + @@ -21,22 +22,29 @@ + 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"/> + + - + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java index 213a4eb587..ebac6c4cac 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java +++ b/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")); + } + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests-context.xml index e184bf91ac..e05534c007 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests-context.xml @@ -1,26 +1,26 @@ + 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"/> @@ -35,7 +35,8 @@ - + @@ -43,4 +44,9 @@ + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests.java index 950a16f31a..e69b6b103c 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests.java +++ b/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")); + } + }