DATAREDIS-1226 - Add support for xClaimJustId using Lettuce.

Original Pull Request: #567

Author: dengliming

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

@@ -229,6 +229,7 @@ default List<ByteRecord> xClaim(byte[] key, String group, String newOwner, Durat
 
 	/**
 	 * @author Christoph Strobl
+	 * @author Dengliming
 	 * @since 2.3
 	 */
 	class XClaimOptions {
@@ -239,16 +240,18 @@ class XClaimOptions {
 		private final @Nullable Instant unixTime;
 		private final @Nullable Long retryCount;
 		private final boolean force;
+		private final boolean justId;
 
 		private XClaimOptions(List<RecordId> ids, Duration minIdleTime, @Nullable Duration idleTime,
-				@Nullable Instant unixTime, @Nullable Long retryCount, boolean force) {
+				@Nullable Instant unixTime, @Nullable Long retryCount, boolean force, boolean justId) {
 
 			this.ids = new ArrayList<>(ids);
 			this.minIdleTime = minIdleTime;
 			this.idleTime = idleTime;
 			this.unixTime = unixTime;
 			this.retryCount = retryCount;
 			this.force = force;
+			this.justId = justId;
 		}
 
 		/**
@@ -281,7 +284,7 @@ public static XClaimOptionsBuilder minIdleMs(long millis) {
 		 * @return {@code this}.
 		 */
 		public XClaimOptions idle(Duration idleTime) {
-			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force);
+			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force, justId);
 		}
 
 		/**
@@ -292,7 +295,7 @@ public XClaimOptions idle(Duration idleTime) {
 		 * @return {@code this}.
 		 */
 		public XClaimOptions time(Instant unixTime) {
-			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force);
+			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force, justId);
 		}
 
 		/**
@@ -302,7 +305,7 @@ public XClaimOptions time(Instant unixTime) {
 		 * @return new instance of {@link XClaimOptions}.
 		 */
 		public XClaimOptions retryCount(long retryCount) {
-			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force);
+			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force, justId);
 		}
 
 		/**
@@ -312,7 +315,16 @@ public XClaimOptions retryCount(long retryCount) {
 		 * @return new instance of {@link XClaimOptions}.
 		 */
 		public XClaimOptions force() {
-			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, true);
+			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, true, justId);
+		}
+
+		/**
+		 * Set the JUSTID flag.
+		 *
+		 * @return new instance of {@link XClaimOptions}.
+		 */
+		public XClaimOptions justId() {
+			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force, true);
 		}
 
 		/**
@@ -381,6 +393,15 @@ public boolean isForce() {
 			return force;
 		}
 
+		/**
+		 * Get the JUSTID flag.
+		 *
+		 * @return
+		 */
+		public boolean isJustId() {
+			return justId;
+		}
+
 		public static class XClaimOptionsBuilder {
 
 			private final Duration minIdleTime;
@@ -404,7 +425,7 @@ public XClaimOptions ids(List<?> ids) {
 						.map(it -> it instanceof RecordId ? (RecordId) it : RecordId.of(it.toString()))
 						.collect(Collectors.toList());
 
-				return new XClaimOptions(idList, minIdleTime, null, null, null, false);
+				return new XClaimOptions(idList, minIdleTime, null, null, null, false, false);
 			}
 
 			/**

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

@@ -123,10 +123,6 @@ public Flux<CommandResponse<AddStreamRecord, RecordId>> xAdd(Publisher<AddStream
 	@Override
 	public Flux<CommandResponse<XClaimCommand, Flux<RecordId>>> xClaimJustId(Publisher<XClaimCommand> commands) {
 
-		if (true /* TODO: set the JUSTID flag */ ) {
-			throw new UnsupportedOperationException("Lettuce does not support XCLAIM with JUSTID. (Ref: lettuce-io#1233)");
-		}
-
 		return connection.execute(cmd -> Flux.from(commands).map(command -> {
 
 			String[] ids = command.getOptions().getIdsAsStringArray();

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

@@ -49,6 +49,7 @@
  * @author Mark Paluch
  * @author Tugdual Grall
  * @author Dejan Jankov
+ * @author Dengliming
  * @since 2.2
  */
 class LettuceStreamCommands implements RedisStreamCommands {
@@ -136,10 +137,6 @@ public List<RecordId> xClaimJustId(byte[] key, String group, String newOwner, XC
 				LettuceConverters.toBytes(newOwner));
 		XClaimArgs args = StreamConverters.toXClaimArgs(options);
 
-		if (true /* TODO: set the JUSTID flag */ ) {
-			throw new UnsupportedOperationException("Lettuce does not support XCLAIM with JUSTID. (Ref: lettuce-io#1233)");
-		}
-
 		try {
 			if (isPipelined()) {
 

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

@@ -236,7 +236,9 @@ public XClaimArgs convert(XClaimOptions source) {
 			if (source.getUnixTime() != null) {
 				args.time(source.getUnixTime());
 			}
-
+			if (source.isJustId()) {
+				args.justid();
+			}
 			return args;
 
 		}

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

@@ -45,6 +45,7 @@
  * @author Mark Paluch
  * @author Christoph Strobl
  * @author Tugdual Grall
+ * @author Dengliming
  */
 public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsTestsBase {
 
@@ -488,4 +489,22 @@ public void xinfoConsumersNoConsumer() {
 		connection.streamCommands().xInfoConsumers(KEY_1_BBUFFER, "my-group").as(StepVerifier::create).verifyComplete();
 	}
 
+	@Test // DATAREDIS-1226
+	public void xClaimJustId() {
+
+		String initialMessage = nativeCommands.xadd(KEY_1, KEY_1, VALUE_1);
+		nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, initialMessage), "my-group");
+
+		String expected = nativeCommands.xadd(KEY_1, KEY_2, VALUE_2);
+
+		connection.streamCommands()
+				.xReadGroup(Consumer.from("my-group", "my-consumer"),
+						StreamOffset.create(KEY_1_BBUFFER, ReadOffset.lastConsumed())) //
+				.delayElements(Duration.ofMillis(5)).next() //
+				.flatMapMany(record -> connection.streamCommands().xClaimJustId(KEY_1_BBUFFER, "my-group", "my-consumer",
+						XClaimOptions.minIdle(Duration.ofMillis(1)).ids(record.getId()).justId())
+				).as(StepVerifier::create) //
+				.assertNext(it -> assertThat(it.getValue()).isEqualTo(expected)) //
+				.verifyComplete();
+	}
 }