Update RedisOperations#convertAndSend to return number of clients receiving the message.

Currently, `RedisOperations#convertAndSend` is void event though the underlying `RedisPubSubCommands#publish` returns a `Long` that represents the number of clients receiving the published message (see https://redis.io/commands/publish).

This commit updates `RedisOperations#convertAndSend` to return the number of clients that received the message.

Closes #2209
Original pull request: #2225.

Author: vpavic

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]]

src/main/java/org/springframework/data/redis/core/RedisOperations.java

@@ -46,6 +46,7 @@
  * @author ihaohong
  * @author Todd Merrill
  * @author Chen Li
+ * @author Vedran Pavic
  */
 public interface RedisOperations<K, V> {
 
@@ -591,9 +592,11 @@ default void restore(K key, byte[] value, long timeToLive, TimeUnit unit) {
 	 *
 	 * @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 <a href="https://redis.io/commands/publish">Redis Documentation: PUBLISH</a>
 	 */
-	void convertAndSend(String destination, Object message);
+	@Nullable
+	Long convertAndSend(String destination, Object message);
 
 	// -------------------------------------------------------------------------
 	// Methods to obtain specific operations interface objects.

src/main/java/org/springframework/data/redis/core/RedisTemplate.java

@@ -86,6 +86,7 @@
  * @author Denis Zavedeev
  * @author ihaohong
  * @author Chen Li
+ * @author Vedran Pavic
  * @param <K> the Redis key type against which the template works (usually a String)
  * @param <V> the Redis value type against which the template works
  * @see StringRedisTemplate
@@ -942,14 +943,14 @@ public void replicaOfNoOne() {
 	}
 
 	@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<RedisConnection> action) {

src/test/java/org/springframework/data/redis/config/NamespaceIntegrationTests.java

@@ -27,6 +27,7 @@
 /**
  * @author Costin Leau
  * @author Mark Paluch
+ * @author Vedran Pavic
  */
 @SpringJUnitConfig(locations = "namespace.xml")
 class NamespaceIntegrationTests {
@@ -43,8 +44,8 @@ void testSanityTest() throws Exception {
 	}
 
 	@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);
 	}
 }

src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java

@@ -63,6 +63,7 @@
  * @author ihaohong
  * @author Hendrik Duerkop
  * @author Chen Li
+ * @author Vedran Pavic
  */
 @MethodSource("testParams")
 public class RedisTemplateIntegrationTests<K, V> {
@@ -815,10 +816,10 @@ public List<Object> execute(RedisOperations operations) throws DataAccessExcepti
 	}
 
 	@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

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 @@
  * @author Jennifer Hickey
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Vedran Pavic
  */
 @MethodSource("testParams")
 @EnabledIfLongRunningTest
@@ -126,7 +128,7 @@ void tearDown() {
 
 	@ParameterizedRedisTest
 	@EnabledIfLongRunningTest
-	void testContainerPatternResubscribe() throws Exception {
+	void testContainerPatternResubscribe() {
 
 		String payload1 = "do";
 		String payload2 = "re mi";
@@ -143,22 +145,22 @@ void testContainerPatternResubscribe() throws Exception {
 		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));
 
 		// bind original listener on another channel
 		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 @@ void testContainerPatternResubscribe() throws Exception {
 	}
 
 	@ParameterizedRedisTest
-	void testContainerChannelResubscribe() throws Exception {
+	void testContainerChannelResubscribe() {
 
 		String payload1 = "do";
 		String payload2 = "re mi";
@@ -183,27 +185,25 @@ void testContainerChannelResubscribe() throws Exception {
 
 		// 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));
 	}
 
 	/**
 	 * 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 @@ 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(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"));
 	}

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

@@ -49,6 +49,7 @@
  * @author Costin Leau
  * @author Jennifer Hickey
  * @author Mark Paluch
+ * @author Vedran Pavic
  */
 @MethodSource("testParams")
 public class PubSubTests<T> {
@@ -124,14 +125,13 @@ T getT() {
 		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 @@ void testContainerSubscribe() throws Exception {
 	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 @@ void testContainerUnsubscribe() throws Exception {
 		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 @@ void testStartNoListeners() {
 		container.start();
 	}
 
-	@SuppressWarnings("unchecked")
 	@ParameterizedRedisTest // DATAREDIS-251, GH-964
-	void testStartListenersToNoSpecificChannelTest() throws InterruptedException {
+	void testStartListenersToNoSpecificChannelTest() {
 
 		assumeThat(isClusterAware(template.getConnectionFactory())).isFalse();
 
@@ -181,7 +180,7 @@ void testStartListenersToNoSpecificChannelTest() throws InterruptedException {
 
 		T payload = getT();
 
-		template.convertAndSend(CHANNEL, payload);
+		assertThat(template.convertAndSend(CHANNEL, payload)).isEqualTo(1L);
 
 		await().atMost(Duration.ofSeconds(2)).until(() -> bag.contains(payload));
 	}