diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java index 18996384da..401b413dea 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java @@ -50,6 +50,7 @@ public class RedisQueueInboundChannelAdapterParser extends AbstractChannelAdapte IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval"); builder.addPropertyReference("outputChannel", channelName); return builder.getBeanDefinition(); diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd index 450916967e..b6ac1e088d 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd @@ -392,6 +392,15 @@ + + + + Specify the time in milliseconds for which the listener task should sleep after catching + an Exception on a Redis operation, before restarting the listener task. + Default is 5 seconds. + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml similarity index 98% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml index 6c7558adf3..5700eee8ef 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml @@ -28,6 +28,7 @@ serializer="serializer" error-channel="errorChannel" receive-timeout="2000" + recovery-interval="3000" task-executor="executor" auto-startup="false" phase="100"/> diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java similarity index 94% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java index 56ba70bba6..31cad85045 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java @@ -46,7 +46,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) -public class RedisMessageDrivenEndpointParserTests { +public class RedisQueueInboundChannelAdapterParserTests { @Autowired @Qualifier("redisConnectionFactory") @@ -90,6 +90,7 @@ public class RedisMessageDrivenEndpointParserTests { assertEquals("si.test.Int3017.Inbound1", TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.key")); assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "expectMessage", Boolean.class)); assertEquals(new Long(1000), TestUtils.getPropertyValue(this.defaultAdapter, "receiveTimeout", Long.class)); + assertEquals(new Long(5000), TestUtils.getPropertyValue(this.defaultAdapter, "recoveryInterval", Long.class)); assertNull(TestUtils.getPropertyValue(this.defaultAdapter, "errorChannel")); assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "taskExecutor"), Matchers.instanceOf(ErrorHandlingTaskExecutor.class)); assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "serializer"), Matchers.instanceOf(JdkSerializationRedisSerializer.class)); @@ -103,6 +104,7 @@ public class RedisMessageDrivenEndpointParserTests { assertEquals("si.test.Int3017.Inbound2", TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.key")); assertTrue(TestUtils.getPropertyValue(this.customAdapter, "expectMessage", Boolean.class)); assertEquals(new Long(2000), TestUtils.getPropertyValue(this.customAdapter, "receiveTimeout", Long.class)); + assertEquals(new Long(3000), TestUtils.getPropertyValue(this.customAdapter, "recoveryInterval", Long.class)); assertSame(this.errorChannel, TestUtils.getPropertyValue(this.customAdapter, "errorChannel")); assertSame(this.taskExecutor, TestUtils.getPropertyValue(this.customAdapter, "taskExecutor")); assertSame(this.serializer, TestUtils.getPropertyValue(this.customAdapter, "serializer")); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java index ec28d11af0..638067ce99 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java @@ -20,12 +20,15 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.Arrays; import java.util.Date; import java.util.List; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.hamcrest.Matchers; import org.junit.Test; @@ -213,6 +216,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { final List exceptionEvents = new ArrayList(); + final CountDownLatch exceptionsLatch = new CountDownLatch(2); + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); endpoint.setApplicationEventPublisher(new ApplicationEventPublisher() { @@ -220,6 +225,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @Override public void publishEvent(ApplicationEvent event) { exceptionEvents.add(event); + exceptionsLatch.countDown(); } }); endpoint.setOutputChannel(channel); @@ -228,16 +234,26 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { endpoint.afterPropertiesSet(); endpoint.start(); + int n = 0; + do { + n++; + if (n == 100) { + break; + } + Thread.sleep(100); + } while (!endpoint.isListening()); + + assertTrue(n < 100); + ((DisposableBean) this.connectionFactory).destroy(); - Thread.sleep(300); + assertTrue(exceptionsLatch.await(10, TimeUnit.SECONDS)); - assertThat(exceptionEvents.size(), Matchers.greaterThan(0)); for (ApplicationEvent exceptionEvent : exceptionEvents) { assertThat(exceptionEvent, Matchers.instanceOf(RedisExceptionEvent.class)); assertSame(endpoint, exceptionEvent.getSource()); assertThat(((IntegrationEvent) exceptionEvent).getCause().getClass(), - Matchers.isIn(Arrays.> asList(RedisSystemException.class, RedisConnectionFailureException.class))); + Matchers.isIn(Arrays.>asList(RedisSystemException.class, RedisConnectionFailureException.class))); } ((InitializingBean) this.connectionFactory).afterPropertiesSet(); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java index 4adb288a48..b476dc7780 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java @@ -68,6 +68,8 @@ public class RedisAvailableTests { while (n++ < 100 && !connection.isSubscribed()) { Thread.sleep(100); } + // TODO: remove this additional delay when/if https://jira.springsource.org/browse/DATAREDIS-242 is resolved + Thread.sleep(250); assertTrue("RedisMessageListenerContainer Failed to Subscribe", n < 100); } diff --git a/src/reference/docbook/redis.xml b/src/reference/docbook/redis.xml index 1bd7c2023a..ff6c822d9e 100644 --- a/src/reference/docbook/redis.xml +++ b/src/reference/docbook/redis.xml @@ -205,6 +205,7 @@ rt.setConnectionFactory(redisConnectionFactory);]]> error-channel="" ]]> ]]> @@ -247,7 +248,9 @@ rt.setConnectionFactory(redisConnectionFactory);]]> The MessageChannel to which to send ErrorMessages with - Exceptions from the listening task of the Endpoint. + Exceptions from the listening task of the Endpoint. By default + the underlying MessagePublishingErrorHandler uses the + default errorChannel from the application context. @@ -262,6 +265,12 @@ rt.setConnectionFactory(redisConnectionFactory);]]> The timeout in milliseconds for 'right pop' operation to wait for a Redis message from the queue. Default is 1 second. + + + The time in milliseconds for which the listener task should sleep after exceptions on the 'right pop' operation, + before restarting the listener task. + + Specify if this Endpoint expects data from the Redis queue to contain entire Messages. @@ -344,6 +353,22 @@ rt.setConnectionFactory(redisConnectionFactory);]]> +
+ Redis Application Events + + Since Spring Integration 3.0, the Redis module provides an implementation + of IntegrationEvent - which, in turn, is a + org.springframework.context.ApplicationEvent. The RedisExceptionEvent + encapsulates an Exceptions from Redis operations (with the Endpoint being the source + of the event). For example, the <int-redis:queue-inbound-channel-adapter/> + emits those events after catching Exceptions from the BoundListOperations.rightPop + operation. + The exception may be any generic org.springframework.data.redis.RedisSystemException or + a org.springframework.data.redis.RedisConnectionFailureException. + Handling these events using an <int-event:inbound-channel-adapter/> can be useful to determine + problems with background Redis tasks and to take administrative actions. + +