From 2546ac3d525647aa6856adf6e4592ab1c1c774e7 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 24 Dec 2015 16:57:44 -0500 Subject: [PATCH] INT-3924: Fix `RedisQueueInboundGateway` JIRA: https://jira.spring.io/browse/INT-3924 * The `RedisQueueInboundGateway` and `RedisQueueMessageDrivenEndpoint` used `MICROSECONDS` for the wait for stop condition. - Change to `MILLISECONDS` * Both of them may pick up data from the Redis List even after the `stop()`. - Add `rightPush()` to the `RedisQueueInboundGateway` to emulate `rollback`. - Even if we have `rollback` there we may "steal" messages after the `context.stop()` causing unexpected race condition. So change the logic between `stopTimeout` and `receiveTimeout` to `max`, thus we wait for stop more than for `rightPop`. * Some attributes have been missed for the XML configuration. - add `stop-timeout` and `recovery-interval` attributes to the XSD and parsers code. - adjust `RedisQueueGatewayIntegrationTests` to use those new attributes. * Some tests needed polishing: - Rework `RedisStoreOutboundChannelAdapterIntegrationTests` to be based on `@ContextConfiguration` for better performance - `RedisChannelParserTests` used to `mock` channels for the same Redis topic. Change of of them to different name to avoid `Dispatcher has no subscribers` from the `testPubSubChannelUsage()` when the second `SubscribableRedisChannel` doesn't have subscribers. * Polishing for `redis.adoc` Address PR comments * Rework logic to the `Math.min(this.stopTimeout, this.receiveTimeout)` on the adapter/gateway `stop()` * Fix JavaDocs and `redis.adoc` to describe `stopTimeout` properly. * Rework `RedisQueueGatewayIntegrationTests` to be based on the `randomUUID` for the queue name to send and receive. Address PR comments. --- .../RedisQueueInboundGatewayParser.java | 4 +- .../inbound/RedisQueueInboundGateway.java | 61 +-- .../RedisQueueMessageDrivenEndpoint.java | 43 +- .../config/spring-integration-redis-4.3.xsd | 18 + .../RedisChannelParserTests-context.xml | 18 +- .../redis/config/RedisChannelParserTests.java | 2 +- ...isQueueGatewayIntegrationTests-context.xml | 9 +- .../RedisQueueGatewayIntegrationTests.java | 18 +- .../RedisQueueMessageDrivenEndpointTests.java | 17 +- ...hannelAdapterIntegrationTests-context.xml} | 5 - ...utboundChannelAdapterIntegrationTests.java | 393 ++++++++---------- src/reference/asciidoc/redis.adoc | 92 ++-- 12 files changed, 334 insertions(+), 346 deletions(-) rename spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/{store-outbound-adapter.xml => RedisStoreOutboundChannelAdapterIntegrationTests-context.xml} (93%) 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 index 5cad5a9cda..92f2ed7051 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2016 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. @@ -57,6 +57,8 @@ public class RedisQueueInboundGatewayParser extends AbstractInboundGatewayParser builder.addConstructorArgReference(connectionFactory); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor"); } 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 index a811614002..b42c44ed65 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors + * Copyright 2014-2016 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. @@ -55,7 +55,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements private static final RedisSerializer stringSerializer = new StringRedisSerializer(); - public static final long DEFAULT_RECEIVE_TIMEOUT = 5000; + public static final long DEFAULT_RECEIVE_TIMEOUT = 1000; public static final long DEFAULT_RECOVERY_INTERVAL = 5000; @@ -75,14 +75,14 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements 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; + private volatile Runnable stopCallback; + /** * @param queueName Must not be an empty String * @param connectionFactory Must not be null @@ -131,11 +131,12 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements /** * @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} + * will be processed or this timeout is reached. Should be less than or equal to {@link #receiveTimeout} + * @deprecated since {@literal 4.3} with no-op in favor of delayer call {@code callback.run()} + * in the {@link #stop(Runnable)}. */ + @Deprecated public void setStopTimeout(long stopTimeout) { - this.stopTimeout = stopTimeout; } public void setTaskExecutor(Executor taskExecutor) { @@ -195,6 +196,10 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements } String uuid = null; if (value != null) { + if (!active) { + this.boundListOperations.rightPush(value); + return; + } uuid = stringSerializer.deserialize(value); try { value = this.template.boundListOps(uuid).rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS); @@ -205,6 +210,11 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements } Message requestMessage = null; if (value != null) { + if (!active) { + this.template.boundListOps(uuid).rightPush(value); + this.boundListOperations.rightPush(stringSerializer.serialize(uuid)); + return; + } if (this.extractPayload) { Object payload = value; if (this.serializer != null) { @@ -248,6 +258,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements @Override protected void doStart() { + super.doStart(); if (!this.active) { this.active = true; this.restart(); @@ -285,23 +296,20 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements this.taskExecutor.execute(new ListenerTask()); } + @Override + protected void doStop(Runnable callback) { + this.stopCallback = callback; + doStop(); + } + @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; - } + super.doStop(); + this.active = this.listening = false; } public boolean isListening() { - return listening; + return this.listening; } /** @@ -336,21 +344,16 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements try { while (RedisQueueInboundGateway.this.active) { RedisQueueInboundGateway.this.listening = true; - RedisQueueInboundGateway.this.receiveAndReply(); + receiveAndReply(); } } finally { if (RedisQueueInboundGateway.this.active) { - RedisQueueInboundGateway.this.restart(); + restart(); } - else { - RedisQueueInboundGateway.this.lifecycleLock.lock(); - try { - RedisQueueInboundGateway.this.lifecycleCondition.signalAll(); - } - finally { - RedisQueueInboundGateway.this.lifecycleLock.unlock(); - } + else if (RedisQueueInboundGateway.this.stopCallback != null) { + RedisQueueInboundGateway.this.stopCallback.run(); + RedisQueueInboundGateway.this.stopCallback = null; } } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java index f1018d8363..fb72d397f2 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2015 the original author or authors + * Copyright 2013-2016 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. @@ -13,6 +13,7 @@ * 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; @@ -73,12 +74,12 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL; - private volatile long stopTimeout = DEFAULT_RECEIVE_TIMEOUT; - private volatile boolean active; private volatile boolean listening; + private volatile Runnable stopCallback; + /** * @param queueName Must not be an empty String * @param connectionFactory Must not be null @@ -134,11 +135,13 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl /** * @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} + * or this timeout is reached. Should be less than or equal to {@link #receiveTimeout} * @since 4.0.3 + * @deprecated since {@literal 4.3} with no-op in favor of delayer call {@code callback.run()} + * in the {@link #stop(Runnable)}. */ + @Deprecated public void setStopTimeout(long stopTimeout) { - this.stopTimeout = stopTimeout; } public void setTaskExecutor(Executor taskExecutor) { @@ -268,19 +271,16 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl this.taskExecutor.execute(new ListenerTask()); } + @Override + protected void doStop(Runnable callback) { + this.stopCallback = callback; + doStop(); + } + @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; - } + super.doStop(); + this.active = this.listening = false; } public boolean isListening() { @@ -327,14 +327,9 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl if (RedisQueueMessageDrivenEndpoint.this.active) { RedisQueueMessageDrivenEndpoint.this.restart(); } - else { - RedisQueueMessageDrivenEndpoint.this.lifecycleLock.lock(); - try { - RedisQueueMessageDrivenEndpoint.this.lifecycleCondition.signalAll(); - } - finally { - RedisQueueMessageDrivenEndpoint.this.lifecycleLock.unlock(); - } + else if (RedisQueueMessageDrivenEndpoint.this.stopCallback != null) { + RedisQueueMessageDrivenEndpoint.this.stopCallback.run(); + RedisQueueMessageDrivenEndpoint.this.stopCallback = null; } } } diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.3.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.3.xsd index e412d18cc7..90405da5d5 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.3.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.3.xsd @@ -506,6 +506,24 @@ + + + + Specify the timeout in milliseconds to wait for the result of the + 'rightPop' operation on Redis queue. + Default is 1 second. + + + + + + + Specify the time in milliseconds for which the listener task should sleep after catching + an Exception on a Redis operation, before restarting the listener task. + Default is 5 seconds. + + + + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xmlns:int-redis="http://www.springframework.org/schema/integration/redis" + xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd + http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd"> - + @@ -19,6 +15,6 @@ + serializer="redisSerializer" max-subscribers="1"/> diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java index 3d89585108..9260dc3af5 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java @@ -56,7 +56,7 @@ public class RedisChannelParserTests extends RedisAvailableTests { RedisSerializer redisSerializer = TestUtils.getPropertyValue(redisChannel, "serializer", RedisSerializer.class); assertEquals(connectionFactory, context.getBean("redisConnectionFactory")); assertEquals(redisSerializer, context.getBean("redisSerializer")); - assertEquals("si.test.topic", TestUtils.getPropertyValue(redisChannel, "topicName")); + assertEquals("si.test.topic.parser", TestUtils.getPropertyValue(redisChannel, "topicName")); assertEquals(Integer.MAX_VALUE, TestUtils.getPropertyValue( TestUtils.getPropertyValue(redisChannel, "dispatcher"), "maxSubscribers", Integer.class).intValue()); redisChannel = context.getBean("redisChannelWithSubLimit", SubscribableChannel.class); 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 index 9943270f23..bd4dccba7d 100644 --- 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 @@ -21,10 +21,12 @@ + + @@ -33,10 +35,11 @@ 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 index ebcd78764f..5870931d09 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -25,6 +25,9 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.channel.DirectChannel; @@ -33,6 +36,7 @@ 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.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; @@ -41,6 +45,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author David Liu + * @author Artem Bilan * @since 4.1 */ @ContextConfiguration @@ -48,6 +53,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @DirtiesContext public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests { + @Value("#{redisQueue.toString().bytes}") + private byte[] queueName; + @Autowired @Qualifier("sendChannel") private DirectChannel sendChannel; @@ -62,6 +70,11 @@ public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests { @Autowired private RedisQueueOutboundGateway outboundGateway; + public void setup() { + RedisConnectionFactory jcf = getConnectionFactoryForTest(); + jcf.getConnection().del(this.queueName); + } + @Test @RedisAvailable public void testRequestWithReply() throws Exception { @@ -74,6 +87,8 @@ public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests { @Test @RedisAvailable public void testInboundGatewayStop() throws Exception { + Long receiveTimeout = TestUtils.getPropertyValue(this.inboundGateway, "receiveTimeout", Long.class); + this.inboundGateway.setReceiveTimeout(1); this.inboundGateway.stop(); try { this.sendChannel.send(new GenericMessage("test1")); @@ -82,6 +97,7 @@ public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests { assertTrue(e.getMessage().contains("No reply produced")); } finally { + this.inboundGateway.setReceiveTimeout(receiveTimeout); this.inboundGateway.start(); } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java index 45d5232102..aae88e6ccb 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2015 the original author or authors + * Copyright 2013-2016 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. @@ -240,7 +240,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); endpoint.setOutputChannel(new DirectChannel()); endpoint.setReceiveTimeout(1000); - endpoint.setStopTimeout(100); ExecutorService executorService = Executors.newCachedThreadPool(); endpoint.setTaskExecutor(executorService); @@ -252,11 +251,23 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { dfa.setPropertyValue("listening", false); redisTemplate.boundListOps(queueName).leftPush("foo"); - endpoint.stop(); + + final CountDownLatch stopLatch = new CountDownLatch(1); + + endpoint.stop(new Runnable() { + + @Override + public void run() { + stopLatch.countDown(); + } + + }); executorService.shutdown(); assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS)); + assertTrue(stopLatch.await(10, TimeUnit.SECONDS)); + Mockito.verify(boundListOperations).rightPush(Mockito.any(byte[].class)); } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/store-outbound-adapter.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests-context.xml similarity index 93% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/store-outbound-adapter.xml rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests-context.xml index 8df4ff38ad..0967285943 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/store-outbound-adapter.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests-context.xml @@ -19,11 +19,6 @@ collection-type="LIST" key="pepboys"/> - - diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java index 5f9b5cceee..921825e37a 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2014 the original author or authors + * Copyright 2007-2015 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. @@ -26,8 +26,15 @@ import java.util.Map; import java.util.Properties; import java.util.Set; +import org.junit.After; +import org.junit.Before; import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; @@ -52,207 +59,221 @@ import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Oleg Zhurakousky * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan * @since 2.2 */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvailableTests { + private final StringRedisTemplate redisTemplate = new StringRedisTemplate(); + + @Autowired + private BeanFactory beanFactory; + + @Autowired + @Qualifier("listWithKeyAsHeader") + private MessageChannel listWithKeyAsHeaderChannel; + + @Autowired + @Qualifier("listWithKeyProvided") + private MessageChannel listWithKeyProvidedChannel; + + @Autowired + @Qualifier("zset") + private MessageChannel zsetChannel; + + @Autowired + @Qualifier("mapToZset") + private MessageChannel mapToZsetChannel; + + @Autowired + @Qualifier("mapToMapA") + private MessageChannel mapToMapAChannel; + + @Autowired + @Qualifier("mapToMapB") + private MessageChannel mapToMapBChannel; + + @Autowired + @Qualifier("simpleMap") + private MessageChannel simpleMapChannel; + + @Autowired + @Qualifier("set") + private MessageChannel setChannel; + + @Autowired + @Qualifier("setNotParsed") + private MessageChannel setNotParsedChannel; + + @Autowired + @Qualifier("pojoIntoSet") + private MessageChannel pojoIntoSetChannel; + + @Autowired + @Qualifier("property") + private MessageChannel propertyChannel; + + @Autowired + @Qualifier("simpleProperty") + private MessageChannel simplePropertyChannel; + + @Before + @After + public void setup() { + RedisConnectionFactory jcf = getConnectionFactoryForTest(); + this.redisTemplate.setConnectionFactory(jcf); + this.redisTemplate.afterPropertiesSet(); + + this.redisTemplate.delete("pepboys"); + this.redisTemplate.delete("foo"); + this.redisTemplate.delete("bar"); + this.redisTemplate.delete("presidents"); + } + @Test @RedisAvailable - public void testListWithKeyAsHeader(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - - this.deleteKey(jcf, "pepboys"); - RedisList redisList = - new DefaultRedisList("pepboys", this.initTemplate(jcf, new StringRedisTemplate())); + public void testListWithKeyAsHeader() { + RedisList redisList = new DefaultRedisList("pepboys", this.redisTemplate); assertEquals(0, redisList.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("listWithKeyAsHeader", MessageChannel.class); List pepboys = new ArrayList(); pepboys.add("Manny"); pepboys.add("Moe"); pepboys.add("Jack"); Message> message = MessageBuilder.withPayload(pepboys).setHeader(RedisHeaders.KEY, "pepboys").build(); - redisChannel.send(message); + this.listWithKeyAsHeaderChannel.send(message); assertEquals(3, redisList.size()); - this.deleteKey(jcf, "pepboys"); - context.close(); } @Test @RedisAvailable - public void testListWithKeyAsHeaderSimple(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - - StringRedisTemplate redisTemplate = new StringRedisTemplate(); - RedisList redisList = - new DefaultRedisList("foo", this.initTemplate(jcf, redisTemplate)); + public void testListWithKeyAsHeaderSimple() { + redisTemplate.delete("foo"); + RedisList redisList = new DefaultRedisList("foo", this.redisTemplate); assertEquals(0, redisList.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("listWithKeyAsHeader", MessageChannel.class); Message message = MessageBuilder.withPayload("bar").setHeader("redis_key", "foo").build(); - redisChannel.send(message); + this.listWithKeyAsHeaderChannel.send(message); assertEquals(1, redisList.size()); - redisTemplate.delete("foo"); - context.close(); } @Test @RedisAvailable - public void testListWithProvidedKey(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "pepboys"); - RedisList redisList = - new DefaultRedisList("pepboys", this.initTemplate(jcf, new StringRedisTemplate())); + public void testListWithProvidedKey() { + RedisList redisList = new DefaultRedisList("pepboys", this.redisTemplate); assertEquals(0, redisList.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("listWithKeyProvided", MessageChannel.class); List pepboys = new ArrayList(); pepboys.add("Manny"); pepboys.add("Moe"); pepboys.add("Jack"); Message> message = MessageBuilder.withPayload(pepboys).build(); - redisChannel.send(message); + this.listWithKeyProvidedChannel.send(message); assertEquals(3, redisList.size()); - this.deleteKey(jcf, "pepboys"); - context.close(); } @Test @RedisAvailable - public void testZsetSimplePayloadIncrement(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - - StringRedisTemplate redisTemplate = new StringRedisTemplate(); - RedisZSet redisZSet = - new DefaultRedisZSet("foo", this.initTemplate(jcf, redisTemplate)); + public void testZsetSimplePayloadIncrement() { + RedisZSet redisZSet = new DefaultRedisZSet("foo", this.redisTemplate); assertEquals(0, redisZSet.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("zset", MessageChannel.class); Message message = MessageBuilder.withPayload("bar").setHeader(RedisHeaders.KEY, "foo").build(); - redisChannel.send(message); + this.zsetChannel.send(message); assertEquals(1, redisZSet.size()); assertEquals(Double.valueOf(1), redisZSet.score("bar")); - redisChannel.send(message); + this.zsetChannel.send(message); assertEquals(1, redisZSet.size()); assertEquals(Double.valueOf(2), redisZSet.score("bar")); - redisTemplate.delete("foo"); - context.close(); } @Test @RedisAvailable - public void testZsetSimplePayloadOverwrite(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - - StringRedisTemplate redisTemplate = new StringRedisTemplate(); - RedisZSet redisZSet = - new DefaultRedisZSet("foo", this.initTemplate(jcf, redisTemplate)); + public void testZsetSimplePayloadOverwrite() { + RedisZSet redisZSet = new DefaultRedisZSet("foo", this.redisTemplate); assertEquals(0, redisZSet.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("zset", MessageChannel.class); Message message = MessageBuilder.withPayload("bar") .setHeader(RedisHeaders.KEY, "foo") .setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, false) .build(); - redisChannel.send(message); + this.zsetChannel.send(message); assertEquals(1, redisZSet.size()); assertEquals(Double.valueOf(1), redisZSet.score("bar")); - redisChannel.send(message); + this.zsetChannel.send(message); assertEquals(1, redisZSet.size()); assertEquals(Double.valueOf(1), redisZSet.score("bar")); - redisTemplate.delete("foo"); - context.close(); } @Test @RedisAvailable - public void testZsetSimplePayloadIncrementBy2(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - - StringRedisTemplate redisTemplate = new StringRedisTemplate(); - RedisZSet redisZSet = - new DefaultRedisZSet("foo", this.initTemplate(jcf, redisTemplate)); + public void testZsetSimplePayloadIncrementBy2() { + RedisZSet redisZSet = new DefaultRedisZSet("foo", this.redisTemplate); assertEquals(0, redisZSet.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("zset", MessageChannel.class); Message message = MessageBuilder.withPayload("bar") .setHeader(RedisHeaders.KEY, "foo") .setHeader(RedisHeaders.ZSET_SCORE, 2) .build(); - redisChannel.send(message); + this.zsetChannel.send(message); assertEquals(1, redisZSet.size()); assertEquals(Double.valueOf(2), redisZSet.score("bar")); - redisChannel.send(message); + this.zsetChannel.send(message); assertEquals(1, redisZSet.size()); assertEquals(Double.valueOf(4), redisZSet.score("bar")); - redisTemplate.delete("foo"); - context.close(); } @Test @RedisAvailable - public void testZsetSimplePayloadOverwriteWithHeaderScore(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - - StringRedisTemplate redisTemplate = new StringRedisTemplate(); - RedisZSet redisZSet = - new DefaultRedisZSet("foo", this.initTemplate(jcf, redisTemplate)); + public void testZsetSimplePayloadOverwriteWithHeaderScore() { + RedisZSet redisZSet = new DefaultRedisZSet("foo", this.redisTemplate); assertEquals(0, redisZSet.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("zset", MessageChannel.class); Message message = MessageBuilder.withPayload("bar") .setHeader(RedisHeaders.KEY, "foo") .setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, false) .setHeader(RedisHeaders.ZSET_SCORE, 2) .build(); - redisChannel.send(message); + this.zsetChannel.send(message); assertEquals(1, redisZSet.size()); assertEquals(Double.valueOf(2), redisZSet.score("bar")); - redisChannel.send(MessageBuilder.fromMessage(message).setHeader(RedisHeaders.ZSET_SCORE, 15).build()); + this.zsetChannel.send(MessageBuilder.fromMessage(message).setHeader(RedisHeaders.ZSET_SCORE, 15).build()); assertEquals(1, redisZSet.size()); assertEquals(Double.valueOf(15), redisZSet.score("bar")); - redisTemplate.delete("foo"); - context.close(); } @Test @RedisAvailable - public void testMapToZsetWithProvidedKey(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deletePresidents(jcf); - RedisZSet redisZset = - new DefaultRedisZSet("presidents", this.initTemplate(jcf, new StringRedisTemplate())); + public void testMapToZsetWithProvidedKey() { + RedisZSet redisZset = new DefaultRedisZSet("presidents", this.redisTemplate); assertEquals(0, redisZset.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("mapToZset", MessageChannel.class); Map presidents = new HashMap(); presidents.put("John Adams", 18); @@ -262,19 +283,21 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail presidents.put("Zachary Taylor", 19); Message> message = MessageBuilder.withPayload(presidents).build(); - redisChannel.send(message); + + this.mapToZsetChannel.send(message); assertEquals(5, redisZset.size()); assertEquals(1, redisZset.rangeByScore(18, 18).size()); assertEquals(4, redisZset.rangeByScore(18, 19).size()); assertEquals(1, redisZset.rangeByScore(21, 21).size()); - RedisStoreWritingMessageHandler handler = context.getBean("mapToZset.handler", - RedisStoreWritingMessageHandler.class); - assertEquals("'presidents'", TestUtils.getPropertyValue(handler, "keyExpression", SpelExpression.class).getExpressionString()); + RedisStoreWritingMessageHandler handler = + this.beanFactory.getBean("mapToZset.handler", RedisStoreWritingMessageHandler.class); + assertEquals("'presidents'", TestUtils.getPropertyValue(handler, "keyExpression.expression")); // test default (increment by score) behavior - redisChannel.send(message); + this.mapToZsetChannel.send(message); + assertEquals(5, redisZset.size()); assertEquals(1, redisZset.rangeByScore(36, 36).size()); assertEquals(4, redisZset.rangeByScore(36, 38).size()); @@ -282,60 +305,51 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail // test overwrite score behavior presidents.put("Barack Obama", 31); - redisChannel.send(MessageBuilder.fromMessage(message).setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, false).build()); + + this.mapToZsetChannel.send(MessageBuilder.fromMessage(message) + .setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, false) + .build()); + assertEquals(5, redisZset.size()); assertEquals(1, redisZset.rangeByScore(18, 18).size()); assertEquals(4, redisZset.rangeByScore(18, 19).size()); assertEquals(1, redisZset.rangeByScore(31, 31).size()); - this.deletePresidents(jcf); - context.close(); } @Test @RedisAvailable - public void testMapToMapWithProvidedKey(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "pepboys"); - RedisMap redisMap = - new DefaultRedisMap("pepboys", - this.initTemplate(jcf, new StringRedisTemplate())); + public void testMapToMapWithProvidedKey() { + RedisMap redisMap = new DefaultRedisMap("pepboys", this.redisTemplate); assertEquals(0, redisMap.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("mapToMapA", MessageChannel.class); Map pepboys = new HashMap(); pepboys.put("1", "Manny"); pepboys.put("2", "Moe"); pepboys.put("3", "Jack"); Message> message = MessageBuilder.withPayload(pepboys).build(); - redisChannel.send(message); + this.mapToMapAChannel.send(message); assertEquals("Manny", redisMap.get("1")); assertEquals("Moe", redisMap.get("2")); assertEquals("Jack", redisMap.get("3")); - RedisStoreWritingMessageHandler handler = context.getBean("mapToMapA.handler", + RedisStoreWritingMessageHandler handler = this.beanFactory.getBean("mapToMapA.handler", RedisStoreWritingMessageHandler.class); - assertEquals("pepboys", TestUtils.getPropertyValue(handler, "keyExpression", LiteralExpression.class).getExpressionString()); - assertEquals("'foo'", TestUtils.getPropertyValue(handler, "mapKeyExpression", SpelExpression.class).getExpressionString()); - this.deleteKey(jcf, "pepboys"); - context.close(); + assertEquals("pepboys", + TestUtils.getPropertyValue(handler, "keyExpression", LiteralExpression.class).getExpressionString()); + assertEquals("'foo'", + TestUtils.getPropertyValue(handler, "mapKeyExpression", SpelExpression.class).getExpressionString()); } - @Test(expected=MessageHandlingException.class) // map key is not provided + @Test(expected = MessageHandlingException.class) // map key is not provided @RedisAvailable - public void testMapToMapAsSingleEntryWithKeyAsHeaderFail(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "pepboys"); + public void testMapToMapAsSingleEntryWithKeyAsHeaderFail() { RedisMap> redisMap = - new DefaultRedisMap>("pepboys", - this.initTemplate(jcf, new RedisTemplate>>())); + new DefaultRedisMap>("pepboys", this.redisTemplate); assertEquals(0, redisMap.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class); Map pepboys = new HashMap(); pepboys.put("1", "Manny"); pepboys.put("2", "Moe"); @@ -343,54 +357,47 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail Message> message = MessageBuilder.withPayload(pepboys). setHeader(RedisHeaders.KEY, "pepboys").build(); - redisChannel.send(message); - this.deleteKey(jcf, "pepboys"); - context.close(); + + this.mapToMapBChannel.send(message); } - @Test(expected=MessageHandlingException.class) // key is not provided + @Test(expected = MessageHandlingException.class) // key is not provided @RedisAvailable - public void testMapToMapNoKey(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "pepboys"); + public void testMapToMapNoKey() { RedisTemplate>> redisTemplate = new RedisTemplate>>(); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); + redisTemplate.setConnectionFactory(getConnectionFactoryForTest()); + redisTemplate.afterPropertiesSet(); + RedisMap> redisMap = - new DefaultRedisMap>("pepboys", - this.initTemplate(jcf, redisTemplate)); + new DefaultRedisMap>("pepboys", redisTemplate); assertEquals(0, redisMap.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class); Map pepboys = new HashMap(); pepboys.put("1", "Manny"); pepboys.put("2", "Moe"); pepboys.put("3", "Jack"); Message> message = MessageBuilder.withPayload(pepboys).build(); - redisChannel.send(message); - this.deleteKey(jcf, "pepboys"); - context.close(); + this.mapToMapBChannel.send(message); } @Test @RedisAvailable - public void testMapToMapAsSingleEntryWithKeyAsHeader(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "pepboys"); + public void testMapToMapAsSingleEntryWithKeyAsHeader() { RedisTemplate>> redisTemplate = new RedisTemplate>>(); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); + redisTemplate.setConnectionFactory(getConnectionFactoryForTest()); + redisTemplate.afterPropertiesSet(); + RedisMap> redisMap = - new DefaultRedisMap>("pepboys", - this.initTemplate(jcf, redisTemplate)); + new DefaultRedisMap>("pepboys", redisTemplate); assertEquals(0, redisMap.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class); Map pepboys = new HashMap(); pepboys.put("1", "Manny"); pepboys.put("2", "Moe"); @@ -398,182 +405,122 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail Message> message = MessageBuilder.withPayload(pepboys). setHeader(RedisHeaders.KEY, "pepboys").setHeader(RedisHeaders.MAP_KEY, "foo").build(); - redisChannel.send(message); + this.mapToMapBChannel.send(message); Map pepboyz = redisMap.get("foo"); assertEquals("Manny", pepboyz.get("1")); assertEquals("Moe", pepboyz.get("2")); assertEquals("Jack", pepboyz.get("3")); - this.deleteKey(jcf, "pepboys"); - context.close(); } @Test @RedisAvailable - public void testStoreSimpleStringInMap(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "bar"); - StringRedisTemplate redisTemplate = new StringRedisTemplate(); - RedisMap redisMap = - new DefaultRedisMap("bar", - this.initTemplate(jcf, redisTemplate)); + public void testStoreSimpleStringInMap() { + RedisMap redisMap = new DefaultRedisMap("bar", this.redisTemplate); assertEquals(0, redisMap.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("simpleMap", MessageChannel.class); - Message message = MessageBuilder.withPayload("hello, world!"). setHeader(RedisHeaders.KEY, "bar").setHeader(RedisHeaders.MAP_KEY, "foo").build(); - redisChannel.send(message); + + this.simpleMapChannel.send(message); String hello = redisMap.get("foo"); assertEquals("hello, world!", hello); - this.deleteKey(jcf, "bar"); - context.close(); } @Test @RedisAvailable - public void testSetWithKeyAsHeader(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "pepboys"); - RedisSet redisSet = - new DefaultRedisSet("pepboys", this.initTemplate(jcf, new StringRedisTemplate())); + public void testSetWithKeyAsHeader() { + RedisSet redisSet = new DefaultRedisSet("pepboys", this.redisTemplate); assertEquals(0, redisSet.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("set", MessageChannel.class); Set pepboys = new HashSet(); pepboys.add("Manny"); pepboys.add("Moe"); pepboys.add("Jack"); Message> message = MessageBuilder.withPayload(pepboys).setHeader("redis_key", "pepboys").build(); - redisChannel.send(message); + this.setChannel.send(message); assertEquals(3, redisSet.size()); - this.deleteKey(jcf, "pepboys"); - context.close(); } @Test @RedisAvailable - public void testSetWithKeyAsHeaderSimple(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - StringRedisTemplate redisTemplate = new StringRedisTemplate(); - RedisSet redisSet = - new DefaultRedisSet("foo", this.initTemplate(jcf, redisTemplate)); + public void testSetWithKeyAsHeaderSimple() { + RedisSet redisSet = new DefaultRedisSet("foo", this.redisTemplate); assertEquals(0, redisSet.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("set", MessageChannel.class); Message message = MessageBuilder.withPayload("foo") .setHeader(RedisHeaders.KEY, "foo").build(); - redisChannel.send(message); + this.setChannel.send(message); assertEquals(1, redisSet.size()); - redisTemplate.delete("foo"); - context.close(); } @Test @RedisAvailable - public void testSetWithKeyAsHeaderNotParsed(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "pepboys"); - RedisTemplate redisTemplate = new RedisTemplate(); - redisTemplate.setKeySerializer(new StringRedisSerializer()); - redisTemplate.setHashKeySerializer(new StringRedisSerializer()); - RedisSet redisSet = - new DefaultRedisSet("pepboys", this.initTemplate(jcf, redisTemplate)); + public void testSetWithKeyAsHeaderNotParsed() { + RedisSet redisSet = new DefaultRedisSet("pepboys", this.redisTemplate); assertEquals(0, redisSet.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("setNotParsed", MessageChannel.class); Set pepboys = new HashSet(); pepboys.add("Manny"); pepboys.add("Moe"); pepboys.add("Jack"); Message> message = MessageBuilder.withPayload(pepboys).setHeader("redis_key", "pepboys").build(); - redisChannel.send(message); + this.setNotParsedChannel.send(message); assertEquals(1, redisSet.size()); - this.deleteKey(jcf, "pepboys"); - context.close(); } @Test @RedisAvailable - public void testPojoIntoSet(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "pepboys"); - RedisSet redisSet = - new DefaultRedisSet("pepboys", this.initTemplate(jcf, new StringRedisTemplate())); + public void testPojoIntoSet() { + RedisSet redisSet = new DefaultRedisSet("pepboys", this.redisTemplate); assertEquals(0, redisSet.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("pojoIntoSet", MessageChannel.class); String pepboy = "Manny"; Message message = MessageBuilder.withPayload(pepboy).setHeader("redis_key", "pepboys").build(); - redisChannel.send(message); + this.pojoIntoSetChannel.send(message); assertEquals(1, redisSet.size()); - this.deleteKey(jcf, "pepboys"); - context.close(); } @Test @RedisAvailable - public void testProperties(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.deleteKey(jcf, "pepboys"); - RedisProperties redisProperties = - new RedisProperties("pepboys", this.initTemplate(jcf, new StringRedisTemplate())); + public void testProperties() { + RedisProperties redisProperties = new RedisProperties("pepboys", this.redisTemplate); assertEquals(0, redisProperties.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("property", MessageChannel.class); Properties pepboys = new Properties(); pepboys.put("1", "Manny"); pepboys.put("2", "Moe"); pepboys.put("3", "Jack"); Message message = MessageBuilder.withPayload(pepboys).build(); - redisChannel.send(message); + this.propertyChannel.send(message); + assertEquals("Manny", redisProperties.get("1")); assertEquals("Moe", redisProperties.get("2")); assertEquals("Jack", redisProperties.get("3")); - this.deleteKey(jcf, "pepboys"); - context.close(); } @Test @RedisAvailable - public void testPropertiesSimple(){ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - StringRedisTemplate redisTemplate = new StringRedisTemplate(); - RedisProperties redisProperties = - new RedisProperties("foo", this.initTemplate(jcf, redisTemplate)); + public void testPropertiesSimple() { + RedisProperties redisProperties = new RedisProperties("foo", this.redisTemplate); assertEquals(0, redisProperties.size()); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); - MessageChannel redisChannel = context.getBean("simpleProperty", MessageChannel.class); Message message = MessageBuilder.withPayload("bar") .setHeader(RedisHeaders.KEY, "foo") .setHeader("baz", "qux") .build(); - redisChannel.send(message); + this.simplePropertyChannel.send(message); assertEquals("bar", redisProperties.get("qux")); - redisTemplate.delete("foo"); - context.close(); } - private RedisTemplate initTemplate(RedisConnectionFactory rcf, RedisTemplate redisTemplate){ - redisTemplate.setConnectionFactory(rcf); - redisTemplate.afterPropertiesSet(); - return redisTemplate; - } } diff --git a/src/reference/asciidoc/redis.adoc b/src/reference/asciidoc/redis.adoc index 328a1c0922..ec96457881 100644 --- a/src/reference/asciidoc/redis.adoc +++ b/src/reference/asciidoc/redis.adoc @@ -24,12 +24,12 @@ To connect to Redis you would use one of the implementations of the `RedisConnec ---- public interface RedisConnectionFactory extends PersistenceExceptionTranslator { - /** - * Provides a suitable connection for interacting with Redis. - * - * @return connection for interacting with Redis. - */ - RedisConnection getConnection(); + /** + * Provides a suitable connection for interacting with Redis. + * + * @return connection for interacting with Redis. + */ + RedisConnection getConnection(); } ---- @@ -72,7 +72,7 @@ Or in Spring's XML configuration:: [source,xml] ---- - + ---- @@ -337,7 +337,7 @@ It is recommended that this store is used for backing channels, instead of the g - + ---- @@ -530,18 +530,17 @@ Since _Spring Integration 4.0_, the Redis Command Gateway is available to perfor [source,xml] ---- - reply-channel="" <2> - requires-reply="" <3> - reply-timeout="" <4> - connection-factory="" <5> - redis-template="" <6> - arguments-serializer="" <7> - command-expression="" <8> - argument-expressions="" <9> - use-command-variable="" <10> - arguments-strategy="" /> <11> - + request-channel="" <1> + reply-channel="" <2> + requires-reply="" <3> + reply-timeout="" <4> + connection-factory="" <5> + redis-template="" <6> + arguments-serializer="" <7> + command-expression="" <8> + argument-expressions="" <9> + use-command-variable="" <10> + arguments-strategy="" /> <11> ---- @@ -597,8 +596,8 @@ For example to get incremented value from Redis Atomic Number: [source,xml] ---- + reply-channel="replyChannel" + command-expression="'INCR'"/> ---- where the Message `payload` should be a name of `redisCounter`, which may be provided by `org.springframework.data.redis.support.atomic.RedisAtomicInteger` bean definition. @@ -615,16 +614,15 @@ A different UUID is used for each interaction. [source,xml] ---- - reply-channel="" <2> - requires-reply="" <3> - reply-timeout="" <4> - connection-factory="" <5> - queue="" <6> - order="" <7> - serializer="" <8> - extract-payload="" <9> - + request-channel="" <1> + reply-channel="" <2> + requires-reply="" <3> + reply-timeout="" <4> + connection-factory="" <5> + queue="" <6> + order="" <7> + serializer="" <8> + extract-payload=""/> <9> ---- @@ -667,21 +665,21 @@ If this attribute is set to `true`, the `serializer` can't be an empty string be === 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'`: +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'`: [source,xml] ---- - reply-channel="" <2> - executor="" <3> - reply-timeout="" <4> - connection-factory="" <5> - queue="" <6> - order="" <7> - serializer="" <8> - receive-timeout="" <9> - expect-message="" <10> - + request-channel="" <1> + reply-channel="" <2> + executor="" <3> + reply-timeout="" <4> + connection-factory="" <5> + queue="" <6> + order="" <7> + serializer="" <8> + receive-timeout="" <9> + expect-message="" <10> + recovery-interval=""/> <11> ---- @@ -706,7 +704,7 @@ Defaults to `redisConnectionFactory`. Mutually exclusive with 'redis-template' attribute. -<6> The name of the Redis List for the _conversation_`UUID` s. +<6> The name of the Redis List for the _conversation_ `UUID` s. <7> The order for this inbound gateway when multiple gateway are registered thereby @@ -725,6 +723,10 @@ Typically is applied for queue-based limited request-channels. <10> Specify if this Endpoint expects data from the Redis queue to contain entire `Message` s. 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). + +<11> The time in milliseconds for which the listener task should sleep after exceptions on the 'right pop' operation, +before restarting the listener task. + [[redis-lock-registry]] === Redis Lock Registry