diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java index 632c72299..b7e907ea6 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java @@ -416,7 +416,8 @@ class DefaultReactiveStreamOperations implements ReactiveStreamOperat return (HV) serializationContext.getHashValueSerializationPair().read(buffer); } - private MapRecord deserializeRecord(ByteBufferRecord record) { + @Override + public MapRecord deserializeRecord(ByteBufferRecord record) { return record.map(it -> it.mapEntries(this::deserializeRecordFields).withStreamKey(readKey(record.getStream()))); } diff --git a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java index e59cf85c7..4db28ee49 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java @@ -347,6 +347,11 @@ class DefaultStreamOperations extends AbstractOperations i return objectMapper.getHashMapper(targetType); } + @Override + public MapRecord 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 extends AbstractOperations i List> result = new ArrayList<>(); for (ByteRecord record : raw) { - result.add(record.deserialize(keySerializer(), hashKeySerializer(), hashValueSerializer())); + result.add(deserializeRecord(record)); } return result; diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java index 0892fa1e1..e78437c0a 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java @@ -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 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 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 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 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 extends HashMapperProvider< */ @Override HashMapper getHashMapper(Class 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 ObjectRecord map(MapRecord record, Class 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 deserializeRecord(ByteBufferRecord record); } diff --git a/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java b/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java index 87604f2cc..7c7067492 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java +++ b/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java @@ -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 ObjectRecord toObjectRecord(HashMapperProvider provider, - MapRecord source, Class targetType) { + static ObjectRecord toObjectRecord(MapRecord source, + HashMapperProvider provider, Class targetType) { return source.toObjectRecord(provider.getHashMapper(targetType)); } @@ -149,7 +149,7 @@ class StreamObjectMapper { * {@literal null}. */ @Nullable - static List> map(@Nullable List> records, + static List> toObjectRecords(@Nullable List> records, HashMapperProvider hashMapperProvider, Class 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> transformed = new ArrayList<>(records.size()); diff --git a/src/main/java/org/springframework/data/redis/core/StreamOperations.java b/src/main/java/org/springframework/data/redis/core/StreamOperations.java index 41537daca..776a9d863 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java @@ -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 extends HashMapperProvider 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 extends HashMapperProvider 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 extends HashMapperProvider 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 extends HashMapperProvider 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 extends HashMapperProvider @Override HashMapper getHashMapper(Class 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 ObjectRecord map(MapRecord record, Class 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 List> map(@Nullable List> records, Class 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 deserializeRecord(ByteRecord record); } diff --git a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java index fce031f4b..b5b372909 100644 --- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java @@ -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> 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> implement return doRegister(getReadTask(streamRequest, listener)); } - @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "rawtypes" }) private StreamPollTask getReadTask(StreamReadRequest streamRequest, StreamListener listener) { - BiFunction>> readFunction = getReadFunction(streamRequest); + Function> readFunction = getReadFunction(streamRequest); + Function 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 getDeserializer() { + + Function> deserializer = streamOperations::deserializeRecord; + + if (containerOptions.getHashMapper() == null) { + return (Function) deserializer; + } + + return source -> { + + MapRecord intermediate = deserializer.apply(source); + return (V) streamOperations.map(intermediate, this.containerOptions.getTargetType()); + }; } @SuppressWarnings("unchecked") - private BiFunction>> getReadFunction(StreamReadRequest streamRequest) { + private Function> getReadFunction(StreamReadRequest streamRequest) { + + byte[] rawKey = ((RedisSerializer) template.getKeySerializer()) + .serialize(streamRequest.getStreamOffset().getKey()); if (streamRequest instanceof StreamMessageListenerContainer.ConsumerStreamReadRequest) { @@ -226,20 +255,12 @@ class DefaultStreamMessageListenerContainer> 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>) 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>) connection -> connection.streamCommands() + .xRead(readOptions, StreamOffset.create(rawKey, offset))); } private Subscription doRegister(Task task) { diff --git a/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java b/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java index 7d2c17159..0eff3f021 100644 --- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java +++ b/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java @@ -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> implements StreamReceiver */ @SuppressWarnings("unchecked") DefaultStreamReceiver(ReactiveRedisConnectionFactory connectionFactory, StreamReceiverOptions options) { + receiverOptions = options; RedisSerializationContext serializationContext = RedisSerializationContext @@ -85,8 +93,8 @@ class DefaultStreamReceiver> 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> implements StreamReceiver logger.debug(String.format("receive(%s)", streamOffset)); } - BiFunction>> readFunction; + RedisSerializationContext.SerializationPair 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> 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> implements StreamReceiver logger.debug(String.format("receiveAutoAck(%s, %s)", consumer, streamOffset)); } - BiFunction>> readFunction = getConsumeReadFunction(consumer, + Function> 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> implements StreamReceiver logger.debug(String.format("receive(%s, %s)", consumer, streamOffset)); } - BiFunction>> readFunction = getConsumeReadFunction(consumer, + Function> 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>> getConsumeReadFunction(Consumer consumer, + private Function> 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 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 getDeserializer() { + + Function> deserializer = streamOperations::deserializeRecord; + + if (receiverOptions.getHashMapper() == null) { + return (Function) deserializer; } - return (key, readOffset) -> streamOperations.read(consumer, readOptions, StreamOffset.create(key, readOffset)); + return source -> { + + MapRecord intermediate = deserializer.apply(source); + return (V) streamOperations.map(intermediate, this.receiverOptions.getTargetType()); + }; } /** @@ -188,15 +214,23 @@ class DefaultStreamReceiver> implements StreamReceiver private final FluxSink sink; private final K key; private final PollState pollState; - private final BiFunction> readFunction; + private final Function> readFunction; + private final Function> resumeFunction; + private final Function deserializer; + private final TypeDescriptor targetType; protected StreamSubscription(FluxSink sink, K key, PollState pollState, - BiFunction> readFunction) { + Function> readFunction, + Function> 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> implements StreamReceiver sink.onCancel(pollState::cancel); } + @SuppressWarnings({ "unchecked", "ConstantConditions" }) private void scheduleIfRequired() { if (logger.isDebugEnabled()) { @@ -290,9 +325,24 @@ class DefaultStreamReceiver> implements StreamReceiver String.format("[stream: %s] scheduleIfRequired(): Activating subscription, offset %s", key, readOffset)); } - Flux poll = readFunction.apply(key, readOffset); + Flux 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> 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) { diff --git a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java index 4c17aad5d..75550d997 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java @@ -417,7 +417,7 @@ public interface StreamMessageListenerContainer> 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> 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> exten return hashMapper; } + public HashMapper getRequiredHashMapper() { + + if (!hasHashMapper()) { + throw new IllegalStateException("No HashMapper configured"); + } + + return hashMapper; + } + + public boolean hasHashMapper() { + return hashMapper != null; + } + public Class getTargetType() { if (this.targetType != null) { @@ -590,6 +604,7 @@ public interface StreamMessageListenerContainer> exten public Executor getExecutor() { return executor; } + } /** diff --git a/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java b/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java index e4ed51ad9..d78a6ab5b 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java @@ -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> implements Task { - private final StreamReadRequest request; private final StreamListener listener; private final ErrorHandler errorHandler; private final Predicate cancelSubscriptionOnError; - private final BiFunction> readFunction; + private final Function> readFunction; + private final Function deserializer; private final PollState pollState; + private final TypeDescriptor targetType; + private volatile boolean isInEventLoop = false; StreamPollTask(StreamReadRequest streamRequest, StreamListener listener, ErrorHandler errorHandler, - BiFunction> readFunction) { + TypeDescriptor targetType, Function> readFunction, + Function 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> 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> implements Task { // allow interruption Thread.sleep(0); - List read = readFunction.apply(key, pollState.getCurrentReadOffset()); + List 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> implements Task { } while (pollState.isSubscriptionActive()); } + private List readRecords() { + return readFunction.apply(pollState.getCurrentReadOffset()); + } + + private void deserializeAndEmitRecords(List 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; diff --git a/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java b/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java index 771d86237..2e5f4e28c 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java @@ -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. * * 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. + *

+ * {@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. *

* See the following example code how to use {@link StreamReceiver}: * @@ -189,6 +197,7 @@ public interface StreamReceiver> { private final Duration pollTimeout; private final @Nullable Integer batchSize; + private final Function> resumeFunction; private final SerializationPair keySerializer; private final SerializationPair hashKeySerializer; private final SerializationPair hashValueSerializer; @@ -196,12 +205,14 @@ public interface StreamReceiver> { private final @Nullable HashMapper hashMapper; @SuppressWarnings({ "unchecked", "rawtypes" }) - private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize, SerializationPair keySerializer, + private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize, + Function> resumeFunction, SerializationPair keySerializer, SerializationPair hashKeySerializer, SerializationPair hashValueSerializer, @Nullable Class targetType, @Nullable HashMapper 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> { return batchSize != null ? OptionalInt.of(batchSize) : OptionalInt.empty(); } + public Function> getResumeFunction() { + return resumeFunction; + } + public SerializationPair getKeySerializer() { return keySerializer; } @@ -266,6 +281,19 @@ public interface StreamReceiver> { return hashMapper; } + public HashMapper getRequiredHashMapper() { + + if (!hasHashMapper()) { + throw new IllegalStateException("No HashMapper configured"); + } + + return hashMapper; + } + + public boolean hasHashMapper() { + return this.hashMapper != null; + } + public Class getTargetType() { if (this.targetType != null) { @@ -274,6 +302,7 @@ public interface StreamReceiver> { return Object.class; } + } /** @@ -288,6 +317,7 @@ public interface StreamReceiver> { private SerializationPair keySerializer; private SerializationPair hashKeySerializer; private SerializationPair hashValueSerializer; + private Function> resumeFunction = Mono::error; private @Nullable HashMapper hashMapper; private @Nullable Class targetType; @@ -311,7 +341,7 @@ public interface StreamReceiver> { /** * 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 batchSize(int recordsPerPoll) { @@ -322,6 +352,25 @@ public interface StreamReceiver> { 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 onErrorResume( + Function> 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> { * @return new {@link StreamReceiverOptions}. */ public StreamReceiverOptions build() { - return new StreamReceiverOptions<>(pollTimeout, batchSize, keySerializer, hashKeySerializer, hashValueSerializer, + return new StreamReceiverOptions<>(pollTimeout, batchSize, resumeFunction, keySerializer, hashKeySerializer, + hashValueSerializer, targetType, hashMapper); } } diff --git a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java index 46dc908c4..5db2b6a57 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java @@ -35,12 +35,14 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.core.convert.ConversionFailedException; import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.RedisStandaloneConfiguration; import org.springframework.data.redis.connection.lettuce.LettuceConnection; import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension; +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.ObjectRecord; @@ -276,7 +278,7 @@ public class StreamMessageListenerContainerIntegrationTests { StreamReadRequest readRequest = StreamReadRequest .builder(StreamOffset.create("my-stream", ReadOffset.lastConsumed())) // - .errorHandler(failures::add) // // + .errorHandler(failures::add) // .cancelOnError(t -> false) // .consumer(Consumer.from("my-group", "my-consumer")) // .build(); @@ -298,6 +300,52 @@ public class StreamMessageListenerContainerIntegrationTests { cancelAwait(subscription); } + @Test // DATAREDIS-1230 + void deserializationShouldContinueStreamRead() throws InterruptedException { + + StreamMessageListenerContainerOptions> containerOptions = StreamMessageListenerContainerOptions + .builder().batchSize(1).pollTimeout(Duration.ofMillis(100)).targetType(Long.class).build(); + + BlockingQueue> records = new LinkedBlockingQueue<>(); + BlockingQueue failures = new LinkedBlockingQueue<>(); + + StreamMessageListenerContainer> container = StreamMessageListenerContainer + .create(connectionFactory, containerOptions); + + StreamReadRequest readRequest = StreamReadRequest + .builder(StreamOffset.create("my-stream", ReadOffset.from("0-0"))) // + .errorHandler(failures::add) // + .cancelOnError(t -> false) // + .build(); + + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "1")); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "foo")); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "3")); + + container.start(); + Subscription subscription = container.register(readRequest, records::add); + + subscription.await(DEFAULT_TIMEOUT); + + ObjectRecord first = records.poll(1, TimeUnit.SECONDS); + Throwable conversionFailure = failures.poll(1, TimeUnit.SECONDS); + ObjectRecord third = records.poll(1, TimeUnit.SECONDS); + + assertThat(first).isNotNull(); + assertThat(first.getValue()).isEqualTo(1L); + + assertThat(conversionFailure).isInstanceOf(ConversionFailedException.class) + .hasCauseInstanceOf(ConversionFailedException.class).hasRootCauseInstanceOf(NumberFormatException.class); + assertThat(((ConversionFailedException) conversionFailure).getValue()).isInstanceOf(ByteRecord.class); + + assertThat(third).isNotNull(); + assertThat(third.getValue()).isEqualTo(3L); + + assertThat(subscription.isActive()).isTrue(); + + cancelAwait(subscription); + } + @Test // DATAREDIS-864 void cancelledStreamShouldNotReceiveMessages() throws InterruptedException { diff --git a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java index 3fc83413f..60ed981df 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java @@ -22,24 +22,29 @@ import static org.mockito.Mockito.*; import lombok.AllArgsConstructor; import lombok.Data; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import java.nio.ByteBuffer; import java.time.Duration; import java.util.Collections; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.core.convert.ConversionFailedException; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension; +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.ObjectRecord; 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.core.ReactiveRedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; @@ -254,6 +259,37 @@ public class StreamReceiverIntegrationTests { .verify(Duration.ofSeconds(5)); } + @Test // DATAREDIS-864 + void shouldResumeFromError() { + + AtomicReference ref = new AtomicReference<>(); + StreamReceiverOptions> options = StreamReceiverOptions.builder() + .pollTimeout(Duration.ofMillis(100)).targetType(Long.class).onErrorResume(throwable -> { + + ref.set(throwable); + return Mono.empty(); + }).build(); + + StreamReceiver> receiver = StreamReceiver.create(connectionFactory, options); + + Flux> messages = receiver.receive(StreamOffset.fromStart("my-stream")); + + redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from("0-0"), "my-group"); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "1")); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "foo")); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "3")); + + messages.map(Record::getValue).as(StepVerifier::create) // + .expectNext(1L) // + .expectNext(3L) // + .thenCancel() // + .verify(); + + assertThat(ref.get()).isInstanceOf(ConversionFailedException.class) + .hasCauseInstanceOf(ConversionFailedException.class).hasRootCauseInstanceOf(NumberFormatException.class); + assertThat(((ConversionFailedException) ref.get()).getValue()).isInstanceOf(ByteBufferRecord.class); + } + @Data @AllArgsConstructor static class LoginEvent {