diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml
index c32aac1c79..361cdd497b 100644
--- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml
@@ -16,9 +16,10 @@
-
@@ -28,8 +29,10 @@
-
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 c223d378d0..099aeafdcc 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
@@ -47,6 +47,7 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
+import org.springframework.context.Lifecycle;
import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
@@ -90,9 +91,15 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@Autowired
private PollableChannel fromChannel;
+ @Autowired
+ private Lifecycle fromChannelEndpoint;
+
@Autowired
private MessageChannel symmetricalInputChannel;
+ @Autowired
+ private Lifecycle symmetricalRedisChannelEndpoint;
+
@Autowired
private PollableChannel symmetricalOutputChannel;
@@ -108,6 +115,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
redisTemplate.afterPropertiesSet();
+ redisTemplate.delete(queueName);
String payload = "testing";
@@ -137,6 +145,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
assertEquals(payload2, receive.getPayload());
endpoint.stop();
+ redisTemplate.delete(queueName);
}
@Test
@@ -151,6 +160,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
redisTemplate.afterPropertiesSet();
+ redisTemplate.delete(queueName);
Message> message = MessageBuilder.withPayload("testing").build();
@@ -189,16 +199,21 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
Matchers.containsString("java.lang.String cannot be cast to org.springframework.messaging.Message"));
endpoint.stop();
+ redisTemplate.delete(queueName);
}
@Test
@RedisAvailable
public void testInt3017IntegrationInbound() {
- String payload = new Date().toString();
-
+ String queueName = "si.test.redisQueueInboundChannelAdapterTests2";
RedisTemplate redisTemplate = new StringRedisTemplate();
redisTemplate.setConnectionFactory(this.connectionFactory);
redisTemplate.afterPropertiesSet();
+ redisTemplate.delete(queueName);
+
+ this.fromChannelEndpoint.start();
+ String payload = new Date().toString();
+
redisTemplate.boundListOps("si.test.Int3017IntegrationInbound")
.leftPush("{\"payload\":\"" + payload + "\",\"headers\":{}}");
@@ -206,14 +221,22 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
Message> receive = this.fromChannel.receive(10000);
assertNotNull(receive);
assertEquals(payload, receive.getPayload());
+ this.fromChannelEndpoint.stop();
+ redisTemplate.delete(queueName);
}
@Test
@RedisAvailable
public void testInt3017IntegrationSymmetrical() {
+ String queueName = "si.test.Int3017IntegrationSymmetrical";
+ RedisTemplate redisTemplate = new StringRedisTemplate();
+ redisTemplate.setConnectionFactory(this.connectionFactory);
+ redisTemplate.afterPropertiesSet();
+ redisTemplate.delete(queueName);
+ this.symmetricalRedisChannelEndpoint.start();
UUID payload = UUID.randomUUID();
Message message = MessageBuilder.withPayload(payload)
- .setHeader("redis_queue", "si.test.Int3017IntegrationSymmetrical")
+ .setHeader("redis_queue", queueName)
.build();
this.symmetricalInputChannel.send(message);
@@ -221,6 +244,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
Message> receive = this.symmetricalOutputChannel.receive(10000);
assertNotNull(receive);
assertEquals(payload, receive.getPayload());
+ this.symmetricalRedisChannelEndpoint.stop();
+ redisTemplate.delete(queueName);
}
@Test
@@ -235,6 +260,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
redisTemplate.afterPropertiesSet();
+ redisTemplate.delete(queueName);
while (redisTemplate.boundListOps(queueName).rightPop() != null) {
// drain
@@ -272,6 +298,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
assertTrue(stopLatch.await(21, TimeUnit.SECONDS));
verify(boundListOperations, atLeastOnce()).rightPush(any(byte[].class));
+ redisTemplate.delete(queueName);
}
@@ -286,7 +313,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
final CountDownLatch exceptionsLatch = new CountDownLatch(2);
- RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
+ RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName,
+ this.connectionFactory);
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
endpoint.setApplicationEventPublisher(event -> {
exceptionEvents.add((ApplicationEvent) event);
@@ -329,6 +357,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
assertEquals(payload, receive.getPayload());
endpoint.stop();
+ redisTemplate.delete(queueName);
}
@Test
@@ -343,6 +372,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
redisTemplate.afterPropertiesSet();
+ redisTemplate.delete(queueName);
String payload = "testing";
@@ -373,6 +403,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
assertEquals(payload2, receive.getPayload());
endpoint.stop();
+ redisTemplate.delete(queueName);
}
private void waitListening(RedisQueueMessageDrivenEndpoint endpoint) throws InterruptedException {