DATAREDIS-1122 - Add support for MAXLEN attribute of XADD.

Original pull request: #523.

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

@@ -3644,11 +3644,11 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.StringRedisConnection#xAdd(StringRecord)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#xAdd(StringRecord, XAddOptions)
 	 */
 	@Override
-	public RecordId xAdd(StringRecord record) {
-		return convertAndReturn(delegate.xAdd(record.serialize(serializer)), identityConverter);
+	public RecordId xAdd(StringRecord record, XAddOptions options) {
+		return convertAndReturn(delegate.xAdd(record.serialize(serializer), options), identityConverter);
 	}
 
 	/*
@@ -3814,11 +3814,11 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xAdd(byte[], MapRecord)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xAdd(MapRecord, XAddOptions)
 	 */
 	@Override
-	public RecordId xAdd(MapRecord<byte[], byte[], byte[]> record) {
-		return delegate.xAdd(record);
+	public RecordId xAdd(MapRecord<byte[], byte[], byte[]> record, XAddOptions options) {
+		return delegate.xAdd(record, options);
 	}
 
 	/*

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

@@ -449,8 +449,8 @@ public interface DefaultedRedisConnection extends RedisConnection {
 	/** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */
 	@Override
 	@Deprecated
-	default RecordId xAdd(MapRecord<byte[], byte[], byte[]> record) {
-		return streamCommands().xAdd(record);
+	default RecordId xAdd(MapRecord<byte[], byte[], byte[]> record, XAddOptions options) {
+		return streamCommands().xAdd(record, options);
 	}
 
 	/** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */

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

@@ -193,11 +193,13 @@ public interface ReactiveStreamCommands {
 	class AddStreamRecord extends KeyCommand {
 
 		private final ByteBufferRecord record;
+		private final @Nullable Long maxlen;
 
-		private AddStreamRecord(ByteBufferRecord record) {
+		private AddStreamRecord(ByteBufferRecord record, @Nullable Long maxlen) {
 
 			super(record.getStream());
 			this.record = record;
+			this.maxlen = maxlen;
 		}
 
 		/**
@@ -210,7 +212,7 @@ public interface ReactiveStreamCommands {
 
 			Assert.notNull(record, "Record must not be null!");
 
-			return new AddStreamRecord(record);
+			return new AddStreamRecord(record, null);
 		}
 
 		/**
@@ -223,7 +225,7 @@ public interface ReactiveStreamCommands {
 
 			Assert.notNull(body, "Body must not be null!");
 
-			return new AddStreamRecord(StreamRecords.rawBuffer(body));
+			return new AddStreamRecord(StreamRecords.rawBuffer(body), null);
 		}
 
 		/**
@@ -233,7 +235,16 @@ public interface ReactiveStreamCommands {
 		 * @return a new {@link ReactiveGeoCommands.GeoAddCommand} with {@literal key} applied.
 		 */
 		public AddStreamRecord to(ByteBuffer key) {
-			return new AddStreamRecord(record.withStreamKey(key));
+			return new AddStreamRecord(record.withStreamKey(key), maxlen);
+		}
+
+		/**
+		 * 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);
 		}
 
 		/**
@@ -246,6 +257,23 @@ public interface ReactiveStreamCommands {
 		public ByteBufferRecord getRecord() {
 			return record;
 		}
+
+		/**
+		 * Limit the size of the stream to the given maximum number of elements.
+		 *
+		 * @return can be {@literal null}.
+		 */
+		@Nullable
+		public Long getMaxlen() {
+			return maxlen;
+		}
+
+		/**
+		 * @return {@literal true} if {@literal MAXLEN} is set.
+		 */
+		public boolean hasMaxLen() {
+			return maxlen != null && maxlen > 0;
+		}
 	}
 
 	/**

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

@@ -28,6 +28,7 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
 import org.springframework.data.redis.connection.stream.*;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -87,7 +88,88 @@ public interface RedisStreamCommands {
 	 * @param record the {@link MapRecord record} to append.
 	 * @return the {@link RecordId id} after save. {@literal null} when used in pipeline / transaction.
 	 */
-	RecordId xAdd(MapRecord<byte[], byte[], byte[]> record);
+	default RecordId xAdd(MapRecord<byte[], byte[], byte[]> record) {
+		return xAdd(record, XAddOptions.none());
+	}
+
+	/**
+	 * Append the given {@link MapRecord record} to the stream stored at {@link Record#getStream()}. <br />
+	 * If you prefer manual id assignment over server generated ones make sure to provide an id via
+	 * {@link Record#withId(RecordId)}.
+	 *
+	 * @param record the {@link MapRecord record} to append.
+	 * @param options additional options (eg. {@literal MAXLEN}). Must not be {@literal null}, use
+	 *          {@link XAddOptions#none()} instead.
+	 * @return the {@link RecordId id} after save. {@literal null} when used in pipeline / transaction.
+	 * @since 2.3
+	 */
+	RecordId xAdd(MapRecord<byte[], byte[], byte[]> record, XAddOptions options);
+
+	/**
+	 * Additional options applicable for {@literal XADD} command.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.3
+	 */
+	class XAddOptions {
+
+		private @Nullable Long maxlen;
+
+		private XAddOptions(@Nullable Long maxlen) {
+			this.maxlen = maxlen;
+		}
+
+		/**
+		 * @return
+		 */
+		public static XAddOptions none() {
+			return new XAddOptions(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);
+		}
+
+		/**
+		 * Limit the size of the stream to the given maximum number of elements.
+		 * 
+		 * @return can be {@literal null}.
+		 */
+		@Nullable
+		public Long getMaxlen() {
+			return maxlen;
+		}
+
+		/**
+		 * @return {@literal true} if {@literal MAXLEN} is set.
+		 */
+		public boolean hasMaxLen() {
+			return maxlen != null && maxlen > 0;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o) {
+				return true;
+			}
+			if (o == null || getClass() != o.getClass()) {
+				return false;
+			}
+
+			XAddOptions that = (XAddOptions) o;
+			return ObjectUtils.nullSafeEquals(this.maxlen, that.maxlen);
+		}
+
+		@Override
+		public int hashCode() {
+			return ObjectUtils.nullSafeHashCode(this.maxlen);
+		}
+	}
 
 	/**
 	 * Change the ownership of a pending message to the given new {@literal consumer} without increasing the delivered

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

@@ -2010,7 +2010,28 @@ public interface StringRedisConnection extends RedisConnection {
 		return xAdd(StreamRecords.newRecord().in(key).ofStrings(body));
 	}
 
-	RecordId xAdd(StringRecord record);
+	/**
+	 * Append the given {@link StringRecord} to the stream stored at {@link StringRecord#getStream()}.
+	 *
+	 * @param record must not be {@literal null}.
+	 * @return the record Id. {@literal null} when used in pipeline / transaction.
+	 * @since 2.2
+	 */
+	@Nullable
+	default RecordId xAdd(StringRecord record) {
+		return xAdd(record, XAddOptions.none());
+	}
+
+	/**
+	 * Append the given {@link StringRecord} to the stream stored at {@link StringRecord#getStream()}.
+	 * 
+	 * @param record must not be {@literal null}.
+	 * @param options must not be {@literal null}, use {@link XAddOptions#none()} instead.
+	 * @return the record Id. {@literal null} when used in pipeline / transaction.
+	 * @since 2.3
+	 */
+	@Nullable
+	RecordId xAdd(StringRecord record, XAddOptions options);
 
 	/**
 	 * Change the ownership of a pending message to the given new {@literal consumer} without increasing the delivered

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

@@ -100,6 +100,9 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands {
 			if (!command.getRecord().getId().shouldBeAutoGenerated()) {
 				args.id(command.getRecord().getId().getValue());
 			}
+			if(command.hasMaxLen()) {
+				args.maxlen(command.getMaxlen());
+			}
 
 			return cmd.xadd(command.getKey(), args, command.getBody())
 					.map(value -> new CommandResponse<>(command, RecordId.of(value)));

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

@@ -81,16 +81,19 @@ class LettuceStreamCommands implements RedisStreamCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xAdd(byte[], MapRecord)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xAdd(byte[], MapRecord, XAddOptions)
 	 */
 	@Override
-	public RecordId xAdd(MapRecord<byte[], byte[], byte[]> record) {
+	public RecordId xAdd(MapRecord<byte[], byte[], byte[]> record, XAddOptions options) {
 
 		Assert.notNull(record.getStream(), "Stream must not be null!");
 		Assert.notNull(record, "Record must not be null!");
 
 		XAddArgs args = new XAddArgs();
 		args.id(record.getId().getValue());
+		if(options.hasMaxLen()) {
+			args.maxlen(options.getMaxlen());
+		}
 
 		try {
 			if (isPipelined()) {

src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java

@@ -46,6 +46,7 @@ import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
 import org.springframework.data.redis.connection.RedisListCommands.Position;
 import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption;
+import org.springframework.data.redis.connection.RedisStreamCommands.XAddOptions;
 import org.springframework.data.redis.connection.RedisStringCommands.BitOperation;
 import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
@@ -2054,10 +2055,10 @@ public class DefaultStringRedisConnectionTests {
 		Assertions.assertThat(getResults()).containsExactly(1L);
 	}
 
-	@Test // DATAREDIS-864
+	@Test // DATAREDIS-864, DATAREDIS-1122
 	public void xAddShouldAppendRecordCorrectly() {
 
-		doReturn(RecordId.of("1-1")).when(nativeConnection).xAdd(any());
+		doReturn(RecordId.of("1-1")).when(nativeConnection).xAdd(any(), eq(XAddOptions.none()));
 		actual.add(connection
 				.xAdd(StreamRecords.newRecord().in("stream-1").ofStrings(Collections.singletonMap("field", "value"))));
 

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java

@@ -19,23 +19,28 @@ import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
 import io.lettuce.core.RedisClient;
+import io.lettuce.core.XAddArgs;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.async.RedisAsyncCommands;
 import io.lettuce.core.api.sync.RedisCommands;
 import io.lettuce.core.codec.RedisCodec;
 
 import java.lang.reflect.InvocationTargetException;
+import java.util.Collections;
 
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Suite;
 
+import org.mockito.ArgumentCaptor;
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.data.redis.connection.AbstractConnectionUnitTestBase;
 import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption;
+import org.springframework.data.redis.connection.RedisStreamCommands.XAddOptions;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionUnitTestSuite.LettuceConnectionUnitTests;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionUnitTestSuite.LettucePipelineConnectionUnitTests;
+import org.springframework.data.redis.connection.stream.MapRecord;
 
 /**
  * @author Christoph Strobl
@@ -164,6 +169,19 @@ public class LettuceConnectionUnitTestSuite {
 			assertThatThrownBy(() -> connection.set("foo".getBytes(), "bar".getBytes()))
 					.hasMessageContaining(exception.getMessage()).hasRootCause(exception);
 		}
+
+		@Test // DATAREDIS-1122
+		public void xaddShouldHonorMaxlen() {
+
+			MapRecord<byte[], byte[], byte[]> record = MapRecord.create("key".getBytes(),
+					Collections.emptyMap());
+
+			connection.streamCommands().xAdd(record, XAddOptions.maxlen(100));
+			ArgumentCaptor<XAddArgs> args = ArgumentCaptor.forClass(XAddArgs.class);
+			verify(syncCommandsMock, times(1)).xadd(any(), args.capture(), anyMap());
+
+			assertThat(args.getValue()).extracting("maxlen").isEqualTo(100L);
+		}
 	}
 
 	public static class LettucePipelineConnectionUnitTests extends LettuceConnectionUnitTests {

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnectionUnitTests.java

@@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
 import io.lettuce.core.RedisConnectionException;
+import io.lettuce.core.XAddArgs;
 import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.reactive.RedisReactiveCommands;
@@ -26,6 +27,8 @@ import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -33,9 +36,13 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Answers;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.connection.ReactiveStreamCommands.AddStreamRecord;
+import org.springframework.data.redis.connection.stream.ByteBufferRecord;
+import org.springframework.data.redis.connection.stream.MapRecord;
 
 /**
  * Unit tests for {@link LettuceReactiveRedisConnection}.
@@ -214,4 +221,22 @@ public class LettuceReactiveRedisConnectionUnitTests {
 
 		StepVerifier.create(connection.serverCommands().bgSave()).expectNextCount(1).verifyComplete();
 	}
+
+	@Test // DATAREDIS-1122
+	public void xaddShouldHonorMaxlen() {
+
+		LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider);
+
+		ArgumentCaptor<XAddArgs> args = ArgumentCaptor.forClass(XAddArgs.class);
+		when(reactiveCommands.xadd(any(ByteBuffer.class), args.capture(), anyMap())).thenReturn(Mono.just("1-1"));
+
+
+		MapRecord<ByteBuffer, ByteBuffer, ByteBuffer> record = MapRecord.create(ByteBuffer.wrap("key".getBytes()),
+				Collections.emptyMap());
+
+		connection.streamCommands().xAdd(Mono.just(AddStreamRecord.of(ByteBufferRecord.of(record)).maxlen(100)))
+				.subscribe();
+
+		assertThat(args.getValue()).extracting("maxlen").isEqualTo(100L);
+	}
 }