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-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-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 6877708810..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()); } -} \ No newline at end of file +} 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..ab0cf50cc3 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java @@ -0,0 +1,83 @@ +/* + * 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.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.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 AbstractSingleBeanDefinitionParser { + + @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) { + 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"); + 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); + + } + +} 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/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java index 587d2277e3..1c33335f27 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,11 @@ */ package org.springframework.integration.redis.inbound; +import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import org.springframework.core.task.SimpleAsyncTaskExecutor; -import org.springframework.core.task.TaskExecutor; +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.MessagingException; import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.endpoint.MessageProducerSupport; 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; @@ -52,7 +54,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { private MessageChannel errorChannel; - private volatile TaskExecutor taskExecutor; + private volatile Executor taskExecutor; private volatile RedisSerializer serializer = new JdkSerializationRedisSerializer(); @@ -117,7 +119,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { this.receiveTimeout = receiveTimeout; } - public void setTaskExecutor(TaskExecutor taskExecutor) { + public void setTaskExecutor(Executor taskExecutor) { this.taskExecutor = taskExecutor; } @@ -138,7 +140,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 +149,25 @@ 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 message = null; - byte[] value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS); + byte[] value = null; + try { + value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS); + } + catch (RedisSystemException e) { + if (this.active) { + throw e; + } + else { + logger.error(e); + } + } if (value != null) { if (this.expectMessage) { 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 bff2a74832..2ff459ec78 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 @@ -84,7 +84,7 @@ public class RedisQueueOutboundChannelAdapter extends AbstractMessageHandler imp @Override public String getComponentType() { - return "int-redis:outbound-channel-adapter"; + return "redis:outbound-channel-adapter"; } @Override 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 0a13868fab..274879d1d4 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 @@ -82,10 +82,7 @@ @@ -329,6 +326,146 @@ + + + + Defines a Message Driven Endpoint for listening a Redis queue. + + + + + + + + + Redis queue name. + + + + + + + + + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer. + It can be specified as an empty String value, which means the Endpoint's 'serializer' property is + set to 'null', in which case the Message will contain the raw byte[] payload. + + + + + + + + + + + Specify the timeout in milliseconds to wait for the result of the + 'rightPop' operation on Redis queue. + Default is 1 second. + + + + + + + When true, specifies that the 'byte[]' from a Redis message should be deserialized + as an entire Spring Integration Message. Otherwise the data becomes just the + payload of the message (deserialized or not). + If this attribute is 'true', the 'serializer' must not be an empty String. + Default is 'false'. + + + + + + + + + + + + + + + + + + + + + + Defines an outbound Redis Queue Message-sending Channel Adapter. + + + + + + + + + + + + + + + + + + + + + + + + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer + + + + + + + + + + + Specifies if the Message payload or the entire (serialized) Message will be send to the Redis queue. + Default is 'true'. + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml new file mode 100644 index 0000000000..6c7558adf3 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java new file mode 100644 index 0000000000..56ba70bba6 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java @@ -0,0 +1,114 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import org.hamcrest.Matchers; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.task.TaskExecutor; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.ErrorHandlingTaskExecutor; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + * @since 3.0 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class RedisMessageDrivenEndpointParserTests { + + @Autowired + @Qualifier("redisConnectionFactory") + private RedisConnectionFactory connectionFactory; + + @Autowired + @Qualifier("customRedisConnectionFactory") + private RedisConnectionFactory customRedisConnectionFactory; + + @Autowired + @Qualifier("defaultAdapter.adapter") + private RedisQueueMessageDrivenEndpoint defaultAdapter; + + @Autowired + @Qualifier("defaultAdapter") + private MessageChannel defaultAdapterChannel; + + @Autowired + @Qualifier("customAdapter") + private RedisQueueMessageDrivenEndpoint customAdapter; + + @Autowired + @Qualifier("errorChannel") + private MessageChannel errorChannel; + + @Autowired + @Qualifier("sendChannel") + private MessageChannel sendChannel; + + @Autowired + @Qualifier("executor") + private TaskExecutor taskExecutor; + + + @Autowired + private RedisSerializer serializer; + + @Test + public void testInt3017DefaultConfig() { + assertSame(this.connectionFactory, TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.ops.template.connectionFactory")); + assertEquals("si.test.Int3017.Inbound1", TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.key")); + assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "expectMessage", Boolean.class)); + assertEquals(new Long(1000), TestUtils.getPropertyValue(this.defaultAdapter, "receiveTimeout", Long.class)); + assertNull(TestUtils.getPropertyValue(this.defaultAdapter, "errorChannel")); + assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "taskExecutor"), Matchers.instanceOf(ErrorHandlingTaskExecutor.class)); + assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "serializer"), Matchers.instanceOf(JdkSerializationRedisSerializer.class)); + assertTrue(TestUtils.getPropertyValue(this.defaultAdapter, "autoStartup", Boolean.class)); + assertSame(this.defaultAdapterChannel, TestUtils.getPropertyValue(this.defaultAdapter, "outputChannel")); + } + + @Test + public void testInt3017CustomConfig() { + assertSame(this.customRedisConnectionFactory, TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.ops.template.connectionFactory")); + assertEquals("si.test.Int3017.Inbound2", TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.key")); + assertTrue(TestUtils.getPropertyValue(this.customAdapter, "expectMessage", Boolean.class)); + assertEquals(new Long(2000), TestUtils.getPropertyValue(this.customAdapter, "receiveTimeout", Long.class)); + assertSame(this.errorChannel, TestUtils.getPropertyValue(this.customAdapter, "errorChannel")); + assertSame(this.taskExecutor, TestUtils.getPropertyValue(this.customAdapter, "taskExecutor")); + assertSame(this.serializer, TestUtils.getPropertyValue(this.customAdapter, "serializer")); + assertFalse(TestUtils.getPropertyValue(this.customAdapter, "autoStartup", Boolean.class)); + assertEquals(new Integer(100), TestUtils.getPropertyValue(this.customAdapter, "phase", Integer.class)); + assertSame(this.sendChannel, TestUtils.getPropertyValue(this.customAdapter, "outputChannel")); + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParserTests-context.xml new file mode 100644 index 0000000000..6fc343aeed --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParserTests-context.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParserTests.java new file mode 100644 index 0000000000..14dcd4ccff --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParserTests.java @@ -0,0 +1,83 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.expression.Expression; +import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter; +import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + * @since 3.0 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class RedisQueueOutboundChannelAdapterParserTests { + + @Autowired + @Qualifier("redisConnectionFactory") + private RedisConnectionFactory connectionFactory; + + @Autowired + @Qualifier("customRedisConnectionFactory") + private RedisConnectionFactory customRedisConnectionFactory; + + @Autowired + @Qualifier("defaultAdapter.handler") + private RedisQueueOutboundChannelAdapter defaultAdapter; + + @Autowired + @Qualifier("customAdapter.handler") + private RedisQueueOutboundChannelAdapter customAdapter; + + @Autowired + private RedisSerializer serializer; + + @Test + public void testInt3017DefaultConfig() { + assertSame(this.connectionFactory, TestUtils.getPropertyValue(this.defaultAdapter, "template.connectionFactory")); + assertEquals("foo", TestUtils.getPropertyValue(this.defaultAdapter, "queueNameExpression", Expression.class).getExpressionString()); + assertTrue(TestUtils.getPropertyValue(this.defaultAdapter, "extractPayload", Boolean.class)); + assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "serializerExplicitlySet", Boolean.class)); + } + + @Test + public void testInt3017CustomConfig() { + assertSame(this.customRedisConnectionFactory, TestUtils.getPropertyValue(this.customAdapter, "template.connectionFactory")); + assertEquals("headers['redis_queue']", TestUtils.getPropertyValue(this.customAdapter, "queueNameExpression", Expression.class).getExpressionString()); + assertFalse(TestUtils.getPropertyValue(this.customAdapter, "extractPayload", Boolean.class)); + assertTrue(TestUtils.getPropertyValue(this.customAdapter, "serializerExplicitlySet", Boolean.class)); + assertSame(this.serializer, TestUtils.getPropertyValue(this.customAdapter, "serializer")); + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml new file mode 100644 index 0000000000..3214554326 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 9184e0eff5..5a1c33c8a5 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 @@ -21,15 +21,22 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import java.util.Date; +import java.util.UUID; import org.hamcrest.Matchers; import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mockito; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; 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.serializer.JdkSerializationRedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; import org.springframework.integration.MessagingException; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.PollableChannel; @@ -37,14 +44,31 @@ import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.support.MessageBuilder; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Gunnar Hillert * @author Artem Bilan * @since 3.0 */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { + @Autowired + private RedisConnectionFactory connectionFactory; + + @Autowired + private PollableChannel fromChannel; + + @Autowired + private MessageChannel symmetricalInputChannel; + + @Autowired + private PollableChannel symmetricalOutputChannel; + + @Test @RedisAvailable @SuppressWarnings("unchecked") @@ -52,10 +76,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { String queueName = "si.test.redisQueueInboundChannelAdapterTests"; - RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); - RedisTemplate redisTemplate = new RedisTemplate(); - redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setConnectionFactory(this.connectionFactory); redisTemplate.setEnableDefaultSerializer(false); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); @@ -71,7 +93,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { PollableChannel channel = new QueueChannel(); - RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, connectionFactory); + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); + endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); endpoint.setOutputChannel(channel); endpoint.setReceiveTimeout(1000); endpoint.afterPropertiesSet(); @@ -86,7 +109,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { assertEquals(payload2, receive.getPayload()); endpoint.stop(); - this.waitUntilListening(endpoint); } @Test @@ -96,10 +118,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { final String queueName = "si.test.redisQueueInboundChannelAdapterTests2"; - RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); - RedisTemplate redisTemplate = new RedisTemplate(); - redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setConnectionFactory(this.connectionFactory); redisTemplate.setEnableDefaultSerializer(false); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); @@ -115,7 +135,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { PollableChannel errorChannel = new QueueChannel(); - RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, connectionFactory); + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); + endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); endpoint.setExpectMessage(true); endpoint.setOutputChannel(channel); endpoint.setErrorChannel(errorChannel); @@ -139,19 +160,38 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { endpoint.stop(); - this.waitUntilListening(endpoint); } + @Test + @RedisAvailable + public void testInt3017IntegrationInbound() throws Exception { - public void waitUntilListening(RedisQueueMessageDrivenEndpoint endpoint) throws Exception { - int n = 0; - while (endpoint.isListening()) { - Thread.sleep(100); - if (n++ > 100) { - throw new Exception("RedisQueueMessageDrivenEndpoint failed to stop."); - } - } + String payload = new Date().toString(); + RedisTemplate redisTemplate = new StringRedisTemplate(); + redisTemplate.setConnectionFactory(this.connectionFactory); + redisTemplate.afterPropertiesSet(); + + redisTemplate.boundListOps("si.test.Int3017IntegrationInbound").leftPush("{\"payload\":\"" + payload + "\",\"headers\":{}}"); + + Message receive = this.fromChannel.receive(2000); + assertNotNull(receive); + assertEquals(payload, receive.getPayload()); + } + + @Test + @RedisAvailable + public void testInt3017IntegrationSymmetrical() throws Exception { + UUID payload = UUID.randomUUID(); + Message message = MessageBuilder.withPayload(payload) + .setHeader("redis_queue", "si.test.Int3017IntegrationSymmetrical") + .build(); + + this.symmetricalInputChannel.send(message); + + Message receive = this.symmetricalOutputChannel.receive(2000); + assertNotNull(receive); + assertEquals(payload, receive.getPayload()); } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests-context.xml new file mode 100644 index 0000000000..87a2b1c3ae --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests-context.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java index 47a0c8c35c..e751911c95 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java @@ -24,7 +24,10 @@ import java.util.Date; import java.util.concurrent.TimeUnit; import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; @@ -32,33 +35,47 @@ import org.springframework.data.redis.serializer.JacksonJsonRedisSerializer; import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.mapping.InboundMessageMapper; import org.springframework.integration.message.GenericMessage; 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.json.Jackson2JsonMessageParser; +import org.springframework.integration.support.json.JsonInboundMessageMapper; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Gunnar Hillert * @author Artem Bilan * @since 3.0 */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests { + @Autowired + private RedisConnectionFactory connectionFactory; + + @Autowired + @Qualifier("toRedisQueueChannel") + private MessageChannel sendChannel; + + @Test @RedisAvailable public void testInt3015Default() throws Exception { final String queueName = "si.test.testRedisQueueOutboundChannelAdapter"; - RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); - - final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory); + final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, this.connectionFactory); String payload = "testing"; handler.handleMessage(MessageBuilder.withPayload(payload).build()); RedisTemplate redisTemplate = new StringRedisTemplate(); - redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setConnectionFactory(this.connectionFactory); redisTemplate.afterPropertiesSet(); Object result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); @@ -70,7 +87,7 @@ public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests { handler.handleMessage(MessageBuilder.withPayload(payload2).build()); RedisTemplate redisTemplate2 = new RedisTemplate(); - redisTemplate2.setConnectionFactory(connectionFactory); + redisTemplate2.setConnectionFactory(this.connectionFactory); redisTemplate2.setEnableDefaultSerializer(false); redisTemplate2.setKeySerializer(new StringRedisSerializer()); redisTemplate2.setValueSerializer(new JdkSerializationRedisSerializer()); @@ -88,16 +105,14 @@ public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests { final String queueName = "si.test.testRedisQueueOutboundChannelAdapter2"; - RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); - - final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory); + final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, this.connectionFactory); handler.setExtractPayload(false); Message message = MessageBuilder.withPayload("testing").build(); handler.handleMessage(message); RedisTemplate redisTemplate = new RedisTemplate(); - redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setConnectionFactory(this.connectionFactory); redisTemplate.setEnableDefaultSerializer(false); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); @@ -116,13 +131,11 @@ public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests { final String queueName = "si.test.testRedisQueueOutboundChannelAdapter2"; - RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); - - final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory); + final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, this.connectionFactory); handler.setSerializer(new JacksonJsonRedisSerializer(Object.class)); RedisTemplate redisTemplate = new StringRedisTemplate(); - redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setConnectionFactory(this.connectionFactory); redisTemplate.afterPropertiesSet(); handler.handleMessage(new GenericMessage(Arrays.asList("foo", "bar", "baz"))); @@ -140,4 +153,24 @@ public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests { assertEquals("\"test\"", result); } + @Test + @RedisAvailable + public void testInt3017IntegrationOutbound() throws Exception { + + final String queueName = "si.test.Int3017IntegrationOutbound"; + + GenericMessage message = new GenericMessage(queueName); + this.sendChannel.send(message); + + RedisTemplate redisTemplate = new StringRedisTemplate(); + redisTemplate.setConnectionFactory(this.connectionFactory); + redisTemplate.afterPropertiesSet(); + + String result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result); + InboundMessageMapper mapper = new JsonInboundMessageMapper(String.class, new Jackson2JsonMessageParser()); + Message resultMessage = mapper.toMessage(result); + assertEquals(message.getPayload(), resultMessage.getPayload()); + } + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/CustomJsonSerializer.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/CustomJsonSerializer.java new file mode 100644 index 0000000000..d76f24a637 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/CustomJsonSerializer.java @@ -0,0 +1,60 @@ +/* + * 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.util; + +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.serializer.SerializationException; +import org.springframework.integration.Message; +import org.springframework.integration.mapping.InboundMessageMapper; +import org.springframework.integration.support.json.Jackson2JsonMessageParser; +import org.springframework.integration.support.json.JsonInboundMessageMapper; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; + +/** +* @author Artem Bilan +* @since 3.0 +*/ +public class CustomJsonSerializer implements RedisSerializer> { + + private final ObjectMapper objectMapper = new ObjectMapper(); + + private final InboundMessageMapper mapper = + new JsonInboundMessageMapper(String.class, new Jackson2JsonMessageParser()); + + @Override + public byte[] serialize(Message message) throws SerializationException { + try { + return this.objectMapper.writeValueAsBytes(message); + } + catch (JsonProcessingException e) { + throw new SerializationException("Fail to serialize 'message' to json.", e); + } + } + + @Override + public Message deserialize(byte[] bytes) throws SerializationException { + try { + return mapper.toMessage(new String(bytes)); + } + catch (Exception e) { + throw new SerializationException("Fail to deserialize 'message' from json.", e); + } + } + +} diff --git a/src/reference/docbook/redis.xml b/src/reference/docbook/redis.xml index 42dcefeadd..367ab41d38 100644 --- a/src/reference/docbook/redis.xml +++ b/src/reference/docbook/redis.xml @@ -179,6 +179,160 @@ rt.setConnectionFactory(redisConnectionFactory);]]> This example also includes the optional, custom MessageConverter (the 'testConverter' bean). +
+ Redis 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. + ]]> + + + + + The component bean name. If the channel attribute isn't provided a DirectChannel + is created and registered with application context with this id attribute as the bean name. + In this case, the endpoint itself is registered with the bean name id + '.adapter'. + + + + + The MessageChannel to which to send Messages from this Endpoint. + + + + + A SmartLifecycle attribute to specify whether this Endpoint should start automatically after + the application context start or not. Default is true. + + + + + A SmartLifecycle attribute to specify the phase in which + this Endpoint will be started. Default is 0. + + + + + A reference to a RedisConnectionFactory bean. Defaults to + redisConnectionFactory. + + + + + The name of the Redis List on which the queue-based 'right pop' operation is performed to get Redis messages. + + + + + The MessageChannel to which to send ErrorMessages with + Exceptions from the listening task of the Endpoint. + + + + + The RedisSerializer bean reference. Can be an empty string, which means 'no serializer'. + In this case the raw byte[] from the inbound Redis message is sent to the channel as the + Message payload. By default it is a JdkSerializationRedisSerializer. + + + + + The timeout in milliseconds for 'right pop' operation to wait for a Redis message from the queue. Default is 1 second. + + + + + Specify if this Endpoint expects data from the Redis queue to contain entire Messages. + If this attribute is set to true, the serializer can't be an empty string because messages + require some form of deserialization (JDK serialization by default). + Default is false. + + + + + A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) + bean. It is used for the underlying listening task. By default a SimpleAsyncTaskExecutor + is used. + + + + +
+
+ Redis Queue Outbound Channel Adapter + + Since Spring Integration 3.0, a Queue Outbound Channel Adapter + is available to 'left push' to a Redis List from Spring Integration messages: + ]]> + + + + + The component bean name. If the channel attribute isn't provided, a DirectChannel + is created and registered with the application context with this id attribute as the bean name. + In this case, the endpoint is registered with the bean name id + '.adapter'. + + + + + The MessageChannel from which this Endpoint receives Messages. + + + + + A reference to a RedisConnectionFactory bean. Defaults to + redisConnectionFactory. + + + + + The name of the Redis List on which the queue-based 'left push' operation is performed to send Redis messages. + This attribute is mutually exclusive with queue-expression. + + + + + A SpEL Expression to determine the name of the Redis List + using the incoming Message at runtime as the #root variable. + This attribute is mutually exclusive with queue. + + + + + A RedisSerializer bean reference. + By default it is a JdkSerializationRedisSerializer. + However, for String payloads, a StringRedisSerializer + is used, if a serializer reference isn't provided. + + + + + Specify if this Endpoint should send just the payload to the Redis queue, + or the entire Message. + Default is true + . + + + + +
@@ -423,4 +577,4 @@ the serialization of values, you may want to consider providing your own
- \ No newline at end of file + diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 8bd71c43ee..614e17e06c 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -143,12 +143,12 @@ For more information see . -
- Redis Metadata Store +
+ Redis: New Components A new Redis-based MetadataStore - implementation was added. The RedisMetadataStore can + implementation has been added. The RedisMetadataStore can be used to maintain state of a MetadataStore across application restarts. This new MetadataStore implementation can be used with adapters such as: @@ -158,7 +158,12 @@ Feed Inbound Channel Adapter - For more information see . + New queue-based components has been added. The <int-redis:queue-inbound-channel-adapter/> + and the <int-redis:queue-outbound-channel-adapter/> components are provided + to perform 'right pop' and 'left push' operations on a Redis List, respectively. + + + For more information see .