diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java index 86591311d..3fa246685 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java +++ b/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 record) { - return delegate.xAdd(record); + public RecordId xAdd(MapRecord record, XAddOptions options) { + return delegate.xAdd(record, options); } /* diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java index 3657a8076..c0c2466d2 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java +++ b/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 record) { - return streamCommands().xAdd(record); + default RecordId xAdd(MapRecord record, XAddOptions options) { + return streamCommands().xAdd(record, options); } /** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */ 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 fdebeedd1..d2f3948fc 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/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; + } } /** 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 50dfa4b73..4aebe5ad5 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/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 record); + default RecordId xAdd(MapRecord record) { + return xAdd(record, XAddOptions.none()); + } + + /** + * Append the given {@link MapRecord record} to the stream stored at {@link Record#getStream()}.
+ * 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 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 diff --git a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java index 2039caa2d..45dde6f5a 100644 --- a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java +++ b/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 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 1ed1fdda1..c916a2c73 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 @@ -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))); 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 769105e83..353b13978 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 @@ -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 record) { + public RecordId xAdd(MapRecord 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()) { diff --git a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java index d66699a3b..344775641 100644 --- a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java +++ b/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")))); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java index 048729047..38f9ea657 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java +++ b/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 record = MapRecord.create("key".getBytes(), + Collections.emptyMap()); + + connection.streamCommands().xAdd(record, XAddOptions.maxlen(100)); + ArgumentCaptor 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 { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnectionUnitTests.java index b7a32e487..7e3e71bd8 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnectionUnitTests.java +++ b/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 args = ArgumentCaptor.forClass(XAddArgs.class); + when(reactiveCommands.xadd(any(ByteBuffer.class), args.capture(), anyMap())).thenReturn(Mono.just("1-1")); + + + MapRecord 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); + } }