GH-8592: Add ReactiveRedisStreamMessageHandler.setAddOptionsFunction

Fixes: #8592

The `XADD` command for the record can be customized with a `RedisStreamCommands.XAddOptions`.

* Expose `ReactiveRedisStreamMessageHandler.setAddOptionsFunction(Function<Message<?>, RedisStreamCommands.XAddOptions> addOptionsFunction)`
to allow to configure those option against the specific request message.

Author: artembilan

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<String, ?, ?> hashMapper;
 
+	@Nullable
+	private Function<Message<?>, 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 void setExtractPayload(boolean extractPayload) {
 		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<Message<?>, RedisStreamCommands.XAddOptions> addOptionsFunction) {
+		this.addOptionsFunction = addOptionsFunction;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "redis:stream-outbound-channel-adapter";
@@ -145,7 +161,12 @@ protected Mono<Void> handleMessageInternal(Message<?> message) {
 							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();
 	}

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<Message<?>, RedisStreamCommands.XAddOptions> addOptionsFunction)` to build `RedisStreamCommands.XAddOptions` based on the request message for the internal `ReactiveStreamOperations.add(Record<K, ?> record, XAddOptions xAddOptions)` call.
+
 [[redis-stream-inbound]]
 == Redis Stream Inbound Channel Adapter
 

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<Message<?>, RedisStreamCommands.XAddOptions>` to provide additional `XADD` option via convenient `RedisStreamCommands.XAddOptions` API.
+See xref:redis.adoc#redis-stream-outbound[Redis Support] for more information.