From 1d4785f3d2d912e7a1085c9eb2a3aaa3a8487f3d Mon Sep 17 00:00:00 2001 From: David Liu Date: Sat, 25 Oct 2014 18:39:04 +0300 Subject: [PATCH] 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 --- .../xml/AbstractInboundGatewayParser.java | 4 - .../gateway/MessagingGatewaySupport.java | 6 +- .../redis/config/RedisNamespaceHandler.java | 2 + .../RedisQueueInboundGatewayParser.java | 63 ++++ .../RedisQueueOutboundGatewayParser.java | 59 +++ .../inbound/RedisQueueInboundGateway.java | 351 ++++++++++++++++++ .../outbound/RedisQueueOutboundGateway.java | 129 +++++++ .../config/spring-integration-redis-4.1.xsd | 242 ++++++++++++ ...isQueueGatewayIntegrationTests-context.xml | 44 +++ .../RedisQueueGatewayIntegrationTests.java | 121 ++++++ ...QueueInboundGatewayParserTests-context.xml | 46 +++ .../RedisQueueInboundGatewayParserTests.java | 76 ++++ ...ueueOutboundGatewayParserTests-context.xml | 49 +++ .../RedisQueueOutboundGatewayParserTests.java | 87 +++++ src/reference/docbook/endpoint-summary.xml | 4 +- src/reference/docbook/redis.xml | 161 +++++++- src/reference/docbook/whats-new.xml | 8 + 17 files changed, 1443 insertions(+), 9 deletions(-) create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParser.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParser.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests-context.xml create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests-context.xml create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParserTests-context.xml create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParserTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractInboundGatewayParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractInboundGatewayParser.java index 18628e7861..c161aa96c8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractInboundGatewayParser.java +++ b/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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 9d7b454a28..3259e6abd5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/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) { diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java index 12b614c4b6..ea99a33c08 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java @@ -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()); } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParser.java new file mode 100644 index 0000000000..5cad5a9cda --- /dev/null +++ b/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"); + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParser.java new file mode 100644 index 0000000000..adde45f8af --- /dev/null +++ b/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; + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java new file mode 100644 index 0000000000..95d2c6399d --- /dev/null +++ b/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 stringSerializer = new StringRedisSerializer(); + + public static final long DEFAULT_RECEIVE_TIMEOUT = 5000; + + public static final long DEFAULT_RECOVERY_INTERVAL = 5000; + + private final RedisTemplate template; + + private final BoundListOperations 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(); + 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}. + *

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. + *

A timeout of zero can be used to block indefinitely. If not set explicitly + * the timeout value will default to {@code 1000} + *

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 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) 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) serializer).serialize(replyMessage.getPayload()); + } + } + else { + value = (byte[]) replyMessage.getPayload(); + } + } + else { + if (this.serializer != null) { + value = ((RedisSerializer) 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 0 + * 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(); + } + } + } + } + + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java new file mode 100644 index 0000000000..a2481c57d1 --- /dev/null +++ b/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 stringSerializer = new StringRedisSerializer(); + + private final RedisTemplate template; + + private final BoundListOperations 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(); + 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) serializer).serialize(value); + } + } + String uuid = defaultIdGenerator.generateId().toString(); + + byte[] uuidByte = uuid.getBytes(); + this.boundListOps.leftPush(uuidByte); + this.template.boundListOps(uuid).leftPush(value); + + BoundListOperations 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; + } + +} diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.1.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.1.xsd index a625d01342..b4e51bbe53 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.1.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.1.xsd @@ -408,6 +408,131 @@ + + + + Configures a gateway that adapts incoming Redis values to Spring Integration Messages + and returns a reply. + + + + + + + Unique ID for this gateway. + + + + + + .reply' + to which this gateway will send the reply. + ]]> + + + + + + + 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. + + + + + + + + + + + Message Channel to which Messages should be sent in order to have them converted and published. + + + + + + + + + + + + + Message Channel to which replies for this gateway should be sent. + + + + + + + + + + + + + + + + + Reference to the Redis ConnectionFactory to be used by this component. + + + + + + + + + + + + + + + + + + + + + + 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'. + + + + + + + @@ -472,6 +597,123 @@ + + + .reply' as its key. A new UUID is used for each interaction. + ]]> + + + + + + + + + + .reply'. + ]]> + + + + + + + + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer + + + + + + + + + + + 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'. + + + + + + + + + + + + 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. + + + + + + + Message Channel where reply Messages will be sent. + + + + + + + Message Channel where request Messages will be expected. + + + + + + + Reference to the Redis ConnectionFactory to be used by this component. + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests-context.xml new file mode 100644 index 0000000000..9943270f23 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests-context.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests.java new file mode 100644 index 0000000000..ebcd78764f --- /dev/null +++ b/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("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("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("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("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); + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests-context.xml new file mode 100644 index 0000000000..e184bf91ac --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests-context.xml @@ -0,0 +1,46 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundGatewayParserTests.java new file mode 100644 index 0000000000..c4cb78a9b4 --- /dev/null +++ b/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")); + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParserTests-context.xml new file mode 100644 index 0000000000..6b6324d791 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParserTests-context.xml @@ -0,0 +1,49 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundGatewayParserTests.java new file mode 100644 index 0000000000..1ab5c56181 --- /dev/null +++ b/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")); + } + +} diff --git a/src/reference/docbook/endpoint-summary.xml b/src/reference/docbook/endpoint-summary.xml index fee82dc72f..870ff1de90 100644 --- a/src/reference/docbook/endpoint-summary.xml +++ b/src/reference/docbook/endpoint-summary.xml @@ -143,7 +143,9 @@ and N - + and + and + Resource diff --git a/src/reference/docbook/redis.xml b/src/reference/docbook/redis.xml index 5f73cefa31..1ce5d35cb7 100644 --- a/src/reference/docbook/redis.xml +++ b/src/reference/docbook/redis.xml @@ -699,7 +699,7 @@ the serialization of values, you may want to consider providing your own Specify whether this outbound gateway must return a non-null value. This value is - false by default, otherwise a ReplyRequiredException will be thrown when + true by default. A ReplyRequiredException will be thrown when the Redis returns a null value. @@ -775,6 +775,165 @@ the serialization of values, you may want to consider providing your own Redis Specification. +
+ Redis Queue Outbound Gateway + + Since Spring Integration 4.1, the Redis Queue Outbound Gateway is available to + perform request and reply scenarios. It pushes a conversation UUID to the + provided queue, then pushes the value to a Redis List with that UUID as its + key and waits for the reply from a Redis List with a key of UUID + '.reply'. A different + UUID is used for each interaction. + + + + + + The MessageChannel from which this Endpoint receivesMessages. + + + + + The MessageChannel where this Endpoint sends replyMessages. + + + + + Specify whether this outbound gateway must return a non-null value. This value is + false by default, otherwise a ReplyRequiredException will be thrown when + the Redis returns a null value. + + + + + The timeout in milliseconds to wait until the reply message will be sent or not. Typically is + applied for queue-based limited reply-channels. + + + + + A reference to a RedisConnectionFactorybean. + Defaults to redisConnectionFactory. Mutually exclusive with 'redis-template' attribute. + + + + + The name of the Redis List to which outbound gateway will send a + conversation UUID. + + + + + The order for this outbound gateway when multiple gateway are registered thereby + + + + + The RedisSerializer bean reference. Can be an empty string, which means 'no serializer'. + In this case the raw byte[] from the inbound Redis message is sent to the channel as the + Message payload. By default it is a JdkSerializationRedisSerializer. + + + + + Specify if this Endpoint expects data from the Redis queue to contain entire Messages. + If this attribute is set to true, the serializer can't be an empty string because messages + require some form of deserialization (JDK serialization by default). + + + + +
+
+ Redis Queue Inbound Gateway + + Since Spring Integration 4.1, the Redis Queue Inbound Gateway is available to + perform request and reply scenarios. It pops a conversation UUID from the + provided queue, then pops the value from the Redis List with that UUID as its + key and pushes the reply to the Redis List with a key of UUID + '.reply': + + + + + + The MessageChannel from which this Endpoint receivesMessages. + + + + + The MessageChannel where this Endpoint sends replyMessages. + + + + + A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) + bean. It is used for the underlying listening task. By default a SimpleAsyncTaskExecutor + is used. + + + + + The timeout in milliseconds to wait until the reply message will be sent or not. Typically is + applied for queue-based limited reply-channels. + + + + + A reference to a RedisConnectionFactorybean. + Defaults to redisConnectionFactory. Mutually exclusive with 'redis-template' attribute. + + + + + The name of the Redis List for the conversation UUIDs. + + + + + The order for this inbound gateway when multiple gateway are registered thereby + + + + + The RedisSerializer bean reference. Can be an empty string, which means 'no serializer'. + In this case the raw byte[] from the inbound Redis message is sent to the channel as the + Message payload. By default it is a StringRedisSerializer. + + + + + The timeout in milliseconds to wait until the receive message will be get or not. Typically is + applied for queue-based limited request-channels. + + + + + Specify if this Endpoint expects data from the Redis queue to contain entire Messages. + If this attribute is set to true, the serializer can't be an empty string because messages + require some form of deserialization (JDK serialization by default). + + + + +
Redis Lock Registry diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index b1bec277b2..d1183ddf88 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -59,6 +59,14 @@ for the JSON transformers. See for more information.
+
+ Redis Queue Gateways + + The <redis-queue-inbound-gateway> and + <redis-queue-outbound-gateway> components are now provided. + See and . + +
PollSkipAdvice