Add NOMKSTREAM option to XADD command

Closes: spring-projects#2047

Author: Mark John Moreno

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

@@ -58,6 +58,7 @@
  * @author Christoph Strobl
  * @author Tugdual Grall
  * @author Dengliming
+ * @author Mark John Moreno
  * @since 2.2
  */
 public interface ReactiveStreamCommands {
@@ -199,12 +200,14 @@ class AddStreamRecord extends KeyCommand {
 
 		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 static AddStreamRecord of(ByteBufferRecord record) {
 
 			Assert.notNull(record, "Record must not be null!");
 
-			return new AddStreamRecord(record, null);
+			return new AddStreamRecord(record, null, false);
 		}
 
 		/**
@@ -230,7 +233,7 @@ public static AddStreamRecord body(Map<ByteBuffer, ByteBuffer> body) {
 
 			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 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);
+			return new AddStreamRecord(record.withStreamKey(key), maxlen, false);
 		}
 
 		/**
@@ -249,7 +252,26 @@ public AddStreamRecord to(ByteBuffer key) {
 		 * @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}.
+		 */
+		public AddStreamRecord makeNoStream() {
+			return new AddStreamRecord(record, maxlen, true);
+		}
+
+		/**
+		 * Disable creation of stream if it does not already exist.
+		 *
+		 * @param maxlen size of the stream to the given maximum number of elements.
+		 * @return new instance of {@link AddStreamRecord}.
+		 */
+		public AddStreamRecord makeNoStream(long maxlen) {
+			return new AddStreamRecord(record, maxlen, true);
 		}
 
 		/**
@@ -281,6 +303,13 @@ public Long getMaxlen() {
 		public boolean hasMaxlen() {
 			return maxlen != null && maxlen > 0;
 		}
+
+		/**
+		 * @return {@literal true} if {@literal NOMKSTREAM} is set.
+		 */
+		public boolean isNoMkStream() {
+			return nomkstream;
+		}
 	}
 
 	/**

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

@@ -41,6 +41,7 @@
  * @author Christoph Strobl
  * @author Tugdual Grall
  * @author Dengliming
+ * @author Mark John Moreno
  * @see <a href="https://redis.io/topics/streams-intro">Redis Documentation - Streams</a>
  * @since 2.2
  */
@@ -116,16 +117,19 @@ default RecordId xAdd(MapRecord<byte[], byte[], byte[]> record) {
 	 * 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,26 @@ public static XAddOptions none() {
 		 * @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}.
+		 */
+		public static XAddOptions makeNoStream() {
+			return new XAddOptions(null, true);
+		}
+
+		/**
+		 * Disable creation of stream if it does not already exist.
+		 *
+		 * @param maxlen size of the stream to the given maximum number of elements.
+		 * @return new instance of {@link XAddOptions}.
+		 */
+		public static XAddOptions makeNoStream(long maxlen) {
+			return new XAddOptions(maxlen, true);
 		}
 
 		/**
@@ -161,6 +184,13 @@ public boolean hasMaxlen() {
 			return maxlen != null && maxlen > 0;
 		}
 
+		/**
+		 * @return {@literal true} if {@literal NOMKSTREAM} is set.
+		 */
+		public boolean isNoMkStream() {
+			return nomkstream;
+		}
+
 		@Override
 		public boolean equals(Object o) {
 			if (this == o) {

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

@@ -55,6 +55,7 @@
  * @author Mark Paluch
  * @author Tugdual Grall
  * @author Dengliming
+ * @author Mark John Moreno
  * @since 2.2
  */
 class LettuceReactiveStreamCommands implements ReactiveStreamCommands {
@@ -110,6 +111,7 @@ public Flux<CommandResponse<AddStreamRecord, RecordId>> xAdd(Publisher<AddStream
 			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)));

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

@@ -49,6 +49,7 @@
  * @author Tugdual Grall
  * @author Dejan Jankov
  * @author Dengliming
+ * @author Mark John Moreno
  * @since 2.2
  */
 class LettuceStreamCommands implements RedisStreamCommands {
@@ -90,6 +91,7 @@ public RecordId xAdd(MapRecord<byte[], byte[], byte[]> record, XAddOptions optio
 		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);

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

@@ -237,6 +237,39 @@ void executeShouldPassThruCustomCommands() {
 
 			verify(asyncCommandsMock).dispatch(eq(command.getType()), eq(command.getOutput()), any(CommandArgs.class));
 		}
+
+		@Test // GH-2047
+		void xaddShouldHonorNoMkStream() {
+
+			MapRecord<byte[], byte[], byte[]> record = MapRecord.create("key".getBytes(), Collections.emptyMap());
+
+			connection.streamCommands().xAdd(record, XAddOptions.makeNoStream());
+			ArgumentCaptor<XAddArgs> 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);
+		}
+
+		@Test // GH-2047
+		void xaddShouldHonorNoMkStreamWithMaxLen() {
+
+			MapRecord<byte[], byte[], byte[]> record = MapRecord.create("key".getBytes(), Collections.emptyMap());
+
+			connection.streamCommands().xAdd(record, XAddOptions.makeNoStream(100));
+			ArgumentCaptor<XAddArgs> 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);
+			assertThat(args.getValue()).extracting("maxlen").isEqualTo(100L);
+		}
 	}
 
 	public static class LettucePipelineConnectionUnitTests extends BasicUnitTests {

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

@@ -241,4 +241,39 @@ void xaddShouldHonorMaxlen() {
 
 		assertThat(args.getValue()).extracting("maxlen").isEqualTo(100L);
 	}
+
+	@Test // GH-2047
+	void xaddShouldHonorNoMkStream() {
+
+		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)).makeNoStream()))
+				.subscribe();
+
+		assertThat(args.getValue()).extracting("nomkstream").isEqualTo(true);
+	}
+
+	@Test // GH-2047
+	void xaddShouldHonorNoMkStreamWithMaxLen() {
+
+		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)).makeNoStream(100)))
+				.subscribe();
+
+		assertThat(args.getValue()).extracting("nomkstream").isEqualTo(true);
+		assertThat(args.getValue()).extracting("maxlen").isEqualTo(100L);
+	}
 }