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