Polishing.

Add support for MINID trimming. Add tests. Extract Jedis XAddParams conversion to StreamConverters.

Reorder methods. Simplify StreamRecord creation.

See #2247

Author: mp911de

src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java

@@ -199,19 +199,20 @@ default Mono<Long> xAck(ByteBuffer key, String group, RecordId... recordIds) {
 	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 static AddStreamRecord of(ByteBufferRecord record) {
 
 			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 static AddStreamRecord body(Map<ByteBuffer, ByteBuffer> body) {
 
 			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,53 +248,57 @@ public static AddStreamRecord body(Map<ByteBuffer, ByteBuffer> body) {
 		 * @return a new {@link ReactiveGeoCommands.GeoAddCommand} with {@literal key} applied.
 		 */
 		public AddStreamRecord to(ByteBuffer key) {
-			return new AddStreamRecord(record.withStreamKey(key), maxlen, false, false);
+			return new AddStreamRecord(record.withStreamKey(key), maxlen, nomkstream, approximateTrimming, minId);
 		}
 
 		/**
-		 * Limit the size of the stream to the given maximum number of elements.
+		 * Disable creation of stream if it does not already exist.
 		 *
 		 * @return new instance of {@link AddStreamRecord}.
+		 * @since 2.6
 		 */
-		public AddStreamRecord maxlen(long maxlen) {
-			return new AddStreamRecord(record, maxlen, false, false);
+		public AddStreamRecord makeNoStream() {
+			return new AddStreamRecord(record, maxlen, true, approximateTrimming, minId);
 		}
 
 		/**
 		 * Disable creation of stream if it does not already exist.
 		 *
+		 * @param makeNoStream {@code true} to not create a stream if it does not already exist.
 		 * @return new instance of {@link AddStreamRecord}.
 		 * @since 2.6
 		 */
-		public AddStreamRecord makeNoStream() {
-			return new AddStreamRecord(record, maxlen, true, false);
+		public AddStreamRecord makeNoStream(boolean makeNoStream) {
+			return new AddStreamRecord(record, maxlen, makeNoStream, approximateTrimming, minId);
 		}
 
 		/**
-		 * Disable creation of stream if it does not already exist.
+		 * Limit the size of the stream to the given maximum number of elements.
 		 *
-		 * @param makeNoStream {@code true} to not create a stream if it does not already exist.
 		 * @return new instance of {@link AddStreamRecord}.
-		 * @since 2.6
 		 */
-		public AddStreamRecord makeNoStream(boolean makeNoStream) {
-			return new AddStreamRecord(record, maxlen, makeNoStream, false);
+		public AddStreamRecord maxlen(long maxlen) {
+			return new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming, minId);
 		}
 
 		/**
-		 * Apply efficient trimming for capped streams using the {@code ~} flag.
+		 * 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 approximateTrimming(boolean approximateTrimming) {
-			return new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming);
+		public AddStreamRecord minId(RecordId minId) {
+			return new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming, minId);
 		}
 
 		/**
-		 * @return {@literal true} if {@literal approximateTrimming} is set.
+		 * Apply efficient trimming for capped streams using the {@code ~} flag.
+		 *
+		 * @return new instance of {@link AddStreamRecord}.
 		 */
-		public boolean isApproximateTrimming() {
-			return approximateTrimming;
+		public AddStreamRecord approximateTrimming(boolean approximateTrimming) {
+			return new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming, minId);
 		}
 
 		/**
@@ -307,6 +312,14 @@ public ByteBufferRecord getRecord() {
 			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 boolean hasMaxlen() {
 		}
 
 		/**
-		 * @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 static GroupCommand deleteConsumer(Consumer consumer) {
 		}
 
 		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 @@ default Mono<String> xGroupCreate(ByteBuffer key, String groupName, ReadOffset r
 	 * @since 2.3
 	 */
 	default Mono<String> 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);
 	}
 
 	/**

src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java

@@ -121,16 +121,19 @@ default RecordId xAdd(MapRecord<byte[], byte[], byte[]> record) {
 	 */
 	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;
 		}
 
 		/**
@@ -141,33 +144,44 @@ public static XAddOptions none() {
 		}
 
 		/**
-		 * Limit the size of the stream to the given maximum number of elements.
+		 * Disable creation of stream if it does not already exist.
 		 *
 		 * @return new instance of {@link XAddOptions}.
+		 * @since 2.6
 		 */
-		public static XAddOptions maxlen(long maxlen) {
-			return new XAddOptions(maxlen, false, false);
+		public static XAddOptions makeNoStream() {
+			return new XAddOptions(null, true, false, null);
 		}
 
 		/**
 		 * Disable creation of stream if it does not already exist.
 		 *
+		 * @param makeNoStream {@code true} to not create a stream if it does not already exist.
 		 * @return new instance of {@link XAddOptions}.
 		 * @since 2.6
 		 */
-		public static XAddOptions makeNoStream() {
-			return new XAddOptions(null, true, false);
+		public static XAddOptions makeNoStream(boolean makeNoStream) {
+			return new XAddOptions(null, makeNoStream, false, null);
 		}
 
 		/**
-		 * Disable creation of stream if it does not already exist.
+		 * Limit the size of the stream to the given maximum number of elements.
 		 *
-		 * @param makeNoStream {@code true} to not create a stream if it does not already exist.
 		 * @return new instance of {@link XAddOptions}.
-		 * @since 2.6
 		 */
-		public static XAddOptions makeNoStream(boolean makeNoStream) {
-			return new XAddOptions(null, makeNoStream, false);
+		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 static XAddOptions makeNoStream(boolean makeNoStream) {
 		 * @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,32 +219,56 @@ public boolean hasMaxlen() {
 		}
 
 		/**
-		 * @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
 		public boolean equals(Object o) {
 			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;
 		}
 	}

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStreamCommands.java

@@ -18,6 +18,7 @@
 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 @@ public RecordId xAdd(MapRecord<byte[], byte[], byte[]> record, XAddOptions optio
 		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);
 		}

src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java

@@ -71,20 +71,10 @@ public RecordId xAdd(MapRecord<byte[], byte[], byte[]> record, XAddOptions optio
 		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)));
 	}