From 945455f360eb06f628e7bf5b26d3e8181a4f3793 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 23 Jan 2014 11:00:10 +0200 Subject: [PATCH] INT-2918: Add Redis Outbound Command Gateway JIRA: https://jira.springsource.org/browse/INT-2918 INT-2918: Improvement according PR comments INT-2918: Polishing according PR comments Minor Doc Polishing INT-2918: Change `notNull` message Polishing --- build.gradle | 2 +- .../redis/config/RedisNamespaceHandler.java | 1 + .../config/RedisOutboundGatewayParser.java | 93 ++++++++++ .../redis/outbound/ArgumentsStrategy.java | 29 ++++ .../outbound/ExpressionArgumentsStrategy.java | 95 ++++++++++ .../redis/outbound/RedisOutboundGateway.java | 141 +++++++++++++++ .../redis/support/RedisHeaders.java | 19 ++ .../config/spring-integration-redis-4.0.xsd | 157 +++++++++++++++++ .../RedisOutboundGatewayTests-context.xml | 67 +++++++ .../outbound/RedisOutboundGatewayTests.java | 163 ++++++++++++++++++ src/reference/docbook/redis.xml | 110 ++++++++++++ src/reference/docbook/whats-new.xml | 8 + 12 files changed, 884 insertions(+), 1 deletion(-) create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundGatewayParser.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ArgumentsStrategy.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ExpressionArgumentsStrategy.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisOutboundGateway.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisOutboundGatewayTests-context.xml create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisOutboundGatewayTests.java diff --git a/build.gradle b/build.gradle index 9467075d20..848e49f626 100644 --- a/build.gradle +++ b/build.gradle @@ -99,7 +99,7 @@ subprojects { subproject -> smackVersion = '3.2.1' springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.3.0.RELEASE' springDataMongoVersion = '1.1.1.RELEASE' - springDataRedisVersion = '1.1.1.RELEASE' + springDataRedisVersion = '1.2.1.RELEASE' springGemfireVersion = '1.3.1.RELEASE' springSecurityVersion = '3.1.3.RELEASE' springSocialTwitterVersion = '1.1.0.M4' 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 c13c32bc06..12b614c4b6 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 @@ -34,5 +34,6 @@ public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler { registerBeanDefinitionParser("outbound-channel-adapter", new RedisOutboundChannelAdapterParser()); registerBeanDefinitionParser("queue-inbound-channel-adapter", new RedisQueueInboundChannelAdapterParser()); registerBeanDefinitionParser("queue-outbound-channel-adapter", new RedisQueueOutboundChannelAdapterParser()); + registerBeanDefinitionParser("outbound-gateway", new RedisOutboundGatewayParser()); } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundGatewayParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundGatewayParser.java new file mode 100644 index 0000000000..be34839e8c --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundGatewayParser.java @@ -0,0 +1,93 @@ +/* + * 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.ExpressionArgumentsStrategy; +import org.springframework.integration.redis.outbound.RedisOutboundGateway; +import org.springframework.util.StringUtils; + +/** + * Parser for the {@code } component. + * + * @author Artem Bilan + * @since 4.0 + */ +public class RedisOutboundGatewayParser extends AbstractConsumerEndpointParser { + + @Override + protected String getInputChannelAttributeName() { + return "request-channel"; + } + + @Override + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisOutboundGateway.class); + + String redisTemplate = element.getAttribute("redis-template"); + String connectionFactory = element.getAttribute("connection-factory"); + if (StringUtils.hasText(redisTemplate) && StringUtils.hasText(connectionFactory)) { + parserContext.getReaderContext().error("Only one of '" + redisTemplate + "' or '" + + connectionFactory + "' is allowed.", element); + } + if (StringUtils.hasText(redisTemplate)) { + builder.addConstructorArgReference(redisTemplate); + } + else { + if (!StringUtils.hasText(connectionFactory)) { + connectionFactory = "redisConnectionFactory"; + } + builder.addConstructorArgReference(connectionFactory); + } + + String argumentExpressions = element.getAttribute("argument-expressions"); + boolean hasArgumentExpressions = StringUtils.hasText(argumentExpressions); + String argumentsStrategy = element.getAttribute("arguments-strategy"); + boolean hasArgumentStrategy = element.hasAttribute("arguments-strategy"); + + if (hasArgumentExpressions & hasArgumentStrategy) { + parserContext.getReaderContext() + .error("'argument-expressions' and 'arguments-strategy' are mutually exclusive.", element); + } + + if (hasArgumentExpressions) { + BeanDefinitionBuilder argumentsBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionArgumentsStrategy.class) + .addConstructorArgValue(argumentExpressions) + .addConstructorArgValue(element.getAttribute("use-command-variable")); + builder.addPropertyValue("argumentsStrategy", argumentsBuilder.getBeanDefinition()); + } + else if (StringUtils.hasLength(argumentsStrategy)) { + builder.addPropertyReference("argumentsStrategy", argumentsStrategy); + } + else if (hasArgumentStrategy) { + builder.addPropertyValue("argumentsStrategy", null); + } + + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "command-expression"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "arguments-serializer"); + + return builder; + } +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ArgumentsStrategy.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ArgumentsStrategy.java new file mode 100644 index 0000000000..f91e097282 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ArgumentsStrategy.java @@ -0,0 +1,29 @@ +/* + * 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 org.springframework.messaging.Message; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public interface ArgumentsStrategy { + + Object[] resolve(String command, Message message); + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ExpressionArgumentsStrategy.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ExpressionArgumentsStrategy.java new file mode 100644 index 0000000000..e8ffb1e308 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ExpressionArgumentsStrategy.java @@ -0,0 +1,95 @@ +/* + * 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.ArrayList; +import java.util.LinkedList; +import java.util.List; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.expression.IntegrationEvaluationContextAware; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public class ExpressionArgumentsStrategy implements ArgumentsStrategy, IntegrationEvaluationContextAware, BeanFactoryAware { + + private static final SpelExpressionParser PARSER = new SpelExpressionParser(); + + private final Expression[] argumentExpressions; + + private EvaluationContext evaluationContext; + + private final boolean useCommandVariable; + + private BeanFactory beanFactory; + + public ExpressionArgumentsStrategy(String[] argumentExpressions) { + this(argumentExpressions, false); + } + + public ExpressionArgumentsStrategy(String[] argumentExpressions, boolean useCommandVariable) { + Assert.notNull(argumentExpressions, "'argumentExpressions' must not be null"); + Assert.noNullElements(argumentExpressions, "'argumentExpressions' cannot have null values."); + List expressions = new LinkedList(); + for (String argumentExpression : argumentExpressions) { + expressions.add(PARSER.parseExpression(argumentExpression)); + } + this.argumentExpressions = expressions.toArray(new Expression[expressions.size()]); + this.useCommandVariable = useCommandVariable; + } + + @Override + public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + Assert.notNull(evaluationContext, "'evaluationContext' must not be null"); + this.evaluationContext = evaluationContext; + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + } + + @Override + public Object[] resolve(String command, Message message) { + EvaluationContext evaluationContext = this.evaluationContext; + + if (this.useCommandVariable) { + evaluationContext = IntegrationContextUtils.getEvaluationContext(this.beanFactory); + evaluationContext.setVariable("cmd", command); + } + + List arguments = new ArrayList(); + for (Expression argumentExpression : this.argumentExpressions) { + Object argument = argumentExpression.getValue(evaluationContext, message); + if (argument != null) { + arguments.add(argument); + } + } + return arguments.toArray(); + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisOutboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisOutboundGateway.java new file mode 100644 index 0000000000..203025c994 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisOutboundGateway.java @@ -0,0 +1,141 @@ +/* + * 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 org.springframework.dao.DataAccessException; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisCallback; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.GenericToStringSerializer; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.expression.IntegrationEvaluationContextAware; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.redis.support.RedisHeaders; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +/** + * The Gateway component implementation to perform Redis commands with provided arguments and to return command result. + * + * @author Artem Bilan + * @since 4.0 + */ +public class RedisOutboundGateway extends AbstractReplyProducingMessageHandler + implements IntegrationEvaluationContextAware { + + private static final SpelExpressionParser PARSER = new SpelExpressionParser(); + + private final RedisTemplate redisTemplate; + + private EvaluationContext evaluationContext; + + private volatile RedisSerializer argumentsSerializer = new GenericToStringSerializer(Object.class); + + private volatile Expression commandExpression = PARSER.parseExpression("headers[" + RedisHeaders.COMMAND + "]"); + + private volatile ArgumentsStrategy argumentsStrategy = new PayloadArgumentsStrategy(); + + public RedisOutboundGateway(RedisTemplate redisTemplate) { + Assert.notNull(redisTemplate, "'redisTemplate' must not be null"); + this.redisTemplate = redisTemplate; + } + + public RedisOutboundGateway(RedisConnectionFactory connectionFactory) { + Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); + this.redisTemplate = new RedisTemplate(); + this.redisTemplate.setConnectionFactory(connectionFactory); + this.redisTemplate.afterPropertiesSet(); + } + + @Override + public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + Assert.notNull(evaluationContext, "'evaluationContext' must not be null"); + this.evaluationContext = evaluationContext; + } + + @SuppressWarnings("unchecked") + public void setArgumentsSerializer(RedisSerializer serializer) { + Assert.notNull(serializer, "'serializer' must not be null"); + this.argumentsSerializer = (RedisSerializer) serializer; + } + + public void setCommandExpression(String commandExpression) { + Assert.hasText(commandExpression, "'commandExpression' must not be an empty string"); + this.commandExpression = PARSER.parseExpression(commandExpression); + } + + public void setArgumentsStrategy(ArgumentsStrategy argumentsStrategy) { + this.argumentsStrategy = argumentsStrategy; + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + final String command = this.commandExpression.getValue(this.evaluationContext, requestMessage, String.class); + Assert.notNull(command, "The 'command' must not evaluate to 'null'."); + byte[][] args = null; + if (this.argumentsStrategy != null) { + Object[] arguments = this.argumentsStrategy.resolve(command, requestMessage); + if (!ObjectUtils.isEmpty(arguments)) { + args = new byte[arguments.length][]; + + for (int i = 0; i < arguments.length; i++) { + Object argument = arguments[i]; + byte[] arg = null; + if (argument instanceof byte[]) { + arg = (byte[]) argument; + } + else { + arg = this.argumentsSerializer.serialize(argument); + } + args[i] = arg; + } + } + } + + final byte[][] actualArgs = args; + + return this.redisTemplate.execute(new RedisCallback() { + + @Override + public Object doInRedis(RedisConnection connection) throws DataAccessException { + return connection.execute(command, actualArgs); + } + + }); + } + + private class PayloadArgumentsStrategy implements ArgumentsStrategy { + + @Override + public Object[] resolve(String command, Message message) { + Object payload = message.getPayload(); + if (payload instanceof Object[]) { + return (Object[]) payload; + } + else { + return new Object[]{payload}; + } + } + + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java index 5f428c3813..541d57246d 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java @@ -1,3 +1,19 @@ +/* + * 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.support; /** @@ -6,6 +22,7 @@ package org.springframework.integration.redis.support; * * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * @since 2.2 */ public class RedisHeaders { @@ -20,4 +37,6 @@ public class RedisHeaders { public static final String ZSET_INCREMENT_SCORE = PREFIX + "zsetIncrementScore"; + public static final String COMMAND = PREFIX + "command"; + } diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.0.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.0.xsd index 8baa528aba..81f07a2cd2 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.0.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.0.xsd @@ -499,6 +499,163 @@ + + + + Defines an outbound Redis Command Sending Gateway. + + + + + + + + + + + + Reference to a RedisConnectionFactory. If none is provided, the default + bean name for the reference will be "redisConnectionFactory". + Mutually exclusive with 'redis-template' attribute. + + + + + + + + + + + + RedisTemplate to be used with this gateway. + Mutually exclusive with 'connection-factory' attribute. + + + + + + + + + + + + + + + + + + + + + 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 underlying service returns a null value. + + + + + + + + + + + + + + + + + + + + + + + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer. + Used to serialize each command argument to byte[] if necessary. + + + + + + + + + + + SpEL expression that returns the command key. Default is the 'redis_command' message header. + The command must not be evaluated to 'null'. + + + + + + + Comma-separated SpEL expressions that will be evaluated as command arguments. + Mutually exclusive with the 'arguments-strategy' attribute. If 'use-command-variable' is + specified to 'true', the '#cmd' variable will be presented within evaluation context. + Argument expressions may evaluate to 'null', to support a variable number of arguments. + + + + + + + Specifies, if the evaluated Redis command string will be made available + as the '#cmd' variable in the SpEL evaluation context in the + org.springframework.integration.redis.outbound.ExpressionArgumentsStrategy + when 'argument-expressions' is + configured, otherwise this attribute is ignored. + + + + + + + + Reference to an instance of org.springframework.integration.redis.outbound.ArgumentsStrategy. + Mutually exclusive with the 'argument-expressions' attribute. By default the 'payload' is used + as the command argument(s). To disable any strategy (no arguments) this attribute should be + configured as an empty string. + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisOutboundGatewayTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisOutboundGatewayTests-context.xml new file mode 100644 index 0000000000..86ed52f741 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisOutboundGatewayTests-context.xml @@ -0,0 +1,67 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisOutboundGatewayTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisOutboundGatewayTests.java new file mode 100644 index 0000000000..e3547e2af2 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisOutboundGatewayTests.java @@ -0,0 +1,163 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.Arrays; +import java.util.List; + +import org.hamcrest.Matchers; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.support.atomic.RedisAtomicInteger; +import org.springframework.integration.handler.ReplyRequiredException; +import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.redis.support.RedisHeaders; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.lambdaworks.redis.protocol.CommandType; + +/** + * @author Artem Bilan + * @since 4.0 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class RedisOutboundGatewayTests extends RedisAvailableTests { + + @Autowired + private PollableChannel replyChannel; + + @Autowired + private MessageChannel pingChannel; + + @Autowired + private MessageChannel leftPushRightPopChannel; + + @Autowired + private MessageChannel incrementAtomicIntegerChannel; + + @Autowired + private RedisAtomicInteger atomicInteger; + + @Autowired + private MessageChannel setDelCommandChannel; + + @Autowired + private MessageChannel getCommandChannel; + + @Autowired + private MessageChannel mgetCommandChannel; + + @Test + @RedisAvailable + public void testPingPongCommand() { + this.pingChannel.send(MessageBuilder.withPayload("foo").setHeader(RedisHeaders.COMMAND, CommandType.PING).build()); + Message receive = this.replyChannel.receive(1000); + assertNotNull(receive); + assertTrue(Arrays.equals("PONG".getBytes(), (byte[]) receive.getPayload())); + } + + @Test + @RedisAvailable + public void testPushAndPopCommands() { + final String queueName = "si.test.testRedisOutboundGateway"; + String payload = "testing"; + this.leftPushRightPopChannel.send(MessageBuilder.withPayload(payload) + .setHeader(RedisHeaders.COMMAND, CommandType.LPUSH) + .setHeader("queue", queueName) + .build()); + Message receive = this.replyChannel.receive(1000); + assertNotNull(receive); + + this.leftPushRightPopChannel.send(MessageBuilder.withPayload(payload) + .setHeader(RedisHeaders.COMMAND, CommandType.RPOP) + .setHeader("queue", queueName) + .build()); + receive = this.replyChannel.receive(1000); + assertNotNull(receive); + assertTrue(Arrays.equals(payload.getBytes(), (byte[]) receive.getPayload())); + } + + @Test + @RedisAvailable + public void testIncrementAtomicCommand() { + this.incrementAtomicIntegerChannel.send(MessageBuilder.withPayload(CommandType.INCR).build()); + System.out.println(); + Message receive = this.replyChannel.receive(1000); + assertNotNull(receive); + assertEquals(11L, receive.getPayload()); + assertEquals(11, this.atomicInteger.get()); + } + + @Test + @RedisAvailable + public void testGetCommand() { + this.setDelCommandChannel.send(MessageBuilder.withPayload(new String[]{"foo", "bar"}).setHeader(RedisHeaders.COMMAND, CommandType.SET).build()); + Message receive = this.replyChannel.receive(1000); + assertNotNull(receive); + assertEquals("OK", receive.getPayload()); + + this.getCommandChannel.send(MessageBuilder.withPayload("foo").build()); + receive = this.replyChannel.receive(1000); + assertNotNull(receive); + assertTrue(Arrays.equals("bar".getBytes(), (byte[]) receive.getPayload())); + + this.setDelCommandChannel.send(MessageBuilder.withPayload("foo").setHeader(RedisHeaders.COMMAND, CommandType.DEL).build()); + receive = this.replyChannel.receive(1000); + assertNotNull(receive); + assertEquals(1L, receive.getPayload()); + + try { + this.getCommandChannel.send(MessageBuilder.withPayload("foo").build()); + fail("ReplyRequiredException expected"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(ReplyRequiredException.class)); + } + } + + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testMGetCommand() { + RedisConnection connection = this.getConnectionFactoryForTest().getConnection(); + byte[] value1 = "bar1".getBytes(); + byte[] value2 = "bar2".getBytes(); + connection.set("foo1".getBytes(), value1); + connection.set("foo2".getBytes(), value2); + this.mgetCommandChannel.send(MessageBuilder.withPayload(new String [] {"foo1", "foo2"}).build()); + Message receive = this.replyChannel.receive(1000); + assertNotNull(receive); + assertThat((List) receive.getPayload(), Matchers.contains(value1, value2)); + } + +} diff --git a/src/reference/docbook/redis.xml b/src/reference/docbook/redis.xml index a3fa1b2bda..2ecf9ae2ed 100644 --- a/src/reference/docbook/redis.xml +++ b/src/reference/docbook/redis.xml @@ -660,4 +660,114 @@ the serialization of values, you may want to consider providing your own +
+ Redis Outbound Command Gateway + + Since Spring Integration 4.0, the Redis Command Gateway is available to + perform any standard Redis command using generic + RedisConnection#execute + method: + ]]> + + + + + 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. + + + + + A reference to a RedisTemplate bean. + Mutually exclusive with 'connection-factory' attribute. + + + + + Reference to an instance oforg.springframework.data.redis.serializer.RedisSerializer. + Used to serialize each command argument to byte[] if necessary. + + + + + The SpEL expression that returns the command key. Default is the redis_command message header. + Must not evaluate to null. + + + + + Comma-separate SpEL expressions that will be evaluated as command arguments. + Mutually exclusive with the arguments-strategy attribute. If neither of them is provided + the payload is used as the command argument(s). + Argument expressions may evaluate to 'null', to support a variable number of arguments. + + + + + A boolean flag to specify if the evaluated Redis command string will be + made available as the #cmd variable + in the expression evaluation context in the + org.springframework.integration.redis.outbound.ExpressionArgumentsStrategy + when argument-expressions is configured, otherwise this attribute is ignored. + + + + + Reference to an instance of org.springframework.integration.redis.outbound.ArgumentsStrategy. + Mutually exclusive with argument-expressions attribute. If neither of them is provided + the payload is used as the command argument(s). + + + + + + The <int-redis:outbound-gateway> can be used as a common component to perform any desired + Redis operation. For example to get incremented value from Redis Atomic Number: + ]]> + where the Message payload should be a name of redisCounter, which may be provided + by org.springframework.data.redis.support.atomic.RedisAtomicInteger bean definition. + + + The RedisConnection#execute has a generic Object as return type and real + result depends on command type, for example MGET returns a List<byte[]>. + For more information about commands, their arguments and result type see + Redis Specification. + +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 46f2e47032..7b04c9cd24 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -117,6 +117,14 @@ For more information, see . +
+ Redis Command Gateway + + The Redis support now provides the <outbound-gateway> component + to perform generic Redis commands using the RedisConnection#execute method. + For more information, see . + +