INT-3341: Add Redis Queue Gateways

Add RedisQueueOutboundGateway and RedisQueueInboundGateway

JIRA: https://jira.spring.io/browse/INT-3341

add test

fix format issue

change expectmessage to extractpayload, fix test potential bug

fix copyright year

fix format and naming issue

fix as Artem's comments

fix as Artem's comments

change boolean condition judgement

INT-3341: Polishing

Minor Doc Polishing

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractInboundGatewayParser.java

@@ -67,10 +67,6 @@ public abstract class AbstractInboundGatewayParser extends AbstractSimpleBeanDef
 		if (StringUtils.hasText(errorChannel)) {
 			builder.addPropertyReference("errorChannel", errorChannel);
 		}
-		String autoStartup = element.getAttribute("auto-startup");
-		if (StringUtils.hasText(autoStartup)) {
-			builder.addPropertyValue("autoStartup", autoStartup);
-		}
 		this.doPostProcess(builder, element);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java

@@ -232,7 +232,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 		}
 	}
 
-	private MessageChannel getRequestChannel() {
+	protected MessageChannel getRequestChannel() {
 		if (this.requestChannelName != null) {
 			synchronized (this) {
 				if (this.requestChannelName != null) {
@@ -252,7 +252,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 		return this.requestChannel;
 	}
 
-	private MessageChannel getReplyChannel() {
+	protected MessageChannel getReplyChannel() {
 		if (this.replyChannelName != null) {
 			synchronized (this) {
 				if (this.replyChannelName != null) {
@@ -272,7 +272,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 		return this.replyChannel;
 	}
 
-	private MessageChannel getErrorChannel() {
+	protected MessageChannel getErrorChannel() {
 		if (this.errorChannelName != null) {
 			synchronized (this) {
 				if (this.errorChannelName != null) {

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

@@ -35,5 +35,7 @@ public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 		registerBeanDefinitionParser("queue-inbound-channel-adapter", new RedisQueueInboundChannelAdapterParser());
 		registerBeanDefinitionParser("queue-outbound-channel-adapter", new RedisQueueOutboundChannelAdapterParser());
 		registerBeanDefinitionParser("outbound-gateway", new RedisOutboundGatewayParser());
+		registerBeanDefinitionParser("queue-inbound-gateway", new RedisQueueInboundGatewayParser());
+		registerBeanDefinitionParser("queue-outbound-gateway", new RedisQueueOutboundGatewayParser());
 	}
 }

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

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2014 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.BeanDefinitionBuilder;
+import org.springframework.integration.config.xml.AbstractInboundGatewayParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.redis.inbound.RedisQueueInboundGateway;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <queue-inbound-gateway> element of the 'redis' namespace.
+ *
+ * @author David Liu
+ * @author Artem Bilan
+ * @since 4.1
+ */
+public class RedisQueueInboundGatewayParser extends AbstractInboundGatewayParser {
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return RedisQueueInboundGateway.class;
+	}
+
+	@Override
+	protected boolean isEligibleAttribute(String attributeName) {
+		return !attributeName.equals("queue")
+				&& !attributeName.equals("connection-factory")
+				&& !attributeName.equals("serializer")
+				&& !attributeName.equals("task-executor")
+				&& super.isEligibleAttribute(attributeName);
+	}
+
+	@Override
+	protected void doPostProcess(BeanDefinitionBuilder builder, Element element) {
+		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");
+	}
+
+}

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

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2014 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.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.redis.outbound.RedisQueueOutboundGateway;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <int-redis:queue-outbound-channel-adapter> element.
+ *
+ * @author Artem Bilan
+ * @author David Liu
+ * @since 3.0
+ */
+public class RedisQueueOutboundGatewayParser extends AbstractConsumerEndpointParser {
+
+	@Override
+	protected String getInputChannelAttributeName() {
+		return "request-channel";
+	}
+
+	@Override
+	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisQueueOutboundGateway.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, "reply-channel", "outputChannel");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "receiveTimeout");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply");
+		return builder;
+	}
+
+}

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

@@ -0,0 +1,351 @@
+/*
+ * Copyright 2014 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.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.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.BoundListOperations;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.integration.channel.MessagePublishingErrorHandler;
+import org.springframework.integration.gateway.MessagingGatewaySupport;
+import org.springframework.integration.redis.event.RedisExceptionEvent;
+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;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
+import org.springframework.util.Assert;
+
+/**
+ * @author David Liu
+ * @author Artem Bilan
+ * @since 4.1
+ */
+@ManagedResource
+public class RedisQueueInboundGateway extends MessagingGatewaySupport implements ApplicationEventPublisherAware {
+
+	private static final String QUEUE_NAME_SUFFIX = ".reply";
+
+	private static final RedisSerializer<String> stringSerializer = new StringRedisSerializer();
+
+	public static final long DEFAULT_RECEIVE_TIMEOUT = 5000;
+
+	public static final long DEFAULT_RECOVERY_INTERVAL = 5000;
+
+	private final RedisTemplate<String, byte[]> template;
+
+	private final BoundListOperations<String, byte[]> boundListOperations;
+
+	private volatile ApplicationEventPublisher applicationEventPublisher;
+
+	private volatile boolean serializerExplicitlySet;
+
+	private volatile Executor taskExecutor;
+
+	private volatile RedisSerializer<?> serializer = new StringRedisSerializer();
+
+	private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
+
+	private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
+
+	private volatile long stopTimeout = DEFAULT_RECEIVE_TIMEOUT;
+
+	private volatile boolean active;
+
+	private volatile boolean listening;
+
+	private volatile boolean extractPayload = true;
+
+	/**
+	 * @param queueName         Must not be an empty String
+	 * @param connectionFactory Must not be null
+	 */
+	public RedisQueueInboundGateway(String queueName, RedisConnectionFactory connectionFactory) {
+		Assert.hasText(queueName, "'queueName' is required");
+		Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
+		this.template = new RedisTemplate<String, byte[]>();
+		this.template.setConnectionFactory(connectionFactory);
+		this.template.setEnableDefaultSerializer(false);
+		this.template.setKeySerializer(new StringRedisSerializer());
+		this.template.afterPropertiesSet();
+		this.boundListOperations = this.template.boundListOps(queueName);
+	}
+
+	public void setExtractPayload(boolean extractPayload) {
+		this.extractPayload = extractPayload;
+	}
+
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
+	public void setSerializer(RedisSerializer<?> serializer) {
+		this.serializer = serializer;
+		this.serializerExplicitlySet = true;
+	}
+
+
+	/**
+	 * This timeout (milliseconds) is used when retrieving elements from the queue
+	 * specified by {@link #boundListOperations}.
+	 * <p> If the queue does contain elements, the data is retrieved immediately. However,
+	 * if the queue is empty, the Redis connection is blocked until either an element
+	 * can be retrieved from the queue or until the specified timeout passes.
+	 * <p> A timeout of zero can be used to block indefinitely. If not set explicitly
+	 * the timeout value will default to {@code 1000}
+	 * <p> See also: http://redis.io/commands/brpop
+	 * @param receiveTimeout Must be non-negative. Specified in milliseconds.
+	 */
+	public void setReceiveTimeout(long receiveTimeout) {
+		Assert.isTrue(receiveTimeout > 0, "'receiveTimeout' must be > 0.");
+		this.receiveTimeout = receiveTimeout;
+	}
+
+	/**
+	 * @param stopTimeout the timeout to block {@link #doStop()} until the last message
+	 * will be processed or this timeout is reached. Should be less then or equal to
+	 * {@link #receiveTimeout}
+	 */
+	public void setStopTimeout(long stopTimeout) {
+		this.stopTimeout = stopTimeout;
+	}
+
+	public void setTaskExecutor(Executor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
+	public void setRecoveryInterval(long recoveryInterval) {
+		this.recoveryInterval = recoveryInterval;
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		if (!this.extractPayload) {
+			Assert.notNull(this.serializer, "'serializer' has to be provided where 'extractPayload == false'.");
+		}
+		if (this.taskExecutor == null) {
+			String beanName = this.getComponentName();
+			this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-")
+					+ this.getComponentType());
+		}
+		if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && this.getBeanFactory() != null) {
+			MessagePublishingErrorHandler errorHandler =
+					new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(getBeanFactory()));
+			errorHandler.setDefaultErrorChannel(getErrorChannel());
+			this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler);
+		}
+	}
+
+	@Override
+	public String getComponentType() {
+		return "redis:queue-inbound-gateway";
+	}
+
+	private void handlePopException(Exception e) {
+		this.listening = false;
+		if (this.active) {
+			logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval
+					+ " milliseconds.", e);
+			this.publishException(e);
+			this.sleepBeforeRecoveryAttempt();
+		}
+		else {
+			logger.debug("Failed to execute listening task. " + e.getClass() + ": " + e.getMessage());
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void receiveAndReply() {
+		byte[] value = null;
+		try {
+			value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
+		}
+		catch (Exception e) {
+			handlePopException(e);
+			return;
+		}
+		String uuid = null;
+		if (value != null) {
+			uuid = stringSerializer.deserialize(value);
+			try {
+				value = this.template.boundListOps(uuid).rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
+			}
+			catch (Exception e) {
+				handlePopException(e);
+				return;
+			}
+			Message<Object> requestMessage = null;
+			if (value != null) {
+				if (this.extractPayload) {
+					Object payload = value;
+					if (this.serializer != null) {
+						payload = this.serializer.deserialize(value);
+					}
+					requestMessage = this.getMessageBuilderFactory().withPayload(payload).build();
+				}
+				else {
+					try {
+						requestMessage = (Message<Object>) this.serializer.deserialize(value);
+					}
+					catch (Exception e) {
+						throw new MessagingException("Deserialization of Message failed.", e);
+					}
+				}
+				Message<?> replyMessage = this.sendAndReceiveMessage(requestMessage);
+				if (replyMessage != null) {
+					if (this.extractPayload) {
+						if (!(replyMessage.getPayload() instanceof byte[])) {
+							if (replyMessage.getPayload() instanceof String && !serializerExplicitlySet) {
+								value = stringSerializer.serialize((String) replyMessage.getPayload());
+							}
+							else {
+								value = ((RedisSerializer<Object>) serializer).serialize(replyMessage.getPayload());
+							}
+						}
+						else {
+							value = (byte[]) replyMessage.getPayload();
+						}
+					}
+					else {
+						if (this.serializer != null) {
+							value = ((RedisSerializer<Object>) serializer).serialize(replyMessage);
+						}
+					}
+					this.template.boundListOps(uuid + QUEUE_NAME_SUFFIX).leftPush(value);
+				}
+			}
+		}
+	}
+
+	@Override
+	protected void doStart() {
+		if (!this.active) {
+			this.active = true;
+			this.restart();
+		}
+	}
+
+	/**
+	 * 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");
+				Thread.currentThread().interrupt();
+			}
+		}
+	}
+
+	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());
+	}
+
+	@Override
+	protected void doStop() {
+		try {
+			this.active = false;
+			this.lifecycleCondition.await(Math.min(this.stopTimeout, this.receiveTimeout), TimeUnit.MICROSECONDS);
+		}
+		catch (InterruptedException e) {
+			logger.debug("Thread interrupted while stopping the endpoint");
+			Thread.currentThread().interrupt();
+		}
+		finally {
+			this.listening = false;
+		}
+	}
+
+	public boolean isListening() {
+		return listening;
+	}
+
+	/**
+	 * Returns the size of the Queue specified by {@link #boundListOperations}. The queue is
+	 * represented by a Redis list. If the queue does not exist <code>0</code>
+	 * is returned. See also http://redis.io/commands/llen
+	 * @return Size of the queue. Never negative.
+	 */
+	@ManagedMetric
+	public long getQueueSize() {
+		return this.boundListOperations.size();
+	}
+
+	/**
+	 * Clear the Redis Queue specified by {@link #boundListOperations}.
+	 */
+	@ManagedOperation
+	public void clearQueue() {
+		this.boundListOperations.getOperations().delete(this.boundListOperations.getKey());
+	}
+
+
+	private class ListenerTask implements Runnable {
+
+		@Override
+		public void run() {
+			try {
+				while (RedisQueueInboundGateway.this.active) {
+					RedisQueueInboundGateway.this.listening = true;
+					RedisQueueInboundGateway.this.receiveAndReply();
+				}
+			}
+			finally {
+				if (RedisQueueInboundGateway.this.active) {
+					RedisQueueInboundGateway.this.restart();
+				}
+				else {
+					RedisQueueInboundGateway.this.lifecycleLock.lock();
+					try {
+						RedisQueueInboundGateway.this.lifecycleCondition.signalAll();
+					}
+					finally {
+						RedisQueueInboundGateway.this.lifecycleLock.unlock();
+					}
+				}
+			}
+		}
+
+	}
+
+}

spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2014 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.outbound;
+
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.BoundListOperations;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.messaging.Message;
+import org.springframework.util.AlternativeJdkIdGenerator;
+import org.springframework.util.Assert;
+import org.springframework.util.IdGenerator;
+
+/**
+ * @author David Liu
+ * @author Artem Bilan
+ * @since 4.1
+ */
+public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHandler {
+
+	private static final String QUEUE_NAME_SUFFIX = ".reply";
+
+	private static final int TIMEOUT = 1000;
+
+	private static final IdGenerator defaultIdGenerator = new AlternativeJdkIdGenerator();
+
+	private final static RedisSerializer<String> stringSerializer = new StringRedisSerializer();
+
+	private final RedisTemplate<String, Object> template;
+
+	private final BoundListOperations<String, Object> boundListOps;
+
+	private volatile boolean extractPayload = true;
+
+	private volatile RedisSerializer<?> serializer = new JdkSerializationRedisSerializer();
+
+	private volatile boolean serializerExplicitlySet;
+
+	private volatile int receiveTimeout = TIMEOUT;
+
+	public RedisQueueOutboundGateway(String queueName, RedisConnectionFactory connectionFactory) {
+		Assert.hasText(queueName, "'queueName' is required");
+		Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
+		this.template = new RedisTemplate<String, Object>();
+		this.template.setConnectionFactory(connectionFactory);
+		this.template.setEnableDefaultSerializer(false);
+		this.template.setKeySerializer(new StringRedisSerializer());
+		this.template.afterPropertiesSet();
+		this.boundListOps = this.template.boundListOps(queueName);
+	}
+
+	public void setReceiveTimeout(int timeout) {
+		this.receiveTimeout = timeout;
+	}
+
+	public void setExtractPayload(boolean extractPayload) {
+		this.extractPayload = extractPayload;
+	}
+
+	public void setSerializer(RedisSerializer<?> serializer) {
+		Assert.notNull(serializer, "'serializer' must not be null");
+		this.serializer = serializer;
+		this.serializerExplicitlySet = true;
+	}
+
+	@Override
+	public String getComponentType() {
+		return "redis:queue-outbound-gatewway";
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	protected Object handleRequestMessage(Message<?> message) {
+		Object value = message;
+
+		if (this.extractPayload) {
+			value = message.getPayload();
+		}
+		if (!(value instanceof byte[])) {
+			if (value instanceof String && !serializerExplicitlySet) {
+				value = stringSerializer.serialize((String) value);
+			}
+			else {
+				value = ((RedisSerializer<Object>) serializer).serialize(value);
+			}
+		}
+		String uuid = defaultIdGenerator.generateId().toString();
+
+		byte[] uuidByte = uuid.getBytes();
+		this.boundListOps.leftPush(uuidByte);
+		this.template.boundListOps(uuid).leftPush(value);
+
+		BoundListOperations<String, Object> boundListOperations = template.boundListOps(uuid + QUEUE_NAME_SUFFIX);
+		byte[] reply = (byte[]) boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
+		if(reply != null && reply.length > 0) {
+			Object replyMessage = this.serializer.deserialize(reply);
+			if (replyMessage == null) {
+				return null;
+			}
+			if (this.extractPayload) {
+				return this.getMessageBuilderFactory().withPayload(replyMessage).build();
+			}
+			else {
+				return replyMessage;
+			}
+		}
+		return null;
+	}
+
+}

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

@@ -408,6 +408,131 @@
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="queue-inbound-gateway">
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures a gateway that adapts incoming Redis values to Spring Integration Messages
+				and returns a reply.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:attribute name="id" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Unique ID for this gateway.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="queue" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						Redis 'control' queue (list). Requests contain a conversation UUID which represents the
+						key of an incoming Redis List to 'pop' a value. The sender sends the actual
+						request to that list and waits for the reply on a list with key '<UUID>.reply'
+						to which this gateway will send the reply.
+					]]></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="request-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Message Channel to which Messages should be sent in order to have them converted and published.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="request-timeout" type="xsd:string"/>
+			<xsd:attribute name="reply-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Message Channel to which replies for this gateway should be sent.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="reply-timeout" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						Allows you to specify how long this gateway will wait for
+						the reply message to be sent successfully to the reply channel
+						before throwing an exception. This attribute only applies when the
+						channel might block, for example when using a bounded queue channel that
+						is currently full.
+
+						Also, keep in mind that when sending to a DirectChannel, the
+						invocation will occur in the sender's thread. Therefore,
+						the failing of the send operation may be caused by other
+						components further downstream.
+
+						The "reply-timeout" attribute maps to the "sendTimeout" property of the
+						underlying 'MessagingTemplate' instance (org.springframework.integration.core.MessagingTemplate).
+						The attribute will default, if not specified, to '-1', meaning that
+						by default, the Gateway will wait indefinitely. The value is
+						specified in milliseconds.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="connection-factory" use="optional" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Reference to the Redis ConnectionFactory to be used by this component.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</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
+						Redis listener invokers. 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:attribute name="extract-payload" type="xsd:string" default="true">
+				<xsd:annotation>
+					<xsd:documentation>
+						When true, specifies that gateway deals only with the payload of the message.
+						Otherwise it expects the Redis 'value' to be a serialized 'Message'.
+						This option is applied to both the request and reply operations.
+						Defaults to 'true'.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup" />
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:element name="queue-outbound-channel-adapter">
 		<xsd:annotation>
 			<xsd:documentation>
@@ -472,6 +597,123 @@
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="queue-outbound-gateway">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+				Configures a gateway that pushes a conversation UUID to the the 'queue', then pushes the value to the
+				Redis List with that UUID as a key and receives a reply from the Redis List with
+				'<UUID>.reply' as its key. A new UUID is used for each interaction.
+			]]></xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:choice minOccurs="0" maxOccurs="3">
+				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
+				<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
+				minOccurs="0" maxOccurs="1" />
+			</xsd:choice>
+			<xsd:attribute name="id" type="xsd:string"/>
+			<xsd:attribute name="queue" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						Redis queue (list) to which send the conversation UUID representing the key of the list
+						to which the actual request is pushed. The is the 'control' queue used by
+						inbound gateways. This gateway will wait for the reply from a list with a key
+						'<UUID>.reply'.
+					]]></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 used as the Redis 'value'.
+						This option is applied for both request and reply operations.
+						Default is 'true'.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="reply-timeout" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						Allows you to specify how long this gateway will wait for
+						the reply message to be sent successfully to the reply channel
+						before throwing an exception. This attribute only applies when the
+						channel might block, for example when using a bounded queue channel that
+						is currently full.
+
+						Also, keep in mind that when sending to a DirectChannel, the
+						invocation will occur in the sender's thread. Therefore,
+						the failing of the send operation may be caused by other
+						components further downstream.
+
+						The "reply-timeout" attribute maps to the "sendTimeout" property of the
+						underlying 'MessagingTemplate' instance (org.springframework.integration.core.MessagingTemplate).
+
+						The attribute will default, if not specified, to '-1', meaning that
+						by default, the Gateway will wait indefinitely. The value is
+						specified in milliseconds.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="requires-reply" type="xsd:string" use="optional" default="true">
+				<xsd:annotation>
+					<xsd:documentation>
+						Specify whether this outbound gateway must return a non-null value. This value is
+						'true' by default, and a ReplyRequiredException will be thrown when
+						the underlying service returns a null value.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="reply-channel" use="optional" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Message Channel where reply Messages will be sent.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="request-channel" use="optional" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Message Channel where request Messages will be expected.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="connection-factory" use="optional" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Reference to the Redis ConnectionFactory to be used by this component.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:element name="outbound-gateway">
 		<xsd:annotation>
 			<xsd:documentation>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests-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"
+	   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.jedis.JedisConnectionFactory">
+		<property name="port" value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
+	</bean>
+
+	<int:channel id="sendChannel"/>
+
+	<int:channel id="outputChannel">
+		<int:queue/>
+	</int:channel>
+
+	<int:channel id="requestChannel"/>
+
+	<int-redis:queue-outbound-gateway id="outboundGateway"
+									  request-channel="sendChannel"
+									  queue="foo"
+									  reply-timeout="2000"
+									  requires-reply="true"
+									  reply-channel="outputChannel"
+									  serializer="serializer"/>
+
+
+	<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
+
+	<int-redis:queue-inbound-gateway id="inboundGateway"
+									 queue="foo"
+									 request-channel="requestChannel"
+									 serializer="serializer"
+									 reply-timeout="20001"
+									 request-timeout="20000"/>
+
+	<int:service-activator input-channel="requestChannel" expression="payload.toUpperCase()"/>
+
+</beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2014 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.assertNotNull;
+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.serializer.JdkSerializationRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.redis.inbound.RedisQueueInboundGateway;
+import org.springframework.integration.redis.outbound.RedisQueueOutboundGateway;
+import org.springframework.integration.redis.rules.RedisAvailable;
+import org.springframework.integration.redis.rules.RedisAvailableTests;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author David Liu
+ * @since 4.1
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests {
+
+	@Autowired
+	@Qualifier("sendChannel")
+	private DirectChannel sendChannel;
+
+	@Autowired
+	@Qualifier("outputChannel")
+	private QueueChannel outputChannel;
+
+	@Autowired
+	private RedisQueueInboundGateway inboundGateway;
+
+	@Autowired
+	private RedisQueueOutboundGateway outboundGateway;
+
+	@Test
+	@RedisAvailable
+	public void testRequestWithReply() throws Exception {
+		this.sendChannel.send(new GenericMessage<String>("test1"));
+		Message<?> receive = this.outputChannel.receive(10000);
+		assertNotNull(receive);
+		assertEquals("test1".toUpperCase(), receive.getPayload());
+	}
+
+	@Test
+	@RedisAvailable
+	public void testInboundGatewayStop() throws Exception {
+		this.inboundGateway.stop();
+		try {
+			this.sendChannel.send(new GenericMessage<String>("test1"));
+		}
+		catch(Exception e) {
+			assertTrue(e.getMessage().contains("No reply produced"));
+		}
+		finally {
+			this.inboundGateway.start();
+		}
+	}
+
+	@Test
+	@RedisAvailable
+	public void testNullSerializer() throws Exception {
+		this.inboundGateway.setSerializer(null);
+		try {
+			this.sendChannel.send(new GenericMessage<String>("test1"));
+		}
+		catch(Exception e) {
+			assertTrue(e.getMessage().contains("No reply produced"));
+		}
+		finally {
+			this.inboundGateway.setSerializer(new StringRedisSerializer());
+		}
+	}
+
+	@Test
+	@RedisAvailable
+	public void testRequestReplyWithMessage() throws Exception {
+		this.inboundGateway.setSerializer(new JdkSerializationRedisSerializer());
+		this.inboundGateway.setExtractPayload(false);
+		this.outboundGateway.setSerializer(new JdkSerializationRedisSerializer());
+		this.outboundGateway.setExtractPayload(false);
+		this.sendChannel.send(new GenericMessage<String>("test1"));
+		Message<?> receive = this.outputChannel.receive(10000);
+		assertNotNull(receive);
+		assertEquals("test1".toUpperCase(), receive.getPayload());
+		this.inboundGateway.setSerializer(new StringRedisSerializer());
+		this.inboundGateway.setExtractPayload(true);
+		this.outboundGateway.setSerializer(new StringRedisSerializer());
+		this.outboundGateway.setExtractPayload(true);
+	}
+
+}

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

@@ -0,0 +1,46 @@
+<?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/task http://www.springframework.org/schema/task/spring-task.xsd
+		http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd">
+
+	<int-redis:queue-inbound-gateway id="inboundGateway"
+		request-channel="requestChannel"
+		connection-factory="redisConnectionFactory"
+		reply-channel="receiveChannel"
+		request-timeout="3000"
+		reply-timeout="2000"
+		queue="si.test.queue"
+		task-executor="executor"
+		serializer="serializer"
+		auto-startup="false"
+		extract-payload="false"
+		phase="3"/>
+
+	<int:channel id="si.test.queue">
+		<int:queue/>
+	</int:channel>
+
+	<int:channel id="requestChannel">
+		<int:queue/>
+	</int:channel>
+
+	<int:channel id="receiveChannel">
+		<int:queue/>
+	</int:channel>
+
+	<bean id="redisConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
+		</constructor-arg>
+	</bean>
+
+	<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
+
+	<task:executor id="executor" pool-size="10"/>
+
+</beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2014 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.serializer.RedisSerializer;
+import org.springframework.integration.redis.inbound.RedisQueueInboundGateway;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import reactor.util.Assert;
+
+/**
+ * @author David Liu
+ * @author Artem Bilan
+ * since 4.1
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class RedisQueueInboundGatewayParserTests {
+
+	@Autowired
+	@Qualifier("inboundGateway")
+	private RedisQueueInboundGateway defaultGateway;
+
+	@Autowired
+	@Qualifier("receiveChannel")
+	private MessageChannel receiveChannel;
+
+	@Autowired
+	@Qualifier("requestChannel")
+	private MessageChannel requestChannel;
+
+	@Autowired
+	private RedisSerializer<?> serializer;
+
+	@Test
+	public void testDefaultConfig() throws Exception {
+		assertFalse(TestUtils.getPropertyValue(this.defaultGateway, "extractPayload", Boolean.class));
+		assertSame(this.serializer, TestUtils.getPropertyValue(this.defaultGateway, "serializer"));
+		assertTrue(TestUtils.getPropertyValue(this.defaultGateway, "serializerExplicitlySet", Boolean.class));
+		assertSame(this.receiveChannel, TestUtils.getPropertyValue(this.defaultGateway, "replyChannel"));
+		assertSame(this.requestChannel, TestUtils.getPropertyValue(this.defaultGateway, "requestChannel"));
+		assertEquals(2000L, TestUtils.getPropertyValue(this.defaultGateway, "replyTimeout"));
+		Assert.notNull(TestUtils.getPropertyValue(this.defaultGateway, "taskExecutor"));
+		assertFalse(TestUtils.getPropertyValue(this.defaultGateway, "autoStartup", Boolean.class));
+		assertEquals(3, TestUtils.getPropertyValue(this.defaultGateway, "phase"));
+	}
+
+}

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

@@ -0,0 +1,49 @@
+<?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">
+
+	<int:channel id="sendChannel"/>
+
+	<int-redis:queue-outbound-gateway id="outboundGateway"
+										request-channel="requestChannel"
+										connection-factory="redisConnectionFactory"
+										reply-channel="receiveChannel"
+										requires-reply="false"
+										reply-timeout="2000"
+										order="2"
+										queue="si.test.queue"
+										extract-payload="false"
+										serializer="serializer"
+										auto-startup="false"
+										phase="3">
+		<int-redis:request-handler-advice-chain>
+			<int:retry-advice/>
+		</int-redis:request-handler-advice-chain>
+		<int:poller fixed-delay="1000" max-messages-per-poll="1" />
+	</int-redis:queue-outbound-gateway>
+
+	<int:channel id="si.test.queue">
+		<int:queue/>
+	</int:channel>
+
+	<int:channel id="requestChannel">
+		<int:queue/>
+	</int:channel>
+
+	<int:channel id="receiveChannel">
+		<int:queue/>
+	</int:channel>
+
+	<bean id="redisConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
+		</constructor-arg>
+	</bean>
+
+	<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
+
+</beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParserTests.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2014 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.context.ApplicationContext;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.integration.endpoint.PollingConsumer;
+import org.springframework.integration.redis.outbound.RedisQueueOutboundGateway;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author David Liu
+ * since 4.1
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class RedisQueueOutboundGatewayParserTests {
+
+	@Autowired
+	private RedisConnectionFactory connectionFactory;
+
+	@Autowired
+	@Qualifier("outboundGateway")
+	private PollingConsumer consumer;
+
+	@Autowired
+	@Qualifier("outboundGateway.handler")
+	private RedisQueueOutboundGateway defaultGateway;
+
+	@Autowired
+	@Qualifier("receiveChannel")
+	private MessageChannel receiveChannel;
+
+	@Autowired
+	@Qualifier("requestChannel")
+	private MessageChannel requestChannel;
+
+	@Autowired
+	private RedisSerializer<?> serializer;
+
+	@Autowired
+	private ApplicationContext context;
+
+	@Test
+	public void testDefaultConfig() throws Exception {
+		assertFalse(TestUtils.getPropertyValue(this.defaultGateway, "extractPayload", Boolean.class));
+		assertSame(this.serializer, TestUtils.getPropertyValue(this.defaultGateway, "serializer"));
+		assertTrue(TestUtils.getPropertyValue(this.defaultGateway, "serializerExplicitlySet", Boolean.class));
+		assertEquals(2, TestUtils.getPropertyValue(this.defaultGateway, "order"));
+		assertSame(this.receiveChannel, TestUtils.getPropertyValue(this.defaultGateway, "outputChannel"));
+		assertSame(this.requestChannel, TestUtils.getPropertyValue(this.consumer, "inputChannel"));
+		assertFalse(TestUtils.getPropertyValue(this.defaultGateway, "requiresReply", Boolean.class));
+		assertEquals(2000, TestUtils.getPropertyValue(this.defaultGateway, "receiveTimeout"));
+		assertFalse(TestUtils.getPropertyValue(this.consumer, "autoStartup", Boolean.class));
+		assertEquals(3, TestUtils.getPropertyValue(this.consumer, "phase"));
+	}
+
+}

src/reference/docbook/endpoint-summary.xml

@@ -143,7 +143,9 @@
 						   <xref linkend="redis-queue-outbound-channel-adapter" /> and
 						   <xref linkend="redis-store-outbound-channel-adapter" /></entry>
 					<entry>N</entry>
-					<entry><xref linkend="redis-outbound-gateway" /></entry>
+					<entry><xref linkend="redis-outbound-gateway" /> and
+						<xref linkend="redis-queue-outbound-gateway" /> and
+						<xref linkend="redis-queue-inbound-gateway" /></entry>
 				</row>
 				<row>
 					<entry><emphasis role="bold">Resource</emphasis></entry>

src/reference/docbook/redis.xml

@@ -699,7 +699,7 @@ the serialization of values, you may want to consider providing your own
 				<callout arearefs="redis-o-g-requires-reply">
 					<para>
 						Specify whether this outbound gateway must return a non-null value. This value is
-						<code>false</code> by default, otherwise a ReplyRequiredException will be thrown when
+						<code>true</code> by default. A ReplyRequiredException will be thrown when
 						the Redis returns a <code>null</code> value.
 					</para>
 				</callout>
@@ -775,6 +775,165 @@ the serialization of values, you may want to consider providing your own
 		  <ulink url="http://redis.io/commands">Redis Specification</ulink>.
 		</para>
 	</section>
+	<section id="redis-queue-outbound-gateway">
+	<title>Redis Queue Outbound Gateway</title>
+		<para>
+			Since <emphasis>Spring Integration 4.1</emphasis>, the Redis Queue Outbound Gateway is available to
+			perform request and reply scenarios. It pushes a <emphasis>conversation</emphasis> <code>UUID</code> to the
+			provided <code>queue</code>, then pushes the value to a Redis List with that <code>UUID</code> as its
+			key and waits for the reply from a Redis List with a key of <code>UUID + '.reply'</code>. A different
+			UUID is used for each interaction.
+			<programlisting language="xml"><![CDATA[<int-redis:queue-outbound-gateway
+		request-channel="" ]]><co id="redis-q-o-g-request-channel"/><![CDATA[
+		reply-channel="" ]]><co id="redis-q-o-g-reply-channel"/><![CDATA[
+		requires-reply="" ]]><co id="redis-q-o-g-requires-reply"/><![CDATA[
+		reply-timeout="" ]]><co id="redis-q-o-g-reply-timeout"/><![CDATA[
+		connection-factory="" ]]><co id="redis-q-o-g-connectionFactory"/><![CDATA[
+		queue="" ]]><co id="redis-q-o-g-queue"/><![CDATA[
+		order="" ]]><co id="redis-q-o-g-order"/><![CDATA[
+		serializer="" ]]><co id="redis-q-o-g-serializer"/><![CDATA[
+		extract-payload="" ]]><co id="redis-q-o-g-extract-payload"/>
+			</programlisting>
+			<calloutlist>
+				<callout arearefs="redis-q-o-g-request-channel">
+					<para>
+						The <interfacename>MessageChannel</interfacename> from which this Endpoint receives<interfacename>Message</interfacename>s.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-o-g-reply-channel">
+					<para>
+						The <interfacename>MessageChannel</interfacename> where this Endpoint sends reply<interfacename>Message</interfacename>s.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-o-g-requires-reply">
+					<para>
+						Specify whether this outbound gateway must return a non-null value. This value is
+						<code>false</code> by default, otherwise a ReplyRequiredException will be thrown when
+						the Redis returns a <code>null</code> value.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-o-g-reply-timeout">
+					<para>
+						The timeout in milliseconds to wait until the reply message will be sent or not. Typically is
+						applied for queue-based limited reply-channels.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-o-g-connectionFactory">
+					<para>
+						A reference to a <interfacename>RedisConnectionFactory</interfacename>bean.
+						Defaults to <code>redisConnectionFactory</code>. Mutually exclusive with 'redis-template' attribute.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-o-g-queue">
+					<para>
+						The name of the Redis List to which outbound gateway will send a
+						<emphasis>conversation</emphasis> <code>UUID</code>.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-o-g-order">
+					<para>
+						The order for this outbound gateway when multiple gateway are registered thereby
+					</para>
+				</callout>
+				<callout arearefs="redis-q-o-g-serializer">
+					<para>
+						  The <interfacename>RedisSerializer</interfacename> bean reference. Can be an empty string, which means 'no serializer'.
+				          In this case the raw <code>byte[]</code> from the inbound Redis message is sent to the <code>channel</code> as the
+				          <interfacename>Message</interfacename> payload. By default it is a <classname>JdkSerializationRedisSerializer</classname>.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-o-g-extract-payload">
+					<para>
+						 Specify if this Endpoint expects data from the Redis queue to contain entire <interfacename>Message</interfacename>s.
+						 If this attribute is set to <code>true</code>, the <code>serializer</code> can't be an empty string because messages
+				   		 require some form of deserialization (JDK serialization by default).
+					</para>
+				</callout>
+			</calloutlist>
+		</para>
+	</section>
+	<section id="redis-queue-inbound-gateway">
+	<title>Redis Queue Inbound Gateway</title>
+		<para>
+			Since <emphasis>Spring Integration 4.1</emphasis>, the Redis Queue Inbound Gateway is available to
+			perform request and reply scenarios. It pops a <emphasis>conversation</emphasis> <code>UUID</code> from the
+			provided <code>queue</code>, then pops the value from the Redis List with that <code>UUID</code> as its
+			key and pushes the reply to the Redis List with a key of <code>UUID + '.reply'</code>:
+			<programlisting language="xml"><![CDATA[<int-redis:queue-inbound-gateway
+		request-channel="" ]]><co id="redis-q-i-g-request-channel"/><![CDATA[
+		reply-channel="" ]]><co id="redis-q-i-g-reply-channel"/><![CDATA[
+		executor="" ]]><co id="redis-q-i-g-task-executor"/><![CDATA[
+		reply-timeout="" ]]><co id="redis-q-i-g-reply-timeout"/><![CDATA[
+		connection-factory="" ]]><co id="redis-q-i-g-connectionFactory"/><![CDATA[
+		queue="" ]]><co id="redis-q-i-g-queue"/><![CDATA[
+		order="" ]]><co id="redis-q-i-g-order"/><![CDATA[
+		serializer="" ]]><co id="redis-q-i-g-serializer"/><![CDATA[
+		receive-timeout="" ]]><co id="redis-q-i-g-receive-timeout"/><![CDATA[
+		expect-message="" ]]><co id="redis-q-i-g-expect-message"/>
+			</programlisting>
+			<calloutlist>
+				<callout arearefs="redis-q-i-g-request-channel">
+					<para>
+						The <interfacename>MessageChannel</interfacename> from which this Endpoint receives<interfacename>Message</interfacename>s.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-i-g-reply-channel">
+					<para>
+						The <interfacename>MessageChannel</interfacename> where this Endpoint sends reply<interfacename>Message</interfacename>s.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-i-g-task-executor">
+					<para>
+						A reference to a Spring <interfacename>TaskExecutor</interfacename> (or standard JDK 1.5+ <interfacename>Executor</interfacename>)
+				   bean. It is used for the underlying listening task. By default a <classname>SimpleAsyncTaskExecutor</classname>
+				   is used.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-i-g-reply-timeout">
+					<para>
+						The timeout in milliseconds to wait until the reply message will be sent or not. Typically is
+						applied for queue-based limited reply-channels.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-i-g-connectionFactory">
+					<para>
+						A reference to a <interfacename>RedisConnectionFactory</interfacename>bean.
+						Defaults to <code>redisConnectionFactory</code>. Mutually exclusive with 'redis-template' attribute.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-i-g-queue">
+					<para>
+						The name of the Redis List for the <emphasis>conversation</emphasis> <code>UUID</code>s.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-i-g-order">
+					<para>
+						The order for this inbound gateway when multiple gateway are registered thereby
+					</para>
+				</callout>
+				<callout arearefs="redis-q-i-g-serializer">
+					<para>
+						  The <interfacename>RedisSerializer</interfacename> bean reference. Can be an empty string, which means 'no serializer'.
+				          In this case the raw <code>byte[]</code> from the inbound Redis message is sent to the <code>channel</code> as the
+				          <interfacename>Message</interfacename> payload. By default it is a <classname>StringRedisSerializer</classname>.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-i-g-receive-timeout">
+					<para>
+						The timeout in milliseconds to wait until the receive message will be get or not. Typically is
+						applied for queue-based limited request-channels.
+					</para>
+				</callout>
+				<callout arearefs="redis-q-i-g-expect-message">
+					<para>
+						 Specify if this Endpoint expects data from the Redis queue to contain entire <interfacename>Message</interfacename>s.
+						 If this attribute is set to <code>true</code>, the <code>serializer</code> can't be an empty string because messages
+				   		 require some form of deserialization (JDK serialization by default).
+					</para>
+				</callout>
+			</calloutlist>
+		</para>
+	</section>
 	<section id="redis-lock-registry">
 		<title>Redis Lock Registry</title>
 		<para>

src/reference/docbook/whats-new.xml

@@ -59,6 +59,14 @@
 				for the JSON transformers. See <xref linkend="transformer"/> for more information.
 			</para>
 		</section>
+		<section id="4.1-redis-queue-gateways">
+			<title>Redis Queue Gateways</title>
+			<para>
+				The <code>&lt;redis-queue-inbound-gateway&gt;</code> and
+				<code>&lt;redis-queue-outbound-gateway&gt;</code> components are now provided.
+				See <xref linkend="redis-queue-inbound-gateway"/> and <xref linkend="redis-queue-outbound-gateway"/>.
+			</para>
+		</section>
 		<section id="4.1-PollSkipAdvice">
 			<title>PollSkipAdvice</title>
 			<para>