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

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

@@ -1809,6 +1809,124 @@ public String toString() {
     }
   };
 
+  public static final Builder<StreamEntryBinary> STREAM_ENTRY_BINARY = new Builder<StreamEntryBinary>() {
+    @Override
+    @SuppressWarnings("unchecked")
+    public StreamEntryBinary build(Object data) {
+      if (null == data) {
+        return null;
+      }
+      List<Object> list = (List<Object>) data;
+      String entryIdString = SafeEncoder.encode((byte[]) list.get(0));
+      StreamEntryID entryID = new StreamEntryID(entryIdString);
+      List<byte[]> hash = (List<byte[]>) list.get(1);
+      if (hash == null) {
+        return new StreamEntryBinary(entryID, null);
+      }
+
+      Iterator<byte[]> hashIterator = hash.iterator();
+      Map<byte[], byte[]> map = new JedisByteHashMap();
+      while (hashIterator.hasNext()) {
+        map.put(hashIterator.next(), hashIterator.next());
+      }
+      return new StreamEntryBinary(entryID, map);
+    }
+
+    @Override
+    public String toString() {
+      return "StreamEntryBinary";
+    }
+  };
+
+  public static final Builder<List<StreamEntryBinary>> STREAM_ENTRY_BINARY_LIST = new Builder<List<StreamEntryBinary>>() {
+    @Override
+    @SuppressWarnings("unchecked")
+    public List<StreamEntryBinary> build(Object data) {
+      if (null == data) {
+        return null;
+      }
+      List<ArrayList<Object>> objectList = (List<ArrayList<Object>>) data;
+
+      List<StreamEntryBinary> responses = new ArrayList<>(objectList.size() / 2);
+      if (objectList.isEmpty()) {
+        return responses;
+      }
+
+      for (ArrayList<Object> res : objectList) {
+        responses.add(STREAM_ENTRY_BINARY.build(res));
+      }
+
+      return responses;
+    }
+
+    @Override
+    public String toString() {
+      return "List<StreamEntryBinary>";
+    }
+  };
+
+  public static final Builder<Map<byte[], List<StreamEntryBinary>>> STREAM_READ_BINARY_MAP_RESPONSE
+          = new Builder<Map<byte[], List<StreamEntryBinary>>>() {
+    @Override
+    @SuppressWarnings("unchecked")
+    public Map<byte[], List<StreamEntryBinary>> build(Object data) {
+      if (data == null) return null;
+      List list = (List) data;
+      if (list.isEmpty()) return Collections.emptyMap();
+
+      if (list.get(0) instanceof KeyValue) {
+        return ((List<KeyValue>) list).stream()
+                .collect(Collectors.toMap(kv -> BINARY.build(kv.getKey()), kv -> STREAM_ENTRY_BINARY_LIST.build(kv.getValue())));
+      } else {
+        Map<byte[], List<StreamEntryBinary>> result = new HashMap<>(list.size());
+        for (Object anObj : list) {
+          List<Object> streamObj = (List<Object>) anObj;
+          byte[] streamKey = (byte[]) streamObj.get(0);
+          List<StreamEntryBinary> streamEntries = STREAM_ENTRY_BINARY_LIST.build(streamObj.get(1));
+          result.put(streamKey, streamEntries);
+        }
+        return result;
+      }
+    }
+
+    @Override
+    public String toString() {
+      return "Map<byte[], List<StreamEntryBinary>>";
+    }
+  };
+
+  public static final Builder<List<Map.Entry<byte[], List<StreamEntryBinary>>>> STREAM_READ_BINARY_RESPONSE
+      = new Builder<List<Map.Entry<byte[], List<StreamEntryBinary>>>>() {
+    @Override
+    @SuppressWarnings("unchecked")
+    public List<Map.Entry<byte[], List<StreamEntryBinary>>> build(Object data) {
+      if (data == null) return null;
+      List list = (List) data;
+      if (list.isEmpty()) return Collections.emptyList();
+
+      if (list.get(0) instanceof KeyValue) {
+        return ((List<KeyValue>) list).stream()
+                .map(kv -> new KeyValue<>(BINARY.build(kv.getKey()),
+                        STREAM_ENTRY_BINARY_LIST.build(kv.getValue())))
+                .collect(Collectors.toList());
+      } else {
+        List<Map.Entry<byte[], List<StreamEntryBinary>>> result = new ArrayList<>(list.size());
+        for (Object anObj : list) {
+          List<Object> streamObj = (List<Object>) anObj;
+          byte[] streamKey = BINARY.build(streamObj.get(0));
+          List<StreamEntryBinary> streamEntries = STREAM_ENTRY_BINARY_LIST.build(streamObj.get(1));
+          result.add(KeyValue.of(streamKey, streamEntries));
+        }
+        return result;
+      }
+    }
+
+    @Override
+    public String toString() {
+      return "List<Entry<byte[], List<StreamEntryBinary>>>";
+    }
+  };
+
   private static final List<Builder> BACKUP_BUILDERS_FOR_DECODING_FUNCTIONS
       = Arrays.asList(STRING, LONG, DOUBLE);
 

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();
+  }
+}