Polishing.

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

Reorder methods. Simplify StreamRecord creation.

See #2247

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);
+			return new AddStreamRecord(record.withStreamKey(key), maxlen, nomkstream, 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, false, false);
 		}
 
 		/**
@@ -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 new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming, minId);
-		}
-
-		/**
-		 * @return {@literal true} if {@literal approximateTrimming} is set.
-		 */
-		public boolean isApproximateTrimming() {
-			return approximateTrimming;
 		}
 
 		/**
@@ -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.
+		 * @return {@literal true} if {@literal approximateTrimming} is set.
-		 * @since 2.6
+		 * @since 2.7
 		 */
-		public boolean isNoMkStream() {
+		public boolean isApproximateTrimming() {
-			return nomkstream;
+			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<String> xGroupCreate(ByteBuffer key, String groupName, ReadOffset readOffset, boolean mkStream) {
-		return xGroup(Mono.just(GroupCommand.createGroup(groupName).forStream(key).at(readOffset).makeStream(mkStream))).next()
+		return xGroup(Mono.just(GroupCommand.createGroup(groupName).forStream(key).at(readOffset).makeStream(mkStream)))
-				.map(CommandResponse::getOutput);
+				.next().map(CommandResponse::getOutput);
 	}
 
 	/**

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() {
+		public boolean isNoMkStream() {
-			return approximateTrimming;
+			return nomkstream;
 		}
 
 		/**
@@ -204,11 +219,27 @@ public interface RedisStreamCommands {
 		}
 
 		/**
-		 * @return {@literal true} if {@literal NOMKSTREAM} is set.
+		 * @return {@literal true} if {@literal approximateTrimming} is set.
-		 * @since 2.6
 		 */
-		public boolean isNoMkStream() {
+		public boolean isApproximateTrimming() {
-			return nomkstream;
+			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 (nomkstream != that.nomkstream) {
-			if (this.approximateTrimming != that.approximateTrimming) return false;
+				return false;
-			return ObjectUtils.nullSafeEquals(this.maxlen, that.maxlen);
+			}
+			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);
+			int result = ObjectUtils.nullSafeHashCode(maxlen);
-			result = 31 * result + ObjectUtils.nullSafeHashCode(this.nomkstream);
+			result = 31 * result + (nomkstream ? 1 : 0);
-			result = 31 * result + ObjectUtils.nullSafeHashCode(this.approximateTrimming);
+			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 @@ 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;
@@ -80,15 +81,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());
+		XAddParams params = StreamConverters.toXAddParams(record, options);
-		long maxLength = Long.MAX_VALUE;
-		if (options.hasMaxlen()) {
-			maxLength = options.getMaxlen();
-		}
 
 		try {
-			return RecordId.of(JedisConverters
+			return RecordId
-					.toString(connection.getCluster().xadd(record.getStream(), id, record.getValue(), maxLength, false)));
+					.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

@@ -79,20 +79,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 params = StreamConverters.toXAddParams(record, options);
-		xAddParams.id(record.getId().getValue());
-		if (options.hasMaxlen()) {
-			xAddParams.maxLen(options.getMaxlen());
-		}
-		if (options.isNoMkStream()) {
-			xAddParams.noMkStream();
-		}
-		if (options.isApproximateTrimming()) {
-			xAddParams.approximateTrimming();
-		}
 
 		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)));
 	}
 

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<byte[], byte[], byte[]> 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;
+	}
 }

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java

@@ -111,6 +111,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());
 
@@ -154,7 +157,6 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands {
 			Flux<ByteBufferRecord> 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);
-
 		}));
 	}
 

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java

@@ -91,6 +91,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());
 

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.
  *
@@ -38,6 +40,17 @@ public interface StringRecord extends MapRecord<String, String, String> {
 	 */
 	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<String, String> source) {
+		return StreamRecords.newRecord().ofStrings(source);
+	}
+
 	/**
 	 * Convert a {@link MapRecord} of {@link String strings} into a {@link StringRecord}.
 	 *

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<Object> 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<Object> 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() {