diff --git a/src/main/asciidoc/reference/reactive-messaging.adoc b/src/main/asciidoc/reference/reactive-messaging.adoc index 56c602dca..34a28cdd2 100644 --- a/src/main/asciidoc/reference/reactive-messaging.adoc +++ b/src/main/asciidoc/reference/reactive-messaging.adoc @@ -4,6 +4,7 @@ Spring Data provides dedicated messaging integration for Redis, very similar in functionality and naming to the JMS integration in Spring Framework; in fact, users familiar with the JMS support in Spring should feel right at home. Redis messaging can be roughly divided into two areas of functionality, namely the production or publication and consumption or subscription of messages, hence the shortcut pubsub (Publish/Subscribe). The `ReactiveRedisTemplate` class is used for message production. For asynchronous reception, Spring Data provides a dedicated message listener container that is used consume a stream of messages. +For the purpose of just subscribing `ReactiveRedisTemplate` offers stripped down alternatives to utilizing a listener container. The package `org.springframework.data.redis.connection` and `org.springframework.data.redis.listener` provide the core functionality for using Redis messaging. @@ -21,7 +22,7 @@ Mono publish = con.publish(msg, channel); // send message through ReactiveRedisTemplate ReactiveRedisTemplate template = … -Mono publish = template.convertAndSend("hello!", "world"); +Mono publish = template.convertAndSend("channel", "message"); ---- [[redis:reactive:pubsub:subscribe]] @@ -31,7 +32,7 @@ On the receiving side, one can subscribe to one or multiple channels either by n At the low-level, `ReactiveRedisConnection` offers `subscribe` and `pSubscribe` methods that map the Redis commands for subscribing by channel respectively by pattern. Note that multiple channels or patterns can be used as arguments. To change a subscription, simply query the channels and patterns of `ReactiveSubscription`. -NOTE: Reactive subscription commands in Spring Data Redis non-blocking and terminate without emitting an element. +NOTE: Reactive subscription commands in Spring Data Redis are non-blocking and may terminate without emitting an element. As mentioned above, once subscribed a connection starts waiting for messages. No other commands can be invoked on it except for adding new subscriptions or modifying/canceling the existing ones. Commands other than `subscribe`, `pSubscribe`, `unsubscribe`, or `pUnsubscribe` are illegal and will cause an exception. @@ -42,7 +43,7 @@ In order to receive messages, one needs to obtain the message stream. Note that Spring Data offers `ReactiveRedisMessageListenerContainer` which does all the heavy lifting of conversion and subscription state management on behalf of the user. -`ReactiveRedisMessageListenerContainer` acts as a message listener container; it is used to receive messages from a Redis channel and expose a stream of messages that emits channel messages with deserialization applied. It takes care of registering to receive messages, resource acquisition and release, exception conversion and the like. This allows you as an application developer to write the (possibly complex) business logic associated with receiving a message (and reacting to it), and delegates boilerplate Redis infrastructure concerns to the framework. Message streams register a subscription in Redis upon publisher subscription and unregister if the subscription gets canceled. +`ReactiveRedisMessageListenerContainer` acts as a message listener container. It is used to receive messages from a Redis channel and expose a stream of messages that emits channel messages with deserialization applied. It takes care of registering to receive messages, resource acquisition and release, exception conversion and the like. This allows you as an application developer to write the (possibly complex) business logic associated with receiving a message (and reacting to it), and delegates boilerplate Redis infrastructure concerns to the framework. Message streams register a subscription in Redis upon publisher subscription and unregister if the subscription gets canceled. Furthermore, to minimize the application footprint, `ReactiveRedisMessageListenerContainer` allows one connection and one thread to be shared by multiple listeners even though they do not share a subscription. Thus no matter how many listeners or channels an application tracks, the runtime cost will remain the same through out its lifetime. Moreover, the container allows runtime configuration changes so one can add or remove listeners while an application is running without the need for restart. Additionally, the container uses a lazy subscription approach, using a `ReactiveRedisConnection` only when needed - if all the listeners are unsubscribed, cleanup is automatically performed. @@ -55,3 +56,18 @@ ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListen Flux> stream = container.receive(ChannelTopic.of("my-chanel")); ---- + +[[redis:reactive:pubsub:subscribe:template]] +=== Subscribing via template API + +As mentioned above you can directly use `ReactiveRedisTemplate` to subscribe to channels / patterns. This approach +offers a straight forward, though limited solution as you loose the option to add subscriptions after the initial +ones. Nevertheless you still can control the message stream via the returned `Flux` using eg. `take(Duration)`. When +done reading, on error or cancellation all bound resources are freed again. + +[source,java] +---- +redisTemplate.listenToChannel("channel1", "channel2").doOnNext(msg -> { + // message processing ... +}).subscribe(); +---- diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisPubSubCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactivePubSubCommands.java similarity index 96% rename from src/main/java/org/springframework/data/redis/connection/ReactiveRedisPubSubCommands.java rename to src/main/java/org/springframework/data/redis/connection/ReactivePubSubCommands.java index 7bdd7817c..03c71cd2e 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisPubSubCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactivePubSubCommands.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,9 +27,10 @@ import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMes * Redis Pub/Sub commands executed using reactive infrastructure. * * @author Mark Paluch + * @author Christoph Strobl * @since 2.1 */ -public interface ReactiveRedisPubSubCommands { +public interface ReactivePubSubCommands { /** * Creates a subscription for this connection. Connections can have multiple {@link ReactiveSubscription}s. @@ -63,7 +64,7 @@ public interface ReactiveRedisPubSubCommands { * Subscribes the connection to the given {@code channels}. Once subscribed, a connection enters listening mode and * can only subscribe to other channels or unsubscribe. No other commands are accepted until the connection is * unsubscribed. - *

+ *

* Note that cancellation of the {@link Flux} will unsubscribe from {@code channels}. * * @param channels channel names, must not be {@literal null}. @@ -82,4 +83,5 @@ public interface ReactiveRedisPubSubCommands { * @see Redis Documentation: PSUBSCRIBE */ Mono pSubscribe(ByteBuffer... patterns); + } diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java index b5d0f017c..415c99852 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java @@ -50,8 +50,21 @@ import org.springframework.util.Assert; */ public interface ReactiveRedisConnection extends Closeable { + /* + * (non-Javadoc) + * @see java.io.Closeable#close() + */ @Override - void close(); + default void close() { + closeLater().block(); + } + + /** + * Asynchronously close the connection and release associated resources. + * + * @return the {@link Mono} signaling when done. + */ + Mono closeLater(); /** * Get {@link ReactiveKeyCommands}. @@ -117,12 +130,12 @@ public interface ReactiveRedisConnection extends Closeable { ReactiveHyperLogLogCommands hyperLogLogCommands(); /** - * Get {@link ReactiveRedisPubSubCommands}. + * Get {@link ReactivePubSubCommands}. * * @return never {@literal null}. * @since 2.1 */ - ReactiveRedisPubSubCommands pubSubCommands(); + ReactivePubSubCommands pubSubCommands(); /** * Get {@link ReactiveScriptingCommands}. diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveSubscription.java b/src/main/java/org/springframework/data/redis/connection/ReactiveSubscription.java index 855d0facd..a0b22728f 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveSubscription.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,19 +20,22 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.nio.ByteBuffer; -import java.util.Collection; +import java.util.Set; + +import org.springframework.util.Assert; /** - * Subscription for Redis channels using reactive infrastructure. A {@link ReactiveSubscription} allows subscription to + * Subscription for Redis channels using reactive infrastructure. A {@link ReactiveSubscription} allows subscribing to * {@link #subscribe(ByteBuffer...) channels} and {@link #pSubscribe(ByteBuffer...) patterns}. It provides access to the * {@link ChannelMessage} {@link #receive() stream} that emits only messages for channels and patterns registered in * this {@link ReactiveSubscription}. - *

+ *

* A reactive Redis connection can have multiple subscriptions. If two or more subscriptions subscribe to the same * target (channel/pattern) and one unsubscribes, then the other one will no longer receive messages for the target due * to how Redis handled Pub/Sub subscription. * * @author Mark Paluch + * @author Christoph Strobl * @since 2.1 */ public interface ReactiveSubscription { @@ -54,8 +57,8 @@ public interface ReactiveSubscription { Mono pSubscribe(ByteBuffer... patterns); /** - * Cancels the current subscription for all {@link #getChannels() channels} given by name. - * + * Cancels the current subscription for all {@link #getChannels() channels}. + * * @return empty {@link Mono} that completes once the channel subscriptions are unregistered. */ Mono unsubscribe(); @@ -64,13 +67,13 @@ public interface ReactiveSubscription { * Cancels the current subscription for all given channels. * * @param channels channel names. Must not be empty. - * @return empty {@link Mono} that completes once the channel subscription is unregistered. + * @return empty {@link Mono} that completes once the channel subscriptions are unregistered. */ Mono unsubscribe(ByteBuffer... channels); /** * Cancels the subscription for all channels matched by {@link #getPatterns()} patterns}. - * + * * @return empty {@link Mono} that completes once the patterns subscriptions are unregistered. */ Mono pUnsubscribe(); @@ -79,91 +82,136 @@ public interface ReactiveSubscription { * Cancels the subscription for all channels matching the given patterns. * * @param patterns must not be empty. - * @return empty {@link Mono} that completes once the patterns subscription is unregistered. + * @return empty {@link Mono} that completes once the patterns subscriptions are unregistered. */ Mono pUnsubscribe(ByteBuffer... patterns); /** * Returns the (named) channels for this subscription. * - * @return collection of named channels + * @return {@link Set} of named channels. */ - Collection getChannels(); + Set getChannels(); /** * Returns the channel patters for this subscription. * - * @return collection of channel patterns + * @return {@link Set} of channel patterns. */ - Collection getPatterns(); + Set getPatterns(); /** - * Retrieve the message stream emitting {@link ChannelMessage} and {@link PatternMessage}. The resulting message - * stream contains only messages for subscribed and registered {@link #getChannels() channels} and - * {@link #getPatterns() patterns}. - *

+ * Retrieve the message stream emitting {@link Message messages}. The resulting message stream contains only messages + * for subscribed and registered {@link #getChannels() channels} and {@link #getPatterns() patterns}. + *

* Stream publishing uses {@link reactor.core.publisher.ConnectableFlux} turning the stream into a hot sequence. * Emission is paused if there is no demand. Messages received in that time are buffered. This stream terminates - * either if all subscribers unsubscribe or if this {@link Subscription} is {@link #terminate() is terminated}. - * - * @return the message stream. + * either if all subscribers unsubscribe or if this {@link Subscription} is {@link #cancel() is terminated}. + * + * @return {@link Flux} emitting the {@link Message} stream. */ - Flux> receive(); + Flux> receive(); /** * Unsubscribe from all {@link #getChannels() channels} and {@link #getPatterns() patterns} and request termination of * all active {@link #receive() message streams}. Active streams will terminate with a * {@link java.util.concurrent.CancellationException}. - * + * * @return a {@link Mono} that completes once termination is finished. */ - Mono terminate(); + Mono cancel(); + + /** + * {@link Message} represents a Redis channel message within Redis pub/sub. + * + * @param channel representation type. + * @param message representation type. + * @author Christoph Strobl + * @since 2.1 + */ + interface Message { + + /** + * Get the channel the message published to. + * + * @return never {@literal null}. + */ + C getChannel(); + + /** + * Get the actual message body. + * + * @return never {@literal null}. + */ + M getMessage(); + } /** * Value object for a Redis channel message. - * + * * @param type of how the channel name is represented. - * @param type of how the message is represented. + * @param type of how the message is represented. * @author Mark Paluch + * @author Christoph Strobl * @since 2.1 */ @EqualsAndHashCode - class ChannelMessage { + class ChannelMessage implements Message { private final C channel; - private final B message; + private final M message; /** * Create a new {@link ChannelMessage}. - * + * * @param channel must not be {@literal null}. * @param message must not be {@literal null}. */ - public ChannelMessage(C channel, B message) { + public ChannelMessage(C channel, M message) { + + Assert.notNull(channel, "Channel must not be null!"); + Assert.notNull(message, "Message must not be null!"); + this.channel = channel; this.message = message; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveSubscription.Message#getChannel() + */ + @Override public C getChannel() { return channel; } - public B getMessage() { + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveSubscription.Message#getMessage() + */ + @Override + public M getMessage() { return message; } + + @Override + public String toString() { + return "ChannelMessage {" + "channel=" + channel + ", message=" + message + '}'; + } } /** * Value object for a Redis channel message received from a pattern subscription. - * - * @param type of how the pattern is represented. + * + * @param

type of how the pattern is represented. * @param type of how the channel name is represented. - * @param type of how the message is represented. + * @param type of how the message is represented. * @author Mark Paluch + * @author Christoph Strobl * @since 2.1 */ @EqualsAndHashCode(callSuper = true) - class PatternMessage extends ChannelMessage { + class PatternMessage extends ChannelMessage { private final P pattern; @@ -174,14 +222,26 @@ public interface ReactiveSubscription { * @param channel must not be {@literal null}. * @param message must not be {@literal null}. */ - public PatternMessage(P pattern, C channel, B message) { + public PatternMessage(P pattern, C channel, M message) { super(channel, message); + + Assert.notNull(pattern, "Pattern must not be null!"); this.pattern = pattern; } + /** + * Get the pattern that matched the channel. + * + * @return never {@literal null}. + */ public P getPattern() { return pattern; } + + @Override + public String toString() { + return "PatternMessage{" + "channel=" + getChannel() + ", pattern=" + pattern + ", message=" + getMessage() + '}'; + } } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java index 4cab537f5..f799e950b 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,69 +25,73 @@ import java.nio.ByteBuffer; import java.util.function.Function; import org.reactivestreams.Publisher; -import org.springframework.data.redis.connection.ReactiveRedisPubSubCommands; +import org.springframework.data.redis.connection.ReactivePubSubCommands; import org.springframework.data.redis.connection.ReactiveSubscription; import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; import org.springframework.util.Assert; /** * @author Mark Paluch + * @author Christoph Strobl * @since 2.1 */ @RequiredArgsConstructor -class LettuceReactivePubSubCommands implements ReactiveRedisPubSubCommands { +class LettuceReactivePubSubCommands implements ReactivePubSubCommands { private final @NonNull LettuceReactiveRedisConnection connection; /* * (non-Javadoc) - * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#createSubscription() + * @see org.springframework.data.redis.connection.ReactivePubSubCommands#createSubscription() */ @Override public Mono createSubscription() { + return connection.getPubSubConnection() - .map(c -> new LettuceReactiveSubscription(c.reactive(), connection.translateException())); + .map(pubSubConnection -> new LettuceReactiveSubscription(pubSubConnection.reactive(), + connection.translateException())); } /* * (non-Javadoc) - * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#publish(org.reactivestreams.Publisher) + * @see org.springframework.data.redis.connection.ReactivePubSubCommands#publish(org.reactivestreams.Publisher) */ @Override public Flux publish(Publisher> messageStream) { Assert.notNull(messageStream, "ChannelMessage stream must not be null!"); - return connection.getCommands().flatMapMany( - c -> Flux.from(messageStream).flatMap(message -> c.publish(message.getChannel(), message.getMessage()))); + return connection.getCommands().flatMapMany(commands -> Flux.from(messageStream) + .flatMap(message -> commands.publish(message.getChannel(), message.getMessage()))); } /* * (non-Javadoc) - * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#subscribe(java.nio.ByteBuffer[]) + * @see org.springframework.data.redis.connection.ReactivePubSubCommands#subscribe(java.nio.ByteBuffer[]) */ @Override public Mono subscribe(ByteBuffer... channels) { Assert.notNull(channels, "Channels must not be null!"); - return doWithPubSub(c -> c.subscribe(channels)); + return doWithPubSub(commands -> commands.subscribe(channels)); } /* * (non-Javadoc) - * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#pSubscribe(java.nio.ByteBuffer[]) + * @see org.springframework.data.redis.connection.ReactivePubSubCommands#pSubscribe(java.nio.ByteBuffer[]) */ @Override public Mono pSubscribe(ByteBuffer... patterns) { Assert.notNull(patterns, "Patterns must not be null!"); - return doWithPubSub(c -> c.psubscribe(patterns)); + return doWithPubSub(commands -> commands.psubscribe(patterns)); } private Mono doWithPubSub(Function, Mono> function) { - return connection.getPubSubConnection().flatMap(c -> function.apply(c.reactive())) + + return connection.getPubSubConnection().flatMap(pubSubConnection -> function.apply(pubSubConnection.reactive())) .onErrorMap(connection.translateException()); } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java index 15426997e..6a11a8bf6 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java @@ -175,7 +175,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { * @see org.springframework.data.redis.connection.ReactiveRedisConnection#pubSubCommands() */ @Override - public ReactiveRedisPubSubCommands pubSubCommands() { + public ReactivePubSubCommands pubSubCommands() { return new LettuceReactivePubSubCommands(this); } @@ -225,11 +225,10 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { /* * (non-Javadoc) - * @see java.io.Closeable#close() + * @see org.springframework.data.redis.connection.ReactiveRedisConnection#closeLater() */ - @Override - public void close() { - dedicatedConnection.close(); + public Mono closeLater() { + return Mono.fromRunnable(() -> dedicatedConnection.close()); } protected Mono> getConnection() { diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java index 87c117f01..b843a34c1 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -24,7 +24,6 @@ import reactor.core.publisher.Mono; import java.nio.ByteBuffer; import java.util.Arrays; -import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Set; @@ -37,11 +36,13 @@ import java.util.function.Supplier; import org.springframework.data.redis.connection.ReactiveSubscription; import org.springframework.lang.Nullable; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; /** * Lettuce-specific implementation of {@link ReactiveSubscription}. * * @author Mark Paluch + * @author Christoph Strobl * @since 2.1 */ class LettuceReactiveSubscription implements ReactiveSubscription { @@ -91,7 +92,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription { */ @Override public Mono unsubscribe() { - return unsubscribe(channelState.getTargets().toArray(new ByteBuffer[0])); + return unsubscribe(channelState.getTargets().toArray(new ByteBuffer[channelState.getTargets().size()])); } /* @@ -104,7 +105,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription { Assert.notNull(channels, "Channels must not be null!"); Assert.noNullElements(channels, "Channels must not contain null elements!"); - return channels.length == 0 ? Mono.empty() : channelState.unsubscribe(channels, commands::unsubscribe); + return ObjectUtils.isEmpty(channels) ? Mono.empty() : channelState.unsubscribe(channels, commands::unsubscribe); } /* @@ -113,7 +114,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription { */ @Override public Mono pUnsubscribe() { - return pUnsubscribe(patternState.getTargets().toArray(new ByteBuffer[0])); + return pUnsubscribe(patternState.getTargets().toArray(new ByteBuffer[patternState.getTargets().size()])); } /* @@ -126,7 +127,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription { Assert.notNull(patterns, "Patterns must not be null!"); Assert.noNullElements(patterns, "Patterns must not contain null elements!"); - return patterns.length == 0 ? Mono.empty() : patternState.unsubscribe(patterns, commands::punsubscribe); + return ObjectUtils.isEmpty(patterns) ? Mono.empty() : patternState.unsubscribe(patterns, commands::punsubscribe); } /* @@ -134,8 +135,8 @@ class LettuceReactiveSubscription implements ReactiveSubscription { * @see org.springframework.data.redis.connection.ReactiveSubscription#getChannels() */ @Override - public Collection getChannels() { - return Collections.unmodifiableCollection(channelState.getTargets()); + public Set getChannels() { + return channelState.getTargets(); } /* @@ -143,8 +144,8 @@ class LettuceReactiveSubscription implements ReactiveSubscription { * @see org.springframework.data.redis.connection.ReactiveSubscription#getPatterns() */ @Override - public Collection getPatterns() { - return Collections.unmodifiableCollection(patternState.getTargets()); + public Set getPatterns() { + return patternState.getTargets(); } /* @@ -152,15 +153,15 @@ class LettuceReactiveSubscription implements ReactiveSubscription { * @see org.springframework.data.redis.connection.ReactiveSubscription#receive() */ @Override - public Flux> receive() { + public Flux> receive() { - Flux> channelMessages = channelState.receive(() -> commands.observeChannels() // - .filter(m -> channelState.getTargets().contains(m.getChannel())) // - .map(m -> new ChannelMessage<>(m.getChannel(), m.getMessage()))); + Flux> channelMessages = channelState.receive(() -> commands.observeChannels() // + .filter(message -> channelState.getTargets().contains(message.getChannel())) // + .map(message -> new ChannelMessage<>(message.getChannel(), message.getMessage()))); - Flux> patternMessages = patternState.receive(() -> commands.observePatterns() // - .filter(m -> patternState.getTargets().contains(m.getPattern())) // - .map(m -> new PatternMessage<>(m.getPattern(), m.getChannel(), m.getMessage()))); + Flux> patternMessages = patternState.receive(() -> commands.observePatterns() // + .filter(message -> patternState.getTargets().contains(message.getPattern())) // + .map(message -> new PatternMessage<>(message.getPattern(), message.getChannel(), message.getMessage()))); return channelMessages.mergeWith(patternMessages); } @@ -170,7 +171,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription { * @see org.springframework.data.redis.connection.ReactiveSubscription#terminate() */ @Override - public Mono terminate() { + public Mono cancel() { return unsubscribe().then(pUnsubscribe()).then(Mono.defer(() -> { @@ -204,9 +205,8 @@ class LettuceReactiveSubscription implements ReactiveSubscription { */ Mono subscribe(ByteBuffer[] targets, Function> subscribeFunction) { - return subscribeFunction.apply(targets).doOnSuccess((v) -> { - this.targets.addAll(Arrays.asList(targets)); - }).onErrorMap(exceptionTranslator); + return subscribeFunction.apply(targets).doOnSuccess((discard) -> this.targets.addAll(Arrays.asList(targets))) + .onErrorMap(exceptionTranslator); } /** @@ -223,14 +223,14 @@ class LettuceReactiveSubscription implements ReactiveSubscription { List targetCollection = Arrays.asList(targets); - return unsubscribeFunction.apply(targets).doOnSuccess((v) -> { + return unsubscribeFunction.apply(targets).doOnSuccess((discard) -> { this.targets.removeAll(targetCollection); }).onErrorMap(exceptionTranslator); }); } - Collection getTargets() { - return targets; + Set getTargets() { + return Collections.unmodifiableSet(targets); } /** @@ -254,13 +254,14 @@ class LettuceReactiveSubscription implements ReactiveSubscription { } ConnectableFlux connectableFlux = connectFunction.get().onErrorMap(exceptionTranslator).publish(); - Flux fluxToUse = connectableFlux.doOnSubscribe(s -> { + Flux fluxToUse = connectableFlux.doOnSubscribe(subscription -> { if (subscribers.incrementAndGet() == 1) { disposable = connectableFlux.connect(); } - }).doFinally(s -> { + }).doFinally(signalType -> { + // TODO: do new need to care about what happened? if (subscribers.decrementAndGet() == 0) { this.flux.compareAndSet(connectableFlux, null); 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 4fdefbd6a..f540c45dd 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java @@ -20,16 +20,22 @@ import reactor.core.publisher.Mono; import java.time.Duration; import java.time.Instant; +import java.util.Arrays; import java.util.Collections; import java.util.List; import org.reactivestreams.Publisher; import org.springframework.data.redis.connection.DataType; +import org.springframework.data.redis.connection.ReactiveSubscription.Message; import org.springframework.data.redis.core.script.RedisScript; +import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.PatternTopic; +import org.springframework.data.redis.listener.Topic; import org.springframework.data.redis.serializer.RedisElementReader; import org.springframework.data.redis.serializer.RedisElementWriter; import org.springframework.data.redis.serializer.RedisSerializationContext; import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.util.Assert; /** * Interface that specified a basic set of Redis operations, implemented by {@link ReactiveRedisTemplate}. Not often @@ -56,17 +62,59 @@ public interface ReactiveRedisOperations { */ Flux execute(ReactiveRedisCallback action); + // ------------------------------------------------------------------------- + // Methods dealing with Redis Pub/Sub + // ------------------------------------------------------------------------- + /** * Publishes the given message to the given channel. * - * @param destination the channel to publish to, must not be {@literal null} or empty. - * @param message message to publish. + * @param destination the channel to publish to, must not be {@literal null} nor empty. + * @param message message to publish. Must not be {@literal null}. * @return the number of clients that received the message * @since 2.1 * @see Redis Documentation: PUBLISH */ Mono convertAndSend(String destination, V message); + /** + * Subscribe to the given Redis {@code channels} and emit {@link Message messages} received for those. + * + * @param channels must not be {@literal null}. + * @return a hot sequence of {@link Message messages}. + * @since 2.1 + */ + default Flux> listenToChannel(String... channels) { + + Assert.notNull(channels, "Channels must not be null!"); + + return listenTo(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. + * + * @param patterns must not be {@literal null}. + * @return a hot sequence of {@link Message messages}. + * @since 2.1 + */ + default Flux> listenToPattern(String... patterns) { + + Assert.notNull(patterns, "Patterns must not be null!"); + return listenTo(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. + * + * @param topics must not be {@literal null}. + * @return a hot sequence of {@link Message messages}. + * @since 2.1 + */ + Flux> listenTo(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 12b1750c2..a88e62851 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java @@ -17,11 +17,13 @@ package org.springframework.data.redis.core; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; import java.lang.reflect.Proxy; import java.nio.ByteBuffer; import java.time.Duration; import java.time.Instant; +import java.util.Arrays; import java.util.List; import org.reactivestreams.Publisher; @@ -30,9 +32,12 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; +import org.springframework.data.redis.connection.ReactiveSubscription.Message; import org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor; import org.springframework.data.redis.core.script.ReactiveScriptExecutor; import org.springframework.data.redis.core.script.RedisScript; +import org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer; +import org.springframework.data.redis.listener.Topic; import org.springframework.data.redis.serializer.RedisElementReader; import org.springframework.data.redis.serializer.RedisElementWriter; import org.springframework.data.redis.serializer.RedisSerializationContext; @@ -193,16 +198,36 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations conn.close()); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveRedisOperations#convertAndSend(java.lang.String, java.lang.Object) + */ @Override public Mono convertAndSend(String destination, V message) { Assert.hasText(destination, "Destination channel must not be empty!"); + Assert.notNull(message, "Message must not be null!"); return createMono(connection -> connection.pubSubCommands().publish( getSerializationContext().getStringSerializationPair().write(destination), getSerializationContext().getValueSerializationPair().write(message))); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveRedisOperations#listenTo(org.springframework.data.redis.listener.Topic[]) + */ + @Override + public Flux> listenTo(Topic... topics) { + + ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(getConnectionFactory()); + + return container + .receive(Arrays.asList(topics), getSerializationContext().getStringSerializationPair(), + getSerializationContext().getValueSerializationPair()) // + .doFinally((signalType) -> container.destroyLater().subscribeOn(Schedulers.elastic())); + } + // ------------------------------------------------------------------------- // Methods dealing with Redis keys // ------------------------------------------------------------------------- diff --git a/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java b/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java index fd3efcfea..fbe37703a 100644 --- a/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java +++ b/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java @@ -21,8 +21,17 @@ import lombok.Getter; import lombok.RequiredArgsConstructor; import java.lang.reflect.Array; -import java.util.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; import java.util.Map.Entry; +import java.util.Optional; +import java.util.Set; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -35,7 +44,6 @@ import org.springframework.core.convert.support.GenericConversionService; import org.springframework.data.convert.CustomConversions; import org.springframework.data.convert.EntityInstantiator; import org.springframework.data.convert.EntityInstantiators; -import org.springframework.data.keyvalue.core.mapping.KeySpaceResolver; import org.springframework.data.mapping.AssociationHandler; import org.springframework.data.mapping.MappingException; import org.springframework.data.mapping.PersistentPropertyAccessor; @@ -450,9 +458,10 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean { targetProperty = getTargetPropertyOrNullForPath(path.replaceAll("\\.\\[.*\\]", ""), update.getTarget()); TypeInformation ti = targetProperty == null ? ClassTypeInformation.OBJECT - : (targetProperty.isMap() ? (targetProperty.getTypeInformation().getMapValueType() != null - ? targetProperty.getTypeInformation().getRequiredMapValueType() - : ClassTypeInformation.OBJECT) : targetProperty.getTypeInformation().getActualType()); + : (targetProperty.isMap() + ? (targetProperty.getTypeInformation().getMapValueType() != null + ? targetProperty.getTypeInformation().getRequiredMapValueType() : ClassTypeInformation.OBJECT) + : targetProperty.getTypeInformation().getActualType()); writeInternal(entity.getKeySpace(), pUpdate.getPropertyPath(), pUpdate.getValue(), ti, sink); return; diff --git a/src/main/java/org/springframework/data/redis/listener/ChannelTopic.java b/src/main/java/org/springframework/data/redis/listener/ChannelTopic.java index 65534c788..0711c7cef 100644 --- a/src/main/java/org/springframework/data/redis/listener/ChannelTopic.java +++ b/src/main/java/org/springframework/data/redis/listener/ChannelTopic.java @@ -44,7 +44,7 @@ public class ChannelTopic implements Topic { /** * Create a new {@link ChannelTopic} for channel subscriptions. - * + * * @param name the channel name, must not be {@literal null} or empty. * @return the {@link ChannelTopic} for {@code channelName}. * @since 2.1 diff --git a/src/main/java/org/springframework/data/redis/listener/PatternTopic.java b/src/main/java/org/springframework/data/redis/listener/PatternTopic.java index 42861dbd6..b118491a9 100644 --- a/src/main/java/org/springframework/data/redis/listener/PatternTopic.java +++ b/src/main/java/org/springframework/data/redis/listener/PatternTopic.java @@ -24,6 +24,7 @@ import org.springframework.util.Assert; * * @author Costin Leau * @author Mark Paluch + * @author Christoph Strobl */ @EqualsAndHashCode public class PatternTopic implements Topic { @@ -49,7 +50,7 @@ public class PatternTopic implements Topic { * @return the {@link PatternTopic} for {@code pattern}. * @since 2.1 */ - static PatternTopic of(String pattern) { + public static PatternTopic of(String pattern) { return new PatternTopic(pattern); } 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 b49d0260b..0501fc449 100644 --- a/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,12 +18,12 @@ package org.springframework.data.redis.listener; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoProcessor; +import reactor.core.scheduler.Schedulers; import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Collection; import java.util.HashMap; -import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -34,16 +34,19 @@ import java.util.stream.StreamSupport; import org.springframework.beans.factory.DisposableBean; import org.springframework.dao.InvalidDataAccessApiUsageException; +import org.springframework.data.redis.connection.ReactivePubSubCommands; import org.springframework.data.redis.connection.ReactiveRedisConnection; import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; import org.springframework.data.redis.connection.ReactiveSubscription; import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; +import org.springframework.data.redis.connection.ReactiveSubscription.Message; import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage; import org.springframework.data.redis.serializer.RedisElementReader; import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.lang.Nullable; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; /** * Container providing a stream of {@link ChannelMessage} for messages received via Redis Pub/Sub listeners. The stream @@ -55,11 +58,12 @@ import org.springframework.util.Assert; * operations. Using reactive infrastructure allows usage of a single connection due to channel multiplexing. *

* This class is thread-safe and allows subscription by multiple concurrent threads. - * + * * @author Mark Paluch + * @author Christoph Strobl * @since 2.1 * @see ReactiveSubscription - * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands + * @see ReactivePubSubCommands */ public class ReactiveRedisMessageListenerContainer implements DisposableBean { @@ -71,7 +75,7 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { /** * Create a new {@link ReactiveRedisMessageListenerContainer} given {@link ReactiveRedisConnectionFactory}. - * + * * @param connectionFactory must not be {@literal null}. */ public ReactiveRedisMessageListenerContainer(ReactiveRedisConnectionFactory connectionFactory) { @@ -86,8 +90,13 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { */ @Override public void destroy() { + destroyLater().block(); + } - ReactiveRedisConnection connection = this.connection; + /** + * @return the {@link Mono} signalling container termination. + */ + public Mono destroyLater() { if (connection != null) { @@ -97,55 +106,49 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { Map local = new HashMap<>(subscriptions); List> monos = local.keySet().stream() // .peek(subscriptions::remove) // - .map(ReactiveSubscription::terminate) // + .map(ReactiveSubscription::cancel) // .collect(Collectors.toList()); if (terminationSignals == null) { - terminationSignals = Flux.merge(monos); + terminationSignals = Flux.concat(monos); } else { - terminationSignals = terminationSignals.mergeWith(Flux.merge(monos)); + terminationSignals = terminationSignals.mergeWith(Flux.concat(monos)); } } if (terminationSignals != null) { - terminationSignals.blockLast(); + return terminationSignals.collectList() + .doFinally(signalType -> connection.closeLater().subscribeOn(Schedulers.immediate())) + .flatMap(all -> Mono.empty()); } - - connection.close(); this.connection = null; } + + return Mono.empty(); } /** * Return the currently active {@link ReactiveSubscription subscriptions}. - * + * * @return {@link Set} of active {@link ReactiveSubscription} */ public Collection getActiveSubscriptions() { - Set subscriptions = new HashSet<>(this.subscriptions.size(), 1); - - this.subscriptions.forEach((subscription, subscribers) -> { - - if (subscribers.hasRegistration()) { - subscriptions.add(subscription); - } - }); - - return subscriptions; + return subscriptions.entrySet().stream().filter(entry -> entry.getValue().hasRegistration()) + .map(entry -> entry.getKey()).collect(Collectors.toList()); } /** * Subscribe to one or more {@link ChannelTopic}s and receive a stream of {@link ChannelMessage}. Messages and channel * names are treated as {@link String}. The message stream subscribes lazily to the Redis channels and unsubscribes if * the {@link org.reactivestreams.Subscription} is {@link org.reactivestreams.Subscription#cancel() cancelled}. - * + * * @param channelTopics the channels to subscribe. * @return the message stream. * @throws InvalidDataAccessApiUsageException if {@code patternTopics} is empty. * @see #receive(Iterable, SerializationPair, SerializationPair) */ - public Flux> receive(ChannelTopic... channelTopics) { + public Flux> receive(ChannelTopic... channelTopics) { Assert.notNull(channelTopics, "ChannelTopics must not be null!"); Assert.noNullElements(channelTopics, "ChannelTopics must not contain null elements!"); @@ -158,8 +161,8 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { * and channel names are treated as {@link String}. The message stream subscribes lazily to the Redis channels and * unsubscribes if the {@link org.reactivestreams.Subscription} is {@link org.reactivestreams.Subscription#cancel() * cancelled}. - * - * @param channelTopics the channels to subscribe. + * + * @param patternTopics the channels to subscribe. * @return the message stream. * @throws InvalidDataAccessApiUsageException if {@code patternTopics} is empty. * @see #receive(Iterable, SerializationPair, SerializationPair) @@ -180,39 +183,35 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { * given {@code channelSerializer} and {@code messageSerializer}. The message stream subscribes lazily to the Redis * channels and unsubscribes if the {@link org.reactivestreams.Subscription} is * {@link org.reactivestreams.Subscription#cancel() cancelled}. - * + * * @param topics the channels to subscribe. * @return the message stream. * @see #receive(Iterable, SerializationPair, SerializationPair) * @throws InvalidDataAccessApiUsageException if {@code topics} is empty. */ - public Flux> receive(Iterable topics, - SerializationPair channelSerializer, SerializationPair messageSerializer) { + public Flux> receive(Iterable topics, SerializationPair channelSerializer, + SerializationPair messageSerializer) { Assert.notNull(topics, "Topics must not be null!"); - ReactiveRedisConnection connection = this.connection; - if (connection == null) { - throw new IllegalStateException("ReactiveRedisMessageListenerContainer is already disposed!"); - } - - Mono subscription = connection.pubSubCommands().createSubscription(); + verifyConnection(); ByteBuffer[] patterns = getTargets(topics, PatternTopic.class); ByteBuffer[] channels = getTargets(topics, ChannelTopic.class); - if (patterns.length == 0 && channels.length == 0) { - throw new InvalidDataAccessApiUsageException("No channels or patterns to subscribe"); + if (ObjectUtils.isEmpty(patterns) && ObjectUtils.isEmpty(channels)) { + throw new InvalidDataAccessApiUsageException("No channels or patterns to subscribe to."); } - return doReceive(channelSerializer, messageSerializer, subscription, patterns, channels); + return doReceive(channelSerializer, messageSerializer, connection.pubSubCommands().createSubscription(), patterns, + channels); } - private Flux> doReceive(SerializationPair channelSerializer, + private Flux> doReceive(SerializationPair channelSerializer, SerializationPair messageSerializer, Mono subscription, ByteBuffer[] patterns, ByteBuffer[] channels) { - Flux> messageStream = subscription.flatMapMany(it -> { + Flux> messageStream = subscription.flatMapMany(it -> { Mono subscribe = subscribe(patterns, channels, it); @@ -238,15 +237,16 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { private static Mono subscribe(ByteBuffer[] patterns, ByteBuffer[] channels, ReactiveSubscription it) { - Assert.isTrue(channels.length != 0 || patterns.length != 0, "Must provide either channels or patterns!"); + Assert.isTrue(!ObjectUtils.isEmpty(channels) || !ObjectUtils.isEmpty(patterns), + "Must provide either channels or patterns!"); Mono subscribe = null; - if (patterns.length != 0) { + if (!ObjectUtils.isEmpty(patterns)) { subscribe = it.pSubscribe(patterns); } - if (channels.length != 0) { + if (!ObjectUtils.isEmpty(channels)) { Mono channelsSubscribe = it.subscribe(channels); @@ -260,6 +260,17 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { return subscribe; } + private boolean isActive() { + return connection != null; + } + + private void verifyConnection() { + + if (!isActive()) { + throw new IllegalStateException("ReactiveRedisMessageListenerContainer is already disposed!"); + } + } + private Subscribers getSubscribers(ReactiveSubscription it) { return subscriptions.computeIfAbsent(it, key -> new Subscribers()); } @@ -274,8 +285,8 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { } @SuppressWarnings("unchecked") - private ChannelMessage readMessage(RedisElementReader channelSerializer, - RedisElementReader messageSerializer, ChannelMessage message) { + private Message readMessage(RedisElementReader channelSerializer, + RedisElementReader messageSerializer, Message message) { if (message instanceof PatternMessage) { @@ -306,7 +317,7 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { /** * Object to track subscriber count and to determine the last unsubscribed subscriber. - * + * * @author Mark Paluch */ static class Subscribers { diff --git a/src/main/java/org/springframework/data/redis/serializer/DefaultRedisElementReader.java b/src/main/java/org/springframework/data/redis/serializer/DefaultRedisElementReader.java index bc1aa1a5b..fd2d27978 100644 --- a/src/main/java/org/springframework/data/redis/serializer/DefaultRedisElementReader.java +++ b/src/main/java/org/springframework/data/redis/serializer/DefaultRedisElementReader.java @@ -19,6 +19,7 @@ import lombok.RequiredArgsConstructor; import java.nio.ByteBuffer; +import org.springframework.data.redis.util.ByteUtils; import org.springframework.lang.Nullable; /** @@ -44,9 +45,7 @@ class DefaultRedisElementReader implements RedisElementReader { return (T) buffer; } - byte[] bytes = new byte[buffer.slice().remaining()]; - buffer.get(bytes); - - return serializer.deserialize(bytes); + return serializer.deserialize(ByteUtils.extractBytes(buffer)); } + } diff --git a/src/main/java/org/springframework/data/redis/util/ByteUtils.java b/src/main/java/org/springframework/data/redis/util/ByteUtils.java index 2e24adb96..a4d30c4cd 100644 --- a/src/main/java/org/springframework/data/redis/util/ByteUtils.java +++ b/src/main/java/org/springframework/data/redis/util/ByteUtils.java @@ -232,4 +232,21 @@ public final class ByteUtils { return charset.encode(theString); } + + /** + * Extract/Transfer bytes from the given {@link ByteBuffer} into an array by duplicating the buffer and fetching its + * content. + * + * @param buffer must not be {@literal null}. + * @return the extracted bytes. + * @since 2.1 + */ + public static byte[] extractBytes(ByteBuffer buffer) { + + ByteBuffer duplicate = buffer.duplicate(); + byte[] bytes = new byte[duplicate.remaining()]; + duplicate.get(bytes); + + return bytes; + } } diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java index 30e38f1b6..0d259d9a3 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java @@ -214,7 +214,6 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati MessageListener listener = (message, pattern) -> { messages.add(message); - System.out.println("Received message '" + new String(message.getBody()) + "'"); }; Thread t = new Thread() { @@ -272,7 +271,6 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati final MessageListener listener = (message, pattern) -> { assertEquals(expectedPattern, new String(pattern)); messages.add(message); - System.out.println("Received message '" + new String(message.getBody()) + "'"); }; Thread th = new Thread() { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java index 7b7fe89b6..00aa4ee75 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -37,13 +37,14 @@ import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; import org.springframework.data.redis.RedisSystemException; -import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; +import org.springframework.data.redis.connection.ReactiveSubscription.Message; import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage; /** * Unit tests for {@link LettuceReactiveSubscription}. - * + * * @author Mark Paluch + * @author Christoph Strobl */ @RunWith(MockitoJUnitRunner.class) public class LettuceReactiveSubscriptionUnitTests { @@ -193,7 +194,7 @@ public class LettuceReactiveSubscriptionUnitTests { when(commandsMock.observePatterns()).thenReturn(Flux.never()); StepVerifier.create(subscription.receive()).then(() -> { - subscription.terminate().subscribe(); + subscription.cancel().subscribe(); }).expectError(CancellationException.class).verify(); assertThat(subscription.getPatterns()).isEmpty(); @@ -211,7 +212,7 @@ public class LettuceReactiveSubscriptionUnitTests { when(commandsMock.observeChannels()).thenReturn(Flux.never()); when(commandsMock.observePatterns()).thenReturn(emitter); - Flux> receive = subscription.receive(); + Flux> receive = subscription.receive(); Disposable subscribe = receive.subscribe(); assertThat(emitter.downstreamCount()).isEqualTo(1); @@ -221,13 +222,15 @@ public class LettuceReactiveSubscriptionUnitTests { } private static io.lettuce.core.pubsub.api.reactive.ChannelMessage createChannelMessage( - String channel, String body) { - return new io.lettuce.core.pubsub.api.reactive.ChannelMessage<>(getByteBuffer(channel), getByteBuffer(body)); + String channel, String message) { + + return new io.lettuce.core.pubsub.api.reactive.ChannelMessage<>(getByteBuffer(channel), getByteBuffer(message)); } private static io.lettuce.core.pubsub.api.reactive.PatternMessage createPatternMessage( - String pattern, String channel, String body) { + String pattern, String channel, String message) { + return new io.lettuce.core.pubsub.api.reactive.PatternMessage<>(getByteBuffer(pattern), getByteBuffer(channel), - getByteBuffer(body)); + getByteBuffer(message)); } } 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 de93aa0d1..bff84e8ac 100644 --- a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java @@ -39,6 +39,9 @@ import org.springframework.data.redis.Person; import org.springframework.data.redis.PersonObjectFactory; import org.springframework.data.redis.connection.DataType; import org.springframework.data.redis.connection.ReactiveRedisClusterConnection; +import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; +import org.springframework.data.redis.connection.ReactiveSubscription.Message; +import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.script.DefaultRedisScript; @@ -386,4 +389,49 @@ public class ReactiveRedisTemplateIntegrationTests { StepVerifier.create(hashOperations.get(key, hashField)).expectNext(hashValue).verifyComplete(); } + + @Test // DATAREDIS-612 + public void listenToChannelShouldReceiveChannelMessagesCorrectly() throws InterruptedException { + + String channel = "my-channel"; + + V message = valueFactory.instance(); + + StepVerifier.create(redisTemplate.listenToChannel(channel)) // + .thenAwait(Duration.ofMillis(500)) // just make sure we the subscription completed + .then(() -> redisTemplate.convertAndSend(channel, message).block()) // + .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)); + } + + @Test // DATAREDIS-612 + public void listenToChannelPatternShouldReceiveChannelMessagesCorrectly() throws InterruptedException { + + String channel = "my-channel"; + String pattern = "my-*"; + + V message = valueFactory.instance(); + + Flux> stream = redisTemplate.listenToPattern(pattern); + + StepVerifier.create(stream) // + .thenAwait(Duration.ofMillis(500)) // just make sure we the subscription completed + .then(() -> redisTemplate.convertAndSend(channel, message).block()) // + .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)); + } } diff --git a/src/test/java/org/springframework/data/redis/listener/ReactiveOperationsTestParams.java b/src/test/java/org/springframework/data/redis/listener/ReactiveOperationsTestParams.java index dae2efeee..84d50d418 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveOperationsTestParams.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveOperationsTestParams.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. 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 45608e5e0..6d9d30949 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.*; import reactor.core.Disposable; import reactor.test.StepVerifier; +import java.time.Duration; import java.util.Collection; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingDeque; @@ -35,6 +36,7 @@ import org.junit.runners.Parameterized; import org.junit.runners.Parameterized.Parameters; import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.connection.ReactiveSubscription; +import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; @@ -44,7 +46,7 @@ import org.springframework.lang.Nullable; /** * Integration tests for {@link ReactiveRedisMessageListenerContainer} via Lettuce. - * + * * @author Mark Paluch */ @RunWith(Parameterized.class) @@ -52,7 +54,7 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests { static final String CHANNEL1 = "my-channel"; static final String PATTERN1 = "my-chan*"; - public static final String MESSAGE = "hello world"; + static final String MESSAGE = "hello world"; private final LettuceConnectionFactory connectionFactory; private @Nullable RedisConnection connection; @@ -154,6 +156,45 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests { container.destroy(); } + @Test // DATAREDIS-612 + public void listenToChannelShouldReceiveChannelMessagesCorrectly() throws InterruptedException { + + ReactiveRedisTemplate template = new ReactiveRedisTemplate<>(connectionFactory, + RedisSerializationContext.string()); + + StepVerifier.create(template.listenToChannel(CHANNEL1)) // + .thenAwait(Duration.ofMillis(100)) // just make sure we the subscription completed + .then(() -> connection.publish(CHANNEL1.getBytes(), MESSAGE.getBytes())) // + .assertNext(message -> { + + assertThat(message).isInstanceOf(ChannelMessage.class); + assertThat(message.getMessage()).isEqualTo(MESSAGE); + assertThat(((ChannelMessage) message).getChannel()).isEqualTo(CHANNEL1); + }) // + .thenCancel() // + .verify(); + } + + @Test // DATAREDIS-612 + public void listenToPatternShouldReceiveMessagesCorrectly() { + + ReactiveRedisTemplate template = new ReactiveRedisTemplate<>(connectionFactory, + RedisSerializationContext.string()); + + StepVerifier.create(template.listenToPattern(PATTERN1)) // + .thenAwait(Duration.ofMillis(100)) // just make sure we the subscription completed + .then(() -> connection.publish(CHANNEL1.getBytes(), MESSAGE.getBytes())) // + .assertNext(message -> { + + assertThat(message).isInstanceOf(PatternMessage.class); + assertThat(((PatternMessage) message).getPattern()).isEqualTo(PATTERN1); + assertThat(((PatternMessage) message).getChannel()).isEqualTo(CHANNEL1); + assertThat(message.getMessage()).isEqualTo(MESSAGE); + }) // + .thenCancel() // + .verify(); + } + private static Runnable awaitSubscription(Supplier> activeSubscriptions) { return () -> { 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 54751e05c..ed1db72d3 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; import static org.springframework.data.redis.util.ByteUtils.*; +import org.springframework.data.redis.connection.ReactiveSubscription.Message; import reactor.core.Disposable; import reactor.core.publisher.DirectProcessor; import reactor.core.publisher.Flux; @@ -34,9 +35,9 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; +import org.springframework.data.redis.connection.ReactivePubSubCommands; import org.springframework.data.redis.connection.ReactiveRedisConnection; import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; -import org.springframework.data.redis.connection.ReactiveRedisPubSubCommands; import org.springframework.data.redis.connection.ReactiveSubscription; import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage; @@ -53,7 +54,7 @@ public class ReactiveRedisMessageListenerContainerUnitTests { @Mock ReactiveRedisConnectionFactory connectionFactoryMock; @Mock ReactiveRedisConnection connectionMock; - @Mock ReactiveRedisPubSubCommands commandsMock; + @Mock ReactivePubSubCommands commandsMock; @Mock ReactiveSubscription subscriptionMock; @Before @@ -61,6 +62,7 @@ public class ReactiveRedisMessageListenerContainerUnitTests { when(connectionFactoryMock.getReactiveConnection()).thenReturn(connectionMock); when(connectionMock.pubSubCommands()).thenReturn(commandsMock); + when(connectionMock.closeLater()).thenReturn(Mono.empty()); when(commandsMock.createSubscription()).thenReturn(Mono.just(subscriptionMock)); when(subscriptionMock.subscribe(any())).thenReturn(Mono.empty()); when(subscriptionMock.pSubscribe(any())).thenReturn(Mono.empty()); @@ -85,8 +87,8 @@ public class ReactiveRedisMessageListenerContainerUnitTests { when(subscriptionMock.receive()).thenReturn(Flux.never()); container = createContainer(); - StepVerifier.create(container.receive(PatternTopic.of("foo*"), PatternTopic.of("bar*"))).thenRequest(1) - .thenAwait().thenCancel().verify(); + StepVerifier.create(container.receive(PatternTopic.of("foo*"), PatternTopic.of("bar*"))).thenRequest(1).thenAwait() + .thenCancel().verify(); verify(subscriptionMock).pSubscribe(getByteBuffer("foo*"), getByteBuffer("bar*")); } @@ -117,12 +119,12 @@ public class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612 public void shouldEmitChannelMessage() { - DirectProcessor> processor = DirectProcessor.create(); + DirectProcessor> processor = DirectProcessor.create(); when(subscriptionMock.receive()).thenReturn(processor); container = createContainer(); - Flux> messageStream = container.receive(ChannelTopic.of("foo")); + Flux> messageStream = container.receive(ChannelTopic.of("foo")); StepVerifier.create(messageStream).then(() -> { processor.onNext(createChannelMessage("foo", "message")); @@ -136,7 +138,7 @@ public class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612 public void shouldEmitPatternMessage() { - DirectProcessor> processor = DirectProcessor.create(); + DirectProcessor> processor = DirectProcessor.create(); when(subscriptionMock.receive()).thenReturn(processor); container = createContainer(); @@ -164,7 +166,7 @@ public class ReactiveRedisMessageListenerContainerUnitTests { when(subscriptionMock.receive()).thenReturn(DirectProcessor.create()); container = createContainer(); - Flux> messageStream = container.receive(ChannelTopic.of("foo*")); + Flux> messageStream = container.receive(ChannelTopic.of("foo*")); Disposable subscription = messageStream.subscribe(); @@ -184,7 +186,7 @@ public class ReactiveRedisMessageListenerContainerUnitTests { when(subscriptionMock.receive()).thenReturn(DirectProcessor.create()); container = createContainer(); - Flux> messageStream = container.receive(new ChannelTopic("foo*")); + Flux> messageStream = container.receive(new ChannelTopic("foo*")); Disposable first = messageStream.subscribe(); Disposable second = messageStream.subscribe(); @@ -220,10 +222,10 @@ public class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612 public void shouldTerminateSubscriptionsOnShutdown() { - DirectProcessor> processor = DirectProcessor.create(); + DirectProcessor> processor = DirectProcessor.create(); when(subscriptionMock.receive()).thenReturn(processor); - when(subscriptionMock.terminate()).thenReturn(Mono.defer(() -> { + when(subscriptionMock.cancel()).thenReturn(Mono.defer(() -> { processor.onError(new CancellationException()); return Mono.empty(); @@ -240,7 +242,7 @@ public class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612 public void shouldCleanupDownstream() { - DirectProcessor> processor = DirectProcessor.create(); + DirectProcessor> processor = DirectProcessor.create(); when(subscriptionMock.receive()).thenReturn(processor); container = createContainer(); @@ -265,6 +267,7 @@ public class ReactiveRedisMessageListenerContainerUnitTests { private static PatternMessage createPatternMessage(String pattern, String channel, String body) { + return new PatternMessage<>(getByteBuffer(pattern), getByteBuffer(channel), getByteBuffer(body)); } }