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 933886f29..40db4e319 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java @@ -199,19 +199,20 @@ public interface ReactiveStreamCommands { class AddStreamRecord extends KeyCommand { private final ByteBufferRecord record; - private final @Nullable Long maxlen; private final boolean nomkstream; + private final @Nullable Long maxlen; private final boolean approximateTrimming; - + private final @Nullable RecordId minId; private AddStreamRecord(ByteBufferRecord record, @Nullable Long maxlen, boolean nomkstream, - boolean approximateTrimming) { + boolean approximateTrimming, @Nullable RecordId minId) { super(record.getStream()); this.record = record; this.maxlen = maxlen; this.nomkstream = nomkstream; this.approximateTrimming = approximateTrimming; + this.minId = minId; } /** @@ -224,7 +225,7 @@ public interface ReactiveStreamCommands { Assert.notNull(record, "Record must not be null!"); - return new AddStreamRecord(record, null, false, false); + return new AddStreamRecord(record, null, false, false, null); } /** @@ -237,7 +238,7 @@ public interface ReactiveStreamCommands { Assert.notNull(body, "Body must not be null!"); - return new AddStreamRecord(StreamRecords.rawBuffer(body), null, false, false); + return new AddStreamRecord(StreamRecords.rawBuffer(body), null, false, false, null); } /** @@ -247,16 +248,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, false); - } - - /** - * Limit the size of the stream to the given maximum number of elements. - * - * @return new instance of {@link AddStreamRecord}. - */ - public AddStreamRecord maxlen(long maxlen) { - return new AddStreamRecord(record, maxlen, false, false); + return new AddStreamRecord(record.withStreamKey(key), maxlen, nomkstream, approximateTrimming, minId); } /** @@ -266,7 +258,7 @@ public interface ReactiveStreamCommands { * @since 2.6 */ public AddStreamRecord makeNoStream() { - return new AddStreamRecord(record, maxlen, true, false); + return new AddStreamRecord(record, maxlen, true, approximateTrimming, minId); } /** @@ -277,7 +269,27 @@ public interface ReactiveStreamCommands { * @since 2.6 */ public AddStreamRecord makeNoStream(boolean makeNoStream) { - return new AddStreamRecord(record, maxlen, makeNoStream, false); + return new AddStreamRecord(record, maxlen, makeNoStream, approximateTrimming, minId); + } + + /** + * Limit the size of the stream to the given maximum number of elements. + * + * @return new instance of {@link AddStreamRecord}. + */ + public AddStreamRecord maxlen(long maxlen) { + return new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming, minId); + } + + /** + * Apply {@code MINID} trimming strategy, that evicts entries with IDs lower than the one specified. + * + * @param minId the minimum record Id to retain. + * @return new instance of {@link AddStreamRecord}. + * @since 2.7 + */ + public AddStreamRecord minId(RecordId minId) { + return new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming, minId); } /** @@ -286,14 +298,7 @@ public interface ReactiveStreamCommands { * @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; + return new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming, minId); } /** @@ -307,6 +312,14 @@ public interface ReactiveStreamCommands { return record; } + /** + * @return {@literal true} if {@literal NOMKSTREAM} is set. + * @since 2.6 + */ + public boolean isNoMkStream() { + return nomkstream; + } + /** * Limit the size of the stream to the given maximum number of elements. * @@ -327,11 +340,28 @@ public interface ReactiveStreamCommands { } /** - * @return {@literal true} if {@literal NOMKSTREAM} is set. - * @since 2.6 + * @return {@literal true} if {@literal approximateTrimming} is set. + * @since 2.7 */ - public boolean isNoMkStream() { - return nomkstream; + public boolean isApproximateTrimming() { + return approximateTrimming; + } + + /** + * @return the minimum record Id to retain during trimming. + * @since 2.7 + */ + @Nullable + public RecordId getMinId() { + return minId; + } + + /** + * @return {@literal true} if {@literal MINID} is set. + * @since 2.7 + */ + public boolean hasMinId() { + return minId != null; } } @@ -1223,7 +1253,7 @@ public interface ReactiveStreamCommands { } public GroupCommand makeStream(boolean mkStream) { - return new GroupCommand(getKey(), action, groupName, consumerName, offset,mkStream); + return new GroupCommand(getKey(), action, groupName, consumerName, offset, mkStream); } public GroupCommand at(ReadOffset offset) { @@ -1291,8 +1321,8 @@ public interface ReactiveStreamCommands { * @since 2.3 */ default Mono xGroupCreate(ByteBuffer key, String groupName, ReadOffset readOffset, boolean mkStream) { - return xGroup(Mono.just(GroupCommand.createGroup(groupName).forStream(key).at(readOffset).makeStream(mkStream))).next() - .map(CommandResponse::getOutput); + return xGroup(Mono.just(GroupCommand.createGroup(groupName).forStream(key).at(readOffset).makeStream(mkStream))) + .next().map(CommandResponse::getOutput); } /** 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 95016e1d6..a7a52b9b5 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java @@ -121,16 +121,19 @@ public interface RedisStreamCommands { */ class XAddOptions { - private static final XAddOptions NONE = new XAddOptions(null, false, false); + private static final XAddOptions NONE = new XAddOptions(null, false, false, null); private final @Nullable Long maxlen; private final boolean nomkstream; private final boolean approximateTrimming; + private final @Nullable RecordId minId; - private XAddOptions(@Nullable Long maxlen, boolean nomkstream, boolean approximateTrimming) { + private XAddOptions(@Nullable Long maxlen, boolean nomkstream, boolean approximateTrimming, + @Nullable RecordId minId) { this.maxlen = maxlen; this.nomkstream = nomkstream; this.approximateTrimming = approximateTrimming; + this.minId = minId; } /** @@ -140,15 +143,6 @@ public interface RedisStreamCommands { return NONE; } - /** - * Limit the size of the stream to the given maximum number of elements. - * - * @return new instance of {@link XAddOptions}. - */ - public static XAddOptions maxlen(long maxlen) { - return new XAddOptions(maxlen, false, false); - } - /** * Disable creation of stream if it does not already exist. * @@ -156,7 +150,7 @@ public interface RedisStreamCommands { * @since 2.6 */ public static XAddOptions makeNoStream() { - return new XAddOptions(null, true, false); + return new XAddOptions(null, true, false, null); } /** @@ -167,7 +161,27 @@ public interface RedisStreamCommands { * @since 2.6 */ public static XAddOptions makeNoStream(boolean makeNoStream) { - return new XAddOptions(null, makeNoStream, false); + return new XAddOptions(null, makeNoStream, false, null); + } + + /** + * Limit the size of the stream to the given maximum number of elements. + * + * @return new instance of {@link XAddOptions}. + */ + public static XAddOptions maxlen(long maxlen) { + return new XAddOptions(maxlen, false, false, null); + } + + /** + * Apply {@code MINID} trimming strategy, that evicts entries with IDs lower than the one specified. + * + * @param minId the minimum record Id to retain. + * @return new instance of {@link XAddOptions}. + * @since 2.7 + */ + public XAddOptions minId(RecordId minId) { + return new XAddOptions(maxlen, nomkstream, approximateTrimming, minId); } /** @@ -176,14 +190,15 @@ public interface RedisStreamCommands { * @return new instance of {@link XAddOptions}. */ public XAddOptions approximateTrimming(boolean approximateTrimming) { - return new XAddOptions(null, nomkstream, approximateTrimming); + return new XAddOptions(maxlen, nomkstream, approximateTrimming, minId); } /** - * @return {@literal true} if {@literal approximateTrimming} is set. + * @return {@literal true} if {@literal NOMKSTREAM} is set. + * @since 2.6 */ - public boolean isApproximateTrimming() { - return approximateTrimming; + public boolean isNoMkStream() { + return nomkstream; } /** @@ -204,11 +219,27 @@ public interface RedisStreamCommands { } /** - * @return {@literal true} if {@literal NOMKSTREAM} is set. - * @since 2.6 + * @return {@literal true} if {@literal approximateTrimming} is set. */ - public boolean isNoMkStream() { - return nomkstream; + public boolean isApproximateTrimming() { + return approximateTrimming; + } + + /** + * @return the minimum record Id to retain during trimming. + * @since 2.7 + */ + @Nullable + public RecordId getMinId() { + return minId; + } + + /** + * @return {@literal true} if {@literal MINID} is set. + * @since 2.7 + */ + public boolean hasMinId() { + return minId != null; } @Override @@ -216,20 +247,28 @@ public interface RedisStreamCommands { if (this == o) { return true; } - if (o == null || getClass() != o.getClass()) { + if (!(o instanceof XAddOptions)) { return false; } 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); + if (nomkstream != that.nomkstream) { + return false; + } + if (approximateTrimming != that.approximateTrimming) { + return false; + } + if (!ObjectUtils.nullSafeEquals(maxlen, that.maxlen)) { + return false; + } + return ObjectUtils.nullSafeEquals(minId, that.minId); } @Override public int hashCode() { - int result = ObjectUtils.nullSafeHashCode(this.maxlen); - result = 31 * result + ObjectUtils.nullSafeHashCode(this.nomkstream); - result = 31 * result + ObjectUtils.nullSafeHashCode(this.approximateTrimming); + int result = ObjectUtils.nullSafeHashCode(maxlen); + result = 31 * result + (nomkstream ? 1 : 0); + result = 31 * result + (approximateTrimming ? 1 : 0); + result = 31 * result + ObjectUtils.nullSafeHashCode(minId); return result; } } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStreamCommands.java index 4f939de40..f659db0f5 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStreamCommands.java @@ -18,6 +18,7 @@ package org.springframework.data.redis.connection.jedis; import static org.springframework.data.redis.connection.jedis.StreamConverters.*; import redis.clients.jedis.BuilderFactory; +import redis.clients.jedis.params.XAddParams; import java.util.Arrays; import java.util.Collections; @@ -72,15 +73,11 @@ class JedisClusterStreamCommands implements RedisStreamCommands { Assert.notNull(record, "Record must not be null!"); Assert.notNull(record.getStream(), "Stream must not be null!"); - byte[] id = JedisConverters.toBytes(record.getId().getValue()); - long maxLength = Long.MAX_VALUE; - if (options.hasMaxlen()) { - maxLength = options.getMaxlen(); - } + XAddParams params = StreamConverters.toXAddParams(record, options); try { - return RecordId.of(JedisConverters - .toString(connection.getCluster().xadd(record.getStream(), id, record.getValue(), maxLength, false))); + return RecordId + .of(JedisConverters.toString(connection.getCluster().xadd(record.getStream(), record.getValue(), params))); } catch (Exception ex) { throw convertJedisAccessException(ex); } 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 3ae3813b6..307f4b1a6 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 @@ -71,20 +71,10 @@ class JedisStreamCommands implements RedisStreamCommands { Assert.notNull(record, "Record must not be null!"); Assert.notNull(record.getStream(), "Stream must not be null!"); - XAddParams xAddParams = new XAddParams(); - xAddParams.id(record.getId().getValue()); - if (options.hasMaxlen()) { - xAddParams.maxLen(options.getMaxlen()); - } - if (options.isNoMkStream()) { - xAddParams.noMkStream(); - } - if (options.isApproximateTrimming()) { - xAddParams.approximateTrimming(); - } + XAddParams params = StreamConverters.toXAddParams(record, options); return connection.invoke() - .from(BinaryJedis::xadd, MultiKeyPipelineBase::xadd, record.getStream(), record.getValue(), xAddParams) + .from(BinaryJedis::xadd, MultiKeyPipelineBase::xadd, record.getStream(), record.getValue(), params) .get(it -> RecordId.of(JedisConverters.toString(it))); } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java b/src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java index bd1e181fe..0214c3f2f 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/StreamConverters.java @@ -18,6 +18,7 @@ package org.springframework.data.redis.connection.jedis; import redis.clients.jedis.StreamEntry; import redis.clients.jedis.StreamEntryID; import redis.clients.jedis.StreamPendingEntry; +import redis.clients.jedis.params.XAddParams; import redis.clients.jedis.util.SafeEncoder; import java.time.Duration; @@ -30,8 +31,10 @@ import java.util.Map; import java.util.stream.Collectors; import org.springframework.data.domain.Range; +import org.springframework.data.redis.connection.RedisStreamCommands; import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.stream.PendingMessage; import org.springframework.data.redis.connection.stream.PendingMessages; import org.springframework.data.redis.connection.stream.RecordId; @@ -166,4 +169,28 @@ class StreamConverters { return new PendingMessages(groupName, messages).withinRange(range); } + + static XAddParams toXAddParams(MapRecord record, RedisStreamCommands.XAddOptions options) { + + XAddParams params = XAddParams.xAddParams(); + params.id(record.getId().getValue()); + + if (options.hasMaxlen()) { + params.maxLen(options.getMaxlen()); + } + + if (options.hasMinId()) { + params.minId(options.getMinId().getValue()); + } + + if (options.isNoMkStream()) { + params.noMkStream(); + } + + if (options.isApproximateTrimming()) { + params.approximateTrimming(); + } + + return params; + } } 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 82453763a..3c8ec5a86 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 @@ -103,6 +103,9 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { if (command.hasMaxlen()) { args.maxlen(command.getMaxlen()); } + if (command.hasMinId()) { + args.minId(command.getMinId().getValue()); + } args.nomkstream(command.isNoMkStream()); args.approximateTrimming(command.isApproximateTrimming()); @@ -138,7 +141,6 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { Flux result = cmd.xclaim(command.getKey(), from, args, ids) .map(it -> StreamRecords.newRecord().in(it.getStream()).withId(it.getId()).ofBuffer(it.getBody())); return new CommandResponse<>(command, result); - })); } 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 2390d50bb..60b266b34 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 @@ -83,6 +83,9 @@ class LettuceStreamCommands implements RedisStreamCommands { if (options.hasMaxlen()) { args.maxlen(options.getMaxlen()); } + if (options.hasMinId()) { + args.minId(options.getMinId().toString()); + } args.nomkstream(options.isNoMkStream()); args.approximateTrimming(options.isApproximateTrimming()); diff --git a/src/main/java/org/springframework/data/redis/connection/stream/StringRecord.java b/src/main/java/org/springframework/data/redis/connection/stream/StringRecord.java index a7d08c3c4..6e893ad60 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/StringRecord.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/StringRecord.java @@ -15,6 +15,8 @@ */ package org.springframework.data.redis.connection.stream; +import java.util.Map; + /** * A {@link Record} within the stream backed by a collection of {@link String} {@literal field/value} paris. * @@ -34,6 +36,17 @@ public interface StringRecord extends MapRecord { */ StringRecord withStreamKey(String key); + /** + * Create a {@link StringRecord} from a {@link Map} of {@link String strings}. + * + * @param source must not be {@literal null}. + * @return new instance of {@link StringRecord}. + * @since 2.7 + */ + static StringRecord of(Map source) { + return StreamRecords.newRecord().ofStrings(source); + } + /** * Convert a {@link MapRecord} of {@link String strings} into a {@link StringRecord}. * diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index ef1db3db5..a385f9ae5 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -76,6 +76,7 @@ import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumer import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroups; import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream; import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.connection.stream.StringRecord; import org.springframework.data.redis.core.Cursor; import org.springframework.data.redis.core.KeyScanOptions; import org.springframework.data.redis.core.ScanOptions; @@ -3602,6 +3603,44 @@ public abstract class AbstractConnectionIntegrationTests { assertThat(results.get(1)).isEqualTo(DataType.STREAM); } + @Test // GH-2247 + @EnabledOnCommand("XADD") + void xAddShouldTrimStreamExactly() { + + RedisStreamCommands.XAddOptions xAddOptions = RedisStreamCommands.XAddOptions.maxlen(1); + actual.add( + connection.xAdd(StringRecord.of(Collections.singletonMap(KEY_2, VALUE_2)).withStreamKey(KEY_1), xAddOptions)); + actual.add( + connection.xAdd(StringRecord.of(Collections.singletonMap(KEY_2, VALUE_2)).withStreamKey(KEY_1), xAddOptions)); + actual.add( + connection.xAdd(StringRecord.of(Collections.singletonMap(KEY_2, VALUE_2)).withStreamKey(KEY_1), xAddOptions)); + actual.add(connection.xLen(KEY_1)); + + List results = getResults(); + assertThat(results).hasSize(4); + assertThat(((RecordId) results.get(0)).getValue()).contains("-"); + assertThat((Long) results.get(3)).isEqualTo(1); + } + + @Test // GH-2247 + @EnabledOnCommand("XADD") + void xAddShouldTrimStreamApprox() { + + RedisStreamCommands.XAddOptions xAddOptions = RedisStreamCommands.XAddOptions.maxlen(1).approximateTrimming(true); + actual.add( + connection.xAdd(StringRecord.of(Collections.singletonMap(KEY_2, VALUE_2)).withStreamKey(KEY_1), xAddOptions)); + actual.add( + connection.xAdd(StringRecord.of(Collections.singletonMap(KEY_2, VALUE_2)).withStreamKey(KEY_1), xAddOptions)); + actual.add( + connection.xAdd(StringRecord.of(Collections.singletonMap(KEY_2, VALUE_2)).withStreamKey(KEY_1), xAddOptions)); + actual.add(connection.xLen(KEY_1)); + + List results = getResults(); + assertThat(results).hasSize(4); + assertThat(((RecordId) results.get(0)).getValue()).contains("-"); + assertThat((Long) results.get(3)).isBetween(1L, 3L); + } + @Test // DATAREDIS-864 @EnabledOnCommand("XADD") public void xReadShouldReadMessage() {