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<Message<?>, RedisStreamCommands.XAddOptions> addOptionsFunction)`
to allow to configure those option against the specific request message.
This commit is contained in:
Artem Bilan
2025-03-24 14:26:33 -04:00
parent 2e830f94e3
commit c10be96396
3 changed files with 30 additions and 1 deletions

View File

@@ -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 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<Message<?>, 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();
}

View File

@@ -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

View File

@@ -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.