Add support for getting Summary info by XPENDING

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

@@ -1666,6 +1666,10 @@ public void xpending(byte[] key, byte[] groupname, byte[] start, byte[] end, int
     }
   }
 
+  public void xpendingSummary(final byte[] key, final byte[] groupname) {
+    sendCommand(XPENDING, key, groupname);
+  }
+
   public void xclaim(byte[] key, byte[] groupname, byte[] consumername, long minIdleTime,
       long newIdleTime, int retries, boolean force, byte[][] ids) {
 

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

@@ -4497,6 +4497,13 @@ public List<Object> xpending(byte[] key, byte[] groupname, byte[] start, byte[]
     return client.getObjectMultiBulkReply();
   }
 
+  @Override
+  public Object xpendingSummary(final byte[] key, final byte[] groupname) {
+    checkIsInMultiOrPipeline();
+    client.xpendingSummary(key, groupname);
+    return client.getOne();
+  }
+
   @Override
   public List<byte[]> xclaim(byte[] key, byte[] groupname, byte[] consumername, long minIdleTime,
       long newIdleTime, int retries, boolean force, byte[]... ids) {

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

@@ -2450,6 +2450,16 @@ public List<Object> execute(Jedis connection) {
     }.runBinary(key);
   }
 
+  @Override
+  public Object xpendingSummary(final byte[] key, final byte[] groupname) {
+    return new JedisClusterCommand<Object>(connectionHandler, maxAttempts) {
+      @Override
+      public Object execute(Jedis connection) {
+        return connection.xpendingSummary(key, groupname);
+      }
+    }.runBinary(key);
+  }
+
   @Override
   public List<byte[]> xclaim(final byte[] key, final byte[] groupname, final byte[] consumername,
       final long minIdleTime, final long newIdleTime, final int retries, final boolean force,

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

@@ -1149,6 +1149,12 @@ public List<Object> xpending(byte[] key, byte[] groupname, byte[] start, byte[]
     return j.xpending(key, groupname, start, end, count, consumername);
   }
 
+  @Override
+  public Object xpendingSummary(final byte[] key, final byte[] groupname) {
+    Jedis j = getShard(key);
+    return j.xpendingSummary(key, groupname);
+  }
+
   @Override
   public List<byte[]> xclaim(byte[] key, byte[] groupname, byte[] consumername, long minIdleTime,
       long newIdleTime, int retries, boolean force, byte[]... ids) {

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

@@ -898,6 +898,32 @@ public String toString() {
     }
   };
 
+  public static final Builder<StreamPendingSummary> STREAM_PENDING_SUMMARY = new Builder<StreamPendingSummary>() {
+    @Override
+    @SuppressWarnings("unchecked")
+    public StreamPendingSummary build(Object data) {
+      if (null == data) {
+        return null;
+      }
+
+      List<Object> objectList = (List<Object>) data;
+      long total = BuilderFactory.LONG.build(objectList.get(0));
+      String minId = SafeEncoder.encode((byte[]) objectList.get(1));
+      String maxId = SafeEncoder.encode((byte[]) objectList.get(2));
+      List<List<Object>> consumerObjList = (List<List<Object>>) objectList.get(3);
+      Map<String, Long> map = new HashMap<>(consumerObjList.size());
+      for (List<Object> consumerObj : consumerObjList) {
+        map.put(SafeEncoder.encode((byte[]) consumerObj.get(0)), Long.parseLong(SafeEncoder.encode((byte[]) consumerObj.get(1))));
+      }
+      return new StreamPendingSummary(total, new StreamEntryID(minId), new StreamEntryID(maxId), map);
+    }
+
+    @Override
+    public String toString() {
+      return "StreamPendingSummary";
+    }
+  };
+
   private static Map<String, Object> createMapFromDecodingFunctions(Iterator<Object> iterator,
       Map<String, Builder> mappingFunctions) {
 

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

@@ -1414,6 +1414,11 @@ public void xpending(String key, String groupname, StreamEntryID start, StreamEn
         SafeEncoder.encode(end==null ? "+" : end.toString()), count, consumername == null? null : SafeEncoder.encode(consumername));
   }
 
+  @Override
+  public void xpendingSummary(String key, String groupname) {
+    xpendingSummary(SafeEncoder.encode(key), SafeEncoder.encode(groupname));
+  }
+
   @Override
   public void xclaim(String key, String group, String consumername, long minIdleTime,
       long newIdleTime, int retries, boolean force, StreamEntryID... ids) {

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

@@ -4117,6 +4117,13 @@ public List<StreamPendingEntry> xpending(final String key, final String groupnam
     return BuilderFactory.STREAM_PENDING_ENTRY_LIST.build(client.getObjectMultiBulkReply());
   }
 
+  @Override
+  public StreamPendingSummary xpendingSummary(final String key, final String groupname) {
+    checkIsInMultiOrPipeline();
+    client.xpendingSummary(key, groupname);
+    return BuilderFactory.STREAM_PENDING_SUMMARY.build(client.getObjectMultiBulkReply());
+  }
+
   @Override
   public List<StreamEntry> xclaim(String key, String group, String consumername, long minIdleTime,
       long newIdleTime, int retries, boolean force, StreamEntryID... ids) {

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

@@ -2522,6 +2522,16 @@ public List<StreamPendingEntry> execute(Jedis connection) {
     }.run(key);
   }
 
+  @Override
+  public StreamPendingSummary xpendingSummary(final String key, final String groupname) {
+    return new JedisClusterCommand<StreamPendingSummary>(connectionHandler, maxAttempts) {
+      @Override
+      public StreamPendingSummary execute(Jedis connection) {
+        return connection.xpendingSummary(key, groupname);
+      }
+    }.run(key);
+  }
+
   @Override
   public Long xdel(final String key, final StreamEntryID... ids) {
     return new JedisClusterCommand<Long>(connectionHandler, maxAttempts) {

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

@@ -1145,6 +1145,12 @@ public List<StreamPendingEntry> xpending(String key, String groupname, StreamEnt
     return j.xpending(key, groupname, start, end, count, consumername);
   }
 
+  @Override
+  public StreamPendingSummary xpendingSummary(String key, String groupname) {
+    Jedis j = getShard(key);
+    return j.xpendingSummary(key, groupname);
+  }
+
   @Override
   public List<StreamEntry> xclaim(String key, String group, String consumername, long minIdleTime,
       long newIdleTime, int retries, boolean force, StreamEntryID... ids) {

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

@@ -0,0 +1,38 @@
+package redis.clients.jedis;
+
+import java.io.Serializable;
+import java.util.Map;
+
+public class StreamPendingSummary implements Serializable {
+
+  private static final long serialVersionUID = 1L;
+
+  private final long total;
+  private final StreamEntryID minId;
+  private final StreamEntryID maxId;
+  private final Map<String, Long> consumerMessageCount;
+
+  public StreamPendingSummary(long total, StreamEntryID minId, StreamEntryID maxId,
+      Map<String, Long> consumerMessageCount) {
+    this.total = total;
+    this.minId = minId;
+    this.maxId = maxId;
+    this.consumerMessageCount = consumerMessageCount;
+  }
+
+  public long getTotal() {
+    return total;
+  }
+
+  public StreamEntryID getMinId() {
+    return minId;
+  }
+
+  public StreamEntryID getMaxId() {
+    return maxId;
+  }
+
+  public Map<String, Long> getConsumerMessageCount() {
+    return consumerMessageCount;
+  }
+}

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

@@ -394,6 +394,8 @@ List<GeoRadiusResponse> georadiusByMemberReadonly(byte[] key, byte[] member, dou
 
   List<Object> xpending(byte[] key, byte[] groupname, byte[] start, byte[] end, int count, byte[] consumername);
 
+  Object xpendingSummary(final byte[] key, final byte[] groupname);
+
   List<byte[]> xclaim(byte[] key, byte[] groupname, byte[] consumername, long minIdleTime, long newIdleTime, int retries, boolean force, byte[][] ids);
 
   Long waitReplicas(byte[] key, int replicas, long timeout);

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

@@ -390,8 +390,8 @@ List<GeoRadiusResponse> georadiusByMemberReadonly(byte[] key, byte[] member, dou
    * @return lenth of the value for key
    */
   Long hstrlen(byte[] key, byte[] field);
-  
-  
+
+
   byte[] xadd(byte[] key, byte[] id, Map<byte[], byte[]> hash, long maxLen, boolean approximateLength);
 
   Long xlen(byte[] key);
@@ -409,21 +409,23 @@ default List<byte[]> xrange(byte[] key, byte[] start, byte[] end, long count) {
   List<byte[]> xrevrange(byte[] key, byte[] end, byte[] start, int count);
 
   Long xack(byte[] key, byte[] group, byte[]... ids);
- 
+
   String xgroupCreate(byte[] key, byte[] consumer, byte[] id, boolean makeStream);
 
   String xgroupSetID(byte[] key, byte[] consumer, byte[] id);
 
   Long xgroupDestroy(byte[] key, byte[] consumer);
 
   Long xgroupDelConsumer(byte[] key, byte[] consumer, byte[] consumerName);
- 
+
   Long xdel(byte[] key, byte[]... ids);
 
   Long xtrim(byte[] key, long maxLen, boolean approximateLength);
 
   List<Object> xpending(byte[] key, byte[] groupname, byte[] start, byte[] end, int count, byte[] consumername);
 
+  Object xpendingSummary(byte[] key, byte[] groupname);
+
   List<byte[]> xclaim(byte[] key, byte[] groupname, byte[] consumername, long minIdleTime, long newIdleTime, int retries, boolean force, byte[]... ids);
 
   /**

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

@@ -450,6 +450,8 @@ default void restoreReplace(String key, int ttl, byte[] serializedValue) {
 
   void xpending(String key, String groupname, StreamEntryID start, StreamEntryID end, int count, String consumername);
 
+  void xpendingSummary(String key, String groupname);
+
   void xclaim(String key, String group, String consumername, long minIdleTime, long newIdleTime,
       int retries, boolean force, StreamEntryID... ids);
 

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

@@ -9,6 +9,7 @@
 import redis.clients.jedis.ScanResult;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.StreamEntry;
+import redis.clients.jedis.StreamPendingSummary;
 import redis.clients.jedis.Tuple;
 import redis.clients.jedis.params.GeoRadiusParam;
 import redis.clients.jedis.params.SetParams;
@@ -517,6 +518,15 @@ List<GeoRadiusResponse> georadiusByMemberReadonly(String key, String member, dou
    */
   List<StreamPendingEntry> xpending(String key, String groupname, StreamEntryID start, StreamEntryID end, int count, String consumername);
 
+  /**
+   * XPENDING key group
+   *
+   * @param key
+   * @param groupname
+   * @return
+   */
+  StreamPendingSummary xpendingSummary(String key, String groupname);
+
   /**
    * XDEL key ID [ID ...]
    * @param key

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

@@ -18,6 +18,7 @@
 import redis.clients.jedis.ScanResult;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.StreamEntry;
+import redis.clients.jedis.StreamPendingSummary;
 import redis.clients.jedis.Tuple;
 import redis.clients.jedis.params.GeoRadiusParam;
 import redis.clients.jedis.params.SetParams;
@@ -518,6 +519,15 @@ List<GeoRadiusResponse> georadiusByMemberReadonly(String key, String member, dou
    */
   List<StreamPendingEntry> xpending(String key, String groupname, StreamEntryID start, StreamEntryID end, int count, String consumername);
 
+  /**
+   * XPENDING key group
+   *
+   * @param key
+   * @param groupname
+   * @return
+   */
+  StreamPendingSummary xpendingSummary(String key, String groupname);
+
   /**
    * XDEL key ID [ID ...]
    * @param key

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

@@ -329,6 +329,12 @@ public void xpendeing() {
     assertEquals(1, range.get(0).getValue().size());
     assertEquals(map, range.get(0).getValue().get(0).getFields());
 
+    // Get the summary about the pending messages
+    StreamPendingSummary pendingSummary = jedis.xpendingSummary("xpendeing-stream", "xpendeing-group");
+    assertEquals(1, pendingSummary.getTotal());
+    assertEquals(id1, pendingSummary.getMinId());
+    assertEquals(1l, pendingSummary.getConsumerMessageCount().get("xpendeing-consumer").longValue());
+
     // Get the pending event
     List<StreamPendingEntry> pendingRange = jedis.xpending("xpendeing-stream", "xpendeing-group",
       null, null, 3, "xpendeing-consumer");