Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Replace poll and assert with await for Pub/Sub tests.

Browse Source 
We now await until a message has arrived in our pub/sub tests. Previously, we polled the message listener which could lead to false positives by delayed messages.

We also exclude raw (byte-array) cases as byte arrays cannot be easily checked for equality without further instrumentation.

Closes: #1834
Original Pull Request: #1978
This commit is contained in:
Mark Paluch
2021-02-18 09:44:03 +01:00
committed by Christoph Strobl
parent 4faf12db2e
commit 31c728a594
3 changed files with 15 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java
View File

@@ -15,20 +15,17 @@
*/
package org.springframework.data.redis.listener;
import static org.assertj.core.api.Assertions.*;
import static org.awaitility.Awaitility.*;
import static org.junit.Assume.*;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.junit.jupiter.api.AfterEach;
@@ -156,18 +153,7 @@ public class PubSubResubscribeTests {
template.convertAndSend(CHANNEL, payload1);
template.convertAndSend(ANOTHER_CHANNEL, payload2);
// anotherListener receives both messages
List<String> msgs = new ArrayList<>();
msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS));
msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS));
assertThat(msgs.size()).isEqualTo(2);
assertThat(msgs).contains(payload1);
assertThat(msgs).contains(payload2);
msgs.clear();
// unsubscribed adapter did not receive message
assertThat(bag.poll(500, TimeUnit.MILLISECONDS)).isNull();
await().atMost(Duration.ofSeconds(2)).until(() -> bag2.contains(payload1) && bag2.contains(payload2));
// bind original listener on another channel
container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL));
@@ -178,21 +164,10 @@ public class PubSubResubscribeTests {
template.convertAndSend(CHANNEL, payload1);
template.convertAndSend(ANOTHER_CHANNEL, payload2);
// original listener received only one message on another channel
msgs.clear();
msgs.add(bag.poll(500, TimeUnit.MILLISECONDS));
msgs.add(bag.poll(500, TimeUnit.MILLISECONDS));
assertThat(msgs).contains(payload2);
assertThat(msgs.contains(null)).isTrue();
await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains(payload2));
// another listener receives messages on both channels
msgs.clear();
msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS));
msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS));
assertThat(msgs.size()).isEqualTo(2);
assertThat(msgs).contains(payload1);
assertThat(msgs).contains(payload2);
await().atMost(Duration.ofSeconds(2)).until(() -> bag2.contains(payload1) && bag2.contains(payload2));
}
@ParameterizedRedisTest
@@ -222,15 +197,7 @@ public class PubSubResubscribeTests {
template.convertAndSend(ANOTHER_CHANNEL, anotherPayload1);
template.convertAndSend(ANOTHER_CHANNEL, anotherPayload2);
Set<String> set = new LinkedHashSet<>();
set.add(bag.poll(500, TimeUnit.MILLISECONDS));
set.add(bag.poll(500, TimeUnit.MILLISECONDS));
assertThat(set.contains(payload1)).isFalse();
assertThat(set.contains(payload2)).isFalse();
assertThat(set.contains(anotherPayload1)).isTrue();
assertThat(set.contains(anotherPayload2)).isTrue();
await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains(anotherPayload1) && bag.contains(anotherPayload2));
}
/**
@@ -246,8 +213,8 @@ public class PubSubResubscribeTests {
container.stop();
String uniqueChannel = "random-" + UUID.randomUUID().toString();
PubSubAwaitUtil.runAndAwaitChannelSubscription(template.getConnectionFactory(), uniqueChannel, () -> {
String uniqueChannel = "random-" + UUID.randomUUID();
PubSubAwaitUtil.runAndAwaitPatternSubscription(template.getConnectionFactory(), () -> {
container.addMessageListener(adapter,
Arrays.asList(new Topic[] { new ChannelTopic(uniqueChannel), new PatternTopic("s*") }));
@@ -256,16 +223,12 @@ public class PubSubResubscribeTests {
// timing: There's currently no other way to synchronize
// than to hope the subscribe/unsubscribe are executed within the time.
Thread.sleep(50);
Thread.sleep(250);
template.convertAndSend("somechannel", "HELLO");
template.convertAndSend(uniqueChannel, "WORLD");
Set<String> set = new LinkedHashSet<>();
set.add(bag.poll(500, TimeUnit.MILLISECONDS));
set.add(bag.poll(500, TimeUnit.MILLISECONDS));
assertThat(set).isEqualTo(new HashSet<>(Arrays.asList(new String[] { "HELLO", "WORLD" })));
await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains("HELLO") && bag.contains("WORLD"));
}
private class MessageHandler {

3
src/test/java/org/springframework/data/redis/listener/PubSubTestParams.java
View File

@@ -21,7 +21,6 @@ import java.util.Collection;
import org.springframework.data.redis.ObjectFactory;
import org.springframework.data.redis.Person;
import org.springframework.data.redis.PersonObjectFactory;
import org.springframework.data.redis.RawObjectFactory;
import org.springframework.data.redis.StringObjectFactory;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.connection.jedis.extension.JedisConnectionFactoryExtension;
@@ -44,7 +43,6 @@ public class PubSubTestParams {
// create Jedis Factory
ObjectFactory<String> stringFactory = new StringObjectFactory();
ObjectFactory<Person> personFactory = new PersonObjectFactory();
ObjectFactory<byte[]> rawFactory = new RawObjectFactory();
JedisConnectionFactory jedisConnFactory = JedisConnectionFactoryExtension
.getNewConnectionFactory(RedisStanalone.class);
@@ -78,7 +76,6 @@ public class PubSubTestParams {
parameters.add(new Object[] { personFactory, personTemplate });
parameters.add(new Object[] { stringFactory, stringTemplateLtc });
parameters.add(new Object[] { personFactory, personTemplateLtc });
parameters.add(new Object[] { rawFactory, rawTemplateLtc });
if (clusterAvailable()) {

17
src/test/java/org/springframework/data/redis/listener/PubSubTests.java
View File

@@ -17,12 +17,12 @@ package org.springframework.data.redis.listener;
import static org.assertj.core.api.Assertions.*;
import static org.assertj.core.api.Assumptions.*;
import static org.awaitility.Awaitility.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Phaser;
@@ -108,7 +108,7 @@ public class PubSubTests<T> {
container.start();
phaser.arriveAndAwaitAdvance();
Thread.sleep(50);
Thread.sleep(250);
}
@AfterEach
@@ -134,11 +134,7 @@ public class PubSubTests<T> {
template.convertAndSend(CHANNEL, payload1);
template.convertAndSend(CHANNEL, payload2);
Set<T> set = new LinkedHashSet<>();
set.add((T) bag.poll(1, TimeUnit.SECONDS));
set.add((T) bag.poll(1, TimeUnit.SECONDS));
assertThat(set).contains(payload1, payload2);
await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains(payload1) && bag.contains(payload2));
}
@ParameterizedRedisTest
@@ -192,10 +188,7 @@ public class PubSubTests<T> {
template.convertAndSend(CHANNEL, payload);
Set<T> set = new LinkedHashSet<>();
set.add((T) bag.poll(3, TimeUnit.SECONDS));
assertThat(set).contains(payload);
await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains(payload));
}
private static boolean isClusterAware(RedisConnectionFactory connectionFactory) {