diff --git a/src/main/antora/modules/ROOT/pages/redis/pubsub.adoc b/src/main/antora/modules/ROOT/pages/redis/pubsub.adoc index 736683835..031425a4e 100644 --- a/src/main/antora/modules/ROOT/pages/redis/pubsub.adoc +++ b/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 --git a/src/main/java/org/springframework/data/redis/listener/Topic.java b/src/main/java/org/springframework/data/redis/listener/Topic.java index 0aad48207..bec4cd3c3 100644 --- a/src/main/java/org/springframework/data/redis/listener/Topic.java +++ b/src/main/java/org/springframework/data/redis/listener/Topic.java @@ -19,13 +19,37 @@ package org.springframework.data.redis.listener; * 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 --git a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java index e990cea5f..4b59bc872 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java @@ -23,12 +23,9 @@ import reactor.test.StepVerifier; 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 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests { 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 class ReactiveRedisMessageListenerContainerIntegrationTests { } }; - 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 class ReactiveRedisMessageListenerContainerIntegrationTests { } }; - 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 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests { ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(connectionFactory); - Flux> c1 = container.receiveLater(ChannelTopic.of(CHANNEL1)) + Flux> c1 = container.receiveLater(Topic.channel(CHANNEL1)) .block(); Flux> 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 --git a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java index f5e99bff4..f22ff8e56 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java @@ -79,7 +79,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { 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 @@ class ReactiveRedisMessageListenerContainerUnitTests { 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 @@ class ReactiveRedisMessageListenerContainerUnitTests { 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 @@ class ReactiveRedisMessageListenerContainerUnitTests { 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 @@ class ReactiveRedisMessageListenerContainerUnitTests { when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); - Flux> messageStream = container.receive(ChannelTopic.of("foo")); + Flux> messageStream = container.receive(Topic.channel("foo")); messageStream.as(StepVerifier::create).then(() -> { sink.tryEmitNext(createChannelMessage("foo", "message")); @@ -146,7 +146,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); - Flux> messageStream = container.receive(PatternTopic.of("foo*")); + Flux> messageStream = container.receive(Topic.pattern("foo*")); messageStream.as(StepVerifier::create).then(() -> { sink.tryEmitNext(createPatternMessage("foo*", "foo", "message")); @@ -171,7 +171,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); - Flux> messageStream = container.receive(ChannelTopic.of("foo*")); + Flux> messageStream = container.receive(Topic.channel("foo*")); Disposable subscription = messageStream.subscribe(); @@ -193,7 +193,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); - Flux> messageStream = container.receive(new ChannelTopic("foo*")); + Flux> messageStream = container.receive(Topic.channel("foo*")); Disposable first = messageStream.subscribe(); Disposable second = messageStream.subscribe(); @@ -216,7 +216,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); - Flux> messageStream = container.receive(PatternTopic.of("foo*")); + Flux> messageStream = container.receive(Topic.pattern("foo*")); messageStream.as(StepVerifier::create).then(() -> { @@ -240,7 +240,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { })); container = createContainer(); - Flux> messageStream = container.receive(PatternTopic.of("foo*")); + Flux> messageStream = container.receive(Topic.pattern("foo*")); messageStream.as(StepVerifier::create).then(() -> { container.destroy(); @@ -255,7 +255,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); - Flux> messageStream = container.receive(PatternTopic.of("foo*")); + Flux> messageStream = container.receive(Topic.pattern("foo*")); messageStream.as(StepVerifier::create).then(() -> { assertThat(sink.currentSubscriberCount()).isGreaterThan(0); diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt index 62a6c968f..90563f411 100644 --- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt +++ b/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>() every { operations.listenTo(any(), any()) } returns Flux.just(message)