Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refine Topic creation for ChannelTopic and PatternTopic.

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

See #3131
This commit is contained in:
Mark Paluch
2025-04-17 15:51:02 +02:00
parent 0de3af0ac5
commit 5029094967
5 changed files with 44 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/antora/modules/ROOT/pages/redis/pubsub.adoc
View File

@@ -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:

24
src/main/java/org/springframework/data/redis/listener/Topic.java
View File

@@ -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();
}

13
src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java
View File

@@ -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<? 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();

22
src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java
View File

@@ -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<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 @@ class ReactiveRedisMessageListenerContainerUnitTests {
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 @@ class ReactiveRedisMessageListenerContainerUnitTests {
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 @@ class ReactiveRedisMessageListenerContainerUnitTests {
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 @@ class ReactiveRedisMessageListenerContainerUnitTests {
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 @@ class ReactiveRedisMessageListenerContainerUnitTests {
}));
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 @@ class ReactiveRedisMessageListenerContainerUnitTests {
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);

6
src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt
View File

@@ -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)