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 dc46f75c1..5dc3e883e 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java @@ -58,6 +58,7 @@ import org.springframework.util.StringUtils; * @author Christoph Strobl * @author Tugdual Grall * @author Dengliming + * @author Mark John Moreno * @since 2.2 */ public interface ReactiveStreamCommands { @@ -199,12 +200,14 @@ public interface ReactiveStreamCommands { private final ByteBufferRecord record; private final @Nullable Long maxlen; + private final boolean nomkstream; - private AddStreamRecord(ByteBufferRecord record, @Nullable Long maxlen) { + private AddStreamRecord(ByteBufferRecord record, @Nullable Long maxlen, boolean nomkstream) { super(record.getStream()); this.record = record; this.maxlen = maxlen; + this.nomkstream = nomkstream; } /** @@ -217,7 +220,7 @@ public interface ReactiveStreamCommands { Assert.notNull(record, "Record must not be null!"); - return new AddStreamRecord(record, null); + return new AddStreamRecord(record, null, false); } /** @@ -230,7 +233,7 @@ public interface ReactiveStreamCommands { Assert.notNull(body, "Body must not be null!"); - return new AddStreamRecord(StreamRecords.rawBuffer(body), null); + return new AddStreamRecord(StreamRecords.rawBuffer(body), null, false); } /** @@ -240,7 +243,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); + return new AddStreamRecord(record.withStreamKey(key), maxlen, false); } /** @@ -249,7 +252,28 @@ public interface ReactiveStreamCommands { * @return new instance of {@link AddStreamRecord}. */ public AddStreamRecord maxlen(long maxlen) { - return new AddStreamRecord(record, maxlen); + return new AddStreamRecord(record, maxlen, false); + } + + /** + * Disable creation of 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); + } + + /** + * 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(boolean makeNoStream) { + return new AddStreamRecord(record, maxlen, makeNoStream); } /** @@ -281,6 +305,14 @@ public interface ReactiveStreamCommands { public boolean hasMaxlen() { return maxlen != null && maxlen > 0; } + + /** + * @return {@literal true} if {@literal NOMKSTREAM} is set. + * @since 2.6 + */ + public boolean isNoMkStream() { + return nomkstream; + } } /** 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 f46cc2732..7f50913c6 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java @@ -41,6 +41,7 @@ import org.springframework.util.StringUtils; * @author Christoph Strobl * @author Tugdual Grall * @author Dengliming + * @author Mark John Moreno * @see Redis Documentation - Streams * @since 2.2 */ @@ -116,16 +117,19 @@ public interface RedisStreamCommands { * Additional options applicable for {@literal XADD} command. * * @author Christoph Strobl + * @author Mark John Moreno * @since 2.3 */ class XAddOptions { - private static final XAddOptions NONE = new XAddOptions(null); + private static final XAddOptions NONE = new XAddOptions(null, false); private final @Nullable Long maxlen; + private final boolean nomkstream; - private XAddOptions(@Nullable Long maxlen) { + private XAddOptions(@Nullable Long maxlen, boolean nomkstream) { this.maxlen = maxlen; + this.nomkstream = nomkstream; } /** @@ -141,7 +145,28 @@ public interface RedisStreamCommands { * @return new instance of {@link XAddOptions}. */ public static XAddOptions maxlen(long maxlen) { - return new XAddOptions(maxlen); + return new XAddOptions(maxlen, false); + } + + /** + * Disable creation of 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); + } + + /** + * 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(boolean makeNoStream) { + return new XAddOptions(null, makeNoStream); } /** @@ -161,6 +186,14 @@ public interface RedisStreamCommands { return maxlen != null && maxlen > 0; } + /** + * @return {@literal true} if {@literal NOMKSTREAM} is set. + * @since 2.6 + */ + public boolean isNoMkStream() { + return nomkstream; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -169,14 +202,16 @@ public interface RedisStreamCommands { if (o == null || getClass() != o.getClass()) { return false; } - XAddOptions that = (XAddOptions) o; + if (this.nomkstream != that.nomkstream) return false; return ObjectUtils.nullSafeEquals(this.maxlen, that.maxlen); } @Override public int hashCode() { - return ObjectUtils.nullSafeHashCode(this.maxlen); + int result = ObjectUtils.nullSafeHashCode(this.maxlen); + result = 31 * result + ObjectUtils.nullSafeHashCode(this.nomkstream); + return result; } } 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 8797d81d2..7fe6d0397 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 @@ -55,6 +55,7 @@ import org.springframework.util.Assert; * @author Mark Paluch * @author Tugdual Grall * @author Dengliming + * @author Mark John Moreno * @since 2.2 */ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { @@ -110,6 +111,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { if (command.hasMaxlen()) { args.maxlen(command.getMaxlen()); } + args.nomkstream(command.isNoMkStream()); 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 36f656419..6f137850b 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 @@ -49,6 +49,7 @@ import org.springframework.util.Assert; * @author Tugdual Grall * @author Dejan Jankov * @author Dengliming + * @author Mark John Moreno * @since 2.2 */ class LettuceStreamCommands implements RedisStreamCommands { @@ -90,6 +91,7 @@ class LettuceStreamCommands implements RedisStreamCommands { if (options.hasMaxlen()) { args.maxlen(options.getMaxlen()); } + args.nomkstream(options.isNoMkStream()); return connection.invoke().from(RedisStreamAsyncCommands::xadd, record.getStream(), args, record.getValue()) .get(RecordId::of); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTests.java index e7882df5b..efc644fc5 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTests.java @@ -237,6 +237,23 @@ public class LettuceConnectionUnitTests { verify(asyncCommandsMock).dispatch(eq(command.getType()), eq(command.getOutput()), any(CommandArgs.class)); } + + @Test // GH-2047 + void xaddShouldHonorNoMkStream() { + + MapRecord record = MapRecord.create("key".getBytes(), Collections.emptyMap()); + + connection.streamCommands().xAdd(record, XAddOptions.makeNoStream()); + ArgumentCaptor args = ArgumentCaptor.forClass(XAddArgs.class); + if (connection.isPipelined()) { + verify(asyncCommandsMock, times(1)).xadd(any(), args.capture(), anyMap()); + } else { + verify(syncCommandsMock, times(1)).xadd(any(), args.capture(), anyMap()); + } + + assertThat(args.getValue()).extracting("nomkstream").isEqualTo(true); + } + } public static class LettucePipelineConnectionUnitTests extends BasicUnitTests { 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 27a48ebed..8ed68c1b4 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 @@ -241,4 +241,22 @@ class LettuceReactiveRedisConnectionUnitTests { assertThat(args.getValue()).extracting("maxlen").isEqualTo(100L); } + + @Test // GH-2047 + void xaddShouldHonorNoMkStream() { + + 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)).makeNoStream())) + .subscribe(); + + assertThat(args.getValue()).extracting("nomkstream").isEqualTo(true); + } + }