Refine Topic creation for ChannelTopic and PatternTopic.

We now expose factory methods to construct ChannelTopic and PatternTopic from the Topic interface.

See #3131

Author: mp911de

Diff for: src/main/antora/modules/ROOT/pages/redis/pubsub.adoc

@@ -162,7 +162,7 @@ XML::
 ----
 ======
 
-NOTE: The listener topic can be either a channel (for example, `topic="chatroom"`) or a pattern (for example, `topic="*room"`). For channels, you should use the `ChannelTopic` class, and for patterns, use the `PatternTopic` class.
+NOTE: The listener topic can be either a channel (for example, `topic="chatroom"` respective `Topic.channel("chatroom")`) or a pattern (for example, `topic="*room"` respective `Topic.pattern("*room")`).
 
 The preceding example uses the Redis namespace to declare the message listener container and automatically register the POJOs as listeners. The full-blown beans definition follows:
 

Diff for: src/main/java/org/springframework/data/redis/listener/Topic.java

@@ -19,13 +19,37 @@
  * Topic for a Redis message. Acts a high-level abstraction on top of Redis low-level channels or patterns.
  *
  * @author Costin Leau
+ * @author Mark Paluch
  */
 public interface Topic {
 
+	/**
+	 * Create a new {@link ChannelTopic} for channel subscriptions.
+	 *
+	 * @param channelName {@link String name} of the Redis channel; must not be {@literal null}.
+	 * @return the {@link ChannelTopic} for the given {@code channelName}.
+	 * @since 3.5
+	 */
+	static ChannelTopic channel(String channelName) {
+		return ChannelTopic.of(channelName);
+	}
+
+	/**
+	 * Create a new {@link PatternTopic} for channel subscriptions based on a {@code pattern}.
+	 *
+	 * @param pattern {@link String pattern} used to match channels; must not be {@literal null} or empty.
+	 * @return the {@link PatternTopic} for the given {@code pattern}.
+	 * @since 3.5
+	 */
+	static PatternTopic pattern(String pattern) {
+		return PatternTopic.of(pattern);
+	}
+
 	/**
 	 * Returns the topic (as a String).
 	 *
 	 * @return the topic
 	 */
 	String getTopic();
+
 }

Diff for: src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java

@@ -23,12 +23,9 @@
 
 import java.nio.ByteBuffer;
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.List;
-import java.util.Queue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.LinkedBlockingDeque;
@@ -110,7 +107,7 @@ void shouldReceiveChannelMessages() {
 
 		ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(connectionFactory);
 
-		container.receiveLater(ChannelTopic.of(CHANNEL1)) //
+		container.receiveLater(Topic.channel(CHANNEL1)) //
 				.doOnNext(it -> doPublish(CHANNEL1.getBytes(), MESSAGE.getBytes())) //
 				.flatMapMany(Function.identity()) //
 				.as(StepVerifier::create) //
@@ -153,7 +150,7 @@ public void onChannelUnsubscribed(byte[] channel, long count) {
 			}
 		};
 
-		container.receive(Collections.singletonList(ChannelTopic.of(CHANNEL1)), listener) //
+		container.receive(Collections.singletonList(Topic.channel(CHANNEL1)), listener) //
 				.as(StepVerifier::create) //
 				.then(awaitSubscription(container::getActiveSubscriptions))
 				.then(() -> doPublish(CHANNEL1.getBytes(), MESSAGE.getBytes())) //
@@ -220,7 +217,7 @@ public void onPatternUnsubscribed(byte[] pattern, long count) {
 			}
 		};
 
-		container.receive(Collections.singletonList(PatternTopic.of(PATTERN1)), listener) //
+		container.receive(Collections.singletonList(Topic.pattern(PATTERN1)), listener) //
 				.cast(PatternMessage.class) //
 				.as(StepVerifier::create) //
 				.then(awaitSubscription(container::getActiveSubscriptions))
@@ -314,10 +311,10 @@ void multipleListenShouldTrackSubscriptions() throws Exception {
 
 		ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(connectionFactory);
 
-		Flux<? extends ReactiveSubscription.Message<String, String>> c1 = container.receiveLater(ChannelTopic.of(CHANNEL1))
+		Flux<? extends ReactiveSubscription.Message<String, String>> c1 = container.receiveLater(Topic.channel(CHANNEL1))
 				.block();
 		Flux<? extends ReactiveSubscription.Message<String, String>> c1p1 = container
-				.receiveLater(Arrays.asList(ChannelTopic.of(CHANNEL1), PatternTopic.of(PATTERN1)),
+				.receiveLater(Arrays.asList(Topic.channel(CHANNEL1), PatternTopic.of(PATTERN1)),
 						SerializationPair.fromSerializer(RedisSerializer.string()),
 						SerializationPair.fromSerializer(RedisSerializer.string()))
 				.block();

Diff for: src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java

@@ -79,7 +79,7 @@ void shouldSubscribeToPattern() {
 
 		container = createContainer();
 
-		container.receive(PatternTopic.of("foo*")).as(StepVerifier::create).thenAwait().thenCancel().verify();
+		container.receive(Topic.pattern("foo*")).as(StepVerifier::create).thenAwait().thenCancel().verify();
 
 		verify(subscriptionMock).pSubscribe(getByteBuffer("foo*"));
 	}
@@ -90,7 +90,7 @@ void shouldSubscribeToMultiplePatterns() {
 		when(subscriptionMock.receive()).thenReturn(Flux.never());
 		container = createContainer();
 
-		container.receive(PatternTopic.of("foo*"), PatternTopic.of("bar*")).as(StepVerifier::create).thenRequest(1)
+		container.receive(Topic.pattern("foo*"), Topic.pattern("bar*")).as(StepVerifier::create).thenRequest(1)
 				.thenAwait().thenCancel().verify();
 
 		verify(subscriptionMock).pSubscribe(getByteBuffer("foo*"), getByteBuffer("bar*"));
@@ -102,7 +102,7 @@ void shouldSubscribeToChannel() {
 		when(subscriptionMock.receive()).thenReturn(Flux.never());
 		container = createContainer();
 
-		container.receive(ChannelTopic.of("foo")).as(StepVerifier::create).thenAwait().thenCancel().verify();
+		container.receive(Topic.channel("foo")).as(StepVerifier::create).thenAwait().thenCancel().verify();
 
 		verify(subscriptionMock).subscribe(getByteBuffer("foo"));
 	}
@@ -113,7 +113,7 @@ void shouldSubscribeToMultipleChannels() {
 		when(subscriptionMock.receive()).thenReturn(Flux.never());
 		container = createContainer();
 
-		container.receive(ChannelTopic.of("foo"), ChannelTopic.of("bar")).as(StepVerifier::create).thenAwait().thenCancel()
+		container.receive(Topic.channel("foo"), Topic.channel("bar")).as(StepVerifier::create).thenAwait().thenCancel()
 				.verify();
 
 		verify(subscriptionMock).subscribe(getByteBuffer("foo"), getByteBuffer("bar"));
@@ -127,7 +127,7 @@ void shouldEmitChannelMessage() {
 		when(subscriptionMock.receive()).thenReturn(sink.asFlux());
 		container = createContainer();
 
-		Flux<Message<String, String>> messageStream = container.receive(ChannelTopic.of("foo"));
+		Flux<Message<String, String>> messageStream = container.receive(Topic.channel("foo"));
 
 		messageStream.as(StepVerifier::create).then(() -> {
 			sink.tryEmitNext(createChannelMessage("foo", "message"));
@@ -146,7 +146,7 @@ void shouldEmitPatternMessage() {
 		when(subscriptionMock.receive()).thenReturn(sink.asFlux());
 		container = createContainer();
 
-		Flux<PatternMessage<String, String, String>> messageStream = container.receive(PatternTopic.of("foo*"));
+		Flux<PatternMessage<String, String, String>> messageStream = container.receive(Topic.pattern("foo*"));
 
 		messageStream.as(StepVerifier::create).then(() -> {
 			sink.tryEmitNext(createPatternMessage("foo*", "foo", "message"));
@@ -171,7 +171,7 @@ void shouldRegisterSubscription() {
 		when(subscriptionMock.receive()).thenReturn(sink.asFlux());
 		container = createContainer();
 
-		Flux<Message<String, String>> messageStream = container.receive(ChannelTopic.of("foo*"));
+		Flux<Message<String, String>> messageStream = container.receive(Topic.channel("foo*"));
 
 		Disposable subscription = messageStream.subscribe();
 
@@ -193,7 +193,7 @@ void shouldRegisterSubscriptionMultipleSubscribers() {
 		when(subscriptionMock.receive()).thenReturn(sink.asFlux());
 		container = createContainer();
 
-		Flux<Message<String, String>> messageStream = container.receive(new ChannelTopic("foo*"));
+		Flux<Message<String, String>> messageStream = container.receive(Topic.channel("foo*"));
 
 		Disposable first = messageStream.subscribe();
 		Disposable second = messageStream.subscribe();
@@ -216,7 +216,7 @@ void shouldUnsubscribeOnCancel() {
 		when(subscriptionMock.receive()).thenReturn(sink.asFlux());
 		container = createContainer();
 
-		Flux<PatternMessage<String, String, String>> messageStream = container.receive(PatternTopic.of("foo*"));
+		Flux<PatternMessage<String, String, String>> messageStream = container.receive(Topic.pattern("foo*"));
 
 		messageStream.as(StepVerifier::create).then(() -> {
 
@@ -240,7 +240,7 @@ void shouldTerminateSubscriptionsOnShutdown() {
 		}));
 		container = createContainer();
 
-		Flux<PatternMessage<String, String, String>> messageStream = container.receive(PatternTopic.of("foo*"));
+		Flux<PatternMessage<String, String, String>> messageStream = container.receive(Topic.pattern("foo*"));
 
 		messageStream.as(StepVerifier::create).then(() -> {
 			container.destroy();
@@ -255,7 +255,7 @@ void shouldCleanupDownstream() {
 		when(subscriptionMock.receive()).thenReturn(sink.asFlux());
 		container = createContainer();
 
-		Flux<PatternMessage<String, String, String>> messageStream = container.receive(PatternTopic.of("foo*"));
+		Flux<PatternMessage<String, String, String>> messageStream = container.receive(Topic.pattern("foo*"));
 
 		messageStream.as(StepVerifier::create).then(() -> {
 			assertThat(sink.currentSubscriberCount()).isGreaterThan(0);

Diff for: src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt

@@ -26,7 +26,7 @@ import org.junit.jupiter.api.Test
 import org.springframework.data.redis.connection.DataType
 import org.springframework.data.redis.connection.ReactiveSubscription
 import org.springframework.data.redis.core.script.RedisScript
-import org.springframework.data.redis.listener.ChannelTopic
+import org.springframework.data.redis.listener.Topic
 import org.springframework.data.redis.serializer.RedisElementReader
 import org.springframework.data.redis.serializer.RedisElementWriter
 import reactor.core.publisher.Flux
@@ -167,8 +167,8 @@ class ReactiveRedisOperationsExtensionsUnitTests {
 	@Test // DATAREDIS-1033
 	fun listenTo() {
 
-		val topic1 = ChannelTopic.of("foo")
-		val topic2 = ChannelTopic.of("bar")
+		val topic1 = Topic.channel("foo")
+		val topic2 = Topic.channel("bar")
 		val message = ReactiveSubscription.ChannelMessage("a", "b")
 		val operations = mockk<ReactiveRedisOperations<String, String>>()
 		every { operations.listenTo(any(), any()) } returns Flux.just(message)