From cd3f13ac91f433ff3d899b3807bb095c5a6a2eee Mon Sep 17 00:00:00 2001
From: Mauro Franceschini
Date: Mon, 4 Nov 2013 10:45:22 +0200
Subject: [PATCH 01/10] INT-3193: Add ObjectToMapTransformer flatten attr
JIRA: https://jira.springsource.org/browse/INT-3193
---
.../xml/ObjectToMapTransformerParser.java | 4 ++-
.../config/xml/spring-integration-3.0.xsd | 9 ++++++
...ectToMapTransformerParserTests-context.xml | 8 ++++++
.../ObjectToMapTransformerParserTests.java | 28 +++++++++++++++++++
src/reference/docbook/transformer.xml | 23 +++++++++++++++
5 files changed, 71 insertions(+), 1 deletion(-)
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ObjectToMapTransformerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ObjectToMapTransformerParser.java
index 6226f63cfa..501ffaf4d4 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ObjectToMapTransformerParser.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ObjectToMapTransformerParser.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -22,6 +22,7 @@ import org.w3c.dom.Element;
/**
* @author Oleg Zhurakousky
+ * @author Mauro Franceschini
* @since 2.0
*/
public class ObjectToMapTransformerParser extends AbstractTransformerParser {
@@ -33,5 +34,6 @@ public class ObjectToMapTransformerParser extends AbstractTransformerParser {
@Override
protected void parseTransformer(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "flatten", "shouldFlattenKeys");
}
}
diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd
index 4c6867c967..a7b47c6edb 100644
--- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd
+++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd
@@ -2169,6 +2169,15 @@
+
+
+
+ Specifies if the result Map of Maps should be transformed further to flat keys of
+ object's property paths.
+ Default is 'true'.
+
+
+
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToMapTransformerParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToMapTransformerParserTests-context.xml
index c0dfd12e5d..11ef5d780d 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToMapTransformerParserTests-context.xml
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToMapTransformerParserTests-context.xml
@@ -15,4 +15,12 @@
+
+
+
+
+
+
+
+
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToMapTransformerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToMapTransformerParserTests.java
index 057ecb086b..75eb4ecd6f 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToMapTransformerParserTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToMapTransformerParserTests.java
@@ -21,6 +21,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
@@ -40,10 +41,12 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
/**
* @author Oleg Zhurakousky
* @author Gunnar Hillert
+ * @author Mauro Franceschini
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@@ -57,6 +60,14 @@ public class ObjectToMapTransformerParserTests {
@Qualifier("output")
private PollableChannel output;
+ @Autowired
+ @Qualifier("nestedInput")
+ private MessageChannel nestedInput;
+
+ @Autowired
+ @Qualifier("nestedOutput")
+ private PollableChannel nestedOutput;
+
@SuppressWarnings("unchecked")
@Test
@@ -90,6 +101,23 @@ public class ObjectToMapTransformerParserTests {
directInput.send(message);
}
+ @Test
+ public void testObjectToNotFlattenedMapTransformer(){
+ Employee employee = this.buildEmployee();
+
+ Message message = MessageBuilder.withPayload(employee).build();
+ nestedInput.send(message);
+
+ @SuppressWarnings("unchecked")
+ Message
*
- * NOTICE: This implementation may change for Spring Integration
- * 3.0. It is provided for use-cases where the current {@link JdbcMessageStore}
- * is not delivering the desired performance characteristics.
- *
- *
- *
* As such, the {@link JdbcChannelMessageStore} uses database specific SQL queries.
*
*
@@ -89,8 +83,8 @@ import org.springframework.util.StringUtils;
* database table only. The SQL scripts to create the necessary table are packaged
* under org/springframework/integration/jdbc/messagestore/channel/schema-*.sql,
* where * denotes the target database type.
- *
+ *
+ *
* @author Gunnar Hillert
* @author Artem Bilan
* @since 2.2
@@ -331,7 +325,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
*
For this to work, you must setup the corresponding
* {@link TransactionSynchronizationFactory}:
*
- *
+ *
* {@code
*
*
@@ -343,7 +337,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
* This {@link TransactionSynchronizationFactory} is then referenced in the
* transaction configuration of the poller:
*
- *
*
diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/MapContentMatchers.java b/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/MapContentMatchers.java
index 98e29142e6..4997dbca7a 100644
--- a/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/MapContentMatchers.java
+++ b/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/MapContentMatchers.java
@@ -33,7 +33,7 @@ import org.hamcrest.core.IsEqual;
* It is possible to match a single entry by value or matcher like this:
*
*
- *
+ *
* assertThat(map, hasEntry(SOME_KEY, is(SOME_VALUE)));
* assertThat(map, hasEntry(SOME_KEY, is(String.class)));
* assertThat(map, hasEntry(SOME_KEY, notNullValue()));
@@ -43,16 +43,18 @@ import org.hamcrest.core.IsEqual;
* It's also possible to match multiple entries in a map:
*
*
- *
* An example using {@link Assert#assertThat(Object, Matcher)} delegating to
* another {@link Matcher}.
*
- *
- * ANY_PAYLOAD = new BigDecimal("1.123");
+ *
+ * ANY_PAYLOAD = new BigDecimal("1.123");
* assertThat(message, PayloadMatcher.hasPayload(is(BigDecimal.class)));
* assertThat(message, PayloadMatcher.hasPayload(notNullValue()));
* assertThat(message, not((PayloadMatcher.hasPayload(is(String.class))))); *
diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/util/SocketUtils.java b/spring-integration-test/src/main/java/org/springframework/integration/test/util/SocketUtils.java
index e54f27cbc4..b57d959ca7 100644
--- a/spring-integration-test/src/main/java/org/springframework/integration/test/util/SocketUtils.java
+++ b/spring-integration-test/src/main/java/org/springframework/integration/test/util/SocketUtils.java
@@ -47,7 +47,7 @@ public final class SocketUtils {
* the need to use the methods of this class multiple times from within your
* Spring Application Context XML file using SpEL. Of course you can do:
*
- *
+ *
* {@code
* ...port="#{T(org.springframework.integration.test.util.SocketUtils).findAvailableServerSocket(12000)}"
* }
@@ -57,7 +57,7 @@ public final class SocketUtils {
* This will be acceptable for single use, but if you need to invoke the
* methods numerous time, you may instead want to do this:
*
- *
+ *
* {@code
*
*
From 156eeeb73876e90424c8ac9fccdb392450615dac Mon Sep 17 00:00:00 2001
From: Artem Bilan
Date: Tue, 5 Nov 2013 14:02:06 +0200
Subject: [PATCH 08/10] INT-3167: Polishing Twitter Documentation
JIRA: https://jira.springsource.org/browse/INT-3167
Doc Polishing
---
src/reference/docbook/twitter.xml | 45 ++++++++++++++++++-------------
1 file changed, 26 insertions(+), 19 deletions(-)
diff --git a/src/reference/docbook/twitter.xml b/src/reference/docbook/twitter.xml
index c78a2e7cc9..f99e602207 100644
--- a/src/reference/docbook/twitter.xml
+++ b/src/reference/docbook/twitter.xml
@@ -17,11 +17,11 @@
subscribers who are known as followers.
-
- Previous versions of Spring Integration were dependent upon the Twitter4J API,
- but with the release of Spring Social 1.0 GA,
- Spring Integration, as of version 2.1, now builds directly upon Spring Social's Twitter support, instead of Twitter4J.
-
+
+ Versions of Spring Integration prior to 2.1 were dependent upon the Twitter4J API,
+ but with the release of Spring Social 1.0 GA,
+ Spring Integration, as of version 2.1, now builds directly upon Spring Social's Twitter support, instead of Twitter4J.
+
@@ -39,9 +39,9 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/twitter
The Twitter API allows for both authenticated and anonymous operations. For authenticated operations Twitter uses OAuth
- an authentication protocol that allows users to approve an application to act on their behalf without
- sharing their password. More information can be found at http://oauth.net/ or
- in this article http://hueniverse.com/oauth/ from Hueniverse.
- Please also see OAuth FAQ for more information about OAuth and Twitter.
+ sharing their password. More information can be found at http://oauth.net or
+ in this article http://hueniverse.com/oauth from Hueniverse.
+ Please also see OAuth FAQ for more information about OAuth and Twitter.
In order to use OAuth authentication/authorization with Twitter you must create a new Application on the Twitter Developers site.
@@ -50,7 +50,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/twitter
- Go to http://dev.twitter.com/
+ Go to http://dev.twitter.comClick on the Register an app link and fill out all required fields on the form provided;
@@ -121,21 +121,28 @@ twitter.oauth.accessTokenSecret=AbRxUAvyNCtqQtxFK8w5ZMtMj20KFhB6o]]>Twitter Inbound Adapters
Twitter inbound adapters allow you to receive Twitter Messages. There are several types of
- twitter messages, or tweets
+ twitter messages, or tweetsSpring Integration version 2.0 and above provides support for receiving tweets as Timeline Updates,
Direct Messages, Mention Messages as well as Search Results.
-
- Every Inbound Twitter Channel Adapter is a Polling Consumer which means you have to provide a poller
- configuration. However, there is one important thing you must understand about Twitter since its inner-workings are slightly
- different than other polling consumers. Twitter defines a concept of Rate Limiting. You can read more about
- it here: Rate Limiting. In a nutshell, Rate Limiting
- is the way Twitter manages how often an application can poll for updates. You should consider this when setting your
- poller intervals, but we are also doing a few things to limit excessively aggressive polling within our adapters.
-
-
+
+
+ Every Inbound Twitter Channel Adapter is a Polling Consumer which means you have to provide a poller
+ configuration.
+ Twitter defines a concept of Rate Limiting. You can read more
+ about it here: Rate Limiting. In a nutshell,
+ Rate Limiting is a mechanism that Twitter uses to manage how often an application can poll for updates. You should consider this when
+ setting your poller intervals so that the adapter polls in compliance with the Twitter policies.
+
+
+ With Spring Integration prior to version 3.0, a hard-coded limit within the adapters was used to ensure
+ the polling interval could not be less than 15 seconds. This is no longer the case and the poller configuration is
+ applied directly.
+
+
+
Another issue that we need to worry about is handling duplicate Tweets. The same adapter (e.g., Search or Timeline Update)
while polling on Twitter may receive the same values more than once. For example if you keep searching on Twitter with the same search
criteria you'll end up with the same set of tweets unless some other new tweet that matches your search criteria was posted
From 3470e3306977539baecbf0b5a1c8fc6abea6ec17 Mon Sep 17 00:00:00 2001
From: Artem Bilan
Date: Thu, 24 Oct 2013 14:42:57 +0300
Subject: [PATCH 09/10] INT-3177: Generic RedisTemplate for PublishingMH
JIRA: https://jira.springsource.org/browse/INT-3177
* Change `RedisPublishingMessageHandler.StringRedisTemplate` to `RedisTemplate, ?>`
* Don't provide `serializer` to the `template`
* Use `serializer` directly for values which are not `byte[]`
* Add `RedisAvailableTests#awaitContainerSubscribed` for tests to avoid race conditions
* Refactor some tests
* Add test for `byte[]` payload
INT-3177: Generic RedisInboundChannelAdapter
Port from Spring XD: allow for `RedisInboundChannelAdapter`
to receive from Redis any object, not only String
* Make `serializer` property 'resettable' to `null`
* Change `RedisInboundChannelAdapterParser` to allow to apply
empty value from `serializer` attribute
* Add parser test and test for `byte[]` messages
INT-3177: Polishing and documentation
JIRA: https://jira.springsource.org/browse/INT-3033
* Add `topic-expression` to ``
* Add tests and docs
INT-3177: Deprecate `RedisPublishMH.defaultTopic`
* Make `topic` and `topic-expression` attributes as mutually exclusive
* Polishing Redis Topic Parsers
* Polishing tests and docs
* Default `serializer` for Redis Topic Adapters is `StringRedisSerializer` for backward compatibility
Polishing according PR's discussion
INT-3177 Doc Polishing
---
.../RedisInboundChannelAdapterParser.java | 8 +-
.../RedisOutboundChannelAdapterParser.java | 17 ++++-
.../inbound/RedisInboundChannelAdapter.java | 12 +--
.../RedisPublishingMessageHandler.java | 70 ++++++++++++-----
.../config/spring-integration-redis-3.0.xsd | 21 +++++-
.../SubscribableRedisChannelTests.java | 12 +--
...boundChannelAdapterParserTests-context.xml | 3 +
...RedisInboundChannelAdapterParserTests.java | 6 ++
...boundChannelAdapterParserTests-context.xml | 8 +-
...edisOutboundChannelAdapterParserTests.java | 14 +++-
.../RedisInboundChannelAdapterTests.java | 75 ++++++++++---------
.../RedisPublishingMessageHandlerTests.java | 17 ++++-
.../redis/rules/RedisAvailableTests.java | 17 ++++-
src/reference/docbook/redis.xml | 13 +++-
src/reference/docbook/whats-new.xml | 18 +++++
15 files changed, 223 insertions(+), 88 deletions(-)
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java
index 6ade9c2c30..13aab72337 100644
--- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java
@@ -23,20 +23,21 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter;
import org.springframework.util.StringUtils;
/**
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Gary Russell
+ * @author Artem Bilan
* @since 2.1
*/
public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterParser {
@Override
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
- BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
- "org.springframework.integration.redis.inbound.RedisInboundChannelAdapter");
+ BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisInboundChannelAdapter.class);
String connectionFactory = element.getAttribute("connection-factory");
if (!StringUtils.hasText(connectionFactory)) {
connectionFactory = "redisConnectionFactory";
@@ -46,7 +47,8 @@ public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterPars
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topics");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
- IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer");
+ IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true);
+
return builder.getBeanDefinition();
}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java
index b97e31572d..eb507557f1 100644
--- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -18,32 +18,41 @@ package org.springframework.integration.redis.config;
import org.w3c.dom.Element;
+import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.redis.outbound.RedisPublishingMessageHandler;
import org.springframework.util.StringUtils;
/**
+ * Parser for the {@code } component.
+ *
* @author Oleg Zhurakousky
* @author Mark Fisher
+ * @author Artem Bilan
* @since 2.1
*/
public class RedisOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
- BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
- "org.springframework.integration.redis.outbound.RedisPublishingMessageHandler");
+ BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisPublishingMessageHandler.class);
String connectionFactory = element.getAttribute("connection-factory");
if (!StringUtils.hasText(connectionFactory)) {
connectionFactory = "redisConnectionFactory";
}
builder.addConstructorArgReference(connectionFactory);
- IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topic", "defaultTopic");
+
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer");
+
+ BeanDefinition topicExpression = IntegrationNamespaceUtils
+ .createExpressionDefinitionFromValueOrExpression("topic", "topic-expression", parserContext, element, true);
+ builder.addPropertyValue("topicExpression", topicExpression);
+
return builder.getBeanDefinition();
}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java
index 0fc7eb78be..d1c7d17dc2 100644
--- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2007-2012 the original author or authors
+ * Copyright 2007-2013 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.
@@ -34,6 +34,7 @@ import org.springframework.util.Assert;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
+ * @author Gary Russell
* @since 2.1
*/
public class RedisInboundChannelAdapter extends MessageProducerSupport {
@@ -52,7 +53,6 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
}
public void setSerializer(RedisSerializer> serializer) {
- Assert.notNull(serializer, "'serializer' must not be null");
this.serializer = serializer;
}
@@ -99,16 +99,16 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
this.container.stop();
}
- private Message> convertMessage(String s) {
- return this.messageConverter.toMessage(s);
+ private Message> convertMessage(Object object) {
+ return this.messageConverter.toMessage(object);
}
private class MessageListenerDelegate {
@SuppressWarnings("unused")
- public void handleMessage(String s) {
- sendMessage(convertMessage(s));
+ public void handleMessage(Object object) {
+ sendMessage(convertMessage(object));
}
}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java
index 731e54265b..2460607608 100644
--- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2007-2011 the original author or authors
+ * Copyright 2007-2013 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.
@@ -17,10 +17,14 @@
package org.springframework.integration.redis.outbound;
import org.springframework.data.redis.connection.RedisConnectionFactory;
-import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.Expression;
+import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.Message;
+import org.springframework.integration.expression.IntegrationEvaluationContextAware;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.support.converter.MessageConverter;
import org.springframework.integration.support.converter.SimpleMessageConverter;
@@ -28,21 +32,32 @@ import org.springframework.util.Assert;
/**
* @author Mark Fisher
+ * @author Artem Bilan
* @since 2.1
*/
-public class RedisPublishingMessageHandler extends AbstractMessageHandler {
+public class RedisPublishingMessageHandler extends AbstractMessageHandler implements IntegrationEvaluationContextAware {
- private final StringRedisTemplate template;
+ private final RedisTemplate, ?> template;
+
+ private volatile EvaluationContext evaluationContext;
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
- private volatile String defaultTopic;
-
private volatile RedisSerializer> serializer = new StringRedisSerializer();
+ private volatile Expression topicExpression;
+
public RedisPublishingMessageHandler(RedisConnectionFactory connectionFactory) {
Assert.notNull(connectionFactory, "connectionFactory must not be null");
- this.template = new StringRedisTemplate(connectionFactory);
+ this.template = new RedisTemplate();
+ this.template.setConnectionFactory(connectionFactory);
+ this.template.setEnableDefaultSerializer(false);
+ this.template.afterPropertiesSet();
+ }
+
+ @Override
+ public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
+ this.evaluationContext = evaluationContext;
}
public void setSerializer(RedisSerializer> serializer) {
@@ -55,27 +70,42 @@ public class RedisPublishingMessageHandler extends AbstractMessageHandler {
this.messageConverter = messageConverter;
}
+ /**
+ * @deprecated in favor of {@link #setTopicExpression(Expression)} or {@link #setTopic(String)}
+ */
+ @Deprecated
public void setDefaultTopic(String defaultTopic) {
- this.defaultTopic = defaultTopic;
+ Assert.hasText(defaultTopic, "'defaultTopic' must not be an empty string.");
+ this.setTopicExpression(new LiteralExpression(defaultTopic));
}
- private String determineTopic(Message> message) {
- // TODO: add support for determining topic by evaluating SpEL against the Message
- Assert.hasText(this.defaultTopic, "Failed to determine Redis topic " +
- "from Message, and no defaultTopic has been provided.");
- return this.defaultTopic;
+ public void setTopic(String topic) {
+ Assert.hasText(topic, "'topic' must not be an empty string.");
+ this.setTopicExpression(new LiteralExpression(topic));
}
- @Override
- protected void handleMessageInternal(Message> message) throws Exception {
- String topic = this.determineTopic(message);
- Object value = this.messageConverter.fromMessage(message);
- this.template.convertAndSend(topic, value.toString());
+ public void setTopicExpression(Expression topicExpression) {
+ Assert.notNull(topicExpression, "'topicExpression' must not be null.");
+ this.topicExpression = topicExpression;
}
@Override
protected void onInit() throws Exception {
- this.template.setValueSerializer(this.serializer);
- this.template.afterPropertiesSet();
+ Assert.notNull(topicExpression, "'topicExpression' must not be null.");
}
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected void handleMessageInternal(Message> message) throws Exception {
+ String topic = this.topicExpression.getValue(this.evaluationContext, message, String.class);
+ Object value = this.messageConverter.fromMessage(message);
+
+ if (value instanceof byte[]) {
+ this.template.convertAndSend(topic, value);
+ }
+ else {
+ this.template.convertAndSend(topic, ((RedisSerializer) this.serializer).serialize(value));
+ }
+ }
+
}
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 274879d1d4..450916967e 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
@@ -174,7 +174,9 @@
- Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer
+ Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer.
+ This attribute can be an empty string, which results in 'null' being used by the underlying adapter,
+ meaning no serializer is used and the raw byte[] will be the message payload.
@@ -197,7 +199,22 @@
-
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java
index e9bbe29fdc..ed0e51b9ed 100644
--- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java
@@ -17,6 +17,8 @@
package org.springframework.integration.redis.config;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import org.junit.Test;
@@ -67,6 +69,10 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests {
Object converterBean = context.getBean("testConverter");
assertEquals(converterBean, accessor.getPropertyValue("messageConverter"));
assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer"));
+
+ Object bean = context.getBean("withoutSerializer.adapter");
+ assertNotNull(bean);
+ assertNull(TestUtils.getPropertyValue(bean, "serializer"));
}
@Test
diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml
index b5126265a3..5041900e54 100644
--- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml
@@ -11,7 +11,7 @@
@@ -21,6 +21,12 @@
+
+
+
+
+
+
diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java
index 52010d1a65..05e8deb1c6 100644
--- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java
@@ -25,6 +25,7 @@ import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
+import org.springframework.expression.Expression;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -33,6 +34,7 @@ import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.redis.outbound.RedisPublishingMessageHandler;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
+import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -53,13 +55,16 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests{
@Test
@RedisAvailable
+ @SuppressWarnings("unchecked")
public void validateConfiguration() {
EventDrivenConsumer adapter = context.getBean("outboundAdapter", EventDrivenConsumer.class);
RedisPublishingMessageHandler handler = (RedisPublishingMessageHandler)
new DirectFieldAccessor(adapter).getPropertyValue("handler");
assertEquals("outboundAdapter", adapter.getComponentName());
DirectFieldAccessor accessor = new DirectFieldAccessor(handler);
- assertEquals("foo", accessor.getPropertyValue("defaultTopic"));
+ Object topicExpression = accessor.getPropertyValue("topicExpression");
+ assertNotNull(topicExpression);
+ assertEquals("headers['topic'] ?: 'foo'", ((Expression) topicExpression).getExpressionString());
Object converterBean = context.getBean("testConverter");
assertEquals(converterBean, accessor.getPropertyValue("messageConverter"));
assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer"));
@@ -74,6 +79,13 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests{
Message> message = receiveChannel.receive(5000);
assertNotNull(message);
assertEquals("Hello Redis", message.getPayload());
+
+ sendChannel = context.getBean("sendChannel", MessageChannel.class);
+ sendChannel.send(MessageBuilder.withPayload("Hello Redis").setHeader("topic", "bar").build());
+ receiveChannel = context.getBean("barChannel", QueueChannel.class);
+ message = receiveChannel.receive(5000);
+ assertNotNull(message);
+ assertEquals("Hello Redis", message.getPayload());
}
@Test //INT-2275
diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java
index a94c429741..279ae1e9aa 100644
--- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java
@@ -18,15 +18,14 @@ package org.springframework.integration.redis.inbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import org.hamcrest.Matchers;
import org.junit.Test;
-import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.integration.Message;
@@ -37,12 +36,11 @@ import org.springframework.integration.test.util.TestUtils;
/**
* @author Mark Fisher
+ * @author Artem Bilan
* @since 2.1
*/
public class RedisInboundChannelAdapterTests extends RedisAvailableTests{
- private final Log logger = LogFactory.getLog(this.getClass());
-
@Test
@RedisAvailable
public void testRedisInboundChannelAdapter() throws Exception {
@@ -59,19 +57,18 @@ public class RedisInboundChannelAdapterTests extends RedisAvailableTests{
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
RedisInboundChannelAdapter adapter = new RedisInboundChannelAdapter(connectionFactory);
- adapter.setTopics("testRedisInboundChannelAdapterChannel");
+ adapter.setTopics(redisChannelName);
adapter.setOutputChannel(channel);
adapter.afterPropertiesSet();
adapter.start();
- RedisMessageListenerContainer container = waitUntilSubscribed(adapter);
+ this.awaitContainerSubscribed(TestUtils.getPropertyValue(adapter, "container", RedisMessageListenerContainer.class));
StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory);
redisTemplate.afterPropertiesSet();
for (int i = 0; i < numToTest; i++) {
String message = "test-" + i + " iteration " + iteration;
redisTemplate.convertAndSend(redisChannelName, message);
- logger.debug("Sent " + message);
}
int counter = 0;
for (int i = 0; i < numToTest; i++) {
@@ -85,34 +82,42 @@ public class RedisInboundChannelAdapterTests extends RedisAvailableTests{
}
assertEquals(numToTest, counter);
adapter.stop();
- container.stop();
- }
- /**
- * Wait until the container has subscribed to the queue and return a
- * reference to it, so we can stop it at the end of the test.
- */
- protected RedisMessageListenerContainer waitUntilSubscribed(
- RedisInboundChannelAdapter adapter) throws Exception {
- RedisMessageListenerContainer container = (RedisMessageListenerContainer) TestUtils
- .getPropertyValue(adapter, "container");
- Object subscriptionTask = TestUtils.getPropertyValue(container, "subscriptionTask");
- RedisConnection connection = (RedisConnection) TestUtils
- .getPropertyValue(subscriptionTask, "connection");
- int n = 0;
- while (true) {
- if (n++ > 50) {
- fail("RMLC Failed to Subscribe");
- }
- if (connection.isSubscribed()) {
- logger.debug("Subscribed OK");
- break;
- }
- logger.debug("Waiting...");
- Thread.sleep(100);
+ redisChannelName = "testRedisBytesInboundChannelAdapterChannel";
+
+ adapter.setTopics(redisChannelName);
+ adapter.setSerializer(null);
+ adapter.afterPropertiesSet();
+ adapter.start();
+
+ this.awaitContainerSubscribed(TestUtils.getPropertyValue(adapter, "container", RedisMessageListenerContainer.class));
+
+ RedisTemplate, ?> template = new RedisTemplate();
+ template.setConnectionFactory(connectionFactory);
+ template.setEnableDefaultSerializer(false);
+ template.afterPropertiesSet();
+
+ for (int i = 0; i < numToTest; i++) {
+ String message = "test-" + i + " iteration " + iteration;
+ template.convertAndSend(redisChannelName, message.getBytes());
}
- Thread.sleep(100); // Wait a little longer due to race condition in connection.isSubscribed()
- return container;
+
+ counter = 0;
+ for (int i = 0; i < numToTest; i++) {
+ Message> message = channel.receive(5000);
+ if (message == null){
+ throw new RuntimeException("Failed to receive message # " + i + " iteration " + iteration);
+ }
+ assertNotNull(message);
+ Object payload = message.getPayload();
+ assertThat(payload, Matchers.instanceOf(byte[].class));
+
+ assertTrue(new String((byte[]) payload).startsWith("test-"));
+ counter++;
+ }
+
+ assertEquals(numToTest, counter);
+ adapter.stop();
}
}
diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java
index 71f4b97f77..dc3648cc06 100644
--- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java
@@ -30,12 +30,14 @@ import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.support.MessageBuilder;
/**
* @author Mark Fisher
+ * @author Artem Bilan
* @since 2.1
*/
public class RedisPublishingMessageHandlerTests extends RedisAvailableTests {
@@ -45,7 +47,7 @@ public class RedisPublishingMessageHandlerTests extends RedisAvailableTests {
public void testRedisPublishingMessageHandler() throws Exception {
int numToTest = 10;
String topic = "si.test.channel";
- final CountDownLatch latch = new CountDownLatch(numToTest);
+ final CountDownLatch latch = new CountDownLatch(numToTest * 2);
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
@@ -59,14 +61,20 @@ public class RedisPublishingMessageHandlerTests extends RedisAvailableTests {
container.afterPropertiesSet();
container.addMessageListener(listener, Collections.singletonList(new ChannelTopic(topic)));
container.start();
- Thread.sleep(1000);
+
+ this.awaitContainerSubscribed(container);
final RedisPublishingMessageHandler handler = new RedisPublishingMessageHandler(connectionFactory);
- handler.setDefaultTopic(topic);
+ handler.setTopicExpression(new LiteralExpression(topic));
+
for (int i = 0; i < numToTest; i++) {
handler.handleMessage(MessageBuilder.withPayload("test-" + i).build());
}
- assertTrue(latch.await(3, TimeUnit.SECONDS));
+
+ for (int i = 0; i < numToTest; i++) {
+ handler.handleMessage(MessageBuilder.withPayload(("test-" + i).getBytes()).build());
+ }
+ assertTrue(latch.await(10, TimeUnit.SECONDS));
container.stop();
}
@@ -83,6 +91,7 @@ public class RedisPublishingMessageHandlerTests extends RedisAvailableTests {
public void handleMessage(String s) {
this.latch.countDown();
}
+
}
}
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 064f5cf3d7..4adb288a48 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
@@ -15,6 +15,8 @@
*/
package org.springframework.integration.redis.rules;
+import static org.junit.Assert.assertTrue;
+
import java.util.UUID;
import org.junit.Rule;
@@ -28,6 +30,8 @@ import org.springframework.data.redis.core.BoundZSetOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.integration.test.util.TestUtils;
/**
* @author Oleg Zhurakousky
@@ -40,7 +44,7 @@ public class RedisAvailableTests {
@Rule
public RedisAvailableRule redisAvailableRule = new RedisAvailableRule();
- public RedisConnectionFactory getConnectionFactoryForTest(){
+ protected RedisConnectionFactory getConnectionFactoryForTest(){
LettuceConnectionFactory connectionFactory = RedisAvailableRule.connectionFactoryResource.get();
RedisTemplate rt = new RedisTemplate();
rt.setConnectionFactory(connectionFactory);
@@ -56,6 +60,17 @@ public class RedisAvailableTests {
return connectionFactory;
}
+ protected void awaitContainerSubscribed(RedisMessageListenerContainer container) throws Exception {
+ RedisConnection connection = TestUtils.getPropertyValue(container, "subscriptionTask.connection",
+ RedisConnection.class);
+
+ int n = 0;
+ while (n++ < 100 && !connection.isSubscribed()) {
+ Thread.sleep(100);
+ }
+ assertTrue("RedisMessageListenerContainer Failed to Subscribe", n < 100);
+ }
+
protected void prepareList(RedisConnectionFactory connectionFactory){
StringRedisTemplate redisTemplate = new StringRedisTemplate();
diff --git a/src/reference/docbook/redis.xml b/src/reference/docbook/redis.xml
index 367ab41d38..1bd7c2023a 100644
--- a/src/reference/docbook/redis.xml
+++ b/src/reference/docbook/redis.xml
@@ -152,6 +152,12 @@ rt.setConnectionFactory(redisConnectionFactory);]]>
Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the
topics attribute.
+
+ Inbound adapters can use a RedisSerializer to deserialize the body of Redis Messages.
+ The serializer attribute of the <int-redis:inbound-channel-adapter> can be set to an
+ empty string, which results in a null value for the RedisSerializer property.
+ In this case the raw byte[] bodies of Redis Messages are provided as the message payloads.
+
@@ -178,11 +184,16 @@ rt.setConnectionFactory(redisConnectionFactory);]]>
a RedisConnectionFactory which was defined with 'redisConnectionFactory' as its bean name.
This example also includes the optional, custom MessageConverter (the 'testConverter' bean).
+
+ Since Spring Integration 3.0, the <int-redis:outbound-channel-adapter>,
+ as an alternative to the topic attribute, has the topic-expression attribute to determine
+ the Redis topic against the Message at runtime. These attributes are mutually exclusive.
+ Redis Queue Inbound Channel Adapter
- Since Spring Integration 3.0, a Queue Inbound Channel Adapter
+ Since Spring Integration 3.0, a Queue Inbound Channel Adapter
is available to 'right pop' messages from a Redis List.
The adapter is message-driven using an internal listener thread and does not use a poller.
.
+
+ Redis Adapters Changers
+
+
+
+ The Redis Inbound Channel Adapter can now use a null value for serializer
+ property, with the raw data being the message payload.
+
+
+ The Redis Outbound Channel Adapter now has the topic-expression property to determine
+ the Redis topic against the Message at runtime.
+
+
+
+
+ For more information, see .
+
+
From 4cf0e2e673a77828851b0d1834d9a692b5d84e4b Mon Sep 17 00:00:00 2001
From: Artem Bilan
Date: Tue, 5 Nov 2013 12:21:47 +0200
Subject: [PATCH 10/10] INT-3196: RedisQueueMDE: Add RecoveryInterval
Catch `RedisSystemException` in the `ListenerTask` and sleep current Thread
with `recoveryInterval` if the Endpoint is `active` before the next `restart()`
JIRA: https://jira.springsource.org/browse/INT-3196
RedisConnFailure & RedisSys Exceptions recovery
Introduce `RedisIntegrationEvent`s
* Catch all `Exception`s on `this.boundListOperations.rightPop`
* Log them and send within `RedisExceptionEvent`
* Mark some fields with `volatile`
---
...RedisQueueInboundChannelAdapterParser.java | 35 ++--------
.../redis/event/RedisExceptionEvent.java | 30 ++++++++
.../redis/event/RedisIntegrationEvent.java | 37 ++++++++++
.../integration/redis/event/package-info.java | 4 ++
.../RedisQueueMessageDrivenEndpoint.java | 62 +++++++++++++---
.../RedisQueueOutboundChannelAdapter.java | 2 +-
.../RedisQueueMessageDrivenEndpointTests.java | 70 ++++++++++++++++++-
7 files changed, 197 insertions(+), 43 deletions(-)
create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java
create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java
create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java
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 ab0cf50cc3..18996384da 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
@@ -18,11 +18,10 @@ package org.springframework.integration.redis.config;
import org.w3c.dom.Element;
-import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
import org.springframework.util.StringUtils;
@@ -33,29 +32,11 @@ import org.springframework.util.StringUtils;
* @author Artem Bilan
* @since 3.0
*/
-public class RedisQueueInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser {
+public class RedisQueueInboundChannelAdapterParser extends AbstractChannelAdapterParser {
@Override
- protected Class> getBeanClass(Element element) {
- return RedisQueueMessageDrivenEndpoint.class;
- }
-
- @Override
- protected final String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
- throws BeanDefinitionStoreException {
- String id = element.getAttribute("id");
- if (!element.hasAttribute("channel")) {
- // the created channel will get the 'id', so the adapter's bean name includes a suffix
- id = id + ".adapter";
- }
- else if (!StringUtils.hasText(id)) {
- id = parserContext.getReaderContext().generateBeanName(definition);
- }
- return id;
- }
-
- @Override
- protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+ protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
+ BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisQueueMessageDrivenEndpoint.class);
builder.addConstructorArgValue(element.getAttribute("queue"));
String connectionFactory = element.getAttribute("connection-factory");
@@ -69,15 +50,9 @@ public class RedisQueueInboundChannelAdapterParser extends AbstractSingleBeanDef
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
- IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
- IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
-
- String channelName = element.getAttribute("channel");
- if (!StringUtils.hasText(channelName)) {
- channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
- }
builder.addPropertyReference("outputChannel", channelName);
+ return builder.getBeanDefinition();
}
}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java
new file mode 100644
index 0000000000..62115bfa6a
--- /dev/null
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2013 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.redis.event;
+
+/**
+ * @author Artem Bilan
+ * @since 3.0
+ */
+@SuppressWarnings("serial")
+public class RedisExceptionEvent extends RedisIntegrationEvent {
+
+ public RedisExceptionEvent(Object source, Throwable cause) {
+ super(source, cause);
+ }
+
+}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java
new file mode 100644
index 0000000000..e2a91068d1
--- /dev/null
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2013 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.redis.event;
+
+import org.springframework.integration.event.IntegrationEvent;
+
+/**
+ * @author Artem Bilan
+ * @since 3.0
+ *
+ */
+@SuppressWarnings("serial")
+public abstract class RedisIntegrationEvent extends IntegrationEvent {
+
+ public RedisIntegrationEvent(Object source) {
+ super(source);
+ }
+
+ public RedisIntegrationEvent(Object source, Throwable cause) {
+ super(source, cause);
+ }
+
+}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java
new file mode 100644
index 0000000000..3c4d6dec86
--- /dev/null
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java
@@ -0,0 +1,4 @@
+/**
+ * Events generated by the redis module
+ */
+package org.springframework.integration.redis.event;
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 1c33335f27..36d7f85a2c 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
@@ -18,8 +18,9 @@ package org.springframework.integration.redis.inbound;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
-import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;
@@ -31,6 +32,7 @@ import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.redis.event.RedisExceptionEvent;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
@@ -46,13 +48,17 @@ import org.springframework.util.Assert;
* @since 3.0
*/
@ManagedResource
-public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
+public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport implements ApplicationEventPublisherAware {
public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
+ public static final long DEFAULT_RECOVERY_INTERVAL = 5000;
+
private final BoundListOperations boundListOperations;
- private MessageChannel errorChannel;
+ private volatile ApplicationEventPublisher applicationEventPublisher;
+
+ private volatile MessageChannel errorChannel;
private volatile Executor taskExecutor;
@@ -62,6 +68,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
+ private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
+
private volatile boolean active;
private volatile boolean listening;
@@ -81,6 +89,11 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
this.boundListOperations = template.boundListOps(queueName);
}
+ @Override
+ public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+ this.applicationEventPublisher = applicationEventPublisher;
+ }
+
public void setSerializer(RedisSerializer> serializer) {
this.serializer = serializer;
}
@@ -129,6 +142,10 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
this.errorChannel = errorChannel;
}
+ public void setRecoveryInterval(long recoveryInterval) {
+ this.recoveryInterval = recoveryInterval;
+ }
+
@Override
protected void onInit() {
super.onInit();
@@ -160,13 +177,12 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
try {
value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
}
- catch (RedisSystemException e) {
- if (this.active) {
- throw e;
- }
- else {
- logger.error(e);
- }
+ catch (Exception e) {
+ logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval + " milliseconds.", e);
+ this.listening = false;
+ this.sleepBeforeRecoveryAttempt();
+ this.publishException(e);
+ return;
}
if (value != null) {
@@ -200,6 +216,32 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
}
}
+ /**
+ * Sleep according to the specified recovery interval.
+ * Called between recovery attempts.
+ */
+ private void sleepBeforeRecoveryAttempt() {
+ if (this.recoveryInterval > 0) {
+ try {
+ Thread.sleep(this.recoveryInterval);
+ }
+ catch (InterruptedException e) {
+ logger.debug("Thread interrupted while sleeping the recovery interval");
+ }
+ }
+ }
+
+ private void publishException(Exception e) {
+ if (this.applicationEventPublisher != null) {
+ this.applicationEventPublisher.publishEvent(new RedisExceptionEvent(this, e));
+ }
+ else {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No application event publisher for exception: " + e.getMessage());
+ }
+ }
+ }
+
private void restart() {
this.taskExecutor.execute(new ListenerTask());
}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java
index 2ff459ec78..b25e058888 100644
--- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java
@@ -43,7 +43,7 @@ public class RedisQueueOutboundChannelAdapter extends AbstractMessageHandler imp
private final Expression queueNameExpression;
- private EvaluationContext evaluationContext;
+ private volatile EvaluationContext evaluationContext;
private volatile boolean extractPayload = true;
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 5a1c33c8a5..ec28d11af0 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
@@ -18,9 +18,13 @@ package org.springframework.integration.redis.inbound;
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 java.util.ArrayList;
+import java.util.Arrays;
import java.util.Date;
+import java.util.List;
import java.util.UUID;
import org.hamcrest.Matchers;
@@ -29,7 +33,13 @@ import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -40,7 +50,9 @@ import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.event.IntegrationEvent;
import org.springframework.integration.message.ErrorMessage;
+import org.springframework.integration.redis.event.RedisExceptionEvent;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.support.MessageBuilder;
@@ -68,7 +80,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@Autowired
private PollableChannel symmetricalOutputChannel;
-
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
@@ -158,7 +169,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
assertThat(((Exception) receive.getPayload()).getCause().getMessage(),
Matchers.containsString("java.lang.String cannot be cast to org.springframework.integration.Message"));
-
endpoint.stop();
}
@@ -194,4 +204,60 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
assertEquals(payload, receive.getPayload());
}
+ @Test
+ @RedisAvailable
+ @SuppressWarnings("unchecked")
+ public void testInt3196Recovery() throws Exception {
+ String queueName = "test.si.Int3196Recovery";
+ QueueChannel channel = new QueueChannel();
+
+ final List exceptionEvents = new ArrayList();
+
+ RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
+ endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
+ endpoint.setApplicationEventPublisher(new ApplicationEventPublisher() {
+
+ @Override
+ public void publishEvent(ApplicationEvent event) {
+ exceptionEvents.add(event);
+ }
+ });
+ endpoint.setOutputChannel(channel);
+ endpoint.setReceiveTimeout(100);
+ endpoint.setRecoveryInterval(200);
+ endpoint.afterPropertiesSet();
+ endpoint.start();
+
+ ((DisposableBean) this.connectionFactory).destroy();
+
+ Thread.sleep(300);
+
+ 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)));
+ }
+
+ ((InitializingBean) this.connectionFactory).afterPropertiesSet();
+
+ RedisTemplate redisTemplate = new RedisTemplate();
+ redisTemplate.setConnectionFactory(this.getConnectionFactoryForTest());
+ redisTemplate.setEnableDefaultSerializer(false);
+ redisTemplate.setKeySerializer(new StringRedisSerializer());
+ redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
+ redisTemplate.afterPropertiesSet();
+
+ String payload = "testing";
+
+ redisTemplate.boundListOps(queueName).leftPush(payload);
+
+ Message> receive = channel.receive(1000);
+ assertNotNull(receive);
+ assertEquals(payload, receive.getPayload());
+
+ endpoint.stop();
+ }
+
}