diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java index f4332bef9..933886f29 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java @@ -201,13 +201,17 @@ public interface ReactiveStreamCommands { private final ByteBufferRecord record; private final @Nullable Long maxlen; private final boolean nomkstream; + private final boolean approximateTrimming; - private AddStreamRecord(ByteBufferRecord record, @Nullable Long maxlen, boolean nomkstream) { + + private AddStreamRecord(ByteBufferRecord record, @Nullable Long maxlen, boolean nomkstream, + boolean approximateTrimming) { super(record.getStream()); this.record = record; this.maxlen = maxlen; this.nomkstream = nomkstream; + this.approximateTrimming = approximateTrimming; } /** @@ -220,7 +224,7 @@ public interface ReactiveStreamCommands { Assert.notNull(record, "Record must not be null!"); - return new AddStreamRecord(record, null, false); + return new AddStreamRecord(record, null, false, false); } /** @@ -233,7 +237,7 @@ public interface ReactiveStreamCommands { Assert.notNull(body, "Body must not be null!"); - return new AddStreamRecord(StreamRecords.rawBuffer(body), null, false); + return new AddStreamRecord(StreamRecords.rawBuffer(body), null, false, false); } /** @@ -243,7 +247,7 @@ public interface ReactiveStreamCommands { * @return a new {@link ReactiveGeoCommands.GeoAddCommand} with {@literal key} applied. */ public AddStreamRecord to(ByteBuffer key) { - return new AddStreamRecord(record.withStreamKey(key), maxlen, false); + return new AddStreamRecord(record.withStreamKey(key), maxlen, false, false); } /** @@ -252,7 +256,7 @@ public interface ReactiveStreamCommands { * @return new instance of {@link AddStreamRecord}. */ public AddStreamRecord maxlen(long maxlen) { - return new AddStreamRecord(record, maxlen, false); + return new AddStreamRecord(record, maxlen, false, false); } /** @@ -262,7 +266,7 @@ public interface ReactiveStreamCommands { * @since 2.6 */ public AddStreamRecord makeNoStream() { - return new AddStreamRecord(record, maxlen, true); + return new AddStreamRecord(record, maxlen, true, false); } /** @@ -273,7 +277,23 @@ public interface ReactiveStreamCommands { * @since 2.6 */ public AddStreamRecord makeNoStream(boolean makeNoStream) { - return new AddStreamRecord(record, maxlen, makeNoStream); + return new AddStreamRecord(record, maxlen, makeNoStream, false); + } + + /** + * Apply efficient trimming for capped streams using the {@code ~} flag. + * + * @return new instance of {@link AddStreamRecord}. + */ + public AddStreamRecord approximateTrimming(boolean approximateTrimming) { + return new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming); + } + + /** + * @return {@literal true} if {@literal approximateTrimming} is set. + */ + public boolean isApproximateTrimming() { + return approximateTrimming; } /** diff --git a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java index ab9550368..95016e1d6 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java @@ -116,18 +116,21 @@ public interface RedisStreamCommands { * * @author Christoph Strobl * @author Mark John Moreno + * @author Liming Deng * @since 2.3 */ class XAddOptions { - private static final XAddOptions NONE = new XAddOptions(null, false); + private static final XAddOptions NONE = new XAddOptions(null, false, false); private final @Nullable Long maxlen; private final boolean nomkstream; + private final boolean approximateTrimming; - private XAddOptions(@Nullable Long maxlen, boolean nomkstream) { + private XAddOptions(@Nullable Long maxlen, boolean nomkstream, boolean approximateTrimming) { this.maxlen = maxlen; this.nomkstream = nomkstream; + this.approximateTrimming = approximateTrimming; } /** @@ -143,7 +146,7 @@ public interface RedisStreamCommands { * @return new instance of {@link XAddOptions}. */ public static XAddOptions maxlen(long maxlen) { - return new XAddOptions(maxlen, false); + return new XAddOptions(maxlen, false, false); } /** @@ -153,7 +156,7 @@ public interface RedisStreamCommands { * @since 2.6 */ public static XAddOptions makeNoStream() { - return new XAddOptions(null, true); + return new XAddOptions(null, true, false); } /** @@ -164,7 +167,23 @@ public interface RedisStreamCommands { * @since 2.6 */ public static XAddOptions makeNoStream(boolean makeNoStream) { - return new XAddOptions(null, makeNoStream); + return new XAddOptions(null, makeNoStream, false); + } + + /** + * Apply efficient trimming for capped streams using the {@code ~} flag. + * + * @return new instance of {@link XAddOptions}. + */ + public XAddOptions approximateTrimming(boolean approximateTrimming) { + return new XAddOptions(null, nomkstream, approximateTrimming); + } + + /** + * @return {@literal true} if {@literal approximateTrimming} is set. + */ + public boolean isApproximateTrimming() { + return approximateTrimming; } /** @@ -202,6 +221,7 @@ public interface RedisStreamCommands { } XAddOptions that = (XAddOptions) o; if (this.nomkstream != that.nomkstream) return false; + if (this.approximateTrimming != that.approximateTrimming) return false; return ObjectUtils.nullSafeEquals(this.maxlen, that.maxlen); } @@ -209,6 +229,7 @@ public interface RedisStreamCommands { public int hashCode() { int result = ObjectUtils.nullSafeHashCode(this.maxlen); result = 31 * result + ObjectUtils.nullSafeHashCode(this.nomkstream); + result = 31 * result + ObjectUtils.nullSafeHashCode(this.approximateTrimming); return result; } } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java index 3bdc8512e..3ae3813b6 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java @@ -79,6 +79,9 @@ class JedisStreamCommands implements RedisStreamCommands { if (options.isNoMkStream()) { xAddParams.noMkStream(); } + if (options.isApproximateTrimming()) { + xAddParams.approximateTrimming(); + } return connection.invoke() .from(BinaryJedis::xadd, MultiKeyPipelineBase::xadd, record.getStream(), record.getValue(), xAddParams) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java index 440522a16..82453763a 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java @@ -104,6 +104,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { args.maxlen(command.getMaxlen()); } args.nomkstream(command.isNoMkStream()); + args.approximateTrimming(command.isApproximateTrimming()); return cmd.xadd(command.getKey(), args, command.getBody()) .map(value -> new CommandResponse<>(command, RecordId.of(value))); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java index d25bc268b..2390d50bb 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java @@ -84,6 +84,7 @@ class LettuceStreamCommands implements RedisStreamCommands { args.maxlen(options.getMaxlen()); } args.nomkstream(options.isNoMkStream()); + args.approximateTrimming(options.isApproximateTrimming()); return connection.invoke().from(RedisStreamAsyncCommands::xadd, record.getStream(), args, record.getValue()) .get(RecordId::of);