Add binary stream support for XREAD and XREADGROUP (#3566)

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

@@ -2892,6 +2892,30 @@ public final CommandObject<List<Object>> xread(XReadParams xReadParams, Map.Entr
     return new CommandObject<>(args, BuilderFactory.RAW_OBJECT_LIST);
   }
 
+  public final CommandObject<List<Map.Entry<byte[], List<StreamEntryBinary>>>> xreadBinary(
+      XReadParams xReadParams, Map.Entry<byte[], byte[]>... streams) {
+    CommandArguments args = commandArguments(XREAD).addParams(xReadParams).add(STREAMS);
+    for (Map.Entry<byte[], byte[]> entry : streams) {
+      args.key(entry.getKey());
+    }
+    for (Map.Entry<byte[], byte[]> entry : streams) {
+      args.add(entry.getValue());
+    }
+    return new CommandObject<>(args, BuilderFactory.STREAM_READ_BINARY_RESPONSE);
+  }
+
+  public final CommandObject<Map<byte[], List<StreamEntryBinary>>> xreadBinaryAsMap(
+      XReadParams xReadParams, Map.Entry<byte[], byte[]>... streams) {
+    CommandArguments args = commandArguments(XREAD).addParams(xReadParams).add(STREAMS);
+    for (Map.Entry<byte[], byte[]> entry : streams) {
+      args.key(entry.getKey());
+    }
+    for (Map.Entry<byte[], byte[]> entry : streams) {
+      args.add(entry.getValue());
+    }
+    return new CommandObject<>(args, BuilderFactory.STREAM_READ_BINARY_MAP_RESPONSE);
+  }
+
   public final CommandObject<List<Object>> xreadGroup(byte[] groupName, byte[] consumer,
       XReadGroupParams xReadGroupParams, Map.Entry<byte[], byte[]>... streams) {
     CommandArguments args = commandArguments(XREADGROUP)
@@ -2905,6 +2929,36 @@ public final CommandObject<List<Object>> xreadGroup(byte[] groupName, byte[] con
     }
     return new CommandObject<>(args, BuilderFactory.RAW_OBJECT_LIST);
   }
+
+  public final CommandObject<List<Map.Entry<byte[], List<StreamEntryBinary>>>> xreadGroupBinary(
+      byte[] groupName, byte[] consumer, XReadGroupParams xReadGroupParams, 
+      Map.Entry<byte[], byte[]>... streams) {
+    CommandArguments args = commandArguments(XREADGROUP)
+        .add(GROUP).add(groupName).add(consumer)
+        .addParams(xReadGroupParams).add(STREAMS);
+    for (Map.Entry<byte[], byte[]> entry : streams) {
+      args.key(entry.getKey());
+    }
+    for (Map.Entry<byte[], byte[]> entry : streams) {
+      args.add(entry.getValue());
+    }
+    return new CommandObject<>(args, BuilderFactory.STREAM_READ_BINARY_RESPONSE);
+  }
+
+  public final CommandObject<Map<byte[], List<StreamEntryBinary>>> xreadGroupBinaryAsMap(
+      byte[] groupName, byte[] consumer, XReadGroupParams xReadGroupParams, 
+      Map.Entry<byte[], byte[]>... streams) {
+    CommandArguments args = commandArguments(XREADGROUP)
+        .add(GROUP).add(groupName).add(consumer)
+        .addParams(xReadGroupParams).add(STREAMS);
+    for (Map.Entry<byte[], byte[]> entry : streams) {
+      args.key(entry.getKey());
+    }
+    for (Map.Entry<byte[], byte[]> entry : streams) {
+      args.add(entry.getValue());
+    }
+    return new CommandObject<>(args, BuilderFactory.STREAM_READ_BINARY_MAP_RESPONSE);
+  }
   // Stream commands
 
   // Scripting commands

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

@@ -1172,7 +1172,7 @@ public long hsetex(byte[] key, HSetExParams params, byte[] field, byte[] value)
     checkIsInMultiOrPipeline();
     return connection.executeCommand(commandObjects.hsetex(key, params, field, value));
   }
-  
+
   @Override
   public long hsetex(byte[] key, HSetExParams params, Map<byte[], byte[]> hash){
     checkIsInMultiOrPipeline();
@@ -1200,13 +1200,13 @@ public List<byte[]> hgetex(byte[] key, HGetExParams params, byte[]... fields){
     checkIsInMultiOrPipeline();
     return connection.executeCommand(commandObjects.hgetex(key, params, fields));
   }
-  
+
   @Override
   public List<byte[]> hgetdel(byte[] key, byte[]... fields){
     checkIsInMultiOrPipeline();
     return connection.executeCommand(commandObjects.hgetdel(key, fields));
   }
-  
+
   /**
    * Set the specified hash field to the specified value if the field not exists. <b>Time
    * complexity:</b> O(1)
@@ -4783,6 +4783,34 @@ public List<Object> xreadGroup(byte[] groupName, byte[] consumer,
     return connection.executeCommand(commandObjects.xreadGroup(groupName, consumer, xReadGroupParams, streams));
   }
 
+  @Override
+  public List<Map.Entry<byte[], List<StreamEntryBinary>>> xreadBinary(XReadParams xReadParams, 
+      Entry<byte[], byte[]>... streams) {
+    checkIsInMultiOrPipeline();
+    return connection.executeCommand(commandObjects.xreadBinary(xReadParams, streams));
+  }
+
+  @Override
+  public Map<byte[], List<StreamEntryBinary>> xreadBinaryAsMap(XReadParams xReadParams, 
+      Entry<byte[], byte[]>... streams) {
+    checkIsInMultiOrPipeline();
+    return connection.executeCommand(commandObjects.xreadBinaryAsMap(xReadParams, streams));
+  }
+
+  @Override
+  public List<Map.Entry<byte[], List<StreamEntryBinary>>> xreadGroupBinary(byte[] groupName, byte[] consumer, 
+      XReadGroupParams xReadGroupParams, Entry<byte[], byte[]>... streams) {
+    checkIsInMultiOrPipeline();
+    return connection.executeCommand(commandObjects.xreadGroupBinary(groupName, consumer, xReadGroupParams, streams));
+  }
+
+  @Override
+  public Map<byte[], List<StreamEntryBinary>> xreadGroupBinaryAsMap(byte[] groupName, byte[] consumer, 
+      XReadGroupParams xReadGroupParams, Entry<byte[], byte[]>... streams) {
+    checkIsInMultiOrPipeline();
+    return connection.executeCommand(commandObjects.xreadGroupBinaryAsMap(groupName, consumer, xReadGroupParams, streams));
+  }
+
   @Override
   public byte[] xadd(final byte[] key, final XAddParams params, final Map<byte[], byte[]> hash) {
     checkIsInMultiOrPipeline();

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

@@ -1548,7 +1548,7 @@ public byte[] hget(byte[] key, byte[] field) {
   public List<byte[]> hgetex(byte[] key, HGetExParams params, byte[]... fields) {
     return executeCommand(commandObjects.hgetex(key, params, fields));
   }
-  
+
   @Override
   public List<byte[]> hgetdel(byte[] key, byte[]... fields) {
     return executeCommand(commandObjects.hgetdel(key, fields));
@@ -3460,6 +3460,30 @@ public List<Object> xread(XReadParams xReadParams, Map.Entry<byte[], byte[]>...
   public List<Object> xreadGroup(byte[] groupName, byte[] consumer, XReadGroupParams xReadGroupParams, Map.Entry<byte[], byte[]>... streams) {
     return executeCommand(commandObjects.xreadGroup(groupName, consumer, xReadGroupParams, streams));
   }
+
+  @Override
+  public List<Map.Entry<byte[], List<StreamEntryBinary>>> xreadBinary(XReadParams xReadParams, 
+      Map.Entry<byte[], byte[]>... streams) {
+    return executeCommand(commandObjects.xreadBinary(xReadParams, streams));
+  }
+
+  @Override
+  public Map<byte[], List<StreamEntryBinary>> xreadBinaryAsMap(XReadParams xReadParams, 
+      Map.Entry<byte[], byte[]>... streams) {
+    return executeCommand(commandObjects.xreadBinaryAsMap(xReadParams, streams));
+  }
+
+  @Override
+  public List<Map.Entry<byte[], List<StreamEntryBinary>>> xreadGroupBinary(byte[] groupName, byte[] consumer, 
+      XReadGroupParams xReadGroupParams, Map.Entry<byte[], byte[]>... streams) {
+    return executeCommand(commandObjects.xreadGroupBinary(groupName, consumer, xReadGroupParams, streams));
+  }
+
+  @Override
+  public Map<byte[], List<StreamEntryBinary>> xreadGroupBinaryAsMap(byte[] groupName, byte[] consumer, 
+      XReadGroupParams xReadGroupParams, Map.Entry<byte[], byte[]>... streams) {
+    return executeCommand(commandObjects.xreadGroupBinaryAsMap(groupName, consumer, xReadGroupParams, streams));
+  }
   // Stream commands
 
   // Scripting commands

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

@@ -4,6 +4,7 @@
 import java.util.Map;
 
 import redis.clients.jedis.params.*;
+import redis.clients.jedis.resps.StreamEntryBinary;
 
 public interface StreamBinaryCommands {
 
@@ -79,4 +80,15 @@ List<Object> xautoclaimJustId(byte[] key, byte[] groupName, byte[] consumerName,
   List<Object> xreadGroup(byte[] groupName, byte[] consumer, XReadGroupParams xReadGroupParams,
       Map.Entry<byte[], byte[]>... streams);
 
+  List<Map.Entry<byte[], List<StreamEntryBinary>>> xreadBinary(XReadParams xReadParams,
+      Map.Entry<byte[], byte[]>... streams);
+
+  Map<byte[], List<StreamEntryBinary>> xreadBinaryAsMap(XReadParams xReadParams,
+      Map.Entry<byte[], byte[]>... streams);
+
+  List<Map.Entry<byte[], List<StreamEntryBinary>>> xreadGroupBinary(byte[] groupName, byte[] consumer,
+      XReadGroupParams xReadGroupParams, Map.Entry<byte[], byte[]>... streams);
+
+  Map<byte[], List<StreamEntryBinary>> xreadGroupBinaryAsMap(byte[] groupName, byte[] consumer,
+      XReadGroupParams xReadGroupParams, Map.Entry<byte[], byte[]>... streams);
 }

src/main/java/redis/clients/jedis/resps/StreamEntryBinary.java

@@ -0,0 +1,42 @@
+package redis.clients.jedis.resps;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
+import redis.clients.jedis.StreamEntryID;
+
+public class StreamEntryBinary implements Serializable {
+
+  private static final long serialVersionUID = 1L;
+
+  private StreamEntryID id;
+  private Map<byte[], byte[]> fields;
+
+  public StreamEntryBinary(StreamEntryID id, Map<byte[], byte[]> fields) {
+    this.id = id;
+    this.fields = fields;
+  }
+
+  public StreamEntryID getID() {
+    return id;
+  }
+
+  public Map<byte[], byte[]> getFields() {
+    return fields;
+  }
+
+  @Override
+  public String toString() {
+    return id + " " + fields;
+  }
+
+  private void writeObject(java.io.ObjectOutputStream out) throws IOException {
+    out.writeUnshared(this.id);
+    out.writeUnshared(this.fields);
+  }
+
+  private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
+    this.id = (StreamEntryID) in.readUnshared();
+    this.fields = (Map<byte[], byte[]>) in.readUnshared();
+  }
+}