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

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

Author: mp911de

src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java

@@ -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 @@ void testContainerPatternResubscribe() throws Exception {
 		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 @@ void testContainerPatternResubscribe() throws Exception {
 		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 @@ void testContainerChannelResubscribe() throws Exception {
 		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 @@ void testInitializeContainerWithMultipleTopicsIncludingPattern() throws Exceptio
 
 		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 @@ void testInitializeContainerWithMultipleTopicsIncludingPattern() throws Exceptio
 
 		// 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 {

src/test/java/org/springframework/data/redis/listener/PubSubTestParams.java

@@ -21,7 +21,6 @@
 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 static Collection<Object[]> testParams() {
 		// 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 static Collection<Object[]> testParams() {
 		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()) {
 

src/test/java/org/springframework/data/redis/listener/PubSubTests.java

@@ -17,12 +17,12 @@
 
 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 @@ protected void doExecute(Runnable task) {
 		container.start();
 
 		phaser.arriveAndAwaitAdvance();
-		Thread.sleep(50);
+		Thread.sleep(250);
 	}
 
 	@AfterEach
@@ -134,11 +134,7 @@ void testContainerSubscribe() throws Exception {
 		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 @@ void testStartListenersToNoSpecificChannelTest() throws InterruptedException {
 
 		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) {