feat(impv): Add idle argument to xPending

Author: whatasame

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

@@ -41,6 +41,7 @@
  * @author Tugdual Grall
  * @author Dengliming
  * @author Mark John Moreno
+ * @author Jeonggyu Choi
  * @see <a href="https://redis.io/topics/streams-intro">Redis Documentation - Streams</a>
  * @since 2.2
  */
@@ -706,6 +707,25 @@ default PendingMessages xPending(byte[] key, String groupName, Range<?> range, L
 		return xPending(key, groupName, XPendingOptions.range(range, count));
 	}
 
+	/**
+	 * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} within a
+	 * {@literal consumer group} and over a given {@link Duration} of idle time.
+	 *
+	 * @param key the {@literal key} the stream is stored at. Must not be {@literal null}.
+	 * @param groupName the name of the {@literal consumer group}. Must not be {@literal null}.
+	 * @param range the range of messages ids to search within. Must not be {@literal null}.
+	 * @param count limit the number of results. Must not be {@literal null}.
+	 * @param idle the minimum idle time to filter pending messages. Must not be {@literal null}.
+	 * @return pending messages for the given {@literal consumer group} or {@literal null} when used in pipeline /
+	 *         transaction.
+	 * @see <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+	 * @since 3.5
+	 */
+	@Nullable
+	default PendingMessages xPending(byte[] key, String groupName, Range<?> range, Long count, Duration idle) {
+		return xPending(key, groupName, XPendingOptions.range(range, count).idle(idle));
+	}
+
 	/**
 	 * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and
 	 * {@link Consumer} within a {@literal consumer group}.
@@ -723,6 +743,24 @@ default PendingMessages xPending(byte[] key, Consumer consumer, Range<?> range,
 		return xPending(key, consumer.getGroup(), consumer.getName(), range, count);
 	}
 
+	/**
+	 * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and
+	 * {@link Consumer} within a {@literal consumer group} and over a given {@link Duration} of idle time.
+	 *
+	 * @param key the {@literal key} the stream is stored at. Must not be {@literal null}.
+	 * @param consumer the name of the {@link Consumer}. Must not be {@literal null}.
+	 * @param range the range of messages ids to search within. Must not be {@literal null}.
+	 * @param count limit the number of results. Must not be {@literal null}.
+	 * @param idle the minimum idle time to filter pending messages. Must not be {@literal null}.
+	 * @return pending messages for the given {@link Consumer} or {@literal null} when used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+	 * @since 3.5
+	 */
+	@Nullable
+	default PendingMessages xPending(byte[] key, Consumer consumer, Range<?> range, Long count, Duration idle) {
+		return xPending(key, consumer.getGroup(), consumer.getName(), range, count, idle);
+	}
+
 	/**
 	 * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and
 	 * {@literal consumer} within a {@literal consumer group}.
@@ -742,6 +780,27 @@ default PendingMessages xPending(byte[] key, String groupName, String consumerNa
 		return xPending(key, groupName, XPendingOptions.range(range, count).consumer(consumerName));
 	}
 
+	/**
+	 * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and
+	 * {@literal consumer} within a {@literal consumer group} and over a given {@link Duration} of idle time.
+	 *
+	 * @param key the {@literal key} the stream is stored at. Must not be {@literal null}.
+	 * @param groupName the name of the {@literal consumer group}. Must not be {@literal null}.
+	 * @param consumerName the name of the {@literal consumer}. Must not be {@literal null}.
+	 * @param range the range of messages ids to search within. Must not be {@literal null}.
+	 * @param count limit the number of results. Must not be {@literal null}.
+	 * @param idle the minimum idle time to filter pending messages. Must not be {@literal null}.
+	 * @return pending messages for the given {@literal consumer} in given {@literal consumer group} or {@literal null}
+	 *         when used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+	 * @since 3.5
+	 */
+	@Nullable
+	default PendingMessages xPending(byte[] key, String groupName, String consumerName, Range<?> range, Long count,
+			Duration idle) {
+		return xPending(key, groupName, XPendingOptions.range(range, count).consumer(consumerName).idle(idle));
+	}
+
 	/**
 	 * Obtain detailed information about pending {@link PendingMessage messages} applying given {@link XPendingOptions
 	 * options}.
@@ -761,19 +820,23 @@ default PendingMessages xPending(byte[] key, String groupName, String consumerNa
 	 * Value Object holding parameters for obtaining pending messages.
 	 *
 	 * @author Christoph Strobl
+	 * @author Jeonggyu Choi
 	 * @since 2.3
 	 */
 	class XPendingOptions {
 
 		private final @Nullable String consumerName;
 		private final Range<?> range;
 		private final @Nullable Long count;
+		private final @Nullable Duration idle;
 
-		private XPendingOptions(@Nullable String consumerName, Range<?> range, @Nullable Long count) {
+		private XPendingOptions(@Nullable String consumerName, Range<?> range, @Nullable Long count,
+				@Nullable Duration idle) {
 
 			this.range = range;
 			this.count = count;
 			this.consumerName = consumerName;
+			this.idle = idle;
 		}
 
 		/**
@@ -782,7 +845,7 @@ private XPendingOptions(@Nullable String consumerName, Range<?> range, @Nullable
 		 * @return new instance of {@link XPendingOptions}.
 		 */
 		public static XPendingOptions unbounded() {
-			return new XPendingOptions(null, Range.unbounded(), null);
+			return new XPendingOptions(null, Range.unbounded(), null, null);
 		}
 
 		/**
@@ -795,7 +858,7 @@ public static XPendingOptions unbounded(Long count) {
 
 			Assert.isTrue(count > -1, "Count must not be negative");
 
-			return new XPendingOptions(null, Range.unbounded(), count);
+			return new XPendingOptions(null, Range.unbounded(), count, null);
 		}
 
 		/**
@@ -810,7 +873,7 @@ public static XPendingOptions range(Range<?> range, Long count) {
 			Assert.notNull(range, "Range must not be null");
 			Assert.isTrue(count > -1, "Count must not be negative");
 
-			return new XPendingOptions(null, range, count);
+			return new XPendingOptions(null, range, count, null);
 		}
 
 		/**
@@ -820,7 +883,20 @@ public static XPendingOptions range(Range<?> range, Long count) {
 		 * @return new instance of {@link XPendingOptions}.
 		 */
 		public XPendingOptions consumer(String consumerName) {
-			return new XPendingOptions(consumerName, range, count);
+			return new XPendingOptions(consumerName, range, count, idle);
+		}
+
+		/**
+		 * Append given idle time.
+		 *
+		 * @param idle must not be {@literal null}.
+		 * @return new instance of {@link} XPendingOptions}.
+		 */
+		public XPendingOptions idle(Duration idle) {
+
+			Assert.notNull(idle, "Idle must not be null");
+
+			return new XPendingOptions(consumerName, range, count, idle);
 		}
 
 		/**
@@ -846,6 +922,20 @@ public String getConsumerName() {
 			return consumerName;
 		}
 
+		@Nullable
+		public Duration getIdle() {
+			return idle;
+		}
+
+		@Nullable
+		public Long getIdleMillis() {
+			if (idle == null) {
+				return null;
+			}
+
+			return idle.toMillis();
+		}
+
 		/**
 		 * @return {@literal true} if a consumer name is present.
 		 */
@@ -859,6 +949,13 @@ public boolean hasConsumer() {
 		public boolean isLimited() {
 			return count != null;
 		}
+
+		/**
+		 * @return {@literal true} if idle time is set.
+		 */
+		public boolean hasIdle() {
+			return idle != null;
+		}
 	}
 
 	/**

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

@@ -46,4 +46,11 @@ void xPendingOptionsRangeShouldThrowExceptionWhenCountIsNegative() {
 
 		assertThatIllegalArgumentException().isThrownBy(() -> XPendingOptions.range(range, -1L));
 	}
+
+	@Test // GH-2046
+	void xPendingOptionsIdleShouldThrowExceptionWhenIdleIsNull() {
+		XPendingOptions xPendingOptions = XPendingOptions.unbounded();
+
+		assertThatIllegalArgumentException().isThrownBy(() -> xPendingOptions.idle(null));
+	}
 }