org.springframework.integration.redis.inbound

Class ReactiveRedisStreamMessageProducer

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.endpoint.AbstractEndpoint

org.springframework.integration.endpoint.MessageProducerSupport

org.springframework.integration.redis.inbound.ReactiveRedisStreamMessageProducer

All Implemented Interfaces:

Aware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, SmartInitializingSingleton, ApplicationContextAware, Lifecycle, Phased, SmartLifecycle, ExpressionCapable, MessageProducer, IntegrationPattern, NamedComponent, ManageableLifecycle, ManageableSmartLifecycle, TrackableComponent

public class ReactiveRedisStreamMessageProducer
extends MessageProducerSupport

A MessageProducerSupport for reading messages from a Redis Stream and publishing them into the provided output channel. By default this adapter reads message as a standalone client XREAD (Redis command) but can be switched to a Consumer Group feature XREADGROUP by setting consumerName field. By default the Consumer Group name is the id of this bean IntegrationObjectSupport.getBeanName().

Since:

5.4

Author:

Attoumane Ahamadi, Artem Bilan, Rohan Mukesh

Field Summary

Fields inherited from class org.springframework.integration.endpoint.AbstractEndpoint

lifecycleCondition, lifecycleLock

Fields inherited from class org.springframework.integration.context.IntegrationObjectSupport

EXPRESSION_PARSER, logger

Fields inherited from interface org.springframework.context.SmartLifecycle

DEFAULT_PHASE

Constructor Summary

Constructors

Constructor and Description
ReactiveRedisStreamMessageProducer(ReactiveRedisConnectionFactory reactiveConnectionFactory, String streamKey)

Method Summary

Modifier and Type	Method and Description
protected void	doStart() Take no action by default.
String	getComponentType() Subclasses may implement this method to provide component type information.
protected void	onInit() Subclasses may implement this for initialization logic.
void	setAutoAck(boolean autoAck) Set whether or not acknowledge message read in the Consumer Group.
void	setConsumerGroup(String consumerGroup) Set the name of the Consumer Group.
void	setConsumerName(String consumerName) Set the name of the consumer.
void	setCreateConsumerGroup(boolean createConsumerGroup) Create the Consumer Group if and only if it does not exist.
void	setExtractPayload(boolean extractPayload) Configure this channel adapter to extract or not value from the Record.
void	setReadOffset(ReadOffset readOffset) Define the offset from which we want to read message.
void	setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions<String,?> streamReceiverOptions) Set ReactiveStreamOperations used to customize the StreamReceiver.

Methods inherited from class org.springframework.integration.endpoint.MessageProducerSupport

afterSingletonsInstantiated, buildErrorMessage, doStop, getErrorChannel, getErrorMessageAttributes, getIntegrationPatternType, getMessagingTemplate, getOutputChannel, sendErrorMessageIfNecessary, sendMessage, setErrorChannel, setErrorChannelName, setErrorMessageStrategy, setOutputChannel, setOutputChannelName, setSendTimeout, setShouldTrack, subscribeToPublisher

Methods inherited from class org.springframework.integration.endpoint.AbstractEndpoint

destroy, doStop, getPhase, getRole, isAutoStartup, isRunning, setAutoStartup, setPhase, setRole, start, stop, stop

Methods inherited from class org.springframework.integration.context.IntegrationObjectSupport

afterPropertiesSet, extractTypeIfPossible, generateId, getApplicationContext, getApplicationContextId, getBeanDescription, getBeanFactory, getBeanName, getChannelResolver, getComponentName, getConversionService, getExpression, getIntegrationProperties, getIntegrationProperty, getMessageBuilderFactory, getTaskScheduler, isInitialized, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentName, setConversionService, setMessageBuilderFactory, setPrimaryExpression, setTaskScheduler, toString

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.integration.support.context.NamedComponent

getBeanName, getComponentName

Constructor Detail

ReactiveRedisStreamMessageProducer

public ReactiveRedisStreamMessageProducer(ReactiveRedisConnectionFactory reactiveConnectionFactory,
                                          String streamKey)

Method Detail

setReadOffset

public void setReadOffset(ReadOffset readOffset)

Define the offset from which we want to read message. By default the ReadOffset.latest() is used. ReadOffset.latest() is equal to '$', which is the Id used with XREAD to get new data added to the stream. Note that when switching to the Consumer Group feature, we set it to ReadOffset.lastConsumed() if it is still equal to ReadOffset.latest().

Parameters:

readOffset - the desired offset

setExtractPayload

public void setExtractPayload(boolean extractPayload)

Configure this channel adapter to extract or not value from the Record.

Parameters:

extractPayload - default true

setAutoAck

public void setAutoAck(boolean autoAck)

Set whether or not acknowledge message read in the Consumer Group. true by default.

Parameters:

autoAck - the acknowledge option.

setConsumerGroup

public void setConsumerGroup(@Nullable
                             String consumerGroup)

Set the name of the Consumer Group. It is possible to create that Consumer Group if desired, see: createConsumerGroup. If not set, the defined bean name IntegrationObjectSupport.getBeanName() is used.

Parameters:

consumerGroup - the Consumer Group on which this adapter should register to listen messages.

setConsumerName

public void setConsumerName(@Nullable
                            String consumerName)

Set the name of the consumer. When a consumer name is provided, this adapter is switched to the Consumer Group feature. Note that this value should be unique in the group.

Parameters:

consumerName - the consumer name in the Consumer Group

setCreateConsumerGroup

public void setCreateConsumerGroup(boolean createConsumerGroup)

Create the Consumer Group if and only if it does not exist. During the creation we also create the stream, see MKSTREAM.

Parameters:

createConsumerGroup - specify if we should create the Consumer Group, false by default

setStreamReceiverOptions

public void setStreamReceiverOptions(@Nullable
                                     StreamReceiver.StreamReceiverOptions<String,?> streamReceiverOptions)

Set ReactiveStreamOperations used to customize the StreamReceiver. It provides a way to set the polling timeout and the serialization context. By default the polling timeout is set to infinite and StringRedisSerializer is used.

Parameters:

streamReceiverOptions - the desired receiver options

getComponentType

public String getComponentType()

Description copied from class: IntegrationObjectSupport

Subclasses may implement this method to provide component type information.

Specified by:

getComponentType in interface NamedComponent

Overrides:

getComponentType in class IntegrationObjectSupport

onInit

protected void onInit()

Description copied from class: IntegrationObjectSupport

Subclasses may implement this for initialization logic.

Overrides:

onInit in class MessageProducerSupport

doStart

protected void doStart()

Description copied from class: MessageProducerSupport

Take no action by default. Subclasses may override this if they need lifecycle-managed behavior. Protected by 'lifecycleLock'.

Overrides:

doStart in class MessageProducerSupport