INT-3177: Generic RedisTemplate for PublishingMH

JIRA: https://jira.springsource.org/browse/INT-3177

* Change `RedisPublishingMessageHandler.StringRedisTemplate` to `RedisTemplate<?, ?>`
* Don't provide `serializer` to the `template`
* Use `serializer` directly for values which are not `byte[]`
* Add `RedisAvailableTests#awaitContainerSubscribed` for tests to avoid race conditions
* Refactor some tests
* Add test for `byte[]` payload

INT-3177: Generic RedisInboundChannelAdapter

Port from Spring XD: allow for `RedisInboundChannelAdapter`
to receive from Redis any object, not only String

* Make `serializer` property 'resettable' to `null`
* Change `RedisInboundChannelAdapterParser` to allow to apply
empty value from `serializer` attribute
* Add parser test and test for `byte[]` messages

INT-3177: Polishing and documentation

JIRA: https://jira.springsource.org/browse/INT-3033

* Add `topic-expression` to `<int-redis:outbound-channel-adapter>`
* Add tests and docs

INT-3177: Deprecate `RedisPublishMH.defaultTopic`

* Make `topic` and `topic-expression` attributes as mutually exclusive
* Polishing Redis Topic Parsers
* Polishing tests and docs
* Default `serializer` for Redis Topic Adapters is `StringRedisSerializer` for backward compatibility

Polishing according PR's discussion

INT-3177 Doc Polishing

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

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 <outbound-channel-adapter/>} 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();
 	}
 

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

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<Object, Object>();
+		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<Object>) this.serializer).serialize(value));
+		}
+	}
+
 }

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

@@ -174,7 +174,9 @@
 				<xsd:annotation>
 					<xsd:appinfo>
 						<xsd:documentation>
-						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.
 						</xsd:documentation>
 						<tool:annotation kind="ref">
 							<tool:expected-type type="org.springframework.data.redis.serializer.RedisSerializer"/>
@@ -197,7 +199,22 @@
 					<xsd:sequence>
 						<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
 					</xsd:sequence>
-					<xsd:attribute name="topic" type="xsd:string"/>
+					<xsd:attribute name="topic" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+								Specifies the Redis topic.
+								This attribute is mutually exclusive with the 'topic-expression' attribute.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="topic-expression" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+								Specifies the SpEL expression to determine the Redis topic using the Message at runtime.
+								This attribute is mutually exclusive with the 'topic' attribute.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="message-converter" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation><![CDATA[

spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java

@@ -30,7 +30,6 @@ import java.util.concurrent.TimeUnit;
 import org.junit.Test;
 
 import org.springframework.beans.factory.BeanFactory;
-import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
@@ -53,7 +52,7 @@ public class SubscribableRedisChannelTests extends RedisAvailableTests {
 
 	@Test
 	@RedisAvailable
-	public void pubSubChannelTest() throws Exception{
+	public void pubSubChannelTest() throws Exception {
 		RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
 
 		SubscribableRedisChannel channel = new SubscribableRedisChannel(connectionFactory, "si.test.channel");
@@ -61,14 +60,7 @@ public class SubscribableRedisChannelTests extends RedisAvailableTests {
 		channel.afterPropertiesSet();
 		channel.start();
 
-		RedisConnection connection = TestUtils.getPropertyValue(channel, "container.subscriptionTask.connection",
-				RedisConnection.class);
-
-		int n = 0;
-		while (n++ < 100 && !connection.isSubscribed()) {
-			Thread.sleep(100);
-		}
-		assertTrue(n < 100);
+		this.awaitContainerSubscribed(TestUtils.getPropertyValue(channel, "container", RedisMessageListenerContainer.class));
 
 		final CountDownLatch latch = new CountDownLatch(3);
 		MessageHandler handler = new MessageHandler() {

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml

@@ -32,4 +32,7 @@
 
 	<int:bridge input-channel="autoChannel" output-channel="nullChannel"/>
 
+	<int-redis:inbound-channel-adapter id="withoutSerializer" topics="foo" serializer=""/>
+
+
 </beans>

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

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml

@@ -11,7 +11,7 @@
 
 	<int-redis:outbound-channel-adapter id="outboundAdapter"
 			channel="sendChannel"
-			topic="foo"
+			topic-expression="headers['topic'] ?: 'foo'"
 			message-converter="testConverter"
 			serializer="serializer"/>
 
@@ -21,6 +21,12 @@
 		<int:queue/>
 	</int:channel>
 
+	<int-redis:inbound-channel-adapter channel="barChannel" topics="bar"/>
+
+	<int:channel id="barChannel">
+		<int:queue/>
+	</int:channel>
+
 	<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory">
 		<property name="port" value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
 	</bean>

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

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<Object, Object>();
+		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();
 	}
 
 }

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.<Topic>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();
 		}
+
 	}
 
 }

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<UUID, Object> rt = new RedisTemplate<UUID, Object>();
 		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();

src/reference/docbook/redis.xml

@@ -152,6 +152,12 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
 
       <para>Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the
       <code>topics</code> attribute.</para>
+      <para>
+         Inbound adapters can use a <classname>RedisSerializer</classname> to deserialize the body of Redis Messages.
+         The <code>serializer</code> attribute of the <code>&lt;int-redis:inbound-channel-adapter&gt;</code> can be set to an
+         empty string, which results in a <code>null</code> value for the <classname>RedisSerializer</classname> property.
+         In this case the raw <code>byte[]</code> bodies of Redis Messages are provided as the message payloads.
+      </para>
     </section>
 
     <section id="redis-outbound-channel-adapter">
@@ -178,11 +184,16 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
       a <classname>RedisConnectionFactory</classname> which was defined with '<code>redisConnectionFactory</code>' as its bean name.
       This example also includes the optional, custom <classname>MessageConverter</classname> (the '<code>testConverter</code>' bean).
       </para>
+      <para>
+         Since <emphasis>Spring Integration 3.0</emphasis>, the <code>&lt;int-redis:outbound-channel-adapter&gt;</code>,
+         as an alternative to the <code>topic</code> attribute, has the <code>topic-expression</code> attribute to determine
+         the Redis topic against the Message at runtime. These attributes are mutually exclusive.
+      </para>
     </section>
     <section id="redis-queue-inbound-channel-adapter">
        <title>Redis Queue Inbound Channel Adapter</title>
        <para>
-          Since <emphasis>Spring Integration 3.0</emphasis>, a Queue Inbound Channel Adapter 
+          Since <emphasis>Spring Integration 3.0</emphasis>, 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.
 		<programlisting language="xml"><![CDATA[<int-redis:queue-inbound-channel-adapter id="" ]]><co id="redis-m-d-c-a-id"/><![CDATA[

src/reference/docbook/whats-new.xml

@@ -572,5 +572,23 @@
 			    result set respectively. For more information see <xref linkend="jpa"/>.
 			</para>
 		</section>
+		<section id="3.0-redis">
+			<title>Redis Adapters Changers</title>
+			<para>
+				<itemizedlist>
+					<listitem>
+						The Redis Inbound Channel Adapter can now use a <code>null</code> value for <code>serializer</code>
+						property, with the raw data being the message payload.
+					</listitem>
+					<listitem>
+						The Redis Outbound Channel Adapter now has the <code>topic-expression</code> property to determine
+						the Redis topic against the Message at runtime.
+					</listitem>
+				</itemizedlist>
+			</para>
+			<para>
+				For more information, see <xref linkend="redis"/>.
+			</para>
+		</section>
 	</section>
 </chapter>