entry : this.headers.entrySet()) {
+ if (!(entry.getValue() instanceof Serializable)) {
+ keysToRemove.add(entry.getKey());
+ }
+ }
+ for (String key : keysToRemove) {
+ if (logger.isInfoEnabled()) {
+ logger.info("removing non-serializable header: " + key);
+ }
+ this.headers.remove(key);
+ }
+ out.defaultWriteObject();
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ in.defaultReadObject();
+ }
+
+ public static interface IdGenerator {
+ UUID generateId();
+ }
+
+ public static class JdkIdGenerator implements IdGenerator {
+
+ @Override
+ public UUID generateId() {
+ return UUID.randomUUID();
+ }
+
+ }
+
+ /**
+ * A variation of {@link UUID#randomUUID()} that uses {@link SecureRandom} only for
+ * the initial seed and {@link Random} thereafter, which provides better performance
+ * in exchange for less securely random id's.
+ */
+ public static class AlternativeJdkIdGenerator implements IdGenerator {
+
+ private final Random random;
+
+ public AlternativeJdkIdGenerator() {
+ byte[] seed = new SecureRandom().generateSeed(8);
+ this.random = new Random(new BigInteger(seed).longValue());
+ }
+
+ public UUID generateId() {
+
+ byte[] randomBytes = new byte[16];
+ this.random.nextBytes(randomBytes);
+
+ long mostSigBits = 0;
+ for (int i = 0; i < 8; i++) {
+ mostSigBits = (mostSigBits << 8) | (randomBytes[i] & 0xff);
+ }
+ long leastSigBits = 0;
+ for (int i = 8; i < 16; i++) {
+ leastSigBits = (leastSigBits << 8) | (randomBytes[i] & 0xff);
+ }
+
+ return new UUID(mostSigBits, leastSigBits);
+ }
+ }
+
+ public static class SimpleIncrementingIdGenerator implements IdGenerator {
+
+ private final AtomicLong topBits = new AtomicLong();
+
+ private final AtomicLong bottomBits = new AtomicLong();
+
+ @Override
+ public UUID generateId() {
+ long bottomBits = this.bottomBits.incrementAndGet();
+ if (bottomBits == 0) {
+ this.topBits.incrementAndGet();
+ }
+ return new UUID(this.topBits.get(), bottomBits);
+ }
+
+ }
+
+}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java
index 7a672c9297..6da6aec529 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java
@@ -19,6 +19,7 @@ package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.BeanMetadataElement;
+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;
@@ -31,18 +32,25 @@ import org.springframework.util.xml.DomUtils;
* @author Mark Fisher
* @author Gary Russell
* @author Oleg Zhurakousky
+ * @author Artem Bilan
*/
public abstract class AbstractPollingInboundChannelAdapterParser extends AbstractChannelAdapterParser {
@Override
+ @SuppressWarnings("unchecked")
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
BeanMetadataElement source = this.parseSource(element, parserContext);
if (source == null) {
parserContext.getReaderContext().error("failed to parse source", element);
}
+
+ String channelAdapterId = this.resolveId(element, (AbstractBeanDefinition) source, parserContext);
+ String sourceBeanName = channelAdapterId + ".source";
+ parserContext.getRegistry().registerBeanDefinition(sourceBeanName, (BeanDefinition) source);
+
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(SourcePollingChannelAdapterFactoryBean.class);
- adapterBuilder.addPropertyValue("source", source);
+ adapterBuilder.addPropertyReference("source", sourceBeanName);
adapterBuilder.addPropertyReference("outputChannel", channelName);
IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "send-timeout");
Element pollerElement = DomUtils.getChildElementByTagName(element, "poller");
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
index 150a3dc054..df6d2f0012 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
@@ -172,9 +172,19 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName, String propertyName) {
- String attributeValue = element.getAttribute(attributeName);
- if (StringUtils.hasText(attributeValue)) {
- builder.addPropertyReference(propertyName, attributeValue);
+ setReferenceIfAttributeDefined(builder, element, attributeName, propertyName, false);
+ }
+
+ public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
+ String attributeName, String propertyName, boolean emptyStringAllowed) {
+ if (element.hasAttribute(attributeName)) {
+ String attributeValue = element.getAttribute(attributeName);
+ if (StringUtils.hasText(attributeValue)) {
+ builder.addPropertyReference(propertyName, attributeValue);
+ }
+ else if (emptyStringAllowed) {
+ builder.addPropertyValue(propertyName, null);
+ }
}
}
@@ -198,8 +208,13 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName) {
+ setReferenceIfAttributeDefined(builder, element, attributeName, false);
+ }
+
+ public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
+ String attributeName, boolean emptyStringAllowed) {
setReferenceIfAttributeDefined(builder, element, attributeName,
- Conventions.attributeNameToPropertyName(attributeName));
+ Conventions.attributeNameToPropertyName(attributeName), emptyStringAllowed);
}
/**
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/java/org/springframework/integration/context/IntegrationContextUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
index 9e80a68860..381a8a896f 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
@@ -19,7 +19,7 @@ package org.springframework.integration.context;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.convert.ConversionService;
import org.springframework.expression.spel.support.StandardEvaluationContext;
-import org.springframework.integration.store.metadata.MetadataStore;
+import org.springframework.integration.metadata.MetadataStore;
import org.springframework.messaging.MessageChannel;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java
index 5f7d7bea7a..9a8a602912 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java
@@ -52,13 +52,15 @@ import org.springframework.util.Assert;
* the configuration by removing channels that can be created implicitly.
*
*
- *
- * <chain>
- * <filter ref="someFilter"/>
- * <bean class="SomeMessageHandlerImplementation"/>
- * <transformer ref="someTransformer"/>
- * <aggregator ... />
- * </chain>
+ *
+ * {@code
+ *
+ *
+ *
+ *
+ *
+ *
+ * }
*
*
* @author Mark Fisher
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java
index 7669b45bb5..3f6395b7ad 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java
@@ -17,6 +17,7 @@
package org.springframework.integration.json;
import org.springframework.beans.factory.BeanClassLoaderAware;
+import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.json.JacksonJsonObjectMapper;
import org.springframework.integration.support.json.JacksonJsonObjectMapperProvider;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java
index 82d8ddc70b..2e2c7430ea 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java
@@ -27,7 +27,7 @@ import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.springframework.integration.json.JsonHeaders;
+import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonHeaders.java b/spring-integration-core/src/main/java/org/springframework/integration/mapping/support/JsonHeaders.java
similarity index 95%
rename from spring-integration-core/src/main/java/org/springframework/integration/json/JsonHeaders.java
rename to spring-integration-core/src/main/java/org/springframework/integration/mapping/support/JsonHeaders.java
index 5381bdf472..fa2c88eef2 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonHeaders.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/mapping/support/JsonHeaders.java
@@ -14,7 +14,7 @@
* limitations under the License.
*/
-package org.springframework.integration.json;
+package org.springframework.integration.mapping.support;
import java.util.Arrays;
import java.util.Collection;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/mapping/support/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/mapping/support/package-info.java
new file mode 100644
index 0000000000..0fa87a6375
--- /dev/null
+++ b/spring-integration-core/src/main/java/org/springframework/integration/mapping/support/package-info.java
@@ -0,0 +1,4 @@
+/**
+ * Support classes for mapping.
+ */
+package org.springframework.integration.mapping.support;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/MetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStore.java
similarity index 69%
rename from spring-integration-core/src/main/java/org/springframework/integration/store/metadata/MetadataStore.java
rename to spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStore.java
index de4a8376aa..06d48c9cab 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/MetadataStore.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStore.java
@@ -14,7 +14,10 @@
* limitations under the License.
*/
-package org.springframework.integration.store.metadata;
+package org.springframework.integration.metadata;
+
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedResource;
/**
* Strategy interface for storing metadata from certain adapters
@@ -25,6 +28,7 @@ package org.springframework.integration.store.metadata;
* @author Mark Fisher
* @since 2.0
*/
+@ManagedResource
public interface MetadataStore {
/**
@@ -35,6 +39,15 @@ public interface MetadataStore {
/**
* Reads a value for the given key from this MetadataStore.
*/
+ @ManagedAttribute
String get(String key);
+ /**
+ * Remove a value for the given key from this MetadataStore.
+ * return the previous value associated with key, or
+ * null if there was no mapping for key.
+ */
+ @ManagedAttribute
+ String remove(String key);
+
}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java
similarity index 96%
rename from spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java
rename to spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java
index 3393468cfd..26c35231ec 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java
@@ -14,7 +14,7 @@
* limitations under the License.
*/
-package org.springframework.integration.store.metadata;
+package org.springframework.integration.metadata;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
@@ -86,6 +86,12 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial
return this.metadata.getProperty(key);
}
+ @Override
+ @SuppressWarnings("uchecked")
+ public String remove(String key) {
+ return (String) this.metadata.remove(key);
+ }
+
public void destroy() throws Exception {
this.saveMetadata();
}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/SimpleMetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/SimpleMetadataStore.java
similarity index 89%
rename from spring-integration-core/src/main/java/org/springframework/integration/store/metadata/SimpleMetadataStore.java
rename to spring-integration-core/src/main/java/org/springframework/integration/metadata/SimpleMetadataStore.java
index 5a04477715..0bf5b50fdd 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/SimpleMetadataStore.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/SimpleMetadataStore.java
@@ -11,7 +11,7 @@
* specific language governing permissions and limitations under the License.
*/
-package org.springframework.integration.store.metadata;
+package org.springframework.integration.metadata;
import java.util.HashMap;
import java.util.Map;
@@ -37,4 +37,9 @@ public class SimpleMetadataStore implements MetadataStore {
return this.metadata.get(key);
}
+ @Override
+ public String remove(String key) {
+ return metadata.remove(key);
+ }
+
}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/package-info.java
similarity index 50%
rename from spring-integration-core/src/main/java/org/springframework/integration/store/metadata/package-info.java
rename to spring-integration-core/src/main/java/org/springframework/integration/metadata/package-info.java
index 0cfc461d34..d27403d529 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/package-info.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/package-info.java
@@ -1,4 +1,4 @@
/**
* Provides classes supporting metadata stores.
*/
-package org.springframework.integration.store.metadata;
+package org.springframework.integration.metadata;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java
index 06b523acbf..8d77b8016b 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java
@@ -27,11 +27,13 @@ import org.springframework.integration.core.MessageSelector;
import org.springframework.util.Assert;
/**
- *
- * <recipient-list-router id="simpleRouter" input-channel="routingChannelA">
- * <recipient channel="channel1"/>
- * <recipient channel="channel2"/>
- * </recipient-list-router>
+ *
+ * {@code
+ *
+ *
+ *
+ *
+ * }
*
*
* A Message Router that sends Messages to a list of recipient channels. The
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java
index 0d802e5bc4..30a6b28a89 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java
@@ -25,7 +25,7 @@ import java.net.URL;
import java.util.Collection;
import java.util.Map;
-import org.springframework.integration.json.JsonHeaders;
+import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.util.Assert;
import com.fasterxml.jackson.core.JsonParser;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonObjectMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonObjectMapper.java
index 5297a2bb33..92342dd621 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonObjectMapper.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonObjectMapper.java
@@ -29,7 +29,7 @@ import org.codehaus.jackson.JsonParser;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.type.JavaType;
-import org.springframework.integration.json.JsonHeaders;
+import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.util.Assert;
/**
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 00dc093ed1..9343147740 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 db49eb9067..6ffcf58cac 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:
*
- *
+ *
* {@code
*
diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java
index 9c7157b896..a9decccdd7 100644
--- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java
+++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.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.
@@ -19,18 +19,15 @@ package org.springframework.integration.jms.config;
import org.w3c.dom.Element;
import org.springframework.beans.BeanMetadataElement;
-import org.springframework.beans.factory.config.BeanDefinition;
-import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
/**
- * Parser for the <inbound-channel-adapter/> element of the 'jms' namespace.
- *
+ * Parser for the <inbound-channel-adapter/> element of the 'jms' namespace.
+ *
* @author Mark Fisher
*/
public class JmsInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@@ -87,9 +84,7 @@ public class JmsInboundChannelAdapterParser extends AbstractPollingInboundChanne
builder.addPropertyReference(JmsAdapterParserUtils.HEADER_MAPPER_PROPERTY, headerMapper);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "selector", "messageSelector");
- BeanDefinition beanDefinition = builder.getBeanDefinition();
- String beanName = BeanDefinitionReaderUtils.generateBeanName(beanDefinition, parserContext.getRegistry());
- return new BeanComponentDefinition(beanDefinition, beanName);
+ return builder.getBeanDefinition();
}
}
diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java
index 0e9a66f7de..3276096978 100644
--- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java
+++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java
@@ -95,7 +95,8 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
String listenerBeanName = this.parseMessageListener(element, parserContext);
builder.addConstructorArgReference(containerBeanName);
builder.addConstructorArgReference(listenerBeanName);
- IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
}
private String parseMessageListenerContainer(Element element, ParserContext parserContext) {
diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml
index d19f5e6bf4..0ab4f1ab17 100644
--- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml
+++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml
@@ -13,18 +13,18 @@
http://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd">
-
+
-
-
+
+
-
+
-
+
diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java
index 25faf51a4d..535f8b12ec 100644
--- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java
+++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java
@@ -1,11 +1,11 @@
/*
* Copyright 2002-2010 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.
@@ -36,7 +36,7 @@ public class PollingAdapterMBeanTests {
@Autowired
private MBeanServer server;
-
+
@Test
public void testMessageSourceMBeanExists() throws Exception {
// System.err.println(server.queryNames(new ObjectName("*:type=MessageSource,*"), null));
diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java
index 216f642ec6..a236ce1b26 100644
--- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java
+++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java
@@ -15,13 +15,11 @@
*/
package org.springframework.integration.mongodb.config;
-import org.springframework.beans.factory.config.BeanDefinition;
import org.w3c.dom.Element;
import org.springframework.beans.BeanMetadataElement;
-import org.springframework.beans.factory.config.RuntimeBeanReference;
+import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
@@ -52,8 +50,7 @@ public class MongoDbInboundChannelAdapterParser extends AbstractPollingInboundCh
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "entity-class");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-single-result");
- String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
- builder.getBeanDefinition(), parserContext.getRegistry());
- return new RuntimeBeanReference(beanName);
+ return builder.getBeanDefinition();
}
+
}
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/RedisNamespaceHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java
index 834863fee4..c13c32bc06 100644
--- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.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.
@@ -21,6 +21,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
* Namespace handler for Spring Integration's 'redis' namespace.
*
* @author Oleg Zhurakousky
+ * @author Artem Bilan
* @since 2.1
*/
public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@@ -31,5 +32,7 @@ public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler {
registerBeanDefinitionParser("store-inbound-channel-adapter", new RedisStoreInboundChannelAdapterParser());
registerBeanDefinitionParser("store-outbound-channel-adapter", new RedisStoreOutboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-channel-adapter", new RedisOutboundChannelAdapterParser());
+ registerBeanDefinitionParser("queue-inbound-channel-adapter", new RedisQueueInboundChannelAdapterParser());
+ registerBeanDefinitionParser("queue-outbound-channel-adapter", new RedisQueueOutboundChannelAdapterParser());
}
}
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/config/RedisQueueInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java
new file mode 100644
index 0000000000..18996384da
--- /dev/null
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java
@@ -0,0 +1,58 @@
+/*
+ * 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.config;
+
+import org.w3c.dom.Element;
+
+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.AbstractChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <queue-inbound-channel-adapter> element of the 'redis' namespace.
+ *
+ * @author Artem Bilan
+ * @since 3.0
+ */
+public class RedisQueueInboundChannelAdapterParser extends AbstractChannelAdapterParser {
+
+ @Override
+ 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");
+ if (!StringUtils.hasText(connectionFactory)) {
+ connectionFactory = "redisConnectionFactory";
+ }
+ builder.addConstructorArgReference(connectionFactory);
+
+ IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true);
+ IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
+ IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message");
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
+ builder.addPropertyReference("outputChannel", channelName);
+
+ return builder.getBeanDefinition();
+ }
+
+}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParser.java
new file mode 100644
index 0000000000..9e0653e814
--- /dev/null
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParser.java
@@ -0,0 +1,57 @@
+/*
+ * 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.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.RedisQueueOutboundChannelAdapter;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <int-redis:queue-outbound-channel-adapter> element.
+ *
+ * @author Artem Bilan
+ * @since 3.0
+ */
+public class RedisQueueOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
+
+ @Override
+ protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
+ BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisQueueOutboundChannelAdapter.class);
+ BeanDefinition queueExpression = IntegrationNamespaceUtils
+ .createExpressionDefinitionFromValueOrExpression("queue", "queue-expression", parserContext, element, true);
+ builder.addConstructorArgValue(queueExpression);
+
+ String connectionFactory = element.getAttribute("connection-factory");
+ if (!StringUtils.hasText(connectionFactory)) {
+ connectionFactory = "redisConnectionFactory";
+ }
+ builder.addConstructorArgReference(connectionFactory);
+
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
+ IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer");
+
+ return builder.getBeanDefinition();
+ }
+
+}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParser.java
index 1920f4ad97..0e73cd3cd7 100644
--- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParser.java
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParser.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.
@@ -20,9 +20,7 @@ import org.w3c.dom.Element;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.config.BeanDefinition;
-import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
@@ -62,9 +60,8 @@ public class RedisStoreInboundChannelAdapterParser extends AbstractPollingInboun
parserContext, element, atLeastOneRequired);
builder.addConstructorArgValue(expressionDef);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type");
- String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
- builder.getBeanDefinition(), parserContext.getRegistry());
- return new RuntimeBeanReference(beanName);
+
+ 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/RedisInboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java
index 920ee7fd9b..737a8c6ca4 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,9 +34,9 @@ import org.springframework.util.Assert;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
+ * @author Gary Russell
* @since 2.1
*/
-@SuppressWarnings("rawtypes")
public class RedisInboundChannelAdapter extends MessageProducerSupport {
private final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
@@ -53,7 +53,6 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
}
public void setSerializer(RedisSerializer> serializer) {
- Assert.notNull(serializer, "'serializer' must not be null");
this.serializer = serializer;
}
@@ -100,17 +99,16 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
this.container.stop();
}
- @SuppressWarnings("unchecked")
- private Message> convertMessage(String s) {
- return this.messageConverter.toMessage(s, null);
+ private Message> convertMessage(Object object) {
+ return this.messageConverter.toMessage(object, null);
}
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/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java
index c63bab5f38..e01afbc08f 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
@@ -15,10 +15,12 @@
*/
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.core.task.TaskExecutor;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;
@@ -27,7 +29,9 @@ import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
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;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
@@ -44,15 +48,19 @@ 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 TaskExecutor taskExecutor;
+ private volatile MessageChannel errorChannel;
+
+ private volatile Executor taskExecutor;
private volatile RedisSerializer> serializer = new JdkSerializationRedisSerializer();
@@ -60,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;
@@ -79,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;
}
@@ -117,7 +132,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
this.receiveTimeout = receiveTimeout;
}
- public void setTaskExecutor(TaskExecutor taskExecutor) {
+ public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
@@ -127,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();
@@ -138,7 +157,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-") + this.getComponentType());
}
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
- MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
+ MessagePublishingErrorHandler errorHandler =
+ new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(this.getBeanFactory()));
errorHandler.setDefaultErrorChannel(this.errorChannel);
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler);
}
@@ -146,14 +166,24 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
@Override
public String getComponentType() {
- return "int-redis:message-driven-channel-adapter";
+ return "redis:queue-inbound-channel-adapter";
}
@SuppressWarnings("unchecked")
private void popMessageAndSend() {
Message