diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java index 6ade9c2c30..13aab72337 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java @@ -23,20 +23,21 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter; import org.springframework.util.StringUtils; /** * @author Oleg Zhurakousky * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan * @since 2.1 */ public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterParser { @Override protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.redis.inbound.RedisInboundChannelAdapter"); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisInboundChannelAdapter.class); String connectionFactory = element.getAttribute("connection-factory"); if (!StringUtils.hasText(connectionFactory)) { connectionFactory = "redisConnectionFactory"; @@ -46,7 +47,8 @@ public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterPars IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topics"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true); + return builder.getBeanDefinition(); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java index b97e31572d..eb507557f1 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,32 +18,41 @@ package org.springframework.integration.redis.config; import org.w3c.dom.Element; +import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.redis.outbound.RedisPublishingMessageHandler; import org.springframework.util.StringUtils; /** + * Parser for the {@code } component. + * * @author Oleg Zhurakousky * @author Mark Fisher + * @author Artem Bilan * @since 2.1 */ public class RedisOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.redis.outbound.RedisPublishingMessageHandler"); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisPublishingMessageHandler.class); String connectionFactory = element.getAttribute("connection-factory"); if (!StringUtils.hasText(connectionFactory)) { connectionFactory = "redisConnectionFactory"; } builder.addConstructorArgReference(connectionFactory); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topic", "defaultTopic"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer"); + + BeanDefinition topicExpression = IntegrationNamespaceUtils + .createExpressionDefinitionFromValueOrExpression("topic", "topic-expression", parserContext, element, true); + builder.addPropertyValue("topicExpression", topicExpression); + return builder.getBeanDefinition(); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java index 0fc7eb78be..d1c7d17dc2 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2012 the original author or authors + * Copyright 2007-2013 the original author or authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -34,6 +34,7 @@ import org.springframework.util.Assert; /** * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.1 */ public class RedisInboundChannelAdapter extends MessageProducerSupport { @@ -52,7 +53,6 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { } public void setSerializer(RedisSerializer serializer) { - Assert.notNull(serializer, "'serializer' must not be null"); this.serializer = serializer; } @@ -99,16 +99,16 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { this.container.stop(); } - private Message convertMessage(String s) { - return this.messageConverter.toMessage(s); + private Message convertMessage(Object object) { + return this.messageConverter.toMessage(object); } private class MessageListenerDelegate { @SuppressWarnings("unused") - public void handleMessage(String s) { - sendMessage(convertMessage(s)); + public void handleMessage(Object object) { + sendMessage(convertMessage(object)); } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java index 731e54265b..2460607608 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2011 the original author or authors + * Copyright 2007-2013 the original author or authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,10 +17,14 @@ package org.springframework.integration.redis.outbound; import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.Message; +import org.springframework.integration.expression.IntegrationEvaluationContextAware; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.support.converter.MessageConverter; import org.springframework.integration.support.converter.SimpleMessageConverter; @@ -28,21 +32,32 @@ import org.springframework.util.Assert; /** * @author Mark Fisher + * @author Artem Bilan * @since 2.1 */ -public class RedisPublishingMessageHandler extends AbstractMessageHandler { +public class RedisPublishingMessageHandler extends AbstractMessageHandler implements IntegrationEvaluationContextAware { - private final StringRedisTemplate template; + private final RedisTemplate template; + + private volatile EvaluationContext evaluationContext; private volatile MessageConverter messageConverter = new SimpleMessageConverter(); - private volatile String defaultTopic; - private volatile RedisSerializer serializer = new StringRedisSerializer(); + private volatile Expression topicExpression; + public RedisPublishingMessageHandler(RedisConnectionFactory connectionFactory) { Assert.notNull(connectionFactory, "connectionFactory must not be null"); - this.template = new StringRedisTemplate(connectionFactory); + this.template = new RedisTemplate(); + this.template.setConnectionFactory(connectionFactory); + this.template.setEnableDefaultSerializer(false); + this.template.afterPropertiesSet(); + } + + @Override + public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + this.evaluationContext = evaluationContext; } public void setSerializer(RedisSerializer serializer) { @@ -55,27 +70,42 @@ public class RedisPublishingMessageHandler extends AbstractMessageHandler { this.messageConverter = messageConverter; } + /** + * @deprecated in favor of {@link #setTopicExpression(Expression)} or {@link #setTopic(String)} + */ + @Deprecated public void setDefaultTopic(String defaultTopic) { - this.defaultTopic = defaultTopic; + Assert.hasText(defaultTopic, "'defaultTopic' must not be an empty string."); + this.setTopicExpression(new LiteralExpression(defaultTopic)); } - private String determineTopic(Message message) { - // TODO: add support for determining topic by evaluating SpEL against the Message - Assert.hasText(this.defaultTopic, "Failed to determine Redis topic " + - "from Message, and no defaultTopic has been provided."); - return this.defaultTopic; + public void setTopic(String topic) { + Assert.hasText(topic, "'topic' must not be an empty string."); + this.setTopicExpression(new LiteralExpression(topic)); } - @Override - protected void handleMessageInternal(Message message) throws Exception { - String topic = this.determineTopic(message); - Object value = this.messageConverter.fromMessage(message); - this.template.convertAndSend(topic, value.toString()); + public void setTopicExpression(Expression topicExpression) { + Assert.notNull(topicExpression, "'topicExpression' must not be null."); + this.topicExpression = topicExpression; } @Override protected void onInit() throws Exception { - this.template.setValueSerializer(this.serializer); - this.template.afterPropertiesSet(); + Assert.notNull(topicExpression, "'topicExpression' must not be null."); } + + @Override + @SuppressWarnings("unchecked") + protected void handleMessageInternal(Message message) throws Exception { + String topic = this.topicExpression.getValue(this.evaluationContext, message, String.class); + Object value = this.messageConverter.fromMessage(message); + + if (value instanceof byte[]) { + this.template.convertAndSend(topic, value); + } + else { + this.template.convertAndSend(topic, ((RedisSerializer) this.serializer).serialize(value)); + } + } + } diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd index 274879d1d4..450916967e 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd @@ -174,7 +174,9 @@ - Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer. + This attribute can be an empty string, which results in 'null' being used by the underlying adapter, + meaning no serializer is used and the raw byte[] will be the message payload. @@ -197,7 +199,22 @@ - + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java index e9bbe29fdc..ed0e51b9ed 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java @@ -17,6 +17,8 @@ package org.springframework.integration.redis.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import org.junit.Test; @@ -67,6 +69,10 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests { Object converterBean = context.getBean("testConverter"); assertEquals(converterBean, accessor.getPropertyValue("messageConverter")); assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer")); + + Object bean = context.getBean("withoutSerializer.adapter"); + assertNotNull(bean); + assertNull(TestUtils.getPropertyValue(bean, "serializer")); } @Test diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml index b5126265a3..5041900e54 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml @@ -11,7 +11,7 @@ @@ -21,6 +21,12 @@ + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java index 52010d1a65..05e8deb1c6 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java @@ -25,6 +25,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.expression.Expression; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.channel.QueueChannel; @@ -33,6 +34,7 @@ import org.springframework.integration.message.GenericMessage; import org.springframework.integration.redis.outbound.RedisPublishingMessageHandler; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -53,13 +55,16 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests{ @Test @RedisAvailable + @SuppressWarnings("unchecked") public void validateConfiguration() { EventDrivenConsumer adapter = context.getBean("outboundAdapter", EventDrivenConsumer.class); RedisPublishingMessageHandler handler = (RedisPublishingMessageHandler) new DirectFieldAccessor(adapter).getPropertyValue("handler"); assertEquals("outboundAdapter", adapter.getComponentName()); DirectFieldAccessor accessor = new DirectFieldAccessor(handler); - assertEquals("foo", accessor.getPropertyValue("defaultTopic")); + Object topicExpression = accessor.getPropertyValue("topicExpression"); + assertNotNull(topicExpression); + assertEquals("headers['topic'] ?: 'foo'", ((Expression) topicExpression).getExpressionString()); Object converterBean = context.getBean("testConverter"); assertEquals(converterBean, accessor.getPropertyValue("messageConverter")); assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer")); @@ -74,6 +79,13 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests{ Message message = receiveChannel.receive(5000); assertNotNull(message); assertEquals("Hello Redis", message.getPayload()); + + sendChannel = context.getBean("sendChannel", MessageChannel.class); + sendChannel.send(MessageBuilder.withPayload("Hello Redis").setHeader("topic", "bar").build()); + receiveChannel = context.getBean("barChannel", QueueChannel.class); + message = receiveChannel.receive(5000); + assertNotNull(message); + assertEquals("Hello Redis", message.getPayload()); } @Test //INT-2275 diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java index a94c429741..279ae1e9aa 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java @@ -18,15 +18,14 @@ package org.springframework.integration.redis.inbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; +import org.hamcrest.Matchers; import org.junit.Test; -import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.integration.Message; @@ -37,12 +36,11 @@ import org.springframework.integration.test.util.TestUtils; /** * @author Mark Fisher + * @author Artem Bilan * @since 2.1 */ public class RedisInboundChannelAdapterTests extends RedisAvailableTests{ - private final Log logger = LogFactory.getLog(this.getClass()); - @Test @RedisAvailable public void testRedisInboundChannelAdapter() throws Exception { @@ -59,19 +57,18 @@ public class RedisInboundChannelAdapterTests extends RedisAvailableTests{ RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); RedisInboundChannelAdapter adapter = new RedisInboundChannelAdapter(connectionFactory); - adapter.setTopics("testRedisInboundChannelAdapterChannel"); + adapter.setTopics(redisChannelName); adapter.setOutputChannel(channel); adapter.afterPropertiesSet(); adapter.start(); - RedisMessageListenerContainer container = waitUntilSubscribed(adapter); + this.awaitContainerSubscribed(TestUtils.getPropertyValue(adapter, "container", RedisMessageListenerContainer.class)); StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory); redisTemplate.afterPropertiesSet(); for (int i = 0; i < numToTest; i++) { String message = "test-" + i + " iteration " + iteration; redisTemplate.convertAndSend(redisChannelName, message); - logger.debug("Sent " + message); } int counter = 0; for (int i = 0; i < numToTest; i++) { @@ -85,34 +82,42 @@ public class RedisInboundChannelAdapterTests extends RedisAvailableTests{ } assertEquals(numToTest, counter); adapter.stop(); - container.stop(); - } - /** - * Wait until the container has subscribed to the queue and return a - * reference to it, so we can stop it at the end of the test. - */ - protected RedisMessageListenerContainer waitUntilSubscribed( - RedisInboundChannelAdapter adapter) throws Exception { - RedisMessageListenerContainer container = (RedisMessageListenerContainer) TestUtils - .getPropertyValue(adapter, "container"); - Object subscriptionTask = TestUtils.getPropertyValue(container, "subscriptionTask"); - RedisConnection connection = (RedisConnection) TestUtils - .getPropertyValue(subscriptionTask, "connection"); - int n = 0; - while (true) { - if (n++ > 50) { - fail("RMLC Failed to Subscribe"); - } - if (connection.isSubscribed()) { - logger.debug("Subscribed OK"); - break; - } - logger.debug("Waiting..."); - Thread.sleep(100); + redisChannelName = "testRedisBytesInboundChannelAdapterChannel"; + + adapter.setTopics(redisChannelName); + adapter.setSerializer(null); + adapter.afterPropertiesSet(); + adapter.start(); + + this.awaitContainerSubscribed(TestUtils.getPropertyValue(adapter, "container", RedisMessageListenerContainer.class)); + + RedisTemplate template = new RedisTemplate(); + template.setConnectionFactory(connectionFactory); + template.setEnableDefaultSerializer(false); + template.afterPropertiesSet(); + + for (int i = 0; i < numToTest; i++) { + String message = "test-" + i + " iteration " + iteration; + template.convertAndSend(redisChannelName, message.getBytes()); } - Thread.sleep(100); // Wait a little longer due to race condition in connection.isSubscribed() - return container; + + counter = 0; + for (int i = 0; i < numToTest; i++) { + Message message = channel.receive(5000); + if (message == null){ + throw new RuntimeException("Failed to receive message # " + i + " iteration " + iteration); + } + assertNotNull(message); + Object payload = message.getPayload(); + assertThat(payload, Matchers.instanceOf(byte[].class)); + + assertTrue(new String((byte[]) payload).startsWith("test-")); + counter++; + } + + assertEquals(numToTest, counter); + adapter.stop(); } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java index 71f4b97f77..dc3648cc06 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java @@ -30,12 +30,14 @@ import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.listener.Topic; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.support.MessageBuilder; /** * @author Mark Fisher + * @author Artem Bilan * @since 2.1 */ public class RedisPublishingMessageHandlerTests extends RedisAvailableTests { @@ -45,7 +47,7 @@ public class RedisPublishingMessageHandlerTests extends RedisAvailableTests { public void testRedisPublishingMessageHandler() throws Exception { int numToTest = 10; String topic = "si.test.channel"; - final CountDownLatch latch = new CountDownLatch(numToTest); + final CountDownLatch latch = new CountDownLatch(numToTest * 2); RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); @@ -59,14 +61,20 @@ public class RedisPublishingMessageHandlerTests extends RedisAvailableTests { container.afterPropertiesSet(); container.addMessageListener(listener, Collections.singletonList(new ChannelTopic(topic))); container.start(); - Thread.sleep(1000); + + this.awaitContainerSubscribed(container); final RedisPublishingMessageHandler handler = new RedisPublishingMessageHandler(connectionFactory); - handler.setDefaultTopic(topic); + handler.setTopicExpression(new LiteralExpression(topic)); + for (int i = 0; i < numToTest; i++) { handler.handleMessage(MessageBuilder.withPayload("test-" + i).build()); } - assertTrue(latch.await(3, TimeUnit.SECONDS)); + + for (int i = 0; i < numToTest; i++) { + handler.handleMessage(MessageBuilder.withPayload(("test-" + i).getBytes()).build()); + } + assertTrue(latch.await(10, TimeUnit.SECONDS)); container.stop(); } @@ -83,6 +91,7 @@ public class RedisPublishingMessageHandlerTests extends RedisAvailableTests { public void handleMessage(String s) { this.latch.countDown(); } + } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java index 064f5cf3d7..4adb288a48 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java @@ -15,6 +15,8 @@ */ package org.springframework.integration.redis.rules; +import static org.junit.Assert.assertTrue; + import java.util.UUID; import org.junit.Rule; @@ -28,6 +30,8 @@ import org.springframework.data.redis.core.BoundZSetOperations; import org.springframework.data.redis.core.RedisCallback; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.integration.test.util.TestUtils; /** * @author Oleg Zhurakousky @@ -40,7 +44,7 @@ public class RedisAvailableTests { @Rule public RedisAvailableRule redisAvailableRule = new RedisAvailableRule(); - public RedisConnectionFactory getConnectionFactoryForTest(){ + protected RedisConnectionFactory getConnectionFactoryForTest(){ LettuceConnectionFactory connectionFactory = RedisAvailableRule.connectionFactoryResource.get(); RedisTemplate rt = new RedisTemplate(); rt.setConnectionFactory(connectionFactory); @@ -56,6 +60,17 @@ public class RedisAvailableTests { return connectionFactory; } + protected void awaitContainerSubscribed(RedisMessageListenerContainer container) throws Exception { + RedisConnection connection = TestUtils.getPropertyValue(container, "subscriptionTask.connection", + RedisConnection.class); + + int n = 0; + while (n++ < 100 && !connection.isSubscribed()) { + Thread.sleep(100); + } + assertTrue("RedisMessageListenerContainer Failed to Subscribe", n < 100); + } + protected void prepareList(RedisConnectionFactory connectionFactory){ StringRedisTemplate redisTemplate = new StringRedisTemplate(); diff --git a/src/reference/docbook/redis.xml b/src/reference/docbook/redis.xml index 367ab41d38..1bd7c2023a 100644 --- a/src/reference/docbook/redis.xml +++ b/src/reference/docbook/redis.xml @@ -152,6 +152,12 @@ rt.setConnectionFactory(redisConnectionFactory);]]> Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the topics attribute. + + Inbound adapters can use a RedisSerializer to deserialize the body of Redis Messages. + The serializer attribute of the <int-redis:inbound-channel-adapter> can be set to an + empty string, which results in a null value for the RedisSerializer property. + In this case the raw byte[] bodies of Redis Messages are provided as the message payloads. +
@@ -178,11 +184,16 @@ rt.setConnectionFactory(redisConnectionFactory);]]> a RedisConnectionFactory which was defined with 'redisConnectionFactory' as its bean name. This example also includes the optional, custom MessageConverter (the 'testConverter' bean). + + Since Spring Integration 3.0, the <int-redis:outbound-channel-adapter>, + as an alternative to the topic attribute, has the topic-expression attribute to determine + the Redis topic against the Message at runtime. These attributes are mutually exclusive. +
Redis Queue Inbound Channel Adapter - Since Spring Integration 3.0, a Queue Inbound Channel Adapter + Since Spring Integration 3.0, a Queue Inbound Channel Adapter is available to 'right pop' messages from a Redis List. The adapter is message-driven using an internal listener thread and does not use a poller. .
+
+ Redis Adapters Changers + + + + The Redis Inbound Channel Adapter can now use a null value for serializer + property, with the raw data being the message payload. + + + The Redis Outbound Channel Adapter now has the topic-expression property to determine + the Redis topic against the Message at runtime. + + + + + For more information, see . + +