Package org.springframework.data.redis.listener

Class ReactiveRedisMessageListenerContainer

java.lang.Object

org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer

All Implemented Interfaces:

DisposableBean

public class ReactiveRedisMessageListenerContainer
extends Object
implements DisposableBean

Container providing a stream of ReactiveSubscription.ChannelMessage for messages received via Redis Pub/Sub listeners. The stream is infinite and registers Redis subscriptions. Handles the low level details of listening, converting and message dispatching.

Note the container allocates a single connection when it is created and releases the connection on destroy(). Connections are allocated eagerly to not interfere with non-blocking use during application 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.

Since:

2.1

Author:

Mark Paluch, Christoph Strobl

See Also:

• ReactiveSubscription
• ReactivePubSubCommands

Constructor Summary

Constructors

Constructor	Description
ReactiveRedisMessageListenerContainer(ReactiveRedisConnectionFactory connectionFactory)	Create a new ReactiveRedisMessageListenerContainer given ReactiveRedisConnectionFactory.

Method Summary

Modifier and Type	Method	Description
void	destroy()
reactor.core.publisher.Mono<Void>	destroyLater()
Collection<ReactiveSubscription>	getActiveSubscriptions()	Return the currently active subscriptions.
reactor.core.publisher.Flux<ReactiveSubscription.Message<String,String>>	receive(Iterable<? extends Topic> topics, SubscriptionListener subscriptionListener)	Subscribe to one or more Topics and receive a stream of ReactiveSubscription.ChannelMessage.
<C, B> reactor.core.publisher.Flux<ReactiveSubscription.Message<C,B>>	receive(Iterable<? extends Topic> topics, RedisSerializationContext.SerializationPair<C> channelSerializer, RedisSerializationContext.SerializationPair<B> messageSerializer)	Subscribe to one or more Topics and receive a stream of ReactiveSubscription.ChannelMessage.
<C, B> reactor.core.publisher.Flux<ReactiveSubscription.Message<C,B>>	receive(Iterable<? extends Topic> topics, RedisSerializationContext.SerializationPair<C> channelSerializer, RedisSerializationContext.SerializationPair<B> messageSerializer, SubscriptionListener subscriptionListener)	Subscribe to one or more Topics and receive a stream of ReactiveSubscription.ChannelMessage.
reactor.core.publisher.Flux<ReactiveSubscription.Message<String,String>>	receive(ChannelTopic... channelTopics)	Subscribe to one or more ChannelTopics and receive a stream of ReactiveSubscription.ChannelMessage.
reactor.core.publisher.Flux<ReactiveSubscription.PatternMessage<String,String,String>>	receive(PatternTopic... patternTopics)	Subscribe to one or more PatternTopics and receive a stream of ReactiveSubscription.PatternMessage.
<C, B> reactor.core.publisher.Mono<reactor.core.publisher.Flux<ReactiveSubscription.Message<C,B>>>	receiveLater(Iterable<? extends Topic> topics, RedisSerializationContext.SerializationPair<C> channelSerializer, RedisSerializationContext.SerializationPair<B> messageSerializer)	Subscribe to one or more Topics and receive a stream of ReactiveSubscription.ChannelMessage.
reactor.core.publisher.Mono<reactor.core.publisher.Flux<ReactiveSubscription.Message<String,String>>>	receiveLater(ChannelTopic... channelTopics)	Subscribe to one or more ChannelTopics and receive a stream of ReactiveSubscription.ChannelMessage once the returned Mono completes.
reactor.core.publisher.Mono<reactor.core.publisher.Flux<ReactiveSubscription.PatternMessage<String,String,String>>>	receiveLater(PatternTopic... patternTopics)	Subscribe to one or more PatternTopics and receive a stream of ReactiveSubscription.PatternMessage once the returned Mono completes.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

ReactiveRedisMessageListenerContainer

public ReactiveRedisMessageListenerContainer(ReactiveRedisConnectionFactory connectionFactory)

Create a new ReactiveRedisMessageListenerContainer given ReactiveRedisConnectionFactory.

Parameters:

connectionFactory - must not be null.

Method Details

destroy

public void destroy()

Specified by:

destroy in interface DisposableBean

destroyLater

public reactor.core.publisher.Mono<Void> destroyLater()

Returns:

the Mono signalling container termination.

getActiveSubscriptions

public Collection<ReactiveSubscription> getActiveSubscriptions()

Return the currently active subscriptions.

Returns:

Set of active ReactiveSubscription

receive

public reactor.core.publisher.Flux<ReactiveSubscription.Message<String,String>> receive(ChannelTopic... channelTopics)

Subscribe to one or more ChannelTopics and receive a stream of ReactiveSubscription.ChannelMessage. Messages and channel names are treated as String. The message stream subscribes lazily to the Redis channels and unsubscribes if the Subscription is cancelled.

Parameters:

channelTopics - the channels to subscribe.

Returns:

the message stream.

Throws:

InvalidDataAccessApiUsageException - if channelTopics is empty.

See Also:

• receive(Iterable, SerializationPair, SerializationPair)

receiveLater

public reactor.core.publisher.Mono<reactor.core.publisher.Flux<ReactiveSubscription.Message<String,String>>> receiveLater(ChannelTopic... channelTopics)

Subscribe to one or more ChannelTopics and receive a stream of ReactiveSubscription.ChannelMessage once the returned Mono completes. Messages and channel names are treated as String. The message stream subscribes lazily to the Redis channels and unsubscribes if the inner Subscription is cancelled.

The returned Mono completes once the connection has been subscribed to the given topics. Note that cancelling the returned Mono can leave the connection in a subscribed state.

Parameters:

channelTopics - the channels to subscribe.

Returns:

the message stream.

Throws:

InvalidDataAccessApiUsageException - if channelTopics is empty.

Since:

2.6

receive

public reactor.core.publisher.Flux<ReactiveSubscription.PatternMessage<String,String,String>> receive(PatternTopic... patternTopics)

Subscribe to one or more PatternTopics and receive a stream of ReactiveSubscription.PatternMessage. Messages, pattern, and channel names are treated as String. The message stream subscribes lazily to the Redis channels and unsubscribes if the Subscription is cancelled.

Parameters:

patternTopics - the channels to subscribe.

Returns:

the message stream.

Throws:

InvalidDataAccessApiUsageException - if patternTopics is empty.

See Also:

• receive(Iterable, SerializationPair, SerializationPair)

receiveLater

public reactor.core.publisher.Mono<reactor.core.publisher.Flux<ReactiveSubscription.PatternMessage<String,String,String>>> receiveLater(PatternTopic... patternTopics)

Subscribe to one or more PatternTopics and receive a stream of ReactiveSubscription.PatternMessage once the returned Mono completes. Messages, pattern, and channel names are treated as String. The message stream subscribes lazily to the Redis channels and unsubscribes if the inner Subscription is cancelled.

The returned Mono completes once the connection has been subscribed to the given topics. Note that cancelling the returned Mono can leave the connection in a subscribed state.

Parameters:

patternTopics - the channels to subscribe.

Returns:

the message stream.

Throws:

InvalidDataAccessApiUsageException - if patternTopics is empty.

Since:

2.6

receive

public reactor.core.publisher.Flux<ReactiveSubscription.Message<String,String>> receive(Iterable<? extends Topic> topics,
 SubscriptionListener subscriptionListener)

Subscribe to one or more Topics and receive a stream of ReactiveSubscription.ChannelMessage. The stream may contain ReactiveSubscription.PatternMessage if subscribed to patterns. Messages, and channel names are serialized/deserialized using the given channelSerializer and messageSerializer. The message stream subscribes lazily to the Redis channels and unsubscribes if the Subscription is cancelled.

Parameters:

topics - the channels/patterns to subscribe.

subscriptionListener - listener to receive subscription/unsubscription notifications.

Returns:

the message stream.

Throws:

InvalidDataAccessApiUsageException - if patternTopics is empty.

Since:

2.6

See Also:

• receive(Iterable, SerializationPair, SerializationPair)

receive

public <C,
B>
reactor.core.publisher.Flux<ReactiveSubscription.Message<C,B>> receive(Iterable<? extends Topic> topics,
 RedisSerializationContext.SerializationPair<C> channelSerializer,
 RedisSerializationContext.SerializationPair<B> messageSerializer)

Subscribe to one or more Topics and receive a stream of ReactiveSubscription.ChannelMessage. The stream may contain ReactiveSubscription.PatternMessage if subscribed to patterns. Messages, and channel names are serialized/deserialized using the given channelSerializer and messageSerializer. The message stream subscribes lazily to the Redis channels and unsubscribes if the Subscription is cancelled.

Parameters:

topics - the channels/patterns to subscribe.

Returns:

the message stream.

Throws:

InvalidDataAccessApiUsageException - if topics is empty.

See Also:

• receive(Iterable, SerializationPair, SerializationPair)

receive

public <C,
B>
reactor.core.publisher.Flux<ReactiveSubscription.Message<C,B>> receive(Iterable<? extends Topic> topics,
 RedisSerializationContext.SerializationPair<C> channelSerializer,
 RedisSerializationContext.SerializationPair<B> messageSerializer,
 SubscriptionListener subscriptionListener)

Subscribe to one or more Topics and receive a stream of ReactiveSubscription.ChannelMessage. The stream may contain ReactiveSubscription.PatternMessage if subscribed to patterns. Messages, and channel names are serialized/deserialized using the given channelSerializer and messageSerializer. The message stream subscribes lazily to the Redis channels and unsubscribes if the Subscription is cancelled. SubscriptionListener is notified upon subscription/unsubscription and can be used for synchronization.

Parameters:

topics - the channels to subscribe.

channelSerializer - serialization pair to decode the channel/pattern name.

messageSerializer - serialization pair to decode the message body.

subscriptionListener - listener to receive subscription/unsubscription notifications.

Returns:

the message stream.

Throws:

InvalidDataAccessApiUsageException - if topics is empty.

Since:

2.6

See Also:

• receive(Iterable, SerializationPair, SerializationPair)

receiveLater

public <C,
B>
reactor.core.publisher.Mono<reactor.core.publisher.Flux<ReactiveSubscription.Message<C,B>>> receiveLater(Iterable<? extends Topic> topics,
 RedisSerializationContext.SerializationPair<C> channelSerializer,
 RedisSerializationContext.SerializationPair<B> messageSerializer)

Subscribe to one or more Topics and receive a stream of ReactiveSubscription.ChannelMessage. The returned Mono completes once the connection has been subscribed to the given topics. Note that cancelling the returned Mono can leave the connection in a subscribed state.

Parameters:

topics - the channels to subscribe.

channelSerializer - serialization pair to decode the channel/pattern name.

messageSerializer - serialization pair to decode the message body.

Returns:

the message stream.

Throws:

InvalidDataAccessApiUsageException - if topics is empty.

Since:

2.6