Support exclusive XPENDING range from params

src/main/java/redis/clients/jedis/CommandObjects.java

@@ -2228,6 +2228,10 @@ public final CommandObject<StreamPendingSummary> xpending(String key, String gro
         BuilderFactory.STREAM_PENDING_SUMMARY);
   }
 
+  /**
+   * @deprecated Use {@link CommandObjects#xpending(java.lang.String, java.lang.String, redis.clients.jedis.params.XPendingParams)}.
+   */
+  @Deprecated
   public final CommandObject<List<StreamPendingEntry>> xpending(String key, String groupname,
       StreamEntryID start, StreamEntryID end, int count, String consumername) {
     CommandArguments args = commandArguments(XPENDING).key(key).add(groupname)
@@ -2246,6 +2250,10 @@ public final CommandObject<Object> xpending(byte[] key, byte[] groupname) {
         BuilderFactory.RAW_OBJECT);
   }
 
+  /**
+   * @deprecated Use {@link CommandObjects#xpending(byte[], byte[], redis.clients.jedis.params.XPendingParams)}.
+   */
+  @Deprecated
   public final CommandObject<List<Object>> xpending(byte[] key, byte[] groupname,
       byte[] start, byte[] end, int count, byte[] consumername) {
     CommandArguments args = commandArguments(XPENDING).key(key).add(groupname)

src/main/java/redis/clients/jedis/Jedis.java

@@ -4566,7 +4566,11 @@ public long xtrim(byte[] key, XTrimParams params) {
     return connection.executeCommand(commandObjects.xtrim(key, params));
   }
 
+  /**
+   * @deprecated Use {@link Jedis#xpending(byte[], byte[], redis.clients.jedis.params.XPendingParams)}.
+   */
   @Override
+  @Deprecated
   public List<Object> xpending(byte[] key, byte[] groupname, byte[] start, byte[] end, int count,
       byte[] consumername) {
     checkIsInMultiOrPipeline();
@@ -8780,7 +8784,11 @@ public StreamPendingSummary xpending(final String key, final String groupname) {
     return connection.executeCommand(commandObjects.xpending(key, groupname));
   }
 
+  /**
+   * @deprecated Use {@link Jedis#xpending(java.lang.String, java.lang.String, redis.clients.jedis.params.XPendingParams)}.
+   */
   @Override
+  @Deprecated
   public List<StreamPendingEntry> xpending(final String key, final String groupname,
       final StreamEntryID start, final StreamEntryID end, final int count, final String consumername) {
     checkIsInMultiOrPipeline();

src/main/java/redis/clients/jedis/commands/StreamBinaryCommands.java

@@ -4,7 +4,6 @@
 import java.util.Map;
 
 import redis.clients.jedis.params.*;
-import redis.clients.jedis.resps.StreamFullInfo;
 
 public interface StreamBinaryCommands {
 
@@ -42,6 +41,10 @@ default byte[] xadd(byte[] key, Map<byte[], byte[]> hash, XAddParams params) {
 
   Object xpending(byte[] key, byte[] groupname);
 
+  /**
+   * @deprecated Use {@link StreamBinaryCommands#xpending(byte[], byte[], redis.clients.jedis.params.XPendingParams)}.
+   */
+  @Deprecated
   List<Object> xpending(byte[] key, byte[] groupname, byte[] start, byte[] end, int count, byte[] consumername);
 
   List<Object> xpending(byte[] key, byte[] groupname, XPendingParams params);

src/main/java/redis/clients/jedis/commands/StreamCommands.java

@@ -153,7 +153,9 @@ default StreamEntryID xadd(String key, Map<String, String> hash, XAddParams para
    * @param end
    * @param count
    * @param consumername
+   * @deprecated Use {@link StreamCommands#xpending(java.lang.String, java.lang.String, redis.clients.jedis.params.XPendingParams)}.
    */
+  @Deprecated
   List<StreamPendingEntry> xpending(String key, String groupname, StreamEntryID start,
       StreamEntryID end, int count, String consumername);
 

src/main/java/redis/clients/jedis/commands/StreamPipelineBinaryCommands.java

@@ -5,7 +5,6 @@
 
 import redis.clients.jedis.Response;
 import redis.clients.jedis.params.*;
-import redis.clients.jedis.resps.StreamFullInfo;
 
 public interface StreamPipelineBinaryCommands {
 //
@@ -45,6 +44,10 @@ default Response<byte[]> xadd(byte[] key, Map<byte[], byte[]> hash, XAddParams p
 
   Response<Object> xpending(byte[] key, byte[] groupname);
 
+  /**
+   * @deprecated Use {@link StreamPipelineBinaryCommands#xpending(byte[], byte[], redis.clients.jedis.params.XPendingParams)}.
+   */
+  @Deprecated
   Response<List<Object>> xpending(byte[] key, byte[] groupname, byte[] start, byte[] end, int count, byte[] consumername);
 
   Response<List<Object>> xpending(byte[] key, byte[] groupname, XPendingParams params);

src/main/java/redis/clients/jedis/commands/StreamPipelineCommands.java

@@ -154,7 +154,9 @@ default Response<StreamEntryID> xadd(String key, Map<String, String> hash, XAddP
    * @param end
    * @param count
    * @param consumername
+   * @deprecated Use {@link StreamPipelineCommands#xpending(java.lang.String, java.lang.String, redis.clients.jedis.params.XPendingParams)}.
    */
+  @Deprecated
   Response<List<StreamPendingEntry>> xpending(String key, String groupname, StreamEntryID start,
       StreamEntryID end, int count, String consumername);
 

src/main/java/redis/clients/jedis/params/XPendingParams.java

@@ -1,40 +1,82 @@
 package redis.clients.jedis.params;
 
 import static redis.clients.jedis.Protocol.Keyword.IDLE;
+import static redis.clients.jedis.Protocol.toByteArray;
+import static redis.clients.jedis.args.RawableFactory.from;
 
 import redis.clients.jedis.CommandArguments;
-import redis.clients.jedis.Protocol;
 import redis.clients.jedis.StreamEntryID;
-import redis.clients.jedis.util.SafeEncoder;
+import redis.clients.jedis.args.Rawable;
 
 public class XPendingParams implements IParams {
 
+  private boolean legacy = true;
   private Long idle;
+  private Rawable start; // TODO: final
+  private Rawable end; // TODO: final
+  private int count = Integer.MIN_VALUE; // TODO: final
+  private Rawable consumer;
+
+  /**
+   * @deprecated Use {@link XPendingParams#XPendingParams(redis.clients.jedis.StreamEntryID, redis.clients.jedis.StreamEntryID, int)}.
+   */
+  @Deprecated
+  public XPendingParams() {
+  }
 
-  private String consumer;
+  /**
+   * @deprecated Use {@link XPendingParams#xPendingParams(redis.clients.jedis.StreamEntryID, redis.clients.jedis.StreamEntryID, int)}.
+   */
+  @Deprecated
+  public static XPendingParams xPendingParams() {
+    return new XPendingParams();
+  }
 
-  private StreamEntryID start;
+  public XPendingParams(StreamEntryID start, StreamEntryID end, int count) {
+    this(start.toString(), end.toString(), count);
+  }
+
+  public XPendingParams(String start, String end, int count) {
+    this(from(start), from(end), count);
+  }
 
-  private StreamEntryID end;
+  public XPendingParams(byte[] start, byte[] end, int count) {
+    this(from(start), from(end), count);
+  }
 
-  private Integer count;
+  private XPendingParams(Rawable start, Rawable end, int count) {
+    this.legacy = false;
+    this.start = start;
+    this.end = end;
+    this.count = count;
+  }
 
-  public static XPendingParams xPendingParams() {
-    return new XPendingParams();
+  public static XPendingParams xPendingParams(StreamEntryID start, StreamEntryID end, int count) {
+    return new XPendingParams(start, end, count);
+  }
+
+  public static XPendingParams xPendingParams(String start, String end, int count) {
+    return new XPendingParams(start, end, count);
+  }
+
+  public static XPendingParams xPendingParams(byte[] start, byte[] end, int count) {
+    return new XPendingParams(start, end, count);
   }
 
   public XPendingParams idle(long idle) {
     this.idle = idle;
     return this;
   }
 
+  @Deprecated
   public XPendingParams start(StreamEntryID start) {
-    this.start = start;
+    this.start = from(start.toString());
     return this;
   }
 
+  @Deprecated
   public XPendingParams end(StreamEntryID end) {
-    this.end = end;
+    this.end = from(end.toString());
     return this;
   }
 
@@ -44,36 +86,44 @@ public XPendingParams count(int count) {
   }
 
   public XPendingParams consumer(String consumer) {
-    this.consumer = consumer;
+    this.consumer = from(consumer);
+    return this;
+  }
+
+  public XPendingParams consumer(byte[] consumer) {
+    this.consumer = from(consumer);
     return this;
   }
 
   @Override
   public void addParams(CommandArguments args) {
 
     if (idle != null) {
-      args.add(IDLE.getRaw());
-      args.add(Protocol.toByteArray(idle));
+      args.add(IDLE).add(toByteArray(idle));
     }
 
-    if (start == null) {
-      args.add(SafeEncoder.encode("-"));
+    if (legacy) {
+      if (start == null) {
+        args.add("-");
+      } else {
+        args.add(start);
+      }
+
+      if (end == null) {
+        args.add("+");
+      } else {
+        args.add(end);
+      }
+
+      if (count != Integer.MIN_VALUE) {
+        args.add(toByteArray(count));
+      }
     } else {
-      args.add(SafeEncoder.encode(start.toString()));
-    }
-
-    if (end == null) {
-      args.add(SafeEncoder.encode("+"));
-    } else {
-      args.add(SafeEncoder.encode(end.toString()));
-    }
-
-    if (count != null) {
-      args.add(Protocol.toByteArray(count));
+      args.add(start).add(end).add(toByteArray(count));
     }
 
     if (consumer != null) {
-      args.add(SafeEncoder.encode(consumer));
+      args.add(consumer);
     }
   }
 }

src/test/java/redis/clients/jedis/commands/jedis/StreamsCommandsTest.java

@@ -423,7 +423,6 @@ public void xreadGroupWithParams() {
 
   @Test
   public void xack() {
-
     Map<String, String> map = new HashMap<String, String>();
     map.put("f1", "v1");
     jedis.xadd("xack-stream", (StreamEntryID) null, map);
@@ -482,6 +481,29 @@ public void xpendingWithParams() {
     assertEquals(0, pendingRange.size());
   }
 
+  @Test
+  public void xpendingRange() {
+    Map<String, String> map = new HashMap<>();
+    map.put("foo", "bar");
+    StreamEntryID m1 = jedis.xadd("xpendeing-stream", (StreamEntryID) null, map);
+    StreamEntryID m2 = jedis.xadd("xpendeing-stream", (StreamEntryID) null, map);
+    jedis.xgroupCreate("xpendeing-stream", "xpendeing-group", null, false);
+
+    // read 1 message from the group with each consumer
+    Map<String, StreamEntryID> streamQeury = Collections.singletonMap(
+        "xpendeing-stream", StreamEntryID.UNRECEIVED_ENTRY);
+    jedis.xreadGroup("xpendeing-group", "consumer1", XReadGroupParams.xReadGroupParams().count(1), streamQeury);
+    jedis.xreadGroup("xpendeing-group", "consumer2", XReadGroupParams.xReadGroupParams().count(1), streamQeury);
+
+    List<StreamPendingEntry> response = jedis.xpending("xpendeing-stream", "xpendeing-group",
+        XPendingParams.xPendingParams("(0", "+", 5));
+    assertEquals(2, response.size());
+    assertEquals(m1, response.get(0).getID());
+    assertEquals("consumer1", response.get(0).getConsumerName());
+    assertEquals(m2, response.get(1).getID());
+    assertEquals("consumer2", response.get(1).getConsumerName());
+  }
+
   @Test
   public void xclaimWithParams() {
     Map<String, String> map = new HashMap<>();