Package org.springframework.data.redis.stream

Interface StreamReceiver<K,V extends Record<K,?>>

Type Parameters:

K - Stream key and Stream field type.

V - Stream value type.

public interface StreamReceiver<K,V extends Record<K,?>>

A receiver to consume Redis Streams using reactive infrastructure.

Once created, a StreamReceiver can subscribe to a Redis Stream and consume incoming records. Consider a Flux of Record infinite. Cancelling the Subscription terminates eventually background polling. Records are converted using key and value serializers to support various serialization strategies.
StreamReceiver supports three modes of stream consumption:

• Standalone
• Using a Consumer with external acknowledge
• Using a Consumer with auto-acknowledge

Reading from a stream requires polling and a strategy to advance stream offsets. Depending on the initial ReadOffset, StreamReceiver applies an individual strategy to obtain the next ReadOffset:
Standalone

• ReadOffset.from(String) Offset using a particular record Id: Start with the given offset and use the last seen record Id.
• ReadOffset.lastConsumed() Last consumed: Start with the latest offset ($) and use the last seen record Id.
• ReadOffset.latest() Last consumed: Start with the latest offset ($) and use latest offset ($) for subsequent reads.

Using Consumer

• ReadOffset.from(String) Offset using a particular record Id: Start with the given offset and use the last seen record Id.
• ReadOffset.lastConsumed() Last consumed: Start with the last consumed record by the consumer (>) and use the last consumed record by the consumer (>) for subsequent reads.
• ReadOffset.latest() Last consumed: Start with the latest offset ($) and use latest offset ($) for subsequent reads.

Note: Using ReadOffset.latest() bears the chance of dropped records as records can arrive in the time during polling is suspended. Use recordId's as offset or ReadOffset.lastConsumed() to minimize the chance of record loss.

StreamReceiver propagates errors during stream reads and deserialization as terminal error signal by default. Configuring a resume function allows conditional resumption by dropping the record or by propagating the error to terminate the subscription.

See the following example code how to use StreamReceiver:

 ReactiveRedisConnectionFactory factory = …;

 StreamReceiver<String, String, String> receiver = StreamReceiver.create(factory);
 Flux<MapRecord<String, String, String>> records = receiver.receive(StreamOffset.fromStart("my-stream"));

 recordFlux.doOnNext(record -> …);
 

Since:

2.2

Author:

Mark Paluch, Eddie McDaniel

See Also:

• StreamReceiver.StreamReceiverOptions.builder()
• ReactiveStreamOperations
• ReactiveRedisConnectionFactory
• StreamMessageListenerContainer

Nested Class Summary

Nested Classes

Modifier and Type	Interface	Description
static class	StreamReceiver.StreamReceiverOptions<K,V extends Record<K,?>>	Options for StreamReceiver.
static class	StreamReceiver.StreamReceiverOptionsBuilder<K,V extends Record<K,?>>	Builder for StreamReceiver.StreamReceiverOptions.

Method Summary

Modifier and Type	Method	Description
static StreamReceiver<String,MapRecord<String,String,String>>	create(ReactiveRedisConnectionFactory connectionFactory)	Create a new StreamReceiver using string serializers given ReactiveRedisConnectionFactory.
static <K, V extends Record<K, ?>> StreamReceiver<K,V>	create(ReactiveRedisConnectionFactory connectionFactory, StreamReceiver.StreamReceiverOptions<K,V> options)	Create a new StreamReceiver given ReactiveRedisConnectionFactory and StreamReceiver.StreamReceiverOptions.
reactor.core.publisher.Flux<V>	receive(Consumer consumer, StreamOffset<K> streamOffset)	Starts a Redis Stream consumer that consumes records from the stream.
reactor.core.publisher.Flux<V>	receive(StreamOffset<K> streamOffset)	Starts a Redis Stream consumer that consumes records from the stream.
reactor.core.publisher.Flux<V>	receiveAutoAck(Consumer consumer, StreamOffset<K> streamOffset)	Starts a Redis Stream consumer that consumes records from the stream.

Method Details

create

static StreamReceiver<String,MapRecord<String,String,String>> create(ReactiveRedisConnectionFactory connectionFactory)

Create a new StreamReceiver using string serializers given ReactiveRedisConnectionFactory.

Parameters:

connectionFactory - must not be null.

Returns:

the new StreamReceiver.

create

static <K,
V extends Record<K,
?>>
StreamReceiver<K,V> create(ReactiveRedisConnectionFactory connectionFactory,
 StreamReceiver.StreamReceiverOptions<K,V> options)

Create a new StreamReceiver given ReactiveRedisConnectionFactory and StreamReceiver.StreamReceiverOptions.

Parameters:

connectionFactory - must not be null.

options - must not be null.

Returns:

the new StreamReceiver.

receive

reactor.core.publisher.Flux<V> receive(StreamOffset<K> streamOffset)

Starts a Redis Stream consumer that consumes records from the stream. Records are consumed from Redis and delivered on the returned Flux when requests are made on the Flux. The receiver is closed when the returned Flux terminates.

Every record must be acknowledged using ReactiveStreamCommands.xAck(ByteBuffer, String, String...)

Parameters:

streamOffset - the stream along its offset.

Returns:

Flux of inbound Records.

See Also:

• StreamOffset.create(Object, ReadOffset)

receiveAutoAck

reactor.core.publisher.Flux<V> receiveAutoAck(Consumer consumer,
 StreamOffset<K> streamOffset)

Starts a Redis Stream consumer that consumes records from the stream. Records are consumed from Redis and delivered on the returned Flux when requests are made on the Flux. The receiver is closed when the returned Flux terminates.

Every record is acknowledged when received.

Parameters:

consumer - consumer group, must not be null.

streamOffset - the stream along its offset.

Returns:

Flux of inbound Records.

See Also:

• StreamOffset.create(Object, ReadOffset)
• ReadOffset.lastConsumed()

receive

reactor.core.publisher.Flux<V> receive(Consumer consumer,
 StreamOffset<K> streamOffset)

Starts a Redis Stream consumer that consumes records from the stream. Records are consumed from Redis and delivered on the returned Flux when requests are made on the Flux. The receiver is closed when the returned Flux terminates.

Every record must be acknowledged using ReactiveStreamOperations.acknowledge(Object, String, String...) after processing.

Parameters:

consumer - consumer group, must not be null.

streamOffset - the stream along its offset.

Returns:

Flux of inbound Records.

See Also:

• StreamOffset.create(Object, ReadOffset)
• ReadOffset.lastConsumed()