Add support for blocking zpopmax, zpopmin commands

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

@@ -692,6 +692,22 @@ public void blpop(final int timeout, final byte[]... keys) {
     blpop(args.toArray(new byte[args.size()][]));
   }
 
+  public void bzpopmax(final int timeout, final byte[]... keys) {
+    final List<byte[]> args = new ArrayList<>();
+    Collections.addAll(args, keys);
+
+    args.add(Protocol.toByteArray(timeout));
+    sendCommand(BZPOPMAX, args.toArray(new byte[args.size()][]));
+  }
+
+  public void bzpopmin(final int timeout, final byte[]... keys) {
+    final List<byte[]> args = new ArrayList<>();
+    Collections.addAll(args, keys);
+
+    args.add(Protocol.toByteArray(timeout));
+    sendCommand(BZPOPMIN, args.toArray(new byte[args.size()][]));
+  }
+
   public void sort(final byte[] key, final SortingParams sortingParameters, final byte[] dstkey) {
     final List<byte[]> args = new ArrayList<>();
     args.add(key);

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

@@ -2363,6 +2363,20 @@ public List<byte[]> blpop(final int timeout, final byte[]... keys) {
     return blpop(getArgsAddTimeout(timeout, keys));
   }
 
+  @Override
+  public ScoreValueTuple bzpopmax(final int timeout, final byte[]... keys) {
+    checkIsInMultiOrPipeline();
+    client.bzpopmax(timeout, keys);
+    return BuilderFactory.SCORE_VALUE_TUPLE.build(client.getBinaryMultiBulkReply());
+  }
+
+  @Override
+  public ScoreValueTuple bzpopmin(final int timeout, final byte[]... keys) {
+    checkIsInMultiOrPipeline();
+    client.bzpopmin(timeout, keys);
+    return BuilderFactory.SCORE_VALUE_TUPLE.build(client.getBinaryMultiBulkReply());
+  }
+
   private byte[][] getArgsAddTimeout(int timeout, byte[][] keys) {
     int size = keys.length;
     final byte[][] args = new byte[size + 1][];

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

@@ -1666,6 +1666,26 @@ public List<byte[]> execute(Jedis connection) {
     }.runBinary(keys.length, keys);
   }
 
+  @Override
+  public ScoreValueTuple bzpopmax(int timeout, byte[]... keys) {
+    return new JedisClusterCommand<ScoreValueTuple>(connectionHandler, maxAttempts) {
+      @Override
+      public ScoreValueTuple execute(Jedis connection) {
+        return connection.bzpopmax(timeout, keys);
+      }
+    }.runBinary(keys.length, keys);
+  }
+
+  @Override
+  public ScoreValueTuple bzpopmin(int timeout, byte[]... keys) {
+    return new JedisClusterCommand<ScoreValueTuple>(connectionHandler, maxAttempts) {
+      @Override
+      public ScoreValueTuple execute(Jedis connection) {
+        return connection.bzpopmin(timeout, keys);
+      }
+    }.runBinary(keys.length, keys);
+  }
+
   @Override
   public List<byte[]> brpop(final int timeout, final byte[]... keys) {
     return new JedisClusterCommand<List<byte[]>>(connectionHandler, maxAttempts) {

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

@@ -326,6 +326,24 @@ public String toString() {
 
   };
 
+  public static final Builder<ScoreValueTuple> SCORE_VALUE_TUPLE = new Builder<ScoreValueTuple>() {
+    @Override
+    @SuppressWarnings("unchecked")
+    public ScoreValueTuple build(Object data) {
+      List<byte[]> l = (List<byte[]>) data; // never null
+      if (l.isEmpty()) {
+        return null;
+      }
+      return new ScoreValueTuple(l.get(0), l.get(1), DOUBLE.build(l.get(2)));
+    }
+
+    @Override
+    public String toString() {
+      return "ScoreValueTuple";
+    }
+
+  };
+
   public static final Builder<Object> EVAL_RESULT = new Builder<Object>() {
 
     @Override

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

@@ -592,6 +592,14 @@ public void blpop(final int timeout, final String... keys) {
     blpop(args.toArray(new String[size]));
   }
 
+  public void bzpopmax(final int timeout, final String... keys) {
+    bzpopmax(timeout, SafeEncoder.encodeMany(keys));
+  }
+
+  public void bzpopmin(final int timeout, final String... keys) {
+    bzpopmin(timeout, SafeEncoder.encodeMany(keys));
+  }
+
   @Override
   public void sort(final String key, final SortingParams sortingParameters, final String dstkey) {
     sort(SafeEncoder.encode(key), sortingParameters, SafeEncoder.encode(dstkey));

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

@@ -2024,6 +2024,20 @@ public List<String> blpop(final int timeout, final String... keys) {
     return blpop(getArgsAddTimeout(timeout, keys));
   }
 
+  @Override
+  public ScoreValueTuple bzpopmax(int timeout, String... keys) {
+    checkIsInMultiOrPipeline();
+    client.bzpopmax(timeout, keys);
+    return BuilderFactory.SCORE_VALUE_TUPLE.build(client.getObjectMultiBulkReply());
+  }
+
+  @Override
+  public ScoreValueTuple bzpopmin(int timeout, String... keys) {
+    checkIsInMultiOrPipeline();
+    client.bzpopmin(timeout, keys);
+    return BuilderFactory.SCORE_VALUE_TUPLE.build(client.getObjectMultiBulkReply());
+  }
+
   private String[] getArgsAddTimeout(int timeout, String[] keys) {
     final int keyCount = keys.length;
     final String[] args = new String[keyCount + 1];

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

@@ -1756,6 +1756,26 @@ public List<String> execute(Jedis connection) {
     }.run(keys.length, keys);
   }
 
+  @Override
+  public ScoreValueTuple bzpopmax(int timeout, String... keys) {
+    return new JedisClusterCommand<ScoreValueTuple>(connectionHandler, maxAttempts) {
+      @Override
+      public ScoreValueTuple execute(Jedis connection) {
+        return connection.bzpopmax(timeout, keys);
+      }
+    }.run(keys.length, keys);
+  }
+
+  @Override
+  public ScoreValueTuple bzpopmin(int timeout, String... keys) {
+    return new JedisClusterCommand<ScoreValueTuple>(connectionHandler, maxAttempts) {
+      @Override
+      public ScoreValueTuple execute(Jedis connection) {
+        return connection.bzpopmin(timeout, keys);
+      }
+    }.run(keys.length, keys);
+  }
+
   @Override
   public List<String> mget(final String... keys) {
     return new JedisClusterCommand<List<String>>(connectionHandler, maxAttempts) {

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

@@ -69,6 +69,30 @@ public Response<List<String>> blpop(int timeout, byte[]... keys) {
     return getResponse(BuilderFactory.STRING_LIST);
   }
 
+  @Override
+  public Response<ScoreValueTuple> bzpopmax(int timeout, String... keys) {
+    client.bzpopmax(timeout, keys);
+    return getResponse(BuilderFactory.SCORE_VALUE_TUPLE);
+  }
+
+  @Override
+  public Response<ScoreValueTuple> bzpopmin(int timeout, String... keys) {
+    client.bzpopmin(timeout, keys);
+    return getResponse(BuilderFactory.SCORE_VALUE_TUPLE);
+  }
+
+  @Override
+  public Response<ScoreValueTuple> bzpopmax(int timeout, byte[]... keys) {
+    client.bzpopmax(timeout, keys);
+    return getResponse(BuilderFactory.SCORE_VALUE_TUPLE);
+  }
+
+  @Override
+  public Response<ScoreValueTuple> bzpopmin(int timeout, byte[]... keys) {
+    client.bzpopmin(timeout, keys);
+    return getResponse(BuilderFactory.SCORE_VALUE_TUPLE);
+  }
+
   @Override
   public Response<Long> del(String... keys) {
     client.del(keys);

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

@@ -261,7 +261,7 @@ public static enum Command implements ProtocolCommand {
     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, ACL, XINFO,
-    BITFIELD_RO, LPOS, SMISMEMBER, ZMSCORE;
+    BITFIELD_RO, LPOS, SMISMEMBER, ZMSCORE, BZPOPMIN, BZPOPMAX;
 
     private final byte[] raw;
 

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

@@ -0,0 +1,50 @@
+package redis.clients.jedis;
+
+import redis.clients.jedis.util.SafeEncoder;
+
+import java.util.Arrays;
+
+public class ScoreValueTuple extends Tuple {
+  private byte[] key;
+
+  public ScoreValueTuple(byte[] key, byte[] element, Double score) {
+    super(element, score);
+    this.key = key;
+  }
+
+  public ScoreValueTuple(String key, String element, Double score) {
+    super(element, score);
+    this.key = SafeEncoder.encode(key);
+  }
+
+  public String getKey() {
+    if (null != key) {
+      return SafeEncoder.encode(key);
+    }
+    return null;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (!(o instanceof ScoreValueTuple)) return false;
+    if (!super.equals(o)) return false;
+
+    ScoreValueTuple that = (ScoreValueTuple) o;
+    return Arrays.equals(key, that.key);
+  }
+
+  @Override
+  public int hashCode() {
+    int result = super.hashCode();
+    result = 31 * result + (key != null ? Arrays.hashCode(key) : 0);
+    return result;
+  }
+
+  @Override
+  public String toString() {
+    return "ScoreValueTuple{" +
+            "key=" + SafeEncoder.encode(key) +
+            "} " + super.toString();
+  }
+}

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

@@ -3,6 +3,7 @@
 import redis.clients.jedis.BinaryJedisPubSub;
 import redis.clients.jedis.BitOP;
 import redis.clients.jedis.GeoUnit;
+import redis.clients.jedis.ScoreValueTuple;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.ZParams;
 import redis.clients.jedis.params.GeoRadiusParam;
@@ -27,6 +28,10 @@ public interface MultiKeyBinaryCommands {
 
   List<byte[]> brpop(byte[]... args);
 
+  ScoreValueTuple bzpopmax(int timeout, byte[]... keys);
+
+  ScoreValueTuple bzpopmin(int timeout, byte[]... keys);
+
   Set<byte[]> keys(byte[] pattern);
 
   List<byte[]> mget(byte[]... keys);

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

@@ -5,6 +5,7 @@
 import redis.clients.jedis.GeoUnit;
 import redis.clients.jedis.ScanParams;
 import redis.clients.jedis.ScanResult;
+import redis.clients.jedis.ScoreValueTuple;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.ZParams;
 import redis.clients.jedis.params.GeoRadiusParam;
@@ -25,6 +26,10 @@ public interface MultiKeyBinaryJedisClusterCommands {
 
   List<byte[]> brpop(int timeout, byte[]... keys);
 
+  ScoreValueTuple bzpopmax(int timeout, byte[]... keys);
+
+  ScoreValueTuple bzpopmin(int timeout, byte[]... keys);
+
   List<byte[]> mget(byte[]... keys);
 
   String mset(byte[]... keysvalues);

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

@@ -3,6 +3,7 @@
 import redis.clients.jedis.BitOP;
 import redis.clients.jedis.GeoUnit;
 import redis.clients.jedis.Response;
+import redis.clients.jedis.ScoreValueTuple;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.ZParams;
 import redis.clients.jedis.params.GeoRadiusParam;
@@ -27,6 +28,10 @@ public interface MultiKeyBinaryRedisPipeline {
 
   Response<List<byte[]>> brpop(byte[]... args);
 
+  Response<ScoreValueTuple> bzpopmax(int timeout, byte[]... keys);
+
+  Response<ScoreValueTuple> bzpopmin(int timeout, byte[]... keys);
+
   Response<Set<byte[]>> keys(byte[] pattern);
 
   Response<List<byte[]>> mget(byte[]... keys);

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

@@ -2,6 +2,7 @@
 
 import redis.clients.jedis.BitOP;
 import redis.clients.jedis.GeoUnit;
+import redis.clients.jedis.ScoreValueTuple;
 import redis.clients.jedis.StreamEntryID;
 import redis.clients.jedis.JedisPubSub;
 import redis.clients.jedis.ScanParams;
@@ -31,6 +32,10 @@ public interface MultiKeyCommands {
 
   List<String> brpop(String... args);
 
+  ScoreValueTuple bzpopmax(int timeout, String... keys);
+
+  ScoreValueTuple bzpopmin(int timeout, String... keys);
+
   /**
    * Returns all the keys matching the glob-style pattern. For example if you have in the database
    * the keys "foo" and "foobar" the command "KEYS foo*" will return "foo foobar".<br>
@@ -118,7 +123,7 @@ public interface MultiKeyCommands {
 
   /**
    * @see #scan(String, ScanParams)
-   * 
+   *
    * @param cursor
    * @return
    */
@@ -178,7 +183,7 @@ public interface MultiKeyCommands {
 
   /**
    * XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
-   * 
+   *
    * @param count
    * @param block
    * @param streams
@@ -189,7 +194,7 @@ List<Map.Entry<String, List<StreamEntry>>> xread(int count, long block,
 
   /**
    * XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
-   * 
+   *
    * @param groupname
    * @param consumer
    * @param count

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

@@ -3,6 +3,7 @@
 import redis.clients.jedis.BitOP;
 import redis.clients.jedis.GeoUnit;
 import redis.clients.jedis.Response;
+import redis.clients.jedis.ScoreValueTuple;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.ZParams;
 import redis.clients.jedis.params.GeoRadiusParam;
@@ -26,6 +27,10 @@ public interface MultiKeyCommandsPipeline {
 
   Response<List<String>> brpop(String... args);
 
+  Response<ScoreValueTuple> bzpopmax(int timeout, String... keys);
+
+  Response<ScoreValueTuple> bzpopmin(int timeout, String... keys);
+
   Response<Set<String>> keys(String pattern);
 
   Response<List<String>> mget(String... keys);

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

@@ -5,6 +5,7 @@
 import redis.clients.jedis.JedisPubSub;
 import redis.clients.jedis.ScanParams;
 import redis.clients.jedis.ScanResult;
+import redis.clients.jedis.ScoreValueTuple;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.StreamEntry;
 import redis.clients.jedis.StreamEntryID;
@@ -28,6 +29,10 @@ public interface MultiKeyJedisClusterCommands {
 
   List<String> brpop(int timeout, String... keys);
 
+  ScoreValueTuple bzpopmax(int timeout, String... keys);
+
+  ScoreValueTuple bzpopmin(int timeout, String... keys);
+
   List<String> mget(String... keys);
 
   String mset(String... keysvalues);