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

INT-3198: Add 'recovery-interval' for redis-q-c-a

Browse Source 
JIRA: https://jira.springsource.org/browse/INT-3198

* Expose `RedisQueueMessageDrivenEndpoint.recoveryInterval` to Namespace support
* Add parser's tests
* Rename parser test class
* Polishing for concurrency of`RedisQueueMessageDrivenEndpointTests#testInt3196Recovery` test
* Add documentation for 'recovery-interval' and Redis Events implementation

Doc Polishing

Add more delay because of unsolvable race condition in the Redis Subscription - see DATAREDIS-242
This commit is contained in:
Artem Bilan
2013-11-06 11:52:31 +02:00
committed by Gary Russell
parent f169555224
commit 93bcfdbc55
7 changed files with 61 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java
View File

@@ -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();

9
spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd
View File

@@ -392,6 +392,15 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="recovery-interval" type="xsd:string" default="5000">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="expect-message" type="xsd:string" default="false">
<xsd:annotation>
<xsd:documentation>

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

@@ -28,6 +28,7 @@
serializer="serializer"
error-channel="errorChannel"
receive-timeout="2000"
recovery-interval="3000"
task-executor="executor"
auto-startup="false"
phase="100"/>

4
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java → spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java
View File

@@ -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"));

22
spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java
View File

@@ -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<ApplicationEvent> exceptionEvents = new ArrayList<ApplicationEvent>();
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.<Class<? extends Throwable>> asList(RedisSystemException.class, RedisConnectionFailureException.class)));
Matchers.isIn(Arrays.<Class<? extends Throwable>>asList(RedisSystemException.class, RedisConnectionFailureException.class)));
}
((InitializingBean) this.connectionFactory).afterPropertiesSet();

2
spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java
View File

@@ -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);
}

27
src/reference/docbook/redis.xml
View File

@@ -205,6 +205,7 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
error-channel="" ]]><co id="redis-m-d-c-a-errorChannel"/><![CDATA[
serializer="" ]]><co id="redis-m-d-c-a-serializer"/><![CDATA[
receive-timeout="" ]]><co id="redis-m-d-c-a-receiveTimeout"/><![CDATA[
recovery-interval="" ]]><co id="redis-m-d-c-a-recoveryInterval"/><![CDATA[
expect-message="" ]]><co id="redis-m-d-c-a-expectMessage"/><![CDATA[
task-executor=""/> ]]><co id="redis-m-d-c-a-task-executor"/>
</programlisting>
@@ -247,7 +248,9 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
<callout arearefs="redis-m-d-c-a-errorChannel">
<para>
The <interfacename>MessageChannel</interfacename> to which to send <interfacename>ErrorMessage</interfacename>s with
<interfacename>Exception</interfacename>s from the listening task of the Endpoint.
<interfacename>Exception</interfacename>s from the listening task of the Endpoint. By default
the underlying <classname>MessagePublishingErrorHandler</classname> uses the
default <code>errorChannel</code> from the application context.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-serializer">
@@ -262,6 +265,12 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
The timeout in milliseconds for 'right pop' operation to wait for a Redis message from the queue. Default is 1 second.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-recoveryInterval">
<para>
The time in milliseconds for which the listener task should sleep after exceptions on the 'right pop' operation,
before restarting the listener task.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-expectMessage">
<para>
Specify if this Endpoint expects data from the Redis queue to contain entire <interfacename>Message</interfacename>s.
@@ -344,6 +353,22 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
</calloutlist>
</para>
</section>
<section id="redis-application-events">
<title>Redis Application Events</title>
<para>
Since <emphasis>Spring Integration 3.0</emphasis>, the Redis module provides an implementation
of <classname>IntegrationEvent</classname> - which, in turn, is a
<interfacename>org.springframework.context.ApplicationEvent</interfacename>. The <classname>RedisExceptionEvent</classname>
encapsulates an <classname>Exception</classname>s from Redis operations (with the Endpoint being the <code>source</code>
of the event). For example, the <code>&lt;int-redis:queue-inbound-channel-adapter/&gt;</code>
emits those events after catching <classname>Exception</classname>s from the <code>BoundListOperations.rightPop</code>
operation.
The exception may be any generic <classname>org.springframework.data.redis.RedisSystemException</classname> or
a <classname>org.springframework.data.redis.RedisConnectionFailureException</classname>.
Handling these events using an <code>&lt;int-event:inbound-channel-adapter/&gt;</code> can be useful to determine
problems with background Redis tasks and to take administrative actions.
</para>
</section>
</section>
<section id="redis-message-store">