diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java index 880ee53fa4..a45968e10b 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,7 +17,11 @@ package org.springframework.integration.redis.inbound; import java.time.Duration; +import java.util.function.Function; +import org.reactivestreams.Publisher; + +import org.springframework.core.convert.ConversionFailedException; import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.ReadOffset; @@ -25,6 +29,7 @@ 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.ReactiveStreamOperations; +import org.springframework.data.redis.hash.HashMapper; import org.springframework.data.redis.serializer.RedisSerializationContext; import org.springframework.data.redis.stream.StreamReceiver; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -34,6 +39,7 @@ import org.springframework.integration.redis.support.RedisHeaders; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; import org.springframework.messaging.converter.MessageConversionException; import org.springframework.util.Assert; @@ -61,12 +67,14 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport { private final String streamKey; - private ReactiveStreamOperations reactiveStreamOperations; - - private StreamReceiver.StreamReceiverOptions streamReceiverOptions = + private final StreamReceiver.StreamReceiverOptionsBuilder streamReceiverOptionsBuilder = StreamReceiver.StreamReceiverOptions.builder() .pollTimeout(Duration.ZERO) - .build(); + .onErrorResume(this::handleReceiverError); + + private ReactiveStreamOperations reactiveStreamOperations; + + private StreamReceiver.StreamReceiverOptions streamReceiverOptions; private StreamReceiver streamReceiver; @@ -84,6 +92,8 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport { private boolean createConsumerGroup; + private boolean receiverBuilderOptionSet; + public ReactiveRedisStreamMessageProducer(ReactiveRedisConnectionFactory reactiveConnectionFactory, String streamKey) { @@ -152,14 +162,105 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport { * It provides a way to set the polling timeout and the serialization context. * By default the polling timeout is set to infinite and * {@link org.springframework.data.redis.serializer.StringRedisSerializer} is used. + * Mutually exclusive with 'pollTimeout', 'batchSize', 'onErrorResume', 'serializer', 'targetType', 'objectMapper'. * @param streamReceiverOptions the desired receiver options * */ public void setStreamReceiverOptions( @Nullable StreamReceiver.StreamReceiverOptions streamReceiverOptions) { + Assert.isTrue(!this.receiverBuilderOptionSet, + "The 'streamReceiverOptions' is mutually exclusive with 'pollTimeout', 'batchSize', " + + "'onErrorResume', 'serializer', 'targetType', 'objectMapper'"); this.streamReceiverOptions = streamReceiverOptions; } + private void assertStreamReceiverOptions(String property) { + Assert.isNull(this.streamReceiverOptions, + () -> "'" + property + "' cannot be set when 'StreamReceiver.StreamReceiverOptions' is provided."); + } + + /** + * Configure a poll timeout for the BLOCK option during reading. + * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}. + * @param pollTimeout the timeout for polling. + * @since 5.5 + * @see org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOptionsBuilder#pollTimeout(Duration) + */ + public void setPollTimeout(Duration pollTimeout) { + assertStreamReceiverOptions("pollTimeout"); + this.streamReceiverOptionsBuilder.pollTimeout(pollTimeout); + this.receiverBuilderOptionSet = true; + } + + /** + * Configure a batch size for the COUNT option during reading. + * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}. + * @param recordsPerPoll must be greater zero. + * @since 5.5 + * @see org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOptionsBuilder#batchSize(int) + */ + public void setBatchSize(int recordsPerPoll) { + assertStreamReceiverOptions("batchSize"); + this.streamReceiverOptionsBuilder.batchSize(recordsPerPoll); + this.receiverBuilderOptionSet = true; + } + + /** + * Configure a resume Function to resume the main sequence when polling the stream fails. + * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}. + * By default this function extract the failed {@link Record} and sends an + * {@link org.springframework.messaging.support.ErrorMessage} to the provided {@link #setErrorChannel(MessageChannel)}. + * The failed message for this record may have a {@link IntegrationMessageHeaderAccessor#ACKNOWLEDGMENT_CALLBACK} + * header when manual acknowledgment is configured for this message producer. + * @param resumeFunction must not be null. + * @since 5.5 + * @see org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOptionsBuilder#onErrorResume(Function) + */ + public void setOnErrorResume(Function> resumeFunction) { + assertStreamReceiverOptions("onErrorResume"); + this.streamReceiverOptionsBuilder.onErrorResume(resumeFunction); + this.receiverBuilderOptionSet = true; + } + + /** + * Configure a key, hash key and hash value serializer. + * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}. + * @param pair must not be null. + * @since 5.5 + * @see StreamReceiver.StreamReceiverOptionsBuilder#serializer(RedisSerializationContext) + */ + public void setSerializer(RedisSerializationContext.SerializationPair pair) { + assertStreamReceiverOptions("serializer"); + this.streamReceiverOptionsBuilder.serializer(pair); + this.receiverBuilderOptionSet = true; + } + + /** + * Configure a hash target type. Changes the emitted Record type to ObjectRecord. + * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}. + * @param targetType must not be null. + * @since 5.5 + * @see StreamReceiver.StreamReceiverOptionsBuilder#targetType(Class) + */ + public void setTargetType(Class targetType) { + assertStreamReceiverOptions("targetType"); + this.streamReceiverOptionsBuilder.targetType(targetType); + this.receiverBuilderOptionSet = true; + } + + /** + * Configure a hash mapper. + * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}. + * @param hashMapper must not be null. + * @since 5.5 + * @see StreamReceiver.StreamReceiverOptionsBuilder#objectMapper(HashMapper) + */ + public void setObjectMapper(HashMapper hashMapper) { + assertStreamReceiverOptions("objectMapper"); + this.streamReceiverOptionsBuilder.objectMapper(hashMapper); + this.receiverBuilderOptionSet = true; + } + @Override public String getComponentType() { return "redis:stream-inbound-channel-adapter"; @@ -168,6 +269,9 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport { @Override protected void onInit() { super.onInit(); + if (this.streamReceiverOptions == null) { + this.streamReceiverOptions = this.streamReceiverOptionsBuilder.build(); + } this.streamReceiver = StreamReceiver.create(this.reactiveConnectionFactory, this.streamReceiverOptions); if (StringUtils.hasText(this.consumerName) && !StringUtils.hasText(this.consumerGroup)) { this.consumerGroup = getBeanName(); @@ -211,17 +315,7 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport { } Flux> messageFlux = - events.map((record) -> buildMessageFromRecord(record, this.extractPayload)) - .onErrorContinue((ex, record) -> { - @SuppressWarnings("unchecked") - Message failedMessage = buildMessageFromRecord((Record) record, false); - MessagingException conversionException = - new MessageConversionException(failedMessage, - "Cannot deserialize Redis Stream Record", ex); - if (!sendErrorMessageIfNecessary(null, conversionException)) { - logger.getLog().error(conversionException); - } - }); + events.map((record) -> buildMessageFromRecord(record, this.extractPayload)); subscribeToPublisher(messageFlux); } @@ -245,4 +339,22 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport { return builder.build(); } + private Publisher handleReceiverError(Throwable error) { + Message failedMessage = null; + if (error instanceof ConversionFailedException) { + @SuppressWarnings("unchecked") + Record record = (Record) ((ConversionFailedException) error).getValue(); + if (record != null) { + failedMessage = buildMessageFromRecord(record, false); + } + } + MessagingException conversionException = + new MessageConversionException(failedMessage, // NOSONAR + "Cannot deserialize Redis Stream Record", error); + if (!sendErrorMessageIfNecessary(null, conversionException)) { + logger.getLog().error(conversionException); + } + return Mono.empty(); + } + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java index 0909eacb79..2e6fd0ee51 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -327,11 +327,8 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests ReactiveRedisStreamMessageProducer reactiveErrorRedisStreamProducer() { ReactiveRedisStreamMessageProducer messageProducer = new ReactiveRedisStreamMessageProducer(RedisAvailableRule.connectionFactory, STREAM_KEY); - messageProducer.setStreamReceiverOptions( - StreamReceiver.StreamReceiverOptions.builder() - .pollTimeout(Duration.ofMillis(100)) - .targetType(Date.class) - .build()); + messageProducer.setTargetType(Date.class); + messageProducer.setPollTimeout(Duration.ofMillis(100)); messageProducer.setCreateConsumerGroup(true); messageProducer.setAutoAck(false); messageProducer.setConsumerName("testConsumer"); diff --git a/src/reference/asciidoc/redis.adoc b/src/reference/asciidoc/redis.adoc index 9623300833..8efbec1a65 100644 --- a/src/reference/asciidoc/redis.adoc +++ b/src/reference/asciidoc/redis.adoc @@ -1,4 +1,4 @@ -[[redis]] +``[[redis]] == Redis Support Spring Integration 2.1 introduced support for https://redis.io/[Redis]: "`an open source advanced key-value store`". @@ -859,6 +859,10 @@ Similar logic is required even when an exception happens during deserialization So, target error handler must decided to ack or nack such a failed message. Alongside with `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK`, the `ReactiveRedisStreamMessageProducer` also populates these headers into the message to produce: `RedisHeaders.STREAM_KEY`, `RedisHeaders.STREAM_MESSAGE_ID`, `RedisHeaders.CONSUMER_GROUP` and `RedisHeaders.CONSUMER`. +Starting with version 5.5, you can configure `StreamReceiver.StreamReceiverOptionsBuilder` options explicitly on the `ReactiveRedisStreamMessageProducer`, including the newly introduced `onErrorResume` function, which is required if the Redis Stream consumer should continue polling when deserialization errors occur. +The default function sends a message to the error channel (if provided) with possible acknowledgement for the failed message as it is described above. +All these `StreamReceiver.StreamReceiverOptionsBuilder` are mutually exclusive with an externally provided `StreamReceiver.StreamReceiverOptions`. + [[redis-lock-registry]] === Redis Lock Registry diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 7ccd2896ad..d930a0ac91 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -23,3 +23,9 @@ If you are interested in more details, see the Issue Tracker tickets that were r The `AmqpInboundChannelAdapter` and `AmqpInboundGateway` (and the respective Java DSL builders) now support an `org.springframework.amqp.rabbit.retry.MessageRecoverer` as an AMQP-specific alternative to the general purpose `RecoveryCallback`. See <<./amqp.adoc#amqp,AMQP Support>> for more information. + +[[x5.5-redis]] +==== Redis Changes + +The `ReactiveRedisStreamMessageProducer` has now setters for all the `StreamReceiver.StreamReceiverOptionsBuilder` options, including an `onErrorResume` function. +See <<./redis.adoc#redis,Redis Support>> for more information.