Xinfo command

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

@@ -1,5 +1,8 @@
 package redis.clients.jedis;
 
+import static redis.clients.jedis.Protocol.CONSUMERS;
+import static redis.clients.jedis.Protocol.GROUPS;
+import static redis.clients.jedis.Protocol.STREAM;
 import static redis.clients.jedis.Protocol.toByteArray;
 import static redis.clients.jedis.Protocol.Command.*;
 import static redis.clients.jedis.Protocol.Keyword.ENCODING;
@@ -1455,4 +1458,21 @@ public void xclaim(byte[] key, byte[] groupname, byte[] consumername, long minId
       sendCommand(XCLAIM, arguments.toArray(new byte[arguments.size()][]));
   }
 
+  public void xinfoStream(byte[] key) {
+
+    sendCommand(XINFO,SafeEncoder.encode(STREAM),key);
+
+  }
+
+  public void xinfoGroup(byte[] key) {
+
+    sendCommand(XINFO,SafeEncoder.encode(GROUPS),key);
+
+  }
+
+  public void xinfoConsumers (byte[] key, byte[] group) {
+
+    sendCommand(XINFO,SafeEncoder.encode(CONSUMERS),key,group);
+  }
+
 }

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

@@ -4051,6 +4051,30 @@ public Object sendCommand(ProtocolCommand cmd, byte[]... args) {
     return client.getOne();
   }
 
+  @Override
+  public byte[] xinfoStream(byte[] key) {
+    checkIsInMultiOrPipeline();
+    client.xinfoStream(key);
+
+    return client.getBinaryBulkReply();
+
+  }
+
+  @Override
+  public List<byte[]> xinfoGroup (byte[] key) {
+    checkIsInMultiOrPipeline();
+    client.xinfoGroup(key);
+
+    return client.getBinaryMultiBulkReply();
+  }
+  @Override
+  public List<byte[]> xinfoConsumers (byte[] key, byte[] group) {
+    checkIsInMultiOrPipeline();
+    client.xinfoConsumers(key,group);
+
+    return client.getBinaryMultiBulkReply();
+  }
+
   public Object sendCommand(ProtocolCommand cmd) {
     return sendCommand(cmd, dummyArray);
   }

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

@@ -1039,6 +1039,24 @@ public List<byte[]> xclaim(byte[] key, byte[] groupname, byte[] consumername, lo
     return j.xclaim(key, groupname, consumername, minIdleTime, newIdleTime, retries, force, ids);
   }
 
+  @Override
+  public byte[] xinfoStream(byte[] key) {
+    Jedis j = getShard(key);
+    return j.xinfoStream(key);
+  }
+
+  @Override
+  public List<byte[]> xinfoGroup(byte[] key) {
+    Jedis j = getShard(key);
+    return j.xinfoGroup(key);
+  }
+
+  @Override
+  public List<byte[]> xinfoConsumers(byte[] key, byte[] group) {
+    Jedis j = getShard(key);
+    return j.xinfoConsumers(key, group);
+  }
+
   public Object sendCommand(ProtocolCommand cmd, byte[]... args) {
     // default since no sample key provided in JedisCommands interface
     byte[] sampleKey = args.length > 0 ? args[0] : cmd.getRaw();

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

@@ -514,7 +514,7 @@ public  List<StreamEntry> build(Object data) {
         Iterator<byte[]> hashIterator = hash.iterator();
         Map<String, String> map = new HashMap<>(hash.size()/2);
         while(hashIterator.hasNext()) {
-          map.put(SafeEncoder.encode((byte[])hashIterator.next()), SafeEncoder.encode((byte[])hashIterator.next()));
+          map.put(SafeEncoder.encode(hashIterator.next()), SafeEncoder.encode(hashIterator.next()));
         }
         responses.add(new StreamEntry(entryID, map));
       }
@@ -527,6 +527,41 @@ public String toString() {
       return "List<StreamEntry>";
     }
   };
+
+  public static final Builder<StreamEntry> STREAM_ENTRY = new Builder<StreamEntry>() {
+    @Override
+    @SuppressWarnings("unchecked")
+    public  StreamEntry build(Object data) {
+      if (null == data) {
+        return null;
+      }
+      List<Object> objectList = (List<Object>) data;
+
+      if (objectList.isEmpty()) {
+        return null;
+      }
+
+      String entryIdString = SafeEncoder.encode((byte[]) objectList.get(0));
+      StreamEntryID entryID = new StreamEntryID(entryIdString);
+      List<byte[]> hash = (List<byte[]>) objectList.get(1);
+
+      Iterator<byte[]> hashIterator = hash.iterator();
+      Map<String, String> map = new HashMap<>(hash.size() / 2);
+      while (hashIterator.hasNext()) {
+        map.put(SafeEncoder.encode(hashIterator.next()),
+            SafeEncoder.encode(hashIterator.next()));
+      }
+      StreamEntry streamEntry = new StreamEntry(entryID, map);
+
+
+      return streamEntry;
+    }
+
+    @Override
+    public String toString() {
+      return "StreamEntry";
+    }
+  };
 
   public static final Builder<List<StreamPendingEntry>> STREAM_PENDING_ENTRY_LIST = new Builder<List<StreamPendingEntry>>() {
     @Override
@@ -555,6 +590,139 @@ public String toString() {
     }
   };
 
+  public static final Builder<StreamInfo> STREAM_INFO = new Builder<StreamInfo>() {
+
+
+    Map<String,Builder> mappingFunctions = createDecoderMap();
+
+    private Map<String, Builder> createDecoderMap() {
+
+      Map<String,Builder> tempMappingFunctions = new HashMap<>();
+      tempMappingFunctions.put(StreamInfo.LAST_GENERATED_ID,STRING);
+      tempMappingFunctions.put(StreamInfo.FIRST_ENTRY,STREAM_ENTRY);
+      tempMappingFunctions.put(StreamInfo.LENGTH, LONG);
+      tempMappingFunctions.put(StreamInfo.RADIX_TREE_KEYS, LONG);
+      tempMappingFunctions.put(StreamInfo.RADIX_TREE_NODES, LONG);
+      tempMappingFunctions.put(StreamInfo.LAST_ENTRY,STREAM_ENTRY);
+      tempMappingFunctions.put(StreamInfo.GROUPS, LONG);
+
+      return  tempMappingFunctions;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public StreamInfo build(Object data) {
+      if (null == data) {
+        return null;
+      }
+
+      List<Object> streamsEntries = (List<Object>)data;
+      Iterator<Object> iterator = streamsEntries.iterator();
+
+      StreamInfo streamInfo = new StreamInfo(
+          createMapFromDecodingFunctions(iterator,mappingFunctions));
+      return streamInfo;
+    }
+
+    @Override
+    public String toString() {
+      return "StreamInfo";
+    }
+  };
+
+  public static final Builder<List<StreamGroupInfo>> STREAM_GROUP_INFO_LIST = new Builder<List<StreamGroupInfo>>() {
+
+    Map<String,Builder> mappingFunctions = createDecoderMap();
+
+    private Map<String, Builder> createDecoderMap() {
+
+      Map<String,Builder> tempMappingFunctions = new HashMap<>();
+      tempMappingFunctions.put(StreamGroupInfo.NAME,STRING);
+      tempMappingFunctions.put(StreamGroupInfo.CONSUMERS, LONG);
+      tempMappingFunctions.put(StreamGroupInfo.PENDING, LONG);
+      tempMappingFunctions.put(StreamGroupInfo.LAST_DELIVERED,STRING);
+
+      return  tempMappingFunctions;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public  List<StreamGroupInfo> build(Object data) {
+      if (null == data) {
+        return null;
+      }
+
+      List<StreamGroupInfo> list = new ArrayList<>();
+      List<Object> streamsEntries = (List<Object>)data;
+      Iterator<Object> groupsArray = streamsEntries.iterator();
+
+      while (groupsArray.hasNext()) {
+
+        List<Object> groupInfo = (List<Object>) groupsArray.next();
+
+        Iterator<Object> groupInfoIterator = groupInfo.iterator();
+
+        StreamGroupInfo streamGroupInfo = new StreamGroupInfo(
+            createMapFromDecodingFunctions(groupInfoIterator,mappingFunctions));
+        list.add(streamGroupInfo);
+
+      }
+      return list;
+
+    }
+
+    @Override
+    public String toString() {
+      return "List<StreamGroupInfo>";
+    }
+  };
+
+  public static final Builder<List<StreamConsumersInfo>> STREAM_CONSUMERS_INFO_LIST = new Builder<List<StreamConsumersInfo>>() {
+
+    Map<String,Builder> mappingFunctions = createDecoderMap();
+
+    private Map<String, Builder> createDecoderMap() {
+      Map<String,Builder> tempMappingFunctions = new HashMap<>();
+      tempMappingFunctions.put(StreamConsumersInfo.NAME,STRING);
+      tempMappingFunctions.put(StreamConsumersInfo.IDLE, LONG);
+      tempMappingFunctions.put(StreamGroupInfo.PENDING, LONG);
+      tempMappingFunctions.put(StreamGroupInfo.LAST_DELIVERED,STRING);
+      return tempMappingFunctions;
+
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public  List<StreamConsumersInfo> build(Object data) {
+      if (null == data) {
+        return null;
+      }
+
+      List<StreamConsumersInfo> list = new ArrayList<>();
+      List<Object> streamsEntries = (List<Object>)data;
+      Iterator<Object> groupsArray = streamsEntries.iterator();
+
+      while (groupsArray.hasNext()) {
+
+        List<Object> groupInfo = (List<Object>) groupsArray.next();
+
+        Iterator<Object> consumerInfoIterator = groupInfo.iterator();
+
+        StreamConsumersInfo streamGroupInfo = new StreamConsumersInfo(
+            createMapFromDecodingFunctions(consumerInfoIterator,mappingFunctions));
+        list.add(streamGroupInfo);
+
+      }
+      return list;
+
+    }
+
+    @Override
+    public String toString() {
+      return "List<StreamConsumersInfo>";
+    }
+  };
+
   public static final Builder<Object> OBJECT = new Builder<Object>() {
     @Override
     public Object build(Object data) {
@@ -572,4 +740,28 @@ private BuilderFactory() {
     throw new InstantiationError( "Must not instantiate this class" );
   }
 
+  private static Map<String,Object> createMapFromDecodingFunctions( Iterator<Object> iterator, Map<String,Builder> mappingFunctions) {
+
+    Map<String,Object> resultMap = new HashMap<>();
+    while (iterator.hasNext()) {
+
+      String mapKey = STRING.build(iterator.next());
+      if (mappingFunctions.containsKey(mapKey)) {
+        resultMap.put(mapKey, mappingFunctions.get(mapKey).build(iterator.next()));
+      } else {   //For future - if we don't find an element in our builder map
+        Object unknownData = iterator.next();
+        for (Builder b:mappingFunctions.values()) {
+          try {
+            resultMap.put(mapKey,b.build(unknownData));
+            break;
+          } catch (ClassCastException e) {
+            //We continue with next builder
+
+          }
+        }
+      }
+    }
+    return resultMap;
+  }
+
 }

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

@@ -1262,5 +1262,25 @@ public void xclaim(String key, String group, String consumername, long minIdleTi
     xclaim(SafeEncoder.encode(key), SafeEncoder.encode(group), SafeEncoder.encode(consumername), minIdleTime, newIdleTime, retries, force, bids);    
   }
 
+  @Override
+  public void xinfoStream(String key) {
+
+    xinfoStream(SafeEncoder.encode(key));
+
+  }
+
+  @Override
+  public void xinfoGroup(String key) {
+
+    xinfoGroup(SafeEncoder.encode(key));
+
+  }
+
+  @Override
+  public void xinfoConsumers(String key, String group) {
+
+    xinfoConsumers(SafeEncoder.encode(key),SafeEncoder.encode(group));
+
+  }
 
 }

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

@@ -3830,6 +3830,30 @@ public List<StreamEntry> xclaim(String key, String group, String consumername, l
     return BuilderFactory.STREAM_ENTRY_LIST.build(client.getObjectMultiBulkReply());
   }
 
+  @Override
+  public StreamInfo xinfoStream(String key) {
+    client.xinfoStream(key);
+
+    return BuilderFactory.STREAM_INFO.build(client.getObjectMultiBulkReply());
+
+  }
+
+  @Override
+  public List<StreamGroupInfo> xinfoGroup(String key) {
+    client.xinfoGroup(key);
+
+    return BuilderFactory.STREAM_GROUP_INFO_LIST.build(client.getObjectMultiBulkReply());
+
+  }
+
+  @Override
+  public List<StreamConsumersInfo> xinfoConsumers(String key, String group) {
+    client.xinfoConsumers(key,group);
+
+    return BuilderFactory.STREAM_CONSUMERS_INFO_LIST.build(client.getObjectMultiBulkReply());
+
+  }
+
   public Object sendCommand(ProtocolCommand cmd, String... args) {
     client.sendCommand(cmd, args);
     return client.getOne();

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

@@ -80,6 +80,9 @@ public final class Protocol {
   public static final byte[] POSITIVE_INFINITY_BYTES = "+inf".getBytes();
   public static final byte[] NEGATIVE_INFINITY_BYTES = "-inf".getBytes();
 
+  public static final String STREAM = "STREAM";
+  public static final String GROUPS = "GROUPS";
+  public static final String CONSUMERS = "CONSUMERS";
   private Protocol() {
     // this prevent the class from instantiation
   }
@@ -259,7 +262,8 @@ public static enum Command implements ProtocolCommand {
     PSETEX, CLIENT, TIME, MIGRATE, HINCRBYFLOAT, SCAN, HSCAN, SSCAN, ZSCAN, WAIT, CLUSTER, ASKING,
     PFADD, PFCOUNT, PFMERGE, READONLY, GEOADD, GEODIST, GEOHASH, GEOPOS, GEORADIUS, GEORADIUS_RO,
     GEORADIUSBYMEMBER, GEORADIUSBYMEMBER_RO, MODULE, BITFIELD, HSTRLEN, TOUCH, SWAPDB, MEMORY,
-    XADD, XLEN, XDEL, XTRIM, XRANGE, XREVRANGE, XREAD, XACK, XGROUP, XREADGROUP, XPENDING, XCLAIM;
+    XADD, XLEN, XDEL, XTRIM, XRANGE, XREVRANGE, XREAD, XACK, XGROUP, XREADGROUP, XPENDING, XCLAIM,
+    XINFO;
 
     private final byte[] raw;