Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-1230 - Revise error handling in StreamReceiver and StreamMessageListenerContainer.

Browse Source 
Reading and deserialization in StreamReceiver and StreamMessageListener is now decoupled from each other to allow fine-grained control over errors and resumption.

Previously, we used the Template API to read and deserialize stream records. Now the actual read happens before the deserialization so that errors during deserialization of individual messages can be handled properly. This split also allows advancing in the stream read. Previously, the failed deserialization prevented of getting hold of the non-serialized Stream record which caused the stream receiver to remain at the offset that fetched the offending record which effectively lead to an infinite loop.

We also support a resume function in StreamReceiver to control whether stream reads should be resumed or terminated.

Original Pull Request: #576
This commit is contained in:
Mark Paluch
2020-11-24 15:09:09 +01:00
committed by Christoph Strobl
parent 7d2bf8f852
commit a5700ba71a
12 changed files with 415 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java
View File

@@ -416,7 +416,8 @@ class DefaultReactiveStreamOperations<K, HK, HV> implements ReactiveStreamOperat
return (HV) serializationContext.getHashValueSerializationPair().read(buffer);
}
private MapRecord<K, HK, HV> deserializeRecord(ByteBufferRecord record) {
@Override
public MapRecord<K, HK, HV> deserializeRecord(ByteBufferRecord record) {
return record.map(it -> it.mapEntries(this::deserializeRecordFields).withStreamKey(readKey(record.getStream())));
}

7
src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java
View File

@@ -347,6 +347,11 @@ class DefaultStreamOperations<K, HK, HV> extends AbstractOperations<K, Object> i
return objectMapper.getHashMapper(targetType);
}
@Override
public MapRecord<K, HK, HV> deserializeRecord(ByteRecord record) {
return record.deserialize(keySerializer(), hashKeySerializer(), hashValueSerializer());
}
protected byte[] serializeHashValueIfRequires(HV value) {
return hashValueSerializerPresent() ? serialize(value, hashValueSerializer())
: objectMapper.getConversionService().convert(value, byte[].class);
@@ -386,7 +391,7 @@ class DefaultStreamOperations<K, HK, HV> extends AbstractOperations<K, Object> i
List<MapRecord<K, HK, HV>> result = new ArrayList<>();
for (ByteRecord record : raw) {
result.add(record.deserialize(keySerializer(), hashKeySerializer(), hashValueSerializer()));
result.add(deserializeRecord(record));
}
return result;

47
src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java
View File

@@ -22,23 +22,13 @@ import java.util.Arrays;
import java.util.Map;
import org.reactivestreams.Publisher;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.PendingMessage;
import org.springframework.data.redis.connection.stream.PendingMessages;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumer;
import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroup;
import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.hash.HashMapper;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -354,7 +344,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
Assert.notNull(targetType, "Target type must not be null");
return range(key, range, limit).map(it -> StreamObjectMapper.toObjectRecord(this, it, targetType));
return range(key, range, limit).map(it -> map(it, targetType));
}
/**
@@ -435,7 +425,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
Assert.notNull(targetType, "Target type must not be null");
return read(readOptions, streams).map(it -> StreamObjectMapper.toObjectRecord(this, it, targetType));
return read(readOptions, streams).map(it -> map(it, targetType));
}
/**
@@ -489,7 +479,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
Assert.notNull(targetType, "Target type must not be null");
return read(consumer, readOptions, streams).map(it -> StreamObjectMapper.toObjectRecord(this, it, targetType));
return read(consumer, readOptions, streams).map(it -> map(it, targetType));
}
/**
@@ -543,7 +533,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
Assert.notNull(targetType, "Target type must not be null");
return reverseRange(key, range, limit).map(it -> StreamObjectMapper.toObjectRecord(this, it, targetType));
return reverseRange(key, range, limit).map(it -> map(it, targetType));
}
/**
@@ -577,4 +567,29 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
*/
@Override
<V> HashMapper<V, HK, HV> getHashMapper(Class<V> targetType);
/**
* Map records from {@link MapRecord} to {@link ObjectRecord}.
*
* @param record the stream records to map.
* @param targetType the target type of the payload.
* @return the mapped {@link ObjectRecord}.
* @since 2.x
*/
default <V> ObjectRecord<K, V> map(MapRecord<K, HK, HV> record, Class<V> targetType) {
Assert.notNull(record, "Records must not be null");
Assert.notNull(targetType, "Target type must not be null");
return StreamObjectMapper.toObjectRecord(record, this, targetType);
}
/**
* Deserialize a {@link ByteBufferRecord} using the configured serialization context into a {@link MapRecord}.
*
* @param record the stream record to map.
* @return deserialized {@link MapRecord}.
* @since 2.x
*/
MapRecord<K, HK, HV> deserializeRecord(ByteBufferRecord record);
}

10
src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java
View File

@@ -128,13 +128,13 @@ class StreamObjectMapper {
/**
* Convert the given {@link Record} into an {@link ObjectRecord}.
*
* @param provider provider for {@link HashMapper} to apply mapping for {@link ObjectRecord}.
* @param source the source value.
* @param provider provider for {@link HashMapper} to apply mapping for {@link ObjectRecord}.
* @param targetType the desired target type.
* @return the converted {@link ObjectRecord}.
*/
static <K, V, HK, HV> ObjectRecord<K, V> toObjectRecord(HashMapperProvider<HK, HV> provider,
MapRecord<K, HK, HV> source, Class<V> targetType) {
static <K, V, HK, HV> ObjectRecord<K, V> toObjectRecord(MapRecord<K, HK, HV> source,
HashMapperProvider<HK, HV> provider, Class<V> targetType) {
return source.toObjectRecord(provider.getHashMapper(targetType));
}
@@ -149,7 +149,7 @@ class StreamObjectMapper {
* {@literal null}.
*/
@Nullable
static <K, V, HK, HV> List<ObjectRecord<K, V>> map(@Nullable List<MapRecord<K, HK, HV>> records,
static <K, V, HK, HV> List<ObjectRecord<K, V>> toObjectRecords(@Nullable List<MapRecord<K, HK, HV>> records,
HashMapperProvider<HK, HV> hashMapperProvider, Class<V> targetType) {
if (records == null) {
@@ -161,7 +161,7 @@ class StreamObjectMapper {
}
if (records.size() == 1) {
return Collections.singletonList(toObjectRecord(hashMapperProvider, records.get(0), targetType));
return Collections.singletonList(toObjectRecord(records.get(0), hashMapperProvider, targetType));
}
List<ObjectRecord<K, V>> transformed = new ArrayList<>(records.size());

62
src/main/java/org/springframework/data/redis/core/StreamOperations.java
View File

@@ -23,21 +23,10 @@ import java.util.Map;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.PendingMessage;
import org.springframework.data.redis.connection.stream.PendingMessages;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumers;
import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroups;
import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.hash.HashMapper;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -356,7 +345,7 @@ public interface StreamOperations<K, HK, HV> extends HashMapperProvider<HK, HV>
Assert.notNull(targetType, "Target type must not be null");
return StreamObjectMapper.map(range(key, range, limit), this, targetType);
return map(range(key, range, limit), targetType);
}
/**
@@ -409,7 +398,7 @@ public interface StreamOperations<K, HK, HV> extends HashMapperProvider<HK, HV>
Assert.notNull(targetType, "Target type must not be null");
return StreamObjectMapper.map(read(readOptions, streams), this, targetType);
return map(read(readOptions, streams), targetType);
}
/**
@@ -467,7 +456,7 @@ public interface StreamOperations<K, HK, HV> extends HashMapperProvider<HK, HV>
Assert.notNull(targetType, "Target type must not be null");
return StreamObjectMapper.map(read(consumer, readOptions, streams), this, targetType);
return map(read(consumer, readOptions, streams), targetType);
}
/**
@@ -523,7 +512,7 @@ public interface StreamOperations<K, HK, HV> extends HashMapperProvider<HK, HV>
Assert.notNull(targetType, "Target type must not be null");
return StreamObjectMapper.map(reverseRange(key, range, limit), this, targetType);
return map(reverseRange(key, range, limit), targetType);
}
/**
@@ -560,4 +549,45 @@ public interface StreamOperations<K, HK, HV> extends HashMapperProvider<HK, HV>
@Override
<V> HashMapper<V, HK, HV> getHashMapper(Class<V> targetType);
/**
* Map record from {@link MapRecord} to {@link ObjectRecord}.
*
* @param record the stream record to map.
* @param targetType the target type of the payload.
* @return the mapped {@link ObjectRecord}.
* @since 2.x
*/
default <V> ObjectRecord<K, V> map(MapRecord<K, HK, HV> record, Class<V> targetType) {
Assert.notNull(record, "Record must not be null");
Assert.notNull(targetType, "Target type must not be null");
return StreamObjectMapper.toObjectRecord(record, this, targetType);
}
/**
* Map records from {@link MapRecord} to {@link ObjectRecord}s.
*
* @param records the stream records to map.
* @param targetType the target type of the payload.
* @return the mapped {@link ObjectRecord object records}.
* @since 2.x
*/
@Nullable
default <V> List<ObjectRecord<K, V>> map(@Nullable List<MapRecord<K, HK, HV>> records, Class<V> targetType) {
Assert.notNull(records, "Records must not be null");
Assert.notNull(targetType, "Target type must not be null");
return StreamObjectMapper.toObjectRecords(records, this, targetType);
}
/**
* Deserialize a {@link ByteRecord} using the configured serializers into a {@link MapRecord}.
*
* @param record the stream record to map.
* @return deserialized {@link MapRecord}.
* @since 2.x
*/
MapRecord<K, HK, HV> deserializeRecord(ByteRecord record);
}

59
src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java
View File

@@ -19,19 +19,25 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.function.BiFunction;
import java.util.function.Function;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.ByteRecord;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import org.springframework.util.ObjectUtils;
@@ -79,8 +85,8 @@ class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implement
this.template = createRedisTemplate(connectionFactory, containerOptions);
this.containerOptions = containerOptions;
if (containerOptions.getHashMapper() != null) {
this.streamOperations = this.template.opsForStream(containerOptions.getHashMapper());
if (containerOptions.hasHashMapper()) {
this.streamOperations = this.template.opsForStream(containerOptions.getRequiredHashMapper());
} else {
this.streamOperations = this.template.opsForStream();
}
@@ -207,16 +213,39 @@ class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implement
return doRegister(getReadTask(streamRequest, listener));
}
@SuppressWarnings("unchecked")
@SuppressWarnings({ "unchecked", "rawtypes" })
private StreamPollTask<K, V> getReadTask(StreamReadRequest<K> streamRequest, StreamListener<K, V> listener) {
BiFunction<K, ReadOffset, List<? extends Record<?, ?>>> readFunction = getReadFunction(streamRequest);
Function<ReadOffset, List<ByteRecord>> readFunction = getReadFunction(streamRequest);
Function<ByteRecord, V> deserializerToUse = getDeserializer();
return new StreamPollTask<>(streamRequest, listener, errorHandler, (BiFunction) readFunction);
TypeDescriptor targetType = TypeDescriptor
.valueOf(containerOptions.hasHashMapper() ? containerOptions.getTargetType() : MapRecord.class);
return new StreamPollTask<>(streamRequest, listener, errorHandler, targetType, readFunction, deserializerToUse);
}
@SuppressWarnings({ "unchecked", "rawtypes" })
private Function<ByteRecord, V> getDeserializer() {
Function<ByteRecord, MapRecord<K, Object, Object>> deserializer = streamOperations::deserializeRecord;
if (containerOptions.getHashMapper() == null) {
return (Function) deserializer;
}
return source -> {
MapRecord<K, Object, Object> intermediate = deserializer.apply(source);
return (V) streamOperations.map(intermediate, this.containerOptions.getTargetType());
};
}
@SuppressWarnings("unchecked")
private BiFunction<K, ReadOffset, List<? extends Record<?, ?>>> getReadFunction(StreamReadRequest<K> streamRequest) {
private Function<ReadOffset, List<ByteRecord>> getReadFunction(StreamReadRequest<K> streamRequest) {
byte[] rawKey = ((RedisSerializer<K>) template.getKeySerializer())
.serialize(streamRequest.getStreamOffset().getKey());
if (streamRequest instanceof StreamMessageListenerContainer.ConsumerStreamReadRequest) {
@@ -226,20 +255,12 @@ class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implement
: this.readOptions;
Consumer consumer = consumerStreamRequest.getConsumer();
if (this.containerOptions.getHashMapper() != null) {
return (key, offset) -> streamOperations.read(this.containerOptions.getTargetType(), consumer, readOptions,
StreamOffset.create(key, offset));
}
return (key, offset) -> streamOperations.read(consumer, readOptions, StreamOffset.create(key, offset));
return (offset) -> template.execute((RedisCallback<List<ByteRecord>>) connection -> connection.streamCommands()
.xReadGroup(consumer, readOptions, StreamOffset.create(rawKey, offset)));
}
if (this.containerOptions.getHashMapper() != null) {
return (key, offset) -> streamOperations.read(this.containerOptions.getTargetType(), readOptions,
StreamOffset.create(key, offset));
}
return (key, offset) -> streamOperations.read(readOptions, StreamOffset.create(key, offset));
return (offset) -> template.execute((RedisCallback<List<ByteRecord>>) connection -> connection.streamCommands()
.xRead(readOptions, StreamOffset.create(rawKey, offset)));
}
private Subscription doRegister(Task task) {

104
src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java
View File

@@ -22,18 +22,25 @@ import reactor.core.publisher.Operators;
import reactor.util.concurrent.Queues;
import reactor.util.context.Context;
import java.nio.ByteBuffer;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.function.Function;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import org.springframework.core.convert.ConversionFailedException;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.stream.ByteBufferRecord;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.StreamOffset;
@@ -66,6 +73,7 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
*/
@SuppressWarnings("unchecked")
DefaultStreamReceiver(ReactiveRedisConnectionFactory connectionFactory, StreamReceiverOptions<K, V> options) {
receiverOptions = options;
RedisSerializationContext<K, Object> serializationContext = RedisSerializationContext
@@ -85,8 +93,8 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
this.readOptions = readOptions;
this.template = new ReactiveRedisTemplate(connectionFactory, serializationContext);
if (options.getHashMapper() != null) {
this.streamOperations = this.template.opsForStream(options.getHashMapper());
if (options.hasHashMapper()) {
this.streamOperations = this.template.opsForStream(options.getRequiredHashMapper());
} else {
this.streamOperations = this.template.opsForStream();
}
@@ -104,20 +112,19 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
logger.debug(String.format("receive(%s)", streamOffset));
}
BiFunction<K, ReadOffset, Flux<? extends Record<?, ?>>> readFunction;
RedisSerializationContext.SerializationPair<K> keySerializer = template.getSerializationContext()
.getKeySerializationPair();
ByteBuffer rawKey = keySerializer.write(streamOffset.getKey());
if (receiverOptions.getHashMapper() != null) {
readFunction = (key, readOffset) -> streamOperations.read(receiverOptions.getTargetType(), readOptions,
StreamOffset.create(key, readOffset));
} else {
readFunction = (key, readOffset) -> streamOperations.read(readOptions, StreamOffset.create(key, readOffset));
}
Function<ReadOffset, Flux<ByteBufferRecord>> readFunction = readOffset -> template.execute(connection -> connection
.streamCommands().xRead(readOptions, StreamOffset.create(rawKey.asReadOnlyBuffer(), readOffset)));
return Flux.defer(() -> {
PollState pollState = PollState.standalone(streamOffset.getOffset());
return Flux.create(
sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, (BiFunction) readFunction).arm());
sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, readFunction,
receiverOptions.getResumeFunction()).arm());
});
}
@@ -133,14 +140,16 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
logger.debug(String.format("receiveAutoAck(%s, %s)", consumer, streamOffset));
}
BiFunction<K, ReadOffset, Flux<? extends Record<?, ?>>> readFunction = getConsumeReadFunction(consumer,
Function<ReadOffset, Flux<ByteBufferRecord>> readFunction = getConsumeReadFunction(streamOffset.getKey(), consumer,
this.readOptions.autoAcknowledge());
return Flux.defer(() -> {
PollState pollState = PollState.consumer(consumer, streamOffset.getOffset());
return Flux.create(
sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, (BiFunction) readFunction).arm());
sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, readFunction,
receiverOptions.getResumeFunction()).arm());
});
}
@@ -156,26 +165,43 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
logger.debug(String.format("receive(%s, %s)", consumer, streamOffset));
}
BiFunction<K, ReadOffset, Flux<? extends Record<?, ?>>> readFunction = getConsumeReadFunction(consumer,
Function<ReadOffset, Flux<ByteBufferRecord>> readFunction = getConsumeReadFunction(streamOffset.getKey(), consumer,
this.readOptions);
return Flux.defer(() -> {
PollState pollState = PollState.consumer(consumer, streamOffset.getOffset());
return Flux.create(
sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, (BiFunction) readFunction).arm());
sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, readFunction,
receiverOptions.getResumeFunction()).arm());
});
}
@SuppressWarnings("unchecked")
private BiFunction<K, ReadOffset, Flux<? extends Record<?, ?>>> getConsumeReadFunction(Consumer consumer,
private Function<ReadOffset, Flux<ByteBufferRecord>> getConsumeReadFunction(K key, Consumer consumer,
StreamReadOptions readOptions) {
if (receiverOptions.getHashMapper() != null) {
return (key, readOffset) -> streamOperations.read(receiverOptions.getTargetType(), consumer, readOptions,
StreamOffset.create(key, readOffset));
RedisSerializationContext.SerializationPair<K> keySerializer = template.getSerializationContext()
.getKeySerializationPair();
ByteBuffer rawKey = keySerializer.write(key);
return readOffset -> template.execute(connection -> connection.streamCommands().xReadGroup(consumer, readOptions,
StreamOffset.create(rawKey.asReadOnlyBuffer(), readOffset)));
}
@SuppressWarnings({ "unchecked", "rawtypes" })
private Function<ByteBufferRecord, V> getDeserializer() {
Function<ByteBufferRecord, MapRecord<K, Object, Object>> deserializer = streamOperations::deserializeRecord;
if (receiverOptions.getHashMapper() == null) {
return (Function) deserializer;
}
return (key, readOffset) -> streamOperations.read(consumer, readOptions, StreamOffset.create(key, readOffset));
return source -> {
MapRecord<K, Object, Object> intermediate = deserializer.apply(source);
return (V) streamOperations.map(intermediate, this.receiverOptions.getTargetType());
};
}
/**
@@ -188,15 +214,23 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
private final FluxSink<V> sink;
private final K key;
private final PollState pollState;
private final BiFunction<K, ReadOffset, Flux<V>> readFunction;
private final Function<ReadOffset, Flux<ByteBufferRecord>> readFunction;
private final Function<? super Throwable, ? extends Publisher<Void>> resumeFunction;
private final Function<ByteBufferRecord, V> deserializer;
private final TypeDescriptor targetType;
protected StreamSubscription(FluxSink<V> sink, K key, PollState pollState,
BiFunction<K, ReadOffset, Flux<V>> readFunction) {
Function<ReadOffset, Flux<ByteBufferRecord>> readFunction,
Function<? super Throwable, ? extends Publisher<Void>> resumeFunction) {
this.sink = sink;
this.key = key;
this.pollState = pollState;
this.readFunction = readFunction;
this.resumeFunction = resumeFunction;
this.deserializer = getDeserializer();
this.targetType = TypeDescriptor
.valueOf(receiverOptions.hasHashMapper() ? receiverOptions.getTargetType() : MapRecord.class);
}
/**
@@ -237,6 +271,7 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
sink.onCancel(pollState::cancel);
}
@SuppressWarnings({ "unchecked", "ConstantConditions" })
private void scheduleIfRequired() {
if (logger.isDebugEnabled()) {
@@ -290,9 +325,24 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
String.format("[stream: %s] scheduleIfRequired(): Activating subscription, offset %s", key, readOffset));
}
Flux<V> poll = readFunction.apply(key, readOffset);
Flux<ByteBufferRecord> poll = readFunction.apply(readOffset)
.onErrorResume(throwable -> Flux.from(resumeFunction.apply(throwable)).then().cast(ByteBufferRecord.class));
poll.subscribe(getSubscriber());
poll.map(it -> {
if (logger.isDebugEnabled()) {
logger.debug(String.format("[stream: %s] onStreamMessage(%s)", key, it));
}
pollState.updateReadOffset(it.getId().getValue());
try {
return deserializer.apply(it);
} catch (RuntimeException e) {
throw new ConversionFailedException(TypeDescriptor.forObject(it), targetType, it, e);
}
}).onErrorResume(throwable -> Flux.from(resumeFunction.apply(throwable)).then().map(it -> (V) new Object())) //
.subscribe(getSubscriber());
}
}
@@ -336,12 +386,6 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
private void onStreamMessage(V message) {
if (logger.isDebugEnabled()) {
logger.debug(String.format("[stream: %s] onStreamMessage(%s)", key, message));
}
pollState.updateReadOffset(message.getId().getValue());
long requested = pollState.getRequested();
if (requested > 0) {

21
src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java
View File

@@ -417,7 +417,7 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
}
/**
* Configure a {@link ErrorHandler} to be notified on {@link Throwable errors}.
* Configure a {@link ErrorHandler} to be notified on {@link Throwable read, deserialization, and listener errors}.
*
* @param errorHandler must not be null.
* @return {@code this} {@link ConsumerStreamReadRequestBuilder}.
@@ -429,8 +429,9 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
}
/**
* Configure a cancellation {@link Predicate} to be notified on {@link Throwable errors}. The outcome of the
* {@link Predicate} decides whether to cancel the subscription by returning {@literal true}.
* Configure a cancellation {@link Predicate} to be notified on {@link Throwable read, deserialization, and listener
* errors}. The outcome of the {@link Predicate} decides whether to cancel the subscription by returning
* {@literal true}.
*
* @param cancelSubscriptionOnError must not be null.
* @return {@code this} {@link ConsumerStreamReadRequestBuilder}.
@@ -568,6 +569,19 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
return hashMapper;
}
public HashMapper<Object, Object, Object> getRequiredHashMapper() {
if (!hasHashMapper()) {
throw new IllegalStateException("No HashMapper configured");
}
return hashMapper;
}
public boolean hasHashMapper() {
return hashMapper != null;
}
public Class<Object> getTargetType() {
if (this.targetType != null) {
@@ -590,6 +604,7 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
public Executor getExecutor() {
return executor;
}
}
/**

66
src/main/java/org/springframework/data/redis/stream/StreamPollTask.java
View File

@@ -21,9 +21,13 @@ import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Predicate;
import org.springframework.core.convert.ConversionFailedException;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.data.redis.connection.stream.ByteRecord;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.Record;
@@ -32,6 +36,7 @@ import org.springframework.data.redis.stream.StreamMessageListenerContainer.Cons
import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamReadRequest;
import org.springframework.util.ErrorHandler;
/**
* {@link Task} that invokes a {@link BiFunction read function} to poll on a Redis Stream.
*
@@ -40,24 +45,28 @@ import org.springframework.util.ErrorHandler;
*/
class StreamPollTask<K, V extends Record<K, ?>> implements Task {
private final StreamReadRequest<K> request;
private final StreamListener<K, V> listener;
private final ErrorHandler errorHandler;
private final Predicate<Throwable> cancelSubscriptionOnError;
private final BiFunction<K, ReadOffset, List<V>> readFunction;
private final Function<ReadOffset, List<ByteRecord>> readFunction;
private final Function<ByteRecord, V> deserializer;
private final PollState pollState;
private final TypeDescriptor targetType;
private volatile boolean isInEventLoop = false;
StreamPollTask(StreamReadRequest<K> streamRequest, StreamListener<K, V> listener, ErrorHandler errorHandler,
BiFunction<K, ReadOffset, List<V>> readFunction) {
TypeDescriptor targetType, Function<ReadOffset, List<ByteRecord>> readFunction,
Function<ByteRecord, V> deserializer) {
this.request = streamRequest;
this.listener = listener;
this.errorHandler = Optional.ofNullable(streamRequest.getErrorHandler()).orElse(errorHandler);
this.cancelSubscriptionOnError = streamRequest.getCancelSubscriptionOnError();
this.readFunction = readFunction;
this.deserializer = deserializer;
this.pollState = createPollState(streamRequest);
this.targetType = targetType;
}
private static PollState createPollState(StreamReadRequest<?> streamRequest) {
@@ -120,13 +129,13 @@ class StreamPollTask<K, V extends Record<K, ?>> implements Task {
isInEventLoop = true;
pollState.running();
doLoop(request.getStreamOffset().getKey());
doLoop();
} finally {
isInEventLoop = false;
}
}
private void doLoop(K key) {
private void doLoop() {
do {
@@ -135,13 +144,9 @@ class StreamPollTask<K, V extends Record<K, ?>> implements Task {
// allow interruption
Thread.sleep(0);
List<V> read = readFunction.apply(key, pollState.getCurrentReadOffset());
List<ByteRecord> raw = readRecords();
deserializeAndEmitRecords(raw);
for (V message : read) {
listener.onMessage(message);
pollState.updateReadOffset(message.getId().getValue());
}
} catch (InterruptedException e) {
cancel();
@@ -157,6 +162,43 @@ class StreamPollTask<K, V extends Record<K, ?>> implements Task {
} while (pollState.isSubscriptionActive());
}
private List<ByteRecord> readRecords() {
return readFunction.apply(pollState.getCurrentReadOffset());
}
private void deserializeAndEmitRecords(List<ByteRecord> records) {
for (ByteRecord raw : records) {
try {
pollState.updateReadOffset(raw.getId().getValue());
V record = convertRecord(raw);
listener.onMessage(record);
} catch (RuntimeException e) {
if (cancelSubscriptionOnError.test(e)) {
cancel();
errorHandler.handleError(e);
return;
}
errorHandler.handleError(e);
}
}
}
private V convertRecord(ByteRecord record) {
try {
return deserializer.apply(record);
} catch (RuntimeException e) {
throw new ConversionFailedException(TypeDescriptor.forObject(record), targetType, record, e);
}
}
@Override
public boolean isActive() {
return State.RUNNING.equals(getState()) || isInEventLoop;

58
src/main/java/org/springframework/data/redis/stream/StreamReceiver.java
View File

@@ -16,10 +16,14 @@
package org.springframework.data.redis.stream;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.OptionalInt;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
@@ -74,8 +78,12 @@ import org.springframework.util.Assert;
* ({@code $}) for subsequent reads.</li>
* </ul>
* <strong>Note: Using {@link ReadOffset#latest()} bears the chance of dropped records as records can arrive in the time
* during polling is suspended. Use recorddId's as offset or {@link ReadOffset#lastConsumed()} to minimize the chance of
* during polling is suspended. Use recordId's as offset or {@link ReadOffset#lastConsumed()} to minimize the chance of
* record loss.</strong>
* <p>
* {@link StreamReceiver} propagates errors during stream reads and deserialization as terminal error signal by default.
* Configuring a {@link StreamReceiverOptions#getResumeFunction() resume function} allows conditional resumption by
* dropping the record or by propagating the error to terminate the subscription.
* <p/>
* See the following example code how to use {@link StreamReceiver}:
*
@@ -189,6 +197,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
private final Duration pollTimeout;
private final @Nullable Integer batchSize;
private final Function<? super Throwable, ? extends Publisher<Void>> resumeFunction;
private final SerializationPair<K> keySerializer;
private final SerializationPair<Object> hashKeySerializer;
private final SerializationPair<Object> hashValueSerializer;
@@ -196,12 +205,14 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
private final @Nullable HashMapper<Object, Object, Object> hashMapper;
@SuppressWarnings({ "unchecked", "rawtypes" })
private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize, SerializationPair<K> keySerializer,
private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize,
Function<? super Throwable, ? extends Publisher<Void>> resumeFunction, SerializationPair<K> keySerializer,
SerializationPair<Object> hashKeySerializer, SerializationPair<Object> hashValueSerializer,
@Nullable Class<?> targetType, @Nullable HashMapper<V, ?, ?> hashMapper) {
this.pollTimeout = pollTimeout;
this.batchSize = batchSize;
this.resumeFunction = resumeFunction;
this.keySerializer = keySerializer;
this.hashKeySerializer = hashKeySerializer;
this.hashValueSerializer = hashValueSerializer;
@@ -249,6 +260,10 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
return batchSize != null ? OptionalInt.of(batchSize) : OptionalInt.empty();
}
public Function<? super Throwable, ? extends Publisher<Void>> getResumeFunction() {
return resumeFunction;
}
public SerializationPair<K> getKeySerializer() {
return keySerializer;
}
@@ -266,6 +281,19 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
return hashMapper;
}
public HashMapper<Object, Object, Object> getRequiredHashMapper() {
if (!hasHashMapper()) {
throw new IllegalStateException("No HashMapper configured");
}
return hashMapper;
}
public boolean hasHashMapper() {
return this.hashMapper != null;
}
public Class<Object> getTargetType() {
if (this.targetType != null) {
@@ -274,6 +302,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
return Object.class;
}
}
/**
@@ -288,6 +317,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
private SerializationPair<K> keySerializer;
private SerializationPair<Object> hashKeySerializer;
private SerializationPair<Object> hashValueSerializer;
private Function<? super Throwable, ? extends Publisher<Void>> resumeFunction = Mono::error;
private @Nullable HashMapper<V, ?, ?> hashMapper;
private @Nullable Class<?> targetType;
@@ -311,7 +341,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
/**
* Configure a batch size for the {@code COUNT} option during reading.
*
* @param recordsPerPoll must not be greater zero.
* @param recordsPerPoll must be greater zero.
* @return {@code this} {@link StreamReceiverOptionsBuilder}.
*/
public StreamReceiverOptionsBuilder<K, V> batchSize(int recordsPerPoll) {
@@ -322,6 +352,25 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
return this;
}
/**
* Configure a resume {@link Function} to resume the main sequence when polling the stream fails. The function can
* either resume by suppressing the error or fail the main sequence by emitting the error to stop receiving. Receive
* errors (Redis errors, Serialization failures) stop receiving by default.
*
* @param resumeFunction must not be {@literal null}.
* @return {@code this} {@link StreamReceiverOptionsBuilder}.
* @since 2.x
* @see Flux#onErrorResume(Function)
*/
public StreamReceiverOptionsBuilder<K, V> onErrorResume(
Function<? super Throwable, ? extends Publisher<Void>> resumeFunction) {
Assert.notNull(resumeFunction, "Resume function must not be null");
this.resumeFunction = resumeFunction;
return this;
}
/**
* Configure a key, hash key and hash value serializer.
*
@@ -450,7 +499,8 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
* @return new {@link StreamReceiverOptions}.
*/
public StreamReceiverOptions<K, V> build() {
return new StreamReceiverOptions<>(pollTimeout, batchSize, keySerializer, hashKeySerializer, hashValueSerializer,
return new StreamReceiverOptions<>(pollTimeout, batchSize, resumeFunction, keySerializer, hashKeySerializer,
hashValueSerializer,
targetType, hashMapper);
}
}