diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java index 8e1dae068..b2b7ada82 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java @@ -116,6 +116,49 @@ public interface ReactiveRedisOperations { */ Flux> listenTo(Topic... topics); + /** + * Subscribe to the given Redis {@code channels} and emit {@link Message messages} received for those. The + * {@link Mono} completes once the {@link Topic topic} subscriptions are registered. + * + * @param channels must not be {@literal null}. + * @return a hot sequence of {@link Message messages}. + * @since 2.6 + * @see org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer#receiveLater(ChannelTopic...) + */ + default Mono>> listenToChannelLater(String... channels) { + + Assert.notNull(channels, "Channels must not be null!"); + + return listenToLater(Arrays.stream(channels).map(ChannelTopic::of).toArray(ChannelTopic[]::new)); + } + + /** + * Subscribe to the Redis channels matching the given {@code pattern} and emit {@link Message messages} received for + * those. The {@link Mono} completes once the {@link Topic topic} subscriptions are registered. + * + * @param patterns must not be {@literal null}. + * @return a hot sequence of {@link Message messages}. + * @since 2.6 + * @see org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer#receiveLater(PatternTopic...) + */ + default Mono>> listenToPatternLater(String... patterns) { + + Assert.notNull(patterns, "Patterns must not be null!"); + return listenToLater(Arrays.stream(patterns).map(PatternTopic::of).toArray(PatternTopic[]::new)); + } + + /** + * Subscribe to the Redis channels for the given {@link Topic topics} and emit {@link Message messages} received for + * those. The {@link Mono} completes once the {@link Topic topic} subscriptions are registered. + * + * @param topics must not be {@literal null}. + * @return a hot sequence of {@link Message messages}. + * @since 2.6 + * @see org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer#receiveLater(Iterable, + * RedisSerializationContext.SerializationPair, RedisSerializationContext.SerializationPair) + */ + Mono>> listenToLater(Topic... topics); + // ------------------------------------------------------------------------- // Methods dealing with Redis Keys // ------------------------------------------------------------------------- diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java index 202998fdd..c9cf241c0 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java @@ -237,6 +237,21 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations container.destroyLater().subscribe()); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveRedisOperations#listenToLater(org.springframework.data.redis.listener.Topic[]) + */ + @Override + @SuppressWarnings({ "unchecked", "rawtypes" }) + public Mono>> listenToLater(Topic... topics) { + + ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(getConnectionFactory()); + + return (Mono) container.receiveLater(Arrays.asList(topics), getSerializationContext().getStringSerializationPair(), + getSerializationContext().getValueSerializationPair()) // + .doFinally((signalType) -> container.destroyLater().subscribe()); + } + // ------------------------------------------------------------------------- // Methods dealing with Redis keys // ------------------------------------------------------------------------- diff --git a/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java index bd7821517..c6d574f9f 100644 --- a/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java @@ -343,7 +343,7 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { * @throws InvalidDataAccessApiUsageException if {@code topics} is empty. * @since 2.6 */ - private Mono>> receiveLater(Iterable topics, + public Mono>> receiveLater(Iterable topics, SerializationPair channelSerializer, SerializationPair messageSerializer) { Assert.notNull(topics, "Topics must not be null!"); diff --git a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java index 6b3cc932f..d38205bc1 100644 --- a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*; import static org.assertj.core.api.Assumptions.*; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import java.time.Duration; @@ -27,6 +28,7 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.function.Function; import org.junit.jupiter.api.BeforeEach; @@ -441,7 +443,29 @@ public class ReactiveRedisTemplateIntegrationTests { redisTemplate.listenToChannel(channel).as(StepVerifier::create) // .thenAwait(Duration.ofMillis(500)) // just make sure we the subscription completed - .then(() -> redisTemplate.convertAndSend(channel, message).block()) // + .then(() -> redisTemplate.convertAndSend(channel, message).subscribe()) // + .assertNext(received -> { + + assertThat(received).isInstanceOf(ChannelMessage.class); + assertThat(received.getMessage()).isEqualTo(message); + assertThat(received.getChannel()).isEqualTo(channel); + }) // + .thenAwait(Duration.ofMillis(10)) // + .thenCancel() // + .verify(Duration.ofSeconds(3)); + } + + @ParameterizedRedisTest // GH-1622 + @EnabledIfLongRunningTest + void listenToLaterChannelShouldReceiveChannelMessagesCorrectly() { + + String channel = "my-channel"; + + V message = valueFactory.instance(); + + redisTemplate.listenToChannelLater(channel) // + .doOnNext(it -> redisTemplate.convertAndSend(channel, message).subscribe()).flatMapMany(Function.identity()) // + .as(StepVerifier::create) // .assertNext(received -> { assertThat(received).isInstanceOf(ChannelMessage.class); @@ -455,7 +479,7 @@ public class ReactiveRedisTemplateIntegrationTests { @ParameterizedRedisTest // DATAREDIS-612 @EnabledIfLongRunningTest - void listenToChannelPatternShouldReceiveChannelMessagesCorrectly() throws InterruptedException { + void listenToPatternShouldReceiveChannelMessagesCorrectly() { String channel = "my-channel"; String pattern = "my-*"; @@ -466,7 +490,32 @@ public class ReactiveRedisTemplateIntegrationTests { stream.as(StepVerifier::create) // .thenAwait(Duration.ofMillis(500)) // just make sure we the subscription completed - .then(() -> redisTemplate.convertAndSend(channel, message).block()) // + .then(() -> redisTemplate.convertAndSend(channel, message).subscribe()) // + .assertNext(received -> { + + assertThat(received).isInstanceOf(PatternMessage.class); + assertThat(received.getMessage()).isEqualTo(message); + assertThat(received.getChannel()).isEqualTo(channel); + assertThat(((PatternMessage) received).getPattern()).isEqualTo(pattern); + }) // + .thenCancel() // + .verify(Duration.ofSeconds(3)); + } + + @ParameterizedRedisTest // GH-1622 + @EnabledIfLongRunningTest + void listenToPatternLaterShouldReceiveChannelMessagesCorrectly() { + + String channel = "my-channel"; + String pattern = "my-*"; + + V message = valueFactory.instance(); + + Mono>> stream = redisTemplate.listenToPatternLater(pattern); + + stream.doOnNext(it -> redisTemplate.convertAndSend(channel, message).subscribe()) // + .flatMapMany(Function.identity()) // + .as(StepVerifier::create) // .assertNext(received -> { assertThat(received).isInstanceOf(PatternMessage.class);