From c10be9639638224a00b307634c7406b052e888eb Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 24 Mar 2025 14:26:33 -0400 Subject: [PATCH] GH-8592: Add `ReactiveRedisStreamMessageHandler.setAddOptionsFunction` Fixes: https://github.com/spring-projects/spring-integration/issues/8592 The `XADD` command for the record can be customized with a `RedisStreamCommands.XAddOptions`. * Expose `ReactiveRedisStreamMessageHandler.setAddOptionsFunction(Function, RedisStreamCommands.XAddOptions> addOptionsFunction)` to allow to configure those option against the specific request message. --- .../ReactiveRedisStreamMessageHandler.java | 23 ++++++++++++++++++- .../antora/modules/ROOT/pages/redis.adoc | 2 ++ .../antora/modules/ROOT/pages/whats-new.adoc | 6 +++++ 3 files changed, 30 insertions(+), 1 deletion(-) diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandler.java index b0c4d11b5a..4db564dd8c 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandler.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandler.java @@ -16,9 +16,12 @@ package org.springframework.integration.redis.outbound; +import java.util.function.Function; + import reactor.core.publisher.Mono; import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; +import org.springframework.data.redis.connection.RedisStreamCommands; import org.springframework.data.redis.connection.stream.Record; import org.springframework.data.redis.connection.stream.StreamRecords; import org.springframework.data.redis.core.ReactiveRedisTemplate; @@ -60,6 +63,9 @@ public class ReactiveRedisStreamMessageHandler extends AbstractReactiveMessageHa @Nullable private HashMapper hashMapper; + @Nullable + private Function, RedisStreamCommands.XAddOptions> addOptionsFunction; + /** * Create an instance based on provided {@link ReactiveRedisConnectionFactory} and key for stream. * @param connectionFactory the {@link ReactiveRedisConnectionFactory} to use @@ -106,6 +112,16 @@ public class ReactiveRedisStreamMessageHandler extends AbstractReactiveMessageHa this.extractPayload = extractPayload; } + /** + * Set a function to create a {@link RedisStreamCommands.XAddOptions} based on the request message. + * Cannot be null and cannot return null. + * @param addOptionsFunction the function to provide a {@link RedisStreamCommands.XAddOptions}. + * @since 6.5 + */ + public void setAddOptionsFunction(Function, RedisStreamCommands.XAddOptions> addOptionsFunction) { + this.addOptionsFunction = addOptionsFunction; + } + @Override public String getComponentType() { return "redis:stream-outbound-channel-adapter"; @@ -145,7 +161,12 @@ public class ReactiveRedisStreamMessageHandler extends AbstractReactiveMessageHa StreamRecords.objectBacked(value) .withStreamKey(streamKey); - return this.reactiveStreamOperations.add(record); + if (this.addOptionsFunction == null) { + return this.reactiveStreamOperations.add(record); + } + else { + return this.reactiveStreamOperations.add(record, this.addOptionsFunction.apply(message)); + } }) .then(); } diff --git a/src/reference/antora/modules/ROOT/pages/redis.adoc b/src/reference/antora/modules/ROOT/pages/redis.adoc index bead76431d..98302aa17d 100644 --- a/src/reference/antora/modules/ROOT/pages/redis.adoc +++ b/src/reference/antora/modules/ROOT/pages/redis.adoc @@ -761,6 +761,8 @@ Another constructor variant is based on a SpEL expression to evaluate a stream k Or use the whole message as a value. It defaults to `true`. +Starting with version 6.5, the `ReactiveRedisStreamMessageHandler` provides a `setAddOptionsFunction(Function, RedisStreamCommands.XAddOptions> addOptionsFunction)` to build `RedisStreamCommands.XAddOptions` based on the request message for the internal `ReactiveStreamOperations.add(Record record, XAddOptions xAddOptions)` call. + [[redis-stream-inbound]] == Redis Stream Inbound Channel Adapter diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 5a33ac838c..12786c14a2 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -83,3 +83,9 @@ The `BeanPropertySqlParameterSourceFactory` uses now internally the `MapSqlParam Also, `JdbcMessageHandler` exposes a `usePayloadAsParameterSource` flag to allow to deal with parameter source only against message payload. That's where the mentioned `MapSqlParameterSource` comes useful for request messages with map payloads. See xref:jdbc.adoc[JDBC Support] for more information. + +[[x6.5-redis-changes]] +== Redis Stream Support + +The `ReactiveRedisStreamMessageHandler` now exposes a `Function, RedisStreamCommands.XAddOptions>` to provide additional `XADD` option via convenient `RedisStreamCommands.XAddOptions` API. +See xref:redis.adoc#redis-stream-outbound[Redis Support] for more information. \ No newline at end of file