Broadcast commands from UnifiedJedis

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

@@ -0,0 +1,28 @@
+package redis.clients.jedis;
+
+import java.util.function.Supplier;
+
+/**
+ * Represents the response from a single node in broadcast mode.
+ */
+public class BroadcastResponse<T> implements Supplier<T> {
+
+  private T response = null;
+  private RuntimeException exception = null;
+
+  public BroadcastResponse(T response) {
+    this.response = response;
+  }
+
+  public BroadcastResponse(RuntimeException exception) {
+    this.exception = exception;
+  }
+
+  @Override
+  public T get() {
+    if (exception != null) {
+      throw exception;
+    }
+    return response;
+  }
+}

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

@@ -2754,32 +2754,62 @@ public final CommandObject<Object> evalshaReadonly(byte[] sha1, List<byte[]> key
         BuilderFactory.RAW_OBJECT);
   }
 
+  public final CommandObject<List<Boolean>> scriptExists(String... sha1s) {
+    return new CommandObject<>(commandArguments(SCRIPT).add(Keyword.EXISTS).addObjects((Object[]) sha1s),
+        BuilderFactory.BOOLEAN_LIST);
+  }
+
   public final CommandObject<List<Boolean>> scriptExists(String sampleKey, String... sha1s) {
     return new CommandObject<>(commandArguments(SCRIPT).add(Keyword.EXISTS).addObjects((Object[]) sha1s)
         .processKey(sampleKey), BuilderFactory.BOOLEAN_LIST);
   }
 
+  public final CommandObject<String> scriptLoad(String script) {
+    return new CommandObject<>(commandArguments(SCRIPT).add(LOAD).add(script), BuilderFactory.STRING);
+  }
+
   public final CommandObject<String> scriptLoad(String script, String sampleKey) {
     return new CommandObject<>(commandArguments(SCRIPT).add(LOAD).add(script).processKey(sampleKey), BuilderFactory.STRING);
   }
 
+  public final CommandObject<String> scriptFlush() {
+    return new CommandObject<>(commandArguments(SCRIPT).add(FLUSH), BuilderFactory.STRING);
+  }
+
   public final CommandObject<String> scriptFlush(String sampleKey) {
     return new CommandObject<>(commandArguments(SCRIPT).add(FLUSH).processKey(sampleKey), BuilderFactory.STRING);
   }
 
+  public final CommandObject<String> scriptFlush(FlushMode flushMode) {
+    return new CommandObject<>(commandArguments(SCRIPT).add(FLUSH).add(flushMode), BuilderFactory.STRING);
+  }
+
   public final CommandObject<String> scriptFlush(String sampleKey, FlushMode flushMode) {
     return new CommandObject<>(commandArguments(SCRIPT).add(FLUSH).add(flushMode).processKey(sampleKey), BuilderFactory.STRING);
   }
 
+  public final CommandObject<String> scriptKill() {
+    return new CommandObject<>(commandArguments(SCRIPT).add(KILL), BuilderFactory.STRING);
+  }
+
   public final CommandObject<String> scriptKill(String sampleKey) {
     return new CommandObject<>(commandArguments(SCRIPT).add(KILL).processKey(sampleKey), BuilderFactory.STRING);
   }
 
+  public final CommandObject<List<Boolean>> scriptExists(byte[]... sha1s) {
+    return new CommandObject<>(commandArguments(SCRIPT).add(Keyword.EXISTS).addObjects((Object[]) sha1s),
+        BuilderFactory.BOOLEAN_LIST);
+  }
+
   public final CommandObject<List<Boolean>> scriptExists(byte[] sampleKey, byte[]... sha1s) {
     return new CommandObject<>(commandArguments(SCRIPT).add(Keyword.EXISTS).addObjects((Object[]) sha1s)
         .processKey(sampleKey), BuilderFactory.BOOLEAN_LIST);
   }
 
+  public final CommandObject<byte[]> scriptLoad(byte[] script) {
+    return new CommandObject<>(commandArguments(SCRIPT).add(LOAD).add(script), BuilderFactory.BINARY);
+  }
+
   public final CommandObject<byte[]> scriptLoad(byte[] script, byte[] sampleKey) {
     return new CommandObject<>(commandArguments(SCRIPT).add(LOAD).add(script).processKey(sampleKey), BuilderFactory.BINARY);
   }

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

@@ -3860,14 +3860,12 @@ public Object evalsha(final byte[] sha1, final int keyCount, final byte[]... par
 
   @Override
   public String scriptFlush() {
-    connection.sendCommand(SCRIPT, FLUSH);
-    return connection.getStatusCodeReply();
+    return connection.executeCommand(commandObjects.scriptFlush());
   }
 
   @Override
   public String scriptFlush(final FlushMode flushMode) {
-    connection.sendCommand(SCRIPT, FLUSH.getRaw(), flushMode.getRaw());
-    return connection.getStatusCodeReply();
+    return connection.executeCommand(commandObjects.scriptFlush(flushMode));
   }
 
   @Override
@@ -3879,20 +3877,17 @@ public Boolean scriptExists(final byte[] sha1) {
 
   @Override
   public List<Boolean> scriptExists(final byte[]... sha1) {
-    connection.sendCommand(SCRIPT, joinParameters(Keyword.EXISTS.getRaw(), sha1));
-    return BuilderFactory.BOOLEAN_LIST.build(connection.getOne());
+    return connection.executeCommand(commandObjects.scriptExists(sha1));
   }
 
   @Override
   public byte[] scriptLoad(final byte[] script) {
-    connection.sendCommand(SCRIPT, LOAD.getRaw(), script);
-    return connection.getBinaryBulkReply();
+    return connection.executeCommand(commandObjects.scriptLoad(script));
   }
 
   @Override
   public String scriptKill() {
-    connection.sendCommand(SCRIPT, KILL);
-    return connection.getStatusCodeReply();
+    return connection.executeCommand(commandObjects.scriptKill());
   }
 
   @Override
@@ -4335,8 +4330,7 @@ public String migrate(String host, int port, int timeout, MigrateParams params,
   @Override
   public long waitReplicas(final int replicas, final long timeout) {
     checkIsInMultiOrPipeline();
-    connection.sendCommand(WAIT, toByteArray(replicas), toByteArray(timeout));
-    return connection.getIntegerReply();
+    return connection.executeCommand(commandObjects.waitReplicas(replicas, timeout));
   }
 
   @Override
@@ -7993,14 +7987,12 @@ public Boolean scriptExists(final String sha1) {
 
   @Override
   public List<Boolean> scriptExists(final String... sha1) {
-    connection.sendCommand(SCRIPT, joinParameters(Keyword.EXISTS.name(), sha1));
-    return BuilderFactory.BOOLEAN_LIST.build(connection.getOne());
+    return connection.executeCommand(commandObjects.scriptExists(sha1));
   }
 
   @Override
   public String scriptLoad(final String script) {
-    connection.sendCommand(SCRIPT, LOAD.name(), script);
-    return connection.getBulkReply();
+    return connection.executeCommand(commandObjects.scriptLoad(script));
   }
 
   @Override

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

@@ -2,9 +2,12 @@
 
 import java.net.URI;
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Supplier;
 import java.util.regex.Pattern;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.json.JSONArray;
@@ -34,6 +37,7 @@
 import redis.clients.jedis.util.IOUtils;
 import redis.clients.jedis.util.JedisURIHelper;
 import redis.clients.jedis.util.KeyValue;
+import redis.clients.jedis.util.Pool;
 
 public class UnifiedJedis implements JedisCommands, JedisBinaryCommands,
     SampleKeyedCommands, SampleBinaryKeyedCommands, RedisModuleCommands,
@@ -167,6 +171,33 @@ public final <T> T executeCommand(CommandObject<T> commandObject) {
     return executor.executeCommand(commandObject);
   }
 
+  public final <T> Map<?, Supplier<T>> broadcastCommand(CommandObject<T> commandObject) {
+    // TODO: Push this implementation in executor interface/classes.
+    Map<?, ?> connectionMap = provider.getConnectionMap();
+    Map<Object, Supplier<T>> responseMap = new HashMap<>(connectionMap.size(), 1f);
+    for (Map.Entry<? extends Object, ? extends Object> entry : connectionMap.entrySet()) {
+      Object key = entry.getKey();
+      Object connection = entry.getValue();
+      try {
+        responseMap.put(key, new BroadcastResponse<>(executeBroadcastCommand(connection, commandObject)));
+      } catch (RuntimeException re) {
+        responseMap.put(key, new BroadcastResponse<>(re));
+      }
+    }
+    return responseMap;
+  }
+
+  private <T> T executeBroadcastCommand(Object connection, CommandObject<T> commandObject) {
+    if (connection instanceof Connection) {
+      return ((Connection) connection).executeCommand(commandObject);
+    } else if (connection instanceof Pool) {
+      try (Connection _conn = ((Pool<Connection>) connection).getResource()) {
+        return _conn.executeCommand(commandObject);
+      }
+    }
+    throw new IllegalStateException(connection.getClass() + "is not supported.");
+  }
+
   // Key commands
   @Override
   public boolean exists(String key) {
@@ -3322,6 +3353,10 @@ public long waitReplicas(byte[] sampleKey, int replicas, long timeout) {
     return executeCommand(commandObjects.waitReplicas(sampleKey, replicas, timeout));
   }
 
+  public Map<?, Supplier<Long>> waitReplicasBroadcast(final int replicas, final long timeout) {
+    return broadcastCommand(commandObjects.waitReplicas(replicas, timeout));
+  }
+
   @Override
   public Object eval(String script, String sampleKey) {
     return executeCommand(commandObjects.eval(script, sampleKey));
@@ -3352,6 +3387,10 @@ public List<Boolean> scriptExists(String sampleKey, String... sha1s) {
     return executeCommand(commandObjects.scriptExists(sampleKey, sha1s));
   }
 
+  public Map<?, Supplier<List<Boolean>>> scriptExistsBroadcast(String... sha1s) {
+    return broadcastCommand(commandObjects.scriptExists(sha1s));
+  }
+
   @Override
   public Boolean scriptExists(byte[] sha1, byte[] sampleKey) {
     return scriptExists(sampleKey, new byte[][]{sha1}).get(0);
@@ -3362,11 +3401,19 @@ public List<Boolean> scriptExists(byte[] sampleKey, byte[]... sha1s) {
     return executeCommand(commandObjects.scriptExists(sampleKey, sha1s));
   }
 
+  public Map<?, Supplier<List<Boolean>>> scriptExistsBroadcast(byte[]... sha1s) {
+    return broadcastCommand(commandObjects.scriptExists(sha1s));
+  }
+
   @Override
   public String scriptLoad(String script, String sampleKey) {
     return executeCommand(commandObjects.scriptLoad(script, sampleKey));
   }
 
+  public Map<?, Supplier<String>> scriptLoadBroadcast(String script) {
+    return broadcastCommand(commandObjects.scriptLoad(script));
+  }
+
   @Override
   public String scriptFlush(String sampleKey) {
     return executeCommand(commandObjects.scriptFlush(sampleKey));
@@ -3377,16 +3424,32 @@ public String scriptFlush(String sampleKey, FlushMode flushMode) {
     return executeCommand(commandObjects.scriptFlush(sampleKey, flushMode));
   }
 
+  public Map<?, Supplier<String>> scriptFlushBroadcast() {
+    return broadcastCommand(commandObjects.scriptFlush());
+  }
+
+  public Map<?, Supplier<String>> scriptFlushBroadcast(FlushMode flushMode) {
+    return broadcastCommand(commandObjects.scriptFlush(flushMode));
+  }
+
   @Override
   public String scriptKill(String sampleKey) {
     return executeCommand(commandObjects.scriptKill(sampleKey));
   }
 
+  public Map<?, Supplier<String>> scriptKillBroadcast() {
+    return broadcastCommand(commandObjects.scriptKill());
+  }
+
   @Override
   public byte[] scriptLoad(byte[] script, byte[] sampleKey) {
     return executeCommand(commandObjects.scriptLoad(script, sampleKey));
   }
 
+  public Map<?, Supplier<byte[]>> scriptLoadBroadcast(byte[] script) {
+    return broadcastCommand(commandObjects.scriptLoad(script));
+  }
+
   @Override
   public String scriptFlush(byte[] sampleKey) {
     return executeCommand(commandObjects.scriptFlush(sampleKey));
@@ -3403,6 +3466,27 @@ public String scriptKill(byte[] sampleKey) {
   }
   // Sample key commands
 
+  public Map<?, Supplier<String>> configSetBroadcast(final String... parameterValues) {
+    if (parameterValues.length > 0 && parameterValues.length % 2 == 0) {
+      // ok
+    } else {
+      throw new IllegalStateException("It requires 'pair's of config parameter-values.");
+    }
+    CommandArguments args = new CommandArguments(Protocol.Command.CONFIG).add(Protocol.Keyword.SET)
+        .addObjects((Object[]) parameterValues);
+    return broadcastCommand(new CommandObject<>(args, BuilderFactory.STRING));
+  }
+
+  public Map<?, Supplier<String>> flushAllBroadcast() {
+    return broadcastCommand(new CommandObject<>(new CommandArguments(Protocol.Command.FLUSHALL),
+        BuilderFactory.STRING));
+  }
+
+  public Map<?, Supplier<String>> flushAllBroadcast(FlushMode flushMode) {
+    return broadcastCommand(new CommandObject<>(new CommandArguments(Protocol.Command.FLUSHALL)
+        .add(flushMode), BuilderFactory.STRING));
+  }
+
   // Random node commands
   public long publish(String channel, String message) {
     return executeCommand(commandObjects.publish(channel, message));
@@ -3451,11 +3535,33 @@ public String ftCreate(String indexName, IndexOptions indexOptions, Schema schem
     return executeCommand(commandObjects.ftCreate(indexName, indexOptions, schema));
   }
 
+  public Map<?, Supplier<String>> ftCreateBroadcast(String indexName, IndexOptions indexOptions, Schema schema) {
+    return broadcastCommand(commandObjects.ftCreate(indexName, indexOptions, schema));
+  }
+
   @Override
   public String ftCreate(String indexName, FTCreateParams createParams, Iterable<SchemaField> schemaFields) {
     return executeCommand(commandObjects.ftCreate(indexName, createParams, schemaFields));
   }
 
+  public Map<?, Supplier<String>> ftCreateBroadcast(String indexName, SchemaField... schemaFields) {
+    return ftCreateBroadcast(indexName, Arrays.asList(schemaFields));
+  }
+
+  public Map<?, Supplier<String>> ftCreateBroadcast(String indexName, FTCreateParams createParams,
+      SchemaField... schemaFields) {
+    return ftCreateBroadcast(indexName, createParams, Arrays.asList(schemaFields));
+  }
+
+  public Map<?, Supplier<String>> ftCreateBroadcast(String indexName, Iterable<SchemaField> schemaFields) {
+    return ftCreateBroadcast(indexName, FTCreateParams.createParams(), schemaFields);
+  }
+
+  public Map<?, Supplier<String>> ftCreateBroadcast(String indexName, FTCreateParams createParams,
+      Iterable<SchemaField> schemaFields) {
+    return broadcastCommand(commandObjects.ftCreate(indexName, createParams, schemaFields));
+  }
+
   @Override
   public String ftAlter(String indexName, Schema schema) {
     return executeCommand(commandObjects.ftAlter(indexName, schema));
@@ -3539,6 +3645,14 @@ public String ftDropIndexDD(String indexName) {
     return executeCommand(commandObjects.ftDropIndexDD(indexName));
   }
 
+  public Map<?, Supplier<String>> ftDropIndexBroadcast(String indexName) {
+    return broadcastCommand(commandObjects.ftDropIndex(indexName));
+  }
+
+  public Map<?, Supplier<String>> ftDropIndexDDBroadcast(String indexName) {
+    return broadcastCommand(commandObjects.ftDropIndexDD(indexName));
+  }
+
   @Override
   public String ftSynUpdate(String indexName, String synonymGroupId, String... terms) {
     return executeCommand(commandObjects.ftSynUpdate(indexName, synonymGroupId, terms));

src/main/java/redis/clients/jedis/providers/ClusterConnectionProvider.java

@@ -135,4 +135,9 @@ public Connection getConnectionFromSlot(int slot) {
       }
     }
   }
+
+  @Override
+  public Map<String, ConnectionPool> getConnectionMap() {
+    return Collections.unmodifiableMap(getNodes());
+  }
 }

src/main/java/redis/clients/jedis/providers/ConnectionProvider.java

@@ -1,5 +1,7 @@
 package redis.clients.jedis.providers;
 
+import java.util.Collections;
+import java.util.Map;
 import redis.clients.jedis.CommandArguments;
 import redis.clients.jedis.Connection;
 
@@ -8,4 +10,9 @@ public interface ConnectionProvider extends AutoCloseable {
   Connection getConnection();
 
   Connection getConnection(CommandArguments args);
+
+  default Map<?, ?> getConnectionMap() {
+    final Connection c = getConnection();
+    return Collections.singletonMap(c.toString(), c);
+  }
 }