diff --git a/src/main/asciidoc/reference/redis-messaging.adoc b/src/main/asciidoc/reference/redis-messaging.adoc index 917bbbee2..bc62ee82f 100644 --- a/src/main/asciidoc/reference/redis-messaging.adoc +++ b/src/main/asciidoc/reference/redis-messaging.adoc @@ -24,7 +24,7 @@ byte[] msg = ... byte[] channel = ... con.publish(msg, channel); // send message through RedisTemplate RedisTemplate template = ... -template.convertAndSend("hello!", "world"); +Long numberOfClients = template.convertAndSend("hello!", "world"); ---- [[redis:pubsub:subscribe]] diff --git a/src/main/java/org/springframework/data/redis/core/RedisOperations.java b/src/main/java/org/springframework/data/redis/core/RedisOperations.java index a013ca3a4..463526fdd 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/RedisOperations.java @@ -46,6 +46,7 @@ import org.springframework.util.Assert; * @author ihaohong * @author Todd Merrill * @author Chen Li + * @author Vedran Pavic */ public interface RedisOperations { @@ -591,9 +592,11 @@ public interface RedisOperations { * * @param destination the channel to publish to, must not be {@literal null}. * @param message message to publish + * @return the number of clients that received the message * @see Redis Documentation: PUBLISH */ - void convertAndSend(String destination, Object message); + @Nullable + Long convertAndSend(String destination, Object message); // ------------------------------------------------------------------------- // Methods to obtain specific operations interface objects. diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java index fe665c648..313a5e898 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java @@ -86,6 +86,7 @@ import org.springframework.util.CollectionUtils; * @author Denis Zavedeev * @author ihaohong * @author Chen Li + * @author Vedran Pavic * @param the Redis key type against which the template works (usually a String) * @param the Redis value type against which the template works * @see StringRedisTemplate @@ -942,14 +943,14 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation } @Override - public void convertAndSend(String channel, Object message) { + public Long convertAndSend(String channel, Object message) { Assert.hasText(channel, "a non-empty channel is required"); byte[] rawChannel = rawString(channel); byte[] rawMessage = rawValue(message); - executeWithoutResult(connection -> connection.publish(rawChannel, rawMessage)); + return execute(connection -> connection.publish(rawChannel, rawMessage), true); } private void executeWithoutResult(Consumer action) { diff --git a/src/test/java/org/springframework/data/redis/config/NamespaceIntegrationTests.java b/src/test/java/org/springframework/data/redis/config/NamespaceIntegrationTests.java index 325e863e2..be11bcc2e 100644 --- a/src/test/java/org/springframework/data/redis/config/NamespaceIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/config/NamespaceIntegrationTests.java @@ -27,6 +27,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Costin Leau * @author Mark Paluch + * @author Vedran Pavic */ @SpringJUnitConfig(locations = "namespace.xml") class NamespaceIntegrationTests { @@ -43,8 +44,8 @@ class NamespaceIntegrationTests { } @Test - void testWithMessages() throws Exception { - template.convertAndSend("x1", "[X]test"); - template.convertAndSend("z1", "[Z]test"); + void testWithMessages() { + assertThat(template.convertAndSend("x1", "[X]test")).isEqualTo(1L); + assertThat(template.convertAndSend("z1", "[Z]test")).isEqualTo(1L); } } diff --git a/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java index 916761bc7..49b29dca3 100644 --- a/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java @@ -63,6 +63,7 @@ import org.springframework.data.redis.test.util.CollectionAwareComparator; * @author ihaohong * @author Hendrik Duerkop * @author Chen Li + * @author Vedran Pavic */ @MethodSource("testParams") public class RedisTemplateIntegrationTests { @@ -815,10 +816,10 @@ public class RedisTemplateIntegrationTests { } @ParameterizedRedisTest - public void testConvertAndSend() { + void testConvertAndSend() { V value1 = valueFactory.instance(); // Make sure basic message sent without Exception on serialization - redisTemplate.convertAndSend("Channel", value1); + assertThat(redisTemplate.convertAndSend("Channel", value1)).isEqualTo(0L); } @ParameterizedRedisTest 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 6db0de0b3..de242de82 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java @@ -15,6 +15,7 @@ */ package org.springframework.data.redis.listener; +import static org.assertj.core.api.Assertions.*; import static org.awaitility.Awaitility.*; import static org.junit.Assume.*; @@ -53,6 +54,7 @@ import org.springframework.data.redis.test.extension.parametrized.ParameterizedR * @author Jennifer Hickey * @author Christoph Strobl * @author Mark Paluch + * @author Vedran Pavic */ @MethodSource("testParams") @EnabledIfLongRunningTest @@ -126,7 +128,7 @@ public class PubSubResubscribeTests { @ParameterizedRedisTest @EnabledIfLongRunningTest - void testContainerPatternResubscribe() throws Exception { + void testContainerPatternResubscribe() { String payload1 = "do"; String payload2 = "re mi"; @@ -143,11 +145,11 @@ public class PubSubResubscribeTests { container.addMessageListener(anotherListener, new PatternTopic(PATTERN)); // Wait for async subscription tasks to setup - Thread.sleep(400); - - // test no messages are sent just to patterns - template.convertAndSend(CHANNEL, payload1); - template.convertAndSend(ANOTHER_CHANNEL, payload2); + await().atMost(Duration.ofMillis(600)).untilAsserted(() -> { + // test no messages are sent just to patterns + assertThat(template.convertAndSend(CHANNEL, payload1)).isEqualTo(1L); + assertThat(template.convertAndSend(ANOTHER_CHANNEL, payload2)).isEqualTo(1L); + }); await().atMost(Duration.ofSeconds(2)).until(() -> bag2.contains(payload1) && bag2.contains(payload2)); @@ -155,10 +157,10 @@ public class PubSubResubscribeTests { container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL)); // Wait for async subscription tasks to setup - Thread.sleep(400); - - template.convertAndSend(CHANNEL, payload1); - template.convertAndSend(ANOTHER_CHANNEL, payload2); + await().atMost(Duration.ofMillis(400)).untilAsserted(() -> { + assertThat(template.convertAndSend(CHANNEL, payload1)).isEqualTo(1L); + assertThat(template.convertAndSend(ANOTHER_CHANNEL, payload2)).isEqualTo(2L); + }); await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains(payload2)); @@ -167,7 +169,7 @@ public class PubSubResubscribeTests { } @ParameterizedRedisTest - void testContainerChannelResubscribe() throws Exception { + void testContainerChannelResubscribe() { String payload1 = "do"; String payload2 = "re mi"; @@ -183,15 +185,15 @@ 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(400); + await().atMost(Duration.ofMillis(400)).untilAsserted(() -> { + // Listener removed from channel + assertThat(template.convertAndSend(CHANNEL, payload1)).isEqualTo(0L); + assertThat(template.convertAndSend(CHANNEL, payload2)).isEqualTo(0L); - // Listener removed from channel - template.convertAndSend(CHANNEL, payload1); - template.convertAndSend(CHANNEL, payload2); - - // Listener receives messages on another channel - template.convertAndSend(ANOTHER_CHANNEL, anotherPayload1); - template.convertAndSend(ANOTHER_CHANNEL, anotherPayload2); + // Listener receives messages on another channel + assertThat(template.convertAndSend(ANOTHER_CHANNEL, anotherPayload1)).isEqualTo(1L); + assertThat(template.convertAndSend(ANOTHER_CHANNEL, anotherPayload2)).isEqualTo(1L); + }); await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains(anotherPayload1) && bag.contains(anotherPayload2)); } @@ -199,11 +201,9 @@ public class PubSubResubscribeTests { /** * Validates the behavior of {@link RedisMessageListenerContainer} when it needs to spin up a thread executing its * PatternSubscriptionTask - * - * @throws Exception */ @ParameterizedRedisTest - void testInitializeContainerWithMultipleTopicsIncludingPattern() throws Exception { + void testInitializeContainerWithMultipleTopicsIncludingPattern() { assumeFalse(isClusterAware(template.getConnectionFactory())); @@ -217,10 +217,10 @@ 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(250); - - template.convertAndSend("somechannel", "HELLO"); - template.convertAndSend(uniqueChannel, "WORLD"); + await().atMost(Duration.ofMillis(250)).untilAsserted(() -> { + assertThat(template.convertAndSend("somechannel", "HELLO")).isEqualTo(1L); + assertThat(template.convertAndSend(uniqueChannel, "WORLD")).isEqualTo(1L); + }); await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains("HELLO") && bag.contains("WORLD")); } 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 1a5fc2688..580b668ed 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubTests.java @@ -49,6 +49,7 @@ import org.springframework.data.redis.test.extension.parametrized.ParameterizedR * @author Costin Leau * @author Jennifer Hickey * @author Mark Paluch + * @author Vedran Pavic */ @MethodSource("testParams") public class PubSubTests { @@ -124,14 +125,13 @@ public class PubSubTests { return factory.instance(); } - @SuppressWarnings("unchecked") @ParameterizedRedisTest - void testContainerSubscribe() throws Exception { + void testContainerSubscribe() { T payload1 = getT(); T payload2 = getT(); - template.convertAndSend(CHANNEL, payload1); - template.convertAndSend(CHANNEL, payload2); + assertThat(template.convertAndSend(CHANNEL, payload1)).isEqualTo(1L); + assertThat(template.convertAndSend(CHANNEL, payload2)).isEqualTo(1L); await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains(payload1) && bag.contains(payload2)); } @@ -140,7 +140,7 @@ public class PubSubTests { void testMessageBatch() throws Exception { int COUNT = 10; for (int i = 0; i < COUNT; i++) { - template.convertAndSend(CHANNEL, getT()); + assertThat(template.convertAndSend(CHANNEL, getT())).isEqualTo(1L); } for (int i = 0; i < COUNT; i++) { @@ -155,8 +155,8 @@ public class PubSubTests { T payload2 = getT(); container.removeMessageListener(adapter, new ChannelTopic(CHANNEL)); - template.convertAndSend(CHANNEL, payload1); - template.convertAndSend(CHANNEL, payload2); + assertThat(template.convertAndSend(CHANNEL, payload1)).isEqualTo(1L); + assertThat(template.convertAndSend(CHANNEL, payload2)).isEqualTo(1L); assertThat(bag.poll(200, TimeUnit.MILLISECONDS)).isNull(); } @@ -169,9 +169,8 @@ public class PubSubTests { container.start(); } - @SuppressWarnings("unchecked") @ParameterizedRedisTest // DATAREDIS-251, GH-964 - void testStartListenersToNoSpecificChannelTest() throws InterruptedException { + void testStartListenersToNoSpecificChannelTest() { assumeThat(isClusterAware(template.getConnectionFactory())).isFalse(); @@ -181,7 +180,7 @@ public class PubSubTests { T payload = getT(); - template.convertAndSend(CHANNEL, payload); + assertThat(template.convertAndSend(CHANNEL, payload)).isEqualTo(1L); await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains(payload)); }