Merge remote-tracking branch 'upstream/master' into 4.0.0-WIP

Conflicts:
	spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java
	spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
	spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java
	spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java
	spring-integration-core/src/main/java/org/springframework/integration/metadata/package-info.java
	spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java
	spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java
	spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java
	spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java
	spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java
	spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java
	spring-integration-test/src/main/java/org/springframework/integration/test/matcher/HeaderMatcher.java
	spring-integration-test/src/main/java/org/springframework/integration/test/matcher/MockitoMessageMatchers.java
	spring-integration-test/src/main/java/org/springframework/integration/test/matcher/PayloadMatcher.java
	spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java
	spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests.java

Resolved.

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

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/config/RedisQueueInboundChannelAdapterParser.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.redis.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <queue-inbound-channel-adapter> element of the 'redis' namespace.
+ *
+ * @author Artem Bilan
+ * @since 3.0
+ */
+public class RedisQueueInboundChannelAdapterParser extends AbstractChannelAdapterParser {
+
+	@Override
+	protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisQueueMessageDrivenEndpoint.class);
+		builder.addConstructorArgValue(element.getAttribute("queue"));
+
+		String connectionFactory = element.getAttribute("connection-factory");
+		if (!StringUtils.hasText(connectionFactory)) {
+			connectionFactory = "redisConnectionFactory";
+		}
+		builder.addConstructorArgReference(connectionFactory);
+
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
+		builder.addPropertyReference("outputChannel", channelName);
+
+		return builder.getBeanDefinition();
+	}
+
+}

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

spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,9 +20,7 @@ import org.w3c.dom.Element;
 
 import org.springframework.beans.BeanMetadataElement;
 import org.springframework.beans.factory.config.BeanDefinition;
-import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
@@ -62,9 +60,8 @@ public class RedisStoreInboundChannelAdapterParser extends AbstractPollingInboun
 						parserContext, element, atLeastOneRequired);
 		builder.addConstructorArgValue(expressionDef);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type");
-		String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
-				builder.getBeanDefinition(), parserContext.getRegistry());
-		return new RuntimeBeanReference(beanName);
+
+		return builder.getBeanDefinition();
 	}
 
 }

spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.redis.event;
+
+/**
+ * @author Artem Bilan
+ * @since 3.0
+ */
+@SuppressWarnings("serial")
+public class RedisExceptionEvent extends RedisIntegrationEvent {
+
+	public RedisExceptionEvent(Object source, Throwable cause) {
+		super(source, cause);
+	}
+
+}

spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.redis.event;
+
+import org.springframework.integration.event.IntegrationEvent;
+
+/**
+ * @author Artem Bilan
+ * @since 3.0
+ *
+ */
+@SuppressWarnings("serial")
+public abstract class RedisIntegrationEvent extends IntegrationEvent {
+
+	public RedisIntegrationEvent(Object source) {
+		super(source);
+	}
+
+	public RedisIntegrationEvent(Object source, Throwable cause) {
+		super(source, cause);
+	}
+
+}

spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Events generated by the redis module
+ */
+package org.springframework.integration.redis.event;

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2007-2012 the original author or authors
+ * Copyright 2007-2013 the original author or authors
  *
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -34,9 +34,9 @@ import org.springframework.util.Assert;
 /**
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  * @since 2.1
  */
-@SuppressWarnings("rawtypes")
 public class RedisInboundChannelAdapter extends MessageProducerSupport {
 
 	private final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
@@ -53,7 +53,6 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
 	}
 
 	public void setSerializer(RedisSerializer<?> serializer) {
-		Assert.notNull(serializer, "'serializer' must not be null");
 		this.serializer = serializer;
 	}
 
@@ -100,17 +99,16 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
 		this.container.stop();
 	}
 
-	@SuppressWarnings("unchecked")
-	private Message<?> convertMessage(String s) {
-		return this.messageConverter.toMessage(s, null);
+	private Message<?> convertMessage(Object object) {
+		return this.messageConverter.toMessage(object, null);
 	}
 
 
 	private class MessageListenerDelegate {
 
 		@SuppressWarnings("unused")
-		public void handleMessage(String s) {
-			sendMessage(convertMessage(s));
+		public void handleMessage(Object object) {
+			sendMessage(convertMessage(object));
 		}
 	}
 

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java

@@ -15,10 +15,12 @@
  */
 package org.springframework.integration.redis.inbound;
 
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
-import org.springframework.core.task.TaskExecutor;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.BoundListOperations;
 import org.springframework.data.redis.core.RedisTemplate;
@@ -27,7 +29,9 @@ import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.redis.event.RedisExceptionEvent;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.util.ErrorHandlingTaskExecutor;
 import org.springframework.jmx.export.annotation.ManagedMetric;
 import org.springframework.jmx.export.annotation.ManagedOperation;
@@ -44,15 +48,19 @@ import org.springframework.util.Assert;
  * @since 3.0
  */
 @ManagedResource
-public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
+public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport implements ApplicationEventPublisherAware {
 
 	public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
 
+	public static final long DEFAULT_RECOVERY_INTERVAL = 5000;
+
 	private final BoundListOperations<String, byte[]> boundListOperations;
 
-	private MessageChannel errorChannel;
+	private volatile ApplicationEventPublisher applicationEventPublisher;
 
-	private volatile TaskExecutor taskExecutor;
+	private volatile MessageChannel errorChannel;
+
+	private volatile Executor taskExecutor;
 
 	private volatile RedisSerializer<?> serializer = new JdkSerializationRedisSerializer();
 
@@ -60,6 +68,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 
 	private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
 
+	private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
+
 	private volatile boolean active;
 
 	private volatile boolean listening;
@@ -79,6 +89,11 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 		this.boundListOperations = template.boundListOps(queueName);
 	}
 
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
 	public void setSerializer(RedisSerializer<?> serializer) {
 		this.serializer = serializer;
 	}
@@ -117,7 +132,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 		this.receiveTimeout = receiveTimeout;
 	}
 
-	public void setTaskExecutor(TaskExecutor taskExecutor) {
+	public void setTaskExecutor(Executor taskExecutor) {
 		this.taskExecutor = taskExecutor;
 	}
 
@@ -127,6 +142,10 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 		this.errorChannel = errorChannel;
 	}
 
+	public void setRecoveryInterval(long recoveryInterval) {
+		this.recoveryInterval = recoveryInterval;
+	}
+
 	@Override
 	protected void onInit() {
 		super.onInit();
@@ -138,7 +157,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 			this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-") + this.getComponentType());
 		}
 		if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
-			MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
+			MessagePublishingErrorHandler errorHandler =
+					new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(this.getBeanFactory()));
 			errorHandler.setDefaultErrorChannel(this.errorChannel);
 			this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler);
 		}
@@ -146,14 +166,24 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 
 	@Override
 	public String getComponentType() {
-		return "int-redis:message-driven-channel-adapter";
+		return "redis:queue-inbound-channel-adapter";
 	}
 
 	@SuppressWarnings("unchecked")
 	private void popMessageAndSend() {
 		Message<Object> message = null;
 
-		byte[] value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
+		byte[] value = null;
+		try {
+			value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
+		}
+		catch (Exception e) {
+			logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval + " milliseconds.", e);
+			this.listening = false;
+			this.sleepBeforeRecoveryAttempt();
+			this.publishException(e);
+			return;
+		}
 
 		if (value != null) {
 			if (this.expectMessage) {
@@ -186,6 +216,32 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 		}
 	}
 
+	/**
+	 * Sleep according to the specified recovery interval.
+	 * Called between recovery attempts.
+	 */
+	private void sleepBeforeRecoveryAttempt() {
+		if (this.recoveryInterval > 0) {
+			try {
+				Thread.sleep(this.recoveryInterval);
+			}
+			catch (InterruptedException e) {
+				logger.debug("Thread interrupted while sleeping the recovery interval");
+			}
+		}
+	}
+
+	private void publishException(Exception e) {
+		if (this.applicationEventPublisher != null) {
+			this.applicationEventPublisher.publishEvent(new RedisExceptionEvent(this, e));
+		}
+		else {
+			if (logger.isDebugEnabled()) {
+				logger.debug("No application event publisher for exception: " + e.getMessage());
+			}
+		}
+	}
+
 	private void restart() {
 		this.taskExecutor.execute(new ListenerTask());
 	}

spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/RedisMetadataStore.java

@@ -11,13 +11,13 @@
  * specific language governing permissions and limitations under the License.
  */
 
-package org.springframework.integration.redis.store.metadata;
+package org.springframework.integration.redis.metadata;
 
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.BoundValueOperations;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.integration.store.metadata.MetadataStore;
+import org.springframework.integration.metadata.MetadataStore;
 import org.springframework.util.Assert;
 
 /**
@@ -65,4 +65,13 @@ public class RedisMetadataStore implements MetadataStore {
 		BoundValueOperations<String, String> ops = this.redisTemplate.boundValueOps(key);
 		return ops.get();
 	}
+
+	@Override
+	public String remove(String key) {
+		Assert.notNull(key, "'key' must not be null.");
+		String value = this.get(key);
+		this.redisTemplate.delete(key);
+		return value;
+	}
+
 }

spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Provides support for Redis-based
+ * {@link org.springframework.integration.metadata.MetadataStore}s.
+ */
+package org.springframework.integration.redis.metadata;

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,9 +17,13 @@
 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.expression.IntegrationEvaluationContextAware;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.support.converter.SimpleMessageConverter;
 import org.springframework.messaging.Message;
@@ -28,22 +32,32 @@ import org.springframework.util.Assert;
 
 /**
  * @author Mark Fisher
+ * @author Artem Bilan
  * @since 2.1
  */
-@SuppressWarnings("rawtypes")
-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) {
@@ -56,28 +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));
 	}
 
-	@SuppressWarnings("unchecked")
-	@Override
-	protected void handleMessageInternal(Message<?> message) throws Exception {
-		String topic = this.determineTopic(message);
-		Object value = this.messageConverter.fromMessage(message, Object.class);
-		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, null);
+
+		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/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java

@@ -43,7 +43,7 @@ public class RedisQueueOutboundChannelAdapter extends AbstractMessageHandler imp
 
 	private final Expression queueNameExpression;
 
-	private EvaluationContext evaluationContext;
+	private volatile EvaluationContext evaluationContext;
 
 	private volatile boolean extractPayload = true;
 
@@ -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

spring-integration-redis/src/main/java/org/springframework/integration/redis/store/metadata/package-info.java

@@ -1,5 +0,0 @@
-/**
- * Provides support for Redis-based
- * {@link org.springframework.integration.store.metadata.MetadataStore}s.
- */
-package org.springframework.integration.redis.store.metadata;

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

@@ -82,10 +82,7 @@
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
 	A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) for executing
-	JMS listener invokers. Default is a SimpleAsyncTaskExecutor in case of a
-	DefaultMessageListenerContainer, using internally managed threads. For a
-	SimpleMessageListenerContainer, listeners will always get invoked within the
-	JMS provider's receive thread by default.
+	Redis listener invokers. Default is a SimpleAsyncTaskExecutor.
 				]]></xsd:documentation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
@@ -177,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"/>
@@ -200,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[
@@ -329,6 +343,146 @@
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="queue-inbound-channel-adapter">
+		<xsd:annotation>
+			<xsd:documentation>
+				Defines a Message Driven Endpoint for listening a Redis queue.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="redisAdapterType">
+					<xsd:attribute name="queue" type="xsd:string" use="required">
+						<xsd:annotation>
+							<xsd:documentation>
+								Redis queue name.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="error-channel" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+						Identifies the channel to which error messages will be sent if a failure occurs in this
+						Endpoint's process. If no "error-channel" reference is provided, this Endpoint will
+						propagate Exceptions to the caller. To completely suppress Exceptions, provide a
+						reference to the "nullChannel" here.
+					]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="serializer" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<xsd:documentation>
+									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.
+								</xsd:documentation>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.data.redis.serializer.RedisSerializer"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="receive-timeout" type="xsd:string" default="1000">
+						<xsd:annotation>
+							<xsd:documentation>
+								Specify the timeout in milliseconds to wait for the result of the
+								'rightPop' operation on Redis queue.
+								Default is 1 second.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="expect-message" type="xsd:string" default="false">
+						<xsd:annotation>
+							<xsd:documentation>
+								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'.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="task-executor" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+								A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) for executing
+								the listening task on the Redis queue. Default is a SimpleAsyncTaskExecutor.
+							]]></xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="java.util.concurrent.Executor"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="queue-outbound-channel-adapter">
+		<xsd:annotation>
+			<xsd:documentation>
+				Defines an outbound Redis Queue Message-sending Channel Adapter.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="redisAdapterType">
+					<xsd:sequence>
+						<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
+					</xsd:sequence>
+					<xsd:attribute name="queue" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+								Specifies the name of the Redis queue.
+								This attribute is mutually exclusive with 'queue-expression' attribute.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="queue-expression" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+								Specifies the expression to determine the name of the Redis queue
+								against the Message at runtime.
+								This attribute is mutually exclusive with 'queue' attribute.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="order" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+								Specifies the order for invocation when this adapter is connected as a
+								subscriber to a SubscribableChannel.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="serializer" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<xsd:documentation>
+									Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer
+								</xsd:documentation>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.data.redis.serializer.RedisSerializer"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="extract-payload" type="xsd:string" default="true">
+						<xsd:annotation>
+							<xsd:documentation>
+								Specifies if the Message payload or the entire (serialized) Message will be send to the Redis queue.
+								Default is 'true'.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:complexType name="redisAdapterType">
 		<xsd:annotation>
 			<xsd:documentation>

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;
@@ -68,6 +70,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/RedisMessageDrivenEndpointParserTests-context.xml

@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
+	   xmlns:task="http://www.springframework.org/schema/task"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
+
+	<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>
+
+	<bean id="customRedisConnectionFactory" parent="redisConnectionFactory"/>
+
+	<int-redis:queue-inbound-channel-adapter id="defaultAdapter" queue="si.test.Int3017.Inbound1"/>
+
+	<int:channel id="sendChannel"/>
+
+	<int-redis:queue-inbound-channel-adapter id="customAdapter"
+											  queue="si.test.Int3017.Inbound2"
+											  channel="sendChannel"
+											  connection-factory="customRedisConnectionFactory"
+											  expect-message="true"
+											  serializer="serializer"
+											  error-channel="errorChannel"
+											  receive-timeout="2000"
+											  task-executor="executor"
+											  auto-startup="false"
+											  phase="100"/>
+
+	<bean id="executor" class="org.springframework.integration.util.ErrorHandlingTaskExecutor">
+		<constructor-arg ref="threadPoolTaskExecutor"/>
+		<constructor-arg value="#{T(org.springframework.scheduling.support.TaskUtils).LOG_AND_SUPPRESS_ERROR_HANDLER}"/>
+	</bean>
+
+	<task:executor id="threadPoolTaskExecutor" pool-size="5"/>
+
+	<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
+
+</beans>

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

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,15 +25,17 @@ 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.messaging.Message;
-import org.springframework.messaging.MessageChannel;
+import org.springframework.expression.Expression;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
-import org.springframework.messaging.support.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.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@@ -59,7 +61,9 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests{
 				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 +78,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/config/RedisQueueOutboundChannelAdapterParserTests-context.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd">
+
+	<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>
+
+	<bean id="customRedisConnectionFactory" parent="redisConnectionFactory"/>
+
+	<int:channel id="sendChannel"/>
+
+	<int-redis:queue-outbound-channel-adapter id="defaultAdapter" channel="sendChannel" queue="foo"/>
+
+	<int-redis:queue-outbound-channel-adapter id="customAdapter" channel="sendChannel"
+											  queue-expression="headers['redis_queue']"
+											  extract-payload="false"
+											  serializer="serializer"
+											  connection-factory="customRedisConnectionFactory"/>
+
+	<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
+
+</beans>

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

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.messaging.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/inbound/RedisQueueMessageDrivenEndpointTests-context.xml

@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
+	   xmlns:task="http://www.springframework.org/schema/task"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
+
+	<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>
+
+	<int:channel id="fromChannel">
+		<int:queue/>
+	</int:channel>
+
+	<int-redis:queue-inbound-channel-adapter queue="si.test.Int3017IntegrationInbound"
+											  channel="fromChannel"
+											  expect-message="true"
+											  serializer="testSerializer"/>
+
+	<bean id="testSerializer" class="org.springframework.integration.redis.util.CustomJsonSerializer"/>
+
+	<int:chain input-channel="symmetricalInputChannel">
+		<int:payload-serializing-transformer/>
+		<int-redis:queue-outbound-channel-adapter queue-expression="headers.redis_queue"/>
+	</int:chain>
+
+	<int-redis:queue-inbound-channel-adapter queue="si.test.Int3017IntegrationSymmetrical"
+											  channel="symmetricalRedisChannel"
+											  serializer=""/>
+
+
+	<int:payload-deserializing-transformer input-channel="symmetricalRedisChannel" output-channel="symmetricalOutputChannel"/>
+
+	<int:channel id="symmetricalOutputChannel">
+		<int:queue/>
+	</int:channel>
+
+</beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java

@@ -18,33 +18,68 @@ package org.springframework.integration.redis.inbound;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Date;
+import java.util.List;
+import java.util.UUID;
 
 import org.hamcrest.Matchers;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
 
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.event.IntegrationEvent;
+import org.springframework.integration.redis.event.RedisExceptionEvent;
 import org.springframework.integration.redis.rules.RedisAvailable;
 import org.springframework.integration.redis.rules.RedisAvailableTests;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ErrorMessage;
+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 +87,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 
 		String queueName = "si.test.redisQueueInboundChannelAdapterTests";
 
-		RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
-
 		RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
-		redisTemplate.setConnectionFactory(connectionFactory);
+		redisTemplate.setConnectionFactory(this.connectionFactory);
 		redisTemplate.setEnableDefaultSerializer(false);
 		redisTemplate.setKeySerializer(new StringRedisSerializer());
 		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
@@ -71,7 +104,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 +120,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 		assertEquals(payload2, receive.getPayload());
 
 		endpoint.stop();
-		this.waitUntilListening(endpoint);
 	}
 
 	@Test
@@ -96,10 +129,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 
 		final String queueName = "si.test.redisQueueInboundChannelAdapterTests2";
 
-		RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
-
 		RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
-		redisTemplate.setConnectionFactory(connectionFactory);
+		redisTemplate.setConnectionFactory(this.connectionFactory);
 		redisTemplate.setEnableDefaultSerializer(false);
 		redisTemplate.setKeySerializer(new StringRedisSerializer());
 		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
@@ -115,7 +146,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);
@@ -137,21 +169,95 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 		assertThat(((Exception) receive.getPayload()).getCause().getMessage(),
 				Matchers.containsString("java.lang.String cannot be cast to org.springframework.messaging.Message"));
 
-
 		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<String, String> 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<UUID> 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());
+	}
+
+	@Test
+	@RedisAvailable
+	@SuppressWarnings("unchecked")
+	public void testInt3196Recovery() throws Exception {
+		String queueName = "test.si.Int3196Recovery";
+		QueueChannel channel = new QueueChannel();
+
+		final List<ApplicationEvent> exceptionEvents = new ArrayList<ApplicationEvent>();
+
+		RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
+		endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
+		endpoint.setApplicationEventPublisher(new ApplicationEventPublisher() {
+
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				exceptionEvents.add(event);
 			}
+		});
+		endpoint.setOutputChannel(channel);
+		endpoint.setReceiveTimeout(100);
+		endpoint.setRecoveryInterval(200);
+		endpoint.afterPropertiesSet();
+		endpoint.start();
+
+		((DisposableBean) this.connectionFactory).destroy();
+
+		Thread.sleep(300);
+
+		assertThat(exceptionEvents.size(), Matchers.greaterThan(0));
+		for (ApplicationEvent exceptionEvent : exceptionEvents) {
+			assertThat(exceptionEvent, Matchers.instanceOf(RedisExceptionEvent.class));
+			assertSame(endpoint, exceptionEvent.getSource());
+			assertThat(((IntegrationEvent) exceptionEvent).getCause().getClass(),
+					Matchers.isIn(Arrays.<Class<? extends Throwable>> asList(RedisSystemException.class, RedisConnectionFailureException.class)));
 		}
 
+		((InitializingBean) this.connectionFactory).afterPropertiesSet();
+
+		RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
+		redisTemplate.setConnectionFactory(this.getConnectionFactoryForTest());
+		redisTemplate.setEnableDefaultSerializer(false);
+		redisTemplate.setKeySerializer(new StringRedisSerializer());
+		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
+		redisTemplate.afterPropertiesSet();
+
+		String payload = "testing";
+
+		redisTemplate.boundListOps(queueName).leftPush(payload);
+
+		Message<?> receive = channel.receive(1000);
+		assertNotNull(receive);
+		assertEquals(payload, receive.getPayload());
+
+		endpoint.stop();
 	}
 
 }

spring-integration-redis/src/test/java/org/springframework/integration/redis/metadata/RedisMetadataStoreTests.java

@@ -13,7 +13,7 @@
  *     See the License for the specific language governing permissions and
  *     limitations under the License.
  */
-package org.springframework.integration.redis.store.metadata;
+package org.springframework.integration.redis.metadata;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
@@ -29,6 +29,7 @@ import org.springframework.integration.redis.rules.RedisAvailableTests;
 
 /**
  * @author Gunnar Hillert
+ * @author Artem Bilan
  * @since 3.0
  *
  */
@@ -143,4 +144,20 @@ public class RedisMetadataStoreTests extends RedisAvailableTests {
 
 		fail("Expected an IllegalArgumentException to be thrown.");
 	}
+
+	@Test
+	@RedisAvailable
+	public void testRemoveFromMetadataStore(){
+		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMetadataStore metadataStore = new RedisMetadataStore(jcf);
+
+		String testKey = "RedisMetadataStoreTests-Remove";
+		String testValue = "Integration";
+
+		metadataStore.put(testKey, testValue);
+
+		assertEquals(testValue, metadataStore.remove(testKey));
+		assertNull(metadataStore.remove(testKey));
+	}
+
 }

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/outbound/RedisQueueOutboundChannelAdapterTests-context.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd">
+
+	<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>
+
+	<int:chain input-channel="toRedisQueueChannel">
+		<int-redis:queue-outbound-channel-adapter queue-expression="payload"
+												  extract-payload="false"
+												  serializer="testSerializer"/>
+	</int:chain>
+
+	<bean id="testSerializer" class="org.springframework.integration.redis.util.CustomJsonSerializer"/>
+
+</beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java

@@ -24,41 +24,58 @@ 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;
 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.mapping.InboundMessageMapper;
 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.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
+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<String, ?> 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<String, ?> redisTemplate2 = new RedisTemplate<String, Object>();
-		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<String> message = MessageBuilder.withPayload("testing").build();
 		handler.handleMessage(message);
 
 		RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
-		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>(Object.class));
 
 		RedisTemplate<String, ?> redisTemplate = new StringRedisTemplate();
-		redisTemplate.setConnectionFactory(connectionFactory);
+		redisTemplate.setConnectionFactory(this.connectionFactory);
 		redisTemplate.afterPropertiesSet();
 
 		handler.handleMessage(new GenericMessage<Object>(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<Object> message = new GenericMessage<Object>(queueName);
+		this.sendChannel.send(message);
+
+		RedisTemplate<String, String> redisTemplate = new StringRedisTemplate();
+		redisTemplate.setConnectionFactory(this.connectionFactory);
+		redisTemplate.afterPropertiesSet();
+
+		String result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS);
+		assertNotNull(result);
+		InboundMessageMapper<String> mapper = new JsonInboundMessageMapper(String.class, new Jackson2JsonMessageParser());
+		Message<?> resultMessage = mapper.toMessage(result);
+		assertEquals(message.getPayload(), resultMessage.getPayload());
+	}
+
 }

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

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<Message<?>> {
+
+	private final ObjectMapper objectMapper = new ObjectMapper();
+
+	private final InboundMessageMapper<String> 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);
+		}
+	}
+
+}