From 0e0519ce254ee20b3acc0fe1f1cc32f3a83d300a Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 23 Apr 2020 09:50:50 +0200 Subject: [PATCH] DATAREDIS-1128 - Improve test stability. Replace condition-based polling/waiting with Awaitility. Improve assertions for timing-based checks. Increase number of HyperLogLog samples to reduce probability of clashing elements. Remove long-waiting sleeps that did not lead to an assertion. Original Pull Request: #525 --- pom.xml | 8 ++ .../data/redis/SpinBarrier.java | 25 ++-- .../data/redis/config/NamespaceTest.java | 15 +-- .../AbstractConnectionIntegrationTests.java | 9 +- .../LettuceClusterConnectionTests.java | 2 +- .../LettuceConnectionFactoryTests.java | 19 --- .../LettuceConnectionIntegrationTests.java | 4 +- ...uceConnectionPipelineIntegrationTests.java | 4 +- ...LettuceReactiveScriptingCommandsTests.java | 3 +- .../DefaultHyperLogLogOperationsTests.java | 14 +-- ...eyExpirationEventMessageListenerTests.java | 49 ++++---- .../data/redis/listener/PubSubAwaitUtil.java | 116 ++++++++++++++++++ .../listener/PubSubResubscribeTests.java | 43 +++---- .../data/redis/listener/PubSubTests.java | 14 ++- ...sageListenerContainerIntegrationTests.java | 13 +- ...sageListenerContainerIntegrationTests.java | 37 +++--- .../StreamReceiverIntegrationTests.java | 14 +-- .../redis/test/util/RedisClusterRule.java | 8 ++ 18 files changed, 243 insertions(+), 154 deletions(-) create mode 100644 src/test/java/org/springframework/data/redis/listener/PubSubAwaitUtil.java diff --git a/pom.xml b/pom.xml index 4c91cc337..42369d674 100644 --- a/pom.xml +++ b/pom.xml @@ -17,6 +17,7 @@ 2.3.0.BUILD-SNAPSHOT + 4.0.2 1.1 1.9.2 1.4.8 @@ -224,6 +225,13 @@ test + + org.awaitility + awaitility + ${awaitility} + test + + javax.transaction jta diff --git a/src/test/java/org/springframework/data/redis/SpinBarrier.java b/src/test/java/org/springframework/data/redis/SpinBarrier.java index cff3a757f..27668fc43 100644 --- a/src/test/java/org/springframework/data/redis/SpinBarrier.java +++ b/src/test/java/org/springframework/data/redis/SpinBarrier.java @@ -15,10 +15,17 @@ */ package org.springframework.data.redis; +import java.util.concurrent.TimeUnit; + +import org.awaitility.Awaitility; + /** * @author Jennifer Hickey + * @author Mark Paluch + * @deprecated Use {@link Awaitility}. */ -abstract public class SpinBarrier { +@Deprecated +public abstract class SpinBarrier { /** * Periodically tests for a condition until it is met or a timeout occurs @@ -28,16 +35,12 @@ abstract public class SpinBarrier { * @return true if condition passes, false if condition does not pass within timeout */ public static boolean waitFor(TestCondition condition, long timeout) { - boolean passes = false; - for (long currentTime = System.currentTimeMillis(); System.currentTimeMillis() - currentTime < timeout;) { - if (condition.passes()) { - passes = true; - break; - } - try { - Thread.sleep(100); - } catch (InterruptedException e) {} + + try { + Awaitility.await().atMost(timeout, TimeUnit.MILLISECONDS).until(condition::passes); + return true; + } catch (Exception e) { + return false; } - return passes; } } diff --git a/src/test/java/org/springframework/data/redis/config/NamespaceTest.java b/src/test/java/org/springframework/data/redis/config/NamespaceTest.java index 499375a0d..bedfedde7 100644 --- a/src/test/java/org/springframework/data/redis/config/NamespaceTest.java +++ b/src/test/java/org/springframework/data/redis/config/NamespaceTest.java @@ -17,8 +17,6 @@ package org.springframework.data.redis.config; import static org.assertj.core.api.Assertions.*; -import java.util.concurrent.TimeUnit; - import org.junit.Test; import org.junit.runner.RunWith; @@ -26,12 +24,12 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.test.util.RelaxedJUnit4ClassRunner; -import org.springframework.test.annotation.IfProfileValue; import org.springframework.test.annotation.ProfileValueSourceConfiguration; import org.springframework.test.context.ContextConfiguration; /** * @author Costin Leau + * @author Mark Paluch */ @RunWith(RelaxedJUnit4ClassRunner.class) @ContextConfiguration("namespace.xml") @@ -45,24 +43,13 @@ public class NamespaceTest { @Autowired private StubErrorHandler handler; @Test - @IfProfileValue(name = "runLongTests", value = "true") public void testSanityTest() throws Exception { assertThat(container.isRunning()).isTrue(); - Thread.sleep(TimeUnit.SECONDS.toMillis(8)); } @Test - @IfProfileValue(name = "runLongTests", value = "true") public void testWithMessages() throws Exception { template.convertAndSend("x1", "[X]test"); template.convertAndSend("z1", "[Z]test"); - Thread.sleep(TimeUnit.SECONDS.toMillis(8)); - } - - public void testErrorHandler() throws Exception { - int index = handler.throwables.size(); - template.convertAndSend("exception", "test1"); - handler.throwables.pollLast(3, TimeUnit.SECONDS); - assertThat(handler.throwables.size()).isEqualTo(index + 1); } } diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 0659a322c..ce42cf4a8 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -35,6 +35,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.assertj.core.data.Offset; +import org.awaitility.Awaitility; import org.junit.After; import org.junit.AfterClass; import org.junit.AssumptionViolatedException; @@ -396,7 +397,8 @@ public abstract class AbstractConnectionIntegrationTests { Thread.sleep(200); connection.scriptKill(); getResults(); - assertThat(waitFor(() -> scriptDead.get(), 3000l)).isTrue(); + + Awaitility.await().untilTrue(scriptDead); } @Test @@ -833,7 +835,7 @@ public abstract class AbstractConnectionIntegrationTests { connection.watch("testitnow".getBytes()); // Give some time for watch to be asynch executed in extending tests - Thread.sleep(500); + Thread.sleep(200); DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(connectionFactory.getConnection()); conn2.set("testitnow", "something"); conn2.close(); @@ -852,11 +854,10 @@ public abstract class AbstractConnectionIntegrationTests { connection.watch("testitnow".getBytes()); connection.unwatch(); - connection.multi(); // Give some time for unwatch to be asynch executed - Thread.sleep(100); + Thread.sleep(200); DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(connectionFactory.getConnection()); conn2.set("testitnow", "something"); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java index ac5c839ed..b00a45e61 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java @@ -2365,7 +2365,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { nativeConnection.set(KEY_1, VALUE_1); nativeConnection.get(KEY_1); - assertThat(clusterConnection.keyCommands().idletime(KEY_1_BYTES)).isEqualTo(Duration.ofSeconds(0)); + assertThat(clusterConnection.keyCommands().idletime(KEY_1_BYTES)).isLessThanOrEqualTo(Duration.ofSeconds(2)); } @Test // DATAREDIS-716 diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java index dd6719950..98df4eb15 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java @@ -315,25 +315,6 @@ public class LettuceConnectionFactoryTests { pool.destroy(); } - @Ignore("Uncomment this test to manually check connection reuse in a pool scenario") - @Test - public void testLotsOfConnections() throws InterruptedException { - // Running a netstat here should show only the 8 conns from the pool (plus 2 from setUp and 1 from factory2 - // afterPropertiesSet for shared conn) - DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort()); - pool.afterPropertiesSet(); - final LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool); - factory2.afterPropertiesSet(); - - ConnectionFactoryTracker.add(factory2); - - for (int i = 1; i < 1000; i++) { - Thread th = new Thread(() -> factory2.getConnection().bRPop(50000, "foo".getBytes())); - th.start(); - } - Thread.sleep(234234234); - } - @Ignore("Redis must have requirepass set to run this test") @Test public void testConnectWithPassword() { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java index 4dde945aa..2c842dd6c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java @@ -27,6 +27,7 @@ import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.pool2.impl.GenericObjectPoolConfig; +import org.awaitility.Awaitility; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; @@ -275,7 +276,8 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra th.start(); Thread.sleep(1000); connection.scriptKill(); - assertThat(waitFor(scriptDead::get, 3000l)).isTrue(); + + Awaitility.await().untilTrue(scriptDead); } @Test diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java index c8bed290b..0f6bd59f8 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java @@ -22,6 +22,7 @@ import static org.springframework.data.redis.SpinBarrier.*; import java.util.Arrays; import java.util.concurrent.atomic.AtomicBoolean; +import org.awaitility.Awaitility; import org.junit.Test; import org.junit.runner.RunWith; @@ -79,7 +80,8 @@ public class LettuceConnectionPipelineIntegrationTests extends AbstractConnectio Thread.sleep(1000); connection.scriptKill(); getResults(); - assertThat(waitFor(scriptDead::get, 3000l)).isTrue(); + + Awaitility.await().untilTrue(scriptDead); } @Test diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommandsTests.java index 4d1722193..90e999156 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommandsTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommandsTests.java @@ -20,6 +20,7 @@ import static org.junit.Assume.*; import static org.springframework.data.redis.SpinBarrier.*; import io.lettuce.core.ScriptOutputType; +import org.awaitility.Awaitility; import reactor.test.StepVerifier; import java.nio.ByteBuffer; @@ -174,7 +175,7 @@ public class LettuceReactiveScriptingCommandsTests extends LettuceReactiveComman connection.scriptingCommands().scriptKill().as(StepVerifier::create).expectNext("OK").verifyComplete(); - assertThat(waitFor(scriptDead::get, 3000L)).isTrue(); + Awaitility.await().untilTrue(scriptDead); } private static ByteBuffer wrap(String content) { diff --git a/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java b/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java index 6b0de8fd9..b05a1fd01 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java @@ -35,6 +35,7 @@ import org.springframework.test.annotation.IfProfileValue; /** * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(Parameterized.class) @IfProfileValue(name = "redisVersion", value = "2.8.9+") @@ -132,10 +133,11 @@ public class DefaultHyperLogLogOperationsTests { K key2 = keyFactory.instance(); V v4 = valueFactory.instance(); + V v5 = valueFactory.instance(); hyperLogLogOps.add(key, v1, v2, v3); - hyperLogLogOps.add(key2, v4); - assertThat(hyperLogLogOps.size(key, key2)).isEqualTo(4L); + hyperLogLogOps.add(key2, v4, v5); + assertThat(hyperLogLogOps.size(key, key2)).isGreaterThan(3L); } @Test // DATAREDIS-308 @@ -149,16 +151,14 @@ public class DefaultHyperLogLogOperationsTests { K sourceKey_2 = keyFactory.instance(); V v4 = valueFactory.instance(); + V v5 = valueFactory.instance(); K desinationKey = keyFactory.instance(); hyperLogLogOps.add(sourceKey_1, v1, v2, v3); - hyperLogLogOps.add(sourceKey_2, v4); - - Thread.sleep(10); // give redis a little time to catch up + hyperLogLogOps.add(sourceKey_2, v4, v5); hyperLogLogOps.union(desinationKey, sourceKey_1, sourceKey_2); - Thread.sleep(10); // give redis a little time to catch up - assertThat(hyperLogLogOps.size(desinationKey)).isEqualTo(4L); + assertThat(hyperLogLogOps.size(desinationKey)).isGreaterThan(3L); } } diff --git a/src/test/java/org/springframework/data/redis/listener/KeyExpirationEventMessageListenerTests.java b/src/test/java/org/springframework/data/redis/listener/KeyExpirationEventMessageListenerTests.java index 017296fba..d10f71a2e 100644 --- a/src/test/java/org/springframework/data/redis/listener/KeyExpirationEventMessageListenerTests.java +++ b/src/test/java/org/springframework/data/redis/listener/KeyExpirationEventMessageListenerTests.java @@ -19,7 +19,9 @@ import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; +import org.awaitility.Awaitility; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -37,6 +39,7 @@ import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; /** * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(MockitoJUnitRunner.class) public class KeyExpirationEventMessageListenerTests { @@ -62,18 +65,15 @@ public class KeyExpirationEventMessageListenerTests { listener = new KeyExpirationEventMessageListener(container); listener.setApplicationEventPublisher(publisherMock); listener.init(); + + try (RedisConnection connection = connectionFactory.getConnection()) { + connection.flushAll(); + } } @After public void tearDown() throws Exception { - RedisConnection connection = connectionFactory.getConnection(); - try { - connection.flushAll(); - } finally { - connection.close(); - } - listener.destroy(); container.destroy(); if (connectionFactory instanceof DisposableBean) { @@ -82,25 +82,23 @@ public class KeyExpirationEventMessageListenerTests { } @Test // DATAREDIS-425 - public void listenerShouldPublishEventCorrectly() throws InterruptedException { + public void listenerShouldPublishEventCorrectly() { byte[] key = ("to-expire:" + UUID.randomUUID().toString()).getBytes(); + AtomicBoolean called = new AtomicBoolean(); + doAnswer(invocation -> { + called.set(true); + return null; + }).when(publisherMock).publishEvent(any(ApplicationEvent.class)); - RedisConnection connection = connectionFactory.getConnection(); - try { - connection.setEx(key, 2, "foo".getBytes()); + try (RedisConnection connection = connectionFactory.getConnection()) { - int iteration = 0; - while (connection.get(key) != null || iteration >= 3) { - - Thread.sleep(2000); - iteration++; - } - } finally { - connection.close(); + connection.pSetEx(key, 1, key); + Awaitility.await().until(() -> !connection.exists(key)); } - Thread.sleep(2000); + Awaitility.await().untilTrue(called); + ArgumentCaptor captor = ArgumentCaptor.forClass(ApplicationEvent.class); verify(publisherMock, times(1)).publishEvent(captor.capture()); @@ -112,18 +110,13 @@ public class KeyExpirationEventMessageListenerTests { byte[] key = ("to-delete:" + UUID.randomUUID().toString()).getBytes(); - RedisConnection connection = connectionFactory.getConnection(); - try { + try (RedisConnection connection = connectionFactory.getConnection()) { connection.setEx(key, 10, "foo".getBytes()); - Thread.sleep(2000); connection.del(key); - Thread.sleep(2000); - } finally { - connection.close(); } - Thread.sleep(2000); - verifyZeroInteractions(publisherMock); + Thread.sleep(500); + verifyNoInteractions(publisherMock); } } diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubAwaitUtil.java b/src/test/java/org/springframework/data/redis/listener/PubSubAwaitUtil.java new file mode 100644 index 000000000..19168607f --- /dev/null +++ b/src/test/java/org/springframework/data/redis/listener/PubSubAwaitUtil.java @@ -0,0 +1,116 @@ +/* + * Copyright 2020 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 + * + * https://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.data.redis.listener; + +import io.lettuce.core.codec.ByteArrayCodec; +import io.lettuce.core.output.ArrayOutput; +import io.lettuce.core.output.IntegerOutput; + +import java.util.List; +import java.util.concurrent.Callable; + +import org.awaitility.Awaitility; + +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.lettuce.LettuceConnection; + +/** + * Utility providing await support. + * + * @author Mark Paluch + */ +class PubSubAwaitUtil { + + /** + * Run the {@link Runnable} and wait until the number of Pub/Sub patterns has increased in comparison to before + * running the callback. + * + * @param connectionFactory + * @param runnable + */ + public static void runAndAwaitPatternSubscription(RedisConnectionFactory connectionFactory, Runnable runnable) { + + try (RedisConnection connection = connectionFactory.getConnection()) { + + Number before = numPat(connection); + + runnable.run(); + + runAndAwait(runnable, () -> { + + Number after = numPat(connection); + + return after.longValue() > before.longValue(); + }); + } + } + + /** + * Run the {@link Runnable} and wait until the number of channel subscribers has increased in comparison to before + * running the callback. + * + * @param connectionFactory + * @param channel + * @param runnable + */ + public static void runAndAwaitChannelSubscription(RedisConnectionFactory connectionFactory, String channel, + Runnable runnable) { + + try (RedisConnection connection = connectionFactory.getConnection()) { + + Number before = numSub(connection, channel); + + runnable.run(); + + runAndAwait(runnable, () -> { + + Number after = numSub(connection, channel); + + return after.longValue() > before.longValue(); + }); + } + } + + private static long numPat(RedisConnection connection) { + + if (connection instanceof LettuceConnection) { + return ((Number) ((LettuceConnection) connection).execute("PUBSUB", new IntegerOutput<>(ByteArrayCodec.INSTANCE), + "NUMPAT".getBytes())).longValue(); + } + + return ((Number) connection.execute("PUBSUB", "NUMPAT".getBytes())).longValue(); + } + + private static long numSub(RedisConnection connection, String channel) { + + List pubsub; + if (connection instanceof LettuceConnection) { + pubsub = (List) ((LettuceConnection) connection).execute("PUBSUB", new ArrayOutput<>(ByteArrayCodec.INSTANCE), + "NUMSUB".getBytes(), channel.getBytes()); + } else { + pubsub = (List) connection.execute("PUBSUB", "NUMSUB".getBytes(), channel.getBytes()); + } + + return ((Number) pubsub.get(1)).longValue(); + } + + private static void runAndAwait(Runnable runnable, Callable predicate) { + + runnable.run(); + Awaitility.await().until(predicate); + } +} diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java index 1346db740..9978383fd 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java @@ -17,7 +17,6 @@ package org.springframework.data.redis.listener; import static org.assertj.core.api.Assertions.*; import static org.junit.Assume.*; -import static org.springframework.data.redis.SpinBarrier.*; import java.util.ArrayList; import java.util.Arrays; @@ -26,6 +25,7 @@ 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; @@ -39,7 +39,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import org.junit.runners.Parameterized.Parameters; -import org.junit.runners.model.Statement; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.SyncTaskExecutor; @@ -150,17 +149,15 @@ public class PubSubResubscribeTests { container = new RedisMessageListenerContainer(); container.setConnectionFactory(template.getConnectionFactory()); container.setBeanName("container"); - container.addMessageListener(adapter, new ChannelTopic(CHANNEL)); container.setTaskExecutor(new SyncTaskExecutor()); container.setSubscriptionExecutor(new SimpleAsyncTaskExecutor()); container.afterPropertiesSet(); container.start(); - - waitFor(container.getConnectionFactory().getConnection()::isSubscribed, 1000); } @After public void tearDown() { + container.stop(); bag.clear(); } @@ -180,7 +177,6 @@ public class PubSubResubscribeTests { // remove adapter from all channels container.addMessageListener(anotherListener, new PatternTopic(PATTERN)); - container.removeMessageListener(adapter); // Wait for async subscription tasks to setup Thread.sleep(400); @@ -275,18 +271,24 @@ public class PubSubResubscribeTests { @Test public void testInitializeContainerWithMultipleTopicsIncludingPattern() throws Exception { - container.removeMessageListener(adapter); + assumeFalse(isClusterAware(template.getConnectionFactory())); + container.stop(); - container.addMessageListener(adapter, - Arrays.asList(new Topic[] { new ChannelTopic(CHANNEL), new PatternTopic("s*") })); - container.start(); + + String uniqueChannel = "random-" + UUID.randomUUID().toString(); + PubSubAwaitUtil.runAndAwaitChannelSubscription(template.getConnectionFactory(), uniqueChannel, () -> { + + container.addMessageListener(adapter, + Arrays.asList(new Topic[] { new ChannelTopic(uniqueChannel), new PatternTopic("s*") })); + container.start(); + }); // timing: There's currently no other way to synchronize // than to hope the subscribe/unsubscribe are executed within the time. - Thread.sleep(1000); + Thread.sleep(50); template.convertAndSend("somechannel", "HELLO"); - template.convertAndSend(CHANNEL, "WORLD"); + template.convertAndSend(uniqueChannel, "WORLD"); Set set = new LinkedHashSet<>(); set.add(bag.poll(500, TimeUnit.MILLISECONDS)); @@ -313,17 +315,16 @@ public class PubSubResubscribeTests { } private static boolean clusterAvailable() { + return new RedisClusterRule().isAvailable(); + } - try { - new RedisClusterRule().apply(new Statement() { - @Override - public void evaluate() throws Throwable { + private static boolean isClusterAware(RedisConnectionFactory connectionFactory) { - } - }, null).evaluate(); - } catch (Throwable throwable) { - return false; + if (connectionFactory instanceof LettuceConnectionFactory) { + return ((LettuceConnectionFactory) connectionFactory).isClusterAware(); + } else if (connectionFactory instanceof JedisConnectionFactory) { + return ((JedisConnectionFactory) connectionFactory).isRedisClusterAware(); } - return true; + return false; } } diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubTests.java b/src/test/java/org/springframework/data/redis/listener/PubSubTests.java index 91316cddf..c4379a1cc 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubTests.java @@ -20,6 +20,7 @@ import static org.junit.Assume.*; 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; @@ -39,6 +40,7 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.ObjectFactory; +import org.springframework.data.redis.connection.ConnectionUtils; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; @@ -185,13 +187,15 @@ public class PubSubTests { @Test // DATAREDIS-251 public void testStartListenersToNoSpecificChannelTest() throws InterruptedException { - assumeTrue(isClusterAware(template.getConnectionFactory())); + assumeFalse(isClusterAware(template.getConnectionFactory())); + assumeTrue(ConnectionUtils.isJedis(template.getConnectionFactory())); - container.removeMessageListener(adapter, new ChannelTopic(CHANNEL)); - container.addMessageListener(adapter, Arrays.asList(new PatternTopic("*"))); - container.start(); + PubSubAwaitUtil.runAndAwaitPatternSubscription(template.getRequiredConnectionFactory(), () -> { - Thread.sleep(1000); // give the container a little time to recover + container.removeMessageListener(adapter, new ChannelTopic(CHANNEL)); + container.addMessageListener(adapter, Collections.singletonList(new PatternTopic(CHANNEL + "*"))); + container.start(); + }); T payload = getT(); diff --git a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java index d4a8f1a76..2cf0f36cd 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java @@ -27,6 +27,7 @@ import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; +import org.awaitility.Awaitility; import org.junit.After; import org.junit.AfterClass; import org.junit.Before; @@ -34,6 +35,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import org.junit.runners.Parameterized.Parameters; + import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.connection.ReactiveSubscription; import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; @@ -198,16 +200,7 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests { private static Runnable awaitSubscription(Supplier> activeSubscriptions) { return () -> { - - try { - - while (activeSubscriptions.get().isEmpty()) { - Thread.sleep(10); - } - - } catch (InterruptedException e) { - return; - } + Awaitility.await().until(() -> !activeSubscriptions.get().isEmpty()); }; } } diff --git a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java index 518f8142b..c2b5d9d28 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java @@ -30,6 +30,7 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import org.awaitility.Awaitility; import org.junit.AfterClass; import org.junit.Before; import org.junit.BeforeClass; @@ -66,6 +67,7 @@ public class StreamMessageListenerContainerIntegrationTests { private static final RedisStandaloneConfiguration standaloneConfiguration = new RedisStandaloneConfiguration( SettingsUtils.getHost(), SettingsUtils.getPort()); + public static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(2); private static RedisConnectionFactory connectionFactory; @@ -115,7 +117,7 @@ public class StreamMessageListenerContainerIntegrationTests { container.start(); Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1")); redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2")); @@ -143,7 +145,7 @@ public class StreamMessageListenerContainerIntegrationTests { container.start(); Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); redisTemplate.opsForStream().add(ObjectRecord.create("my-stream", "value1")); @@ -167,7 +169,7 @@ public class StreamMessageListenerContainerIntegrationTests { container.start(); Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); redisTemplate.opsForStream().add(ObjectRecord.create("my-stream", new LoginEvent("Walter", "White"))); @@ -192,7 +194,7 @@ public class StreamMessageListenerContainerIntegrationTests { Subscription subscription = container.receive(Consumer.from("my-group", "my-consumer"), StreamOffset.create("my-stream", ReadOffset.lastConsumed()), queue::add); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2")); @@ -218,7 +220,7 @@ public class StreamMessageListenerContainerIntegrationTests { Subscription subscription = container.receiveAutoAck(Consumer.from("my-group", "my-consumer"), StreamOffset.create("my-stream", ReadOffset.lastConsumed()), queue::add); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2")); @@ -245,7 +247,7 @@ public class StreamMessageListenerContainerIntegrationTests { Subscription subscription = container.receive(Consumer.from("my-group", "my-consumer"), StreamOffset.create("my-stream", ReadOffset.lastConsumed()), it -> {}); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); Throwable error = failures.poll(1, TimeUnit.SECONDS); assertThat(failures).isEmpty(); @@ -271,14 +273,11 @@ public class StreamMessageListenerContainerIntegrationTests { container.start(); Subscription subscription = container.register(readRequest, it -> {}); - subscription.await(Duration.ofSeconds(1)); redisTemplate.delete("my-stream"); - subscription.await(Duration.ofSeconds(1)); - - assertThat(failures.poll(1, TimeUnit.SECONDS)).isNotNull(); + assertThat(failures.poll(3, TimeUnit.SECONDS)).isNotNull(); assertThat(subscription.isActive()).isFalse(); cancelAwait(subscription); @@ -305,7 +304,7 @@ public class StreamMessageListenerContainerIntegrationTests { container.start(); Subscription subscription = container.register(readRequest, it -> {}); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); redisTemplate.delete("my-stream"); @@ -326,7 +325,7 @@ public class StreamMessageListenerContainerIntegrationTests { container.start(); Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); cancelAwait(subscription); redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value4")); @@ -344,17 +343,15 @@ public class StreamMessageListenerContainerIntegrationTests { container.start(); Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); container.stop(); - while (subscription.isActive()) { - Thread.sleep(10); - } + Awaitility.await().atMost(DEFAULT_TIMEOUT).until(() -> !subscription.isActive()); container.start(); - subscription.await(Duration.ofSeconds(2)); + subscription.await(DEFAULT_TIMEOUT); redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1")); @@ -363,13 +360,11 @@ public class StreamMessageListenerContainerIntegrationTests { cancelAwait(subscription); } - private static void cancelAwait(Subscription subscription) throws InterruptedException { + private static void cancelAwait(Subscription subscription) { subscription.cancel(); - while (subscription.isActive()) { - Thread.sleep(10); - } + Awaitility.await().atMost(DEFAULT_TIMEOUT).until(() -> !subscription.isActive()); } private int getNumberOfPending(String stream, String group) { diff --git a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java index ba94187e7..354af8d65 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java @@ -183,24 +183,18 @@ public class StreamReceiverIntegrationTests { messages.as(publisher -> StepVerifier.create(publisher, 0)) // .thenRequest(1) // + .thenAwait(Duration.ofMillis(500)) // .then(() -> { - try { - Thread.sleep(500); - reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1")) - .subscribe(); - } catch (InterruptedException e) {} + reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1")).subscribe(); }) // .expectNextCount(1) // .then(() -> { reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2")).subscribe(); }) // .thenRequest(1) // + .thenAwait(Duration.ofMillis(500)) // .then(() -> { - try { - Thread.sleep(500); - reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value3")) - .subscribe(); - } catch (InterruptedException e) {} + reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value3")).subscribe(); }).consumeNextWith(it -> { assertThat(it.getStream()).isEqualTo("my-stream"); diff --git a/src/test/java/org/springframework/data/redis/test/util/RedisClusterRule.java b/src/test/java/org/springframework/data/redis/test/util/RedisClusterRule.java index 874069e69..7e47f1d4f 100644 --- a/src/test/java/org/springframework/data/redis/test/util/RedisClusterRule.java +++ b/src/test/java/org/springframework/data/redis/test/util/RedisClusterRule.java @@ -30,6 +30,7 @@ import org.springframework.data.redis.connection.jedis.JedisConverters; * Simple {@link TestRule} implementation that check Redis is running in cluster mode. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ public class RedisClusterRule extends ExternalResource { @@ -67,6 +68,13 @@ public class RedisClusterRule extends ExternalResource { return this.clusterConfig; } + /** + * @return {@literal true} if Redis Cluster is available. + */ + public boolean isAvailable() { + return "cluster".equals(mode); + } + private void init() { if (clusterConfig == null) {