CLIENT PAUSE command with mode option (#2601)

redis · Jul 26, 2021 · 6554807 · 6554807
1 parent 468bf8f
commit 6554807
Show file tree

Hide file tree

Showing 6 changed files with 114 additions and 0 deletions.
diff --git a/src/main/java/redis/clients/jedis/BinaryClient.java b/src/main/java/redis/clients/jedis/BinaryClient.java
@@ -26,6 +26,7 @@
 
 import redis.clients.jedis.Protocol.Keyword;
 import redis.clients.jedis.Protocol.SentinelKeyword;
+import redis.clients.jedis.args.ClientPauseMode;
 import redis.clients.jedis.args.ClientType;
 import redis.clients.jedis.args.ListDirection;
 import redis.clients.jedis.args.FlushMode;
@@ -1387,6 +1388,10 @@ public void clientUnblock(final long clientId, final UnblockType unblockType) {
     }
   }
 
+  public void clientPause(final long timeout, final ClientPauseMode mode) {
+    sendCommand(CLIENT, Keyword.PAUSE.getRaw(), toByteArray(timeout), mode.getRaw());
+  }
+
   public void time() {
     sendCommand(TIME);
   }

diff --git a/src/main/java/redis/clients/jedis/BinaryJedis.java b/src/main/java/redis/clients/jedis/BinaryJedis.java
@@ -4313,12 +4313,20 @@ public long clientUnblock(final long clientId, final UnblockType unblockType) {
     return client.getIntegerReply();
   }
 
+  @Override
   public String clientPause(final long timeout) {
     checkIsInMultiOrPipeline();
     client.clientPause(timeout);
     return client.getBulkReply();
   }
 
+  @Override
+  public String clientPause(final long timeout, final ClientPauseMode mode) {
+    checkIsInMultiOrPipeline();
+    client.clientPause(timeout, mode);
+    return client.getBulkReply();
+  }
+
   public List<String> time() {
     checkIsInMultiOrPipeline();
     client.time();

diff --git a/src/main/java/redis/clients/jedis/args/ClientPauseMode.java b/src/main/java/redis/clients/jedis/args/ClientPauseMode.java
@@ -0,0 +1,19 @@
+package redis.clients.jedis.args;
+
+import redis.clients.jedis.util.SafeEncoder;
+
+public enum ClientPauseMode implements Rawable {
+
+  ALL, WRITE;
+
+  private final byte[] raw;
+
+  private ClientPauseMode() {
+    raw = SafeEncoder.encode(name());
+  }
+
+  @Override
+  public byte[] getRaw() {
+    return raw;
+  }
+}
diff --git a/src/main/java/redis/clients/jedis/commands/AdvancedBinaryJedisCommands.java b/src/main/java/redis/clients/jedis/commands/AdvancedBinaryJedisCommands.java
@@ -3,6 +3,7 @@
 import java.util.List;
 
 import redis.clients.jedis.AccessControlUser;
+import redis.clients.jedis.args.ClientPauseMode;
 import redis.clients.jedis.args.ClientType;
 import redis.clients.jedis.args.UnblockType;
 import redis.clients.jedis.params.MigrateParams;
@@ -69,6 +70,10 @@ String migrate(String host, int port, int destinationDB, int timeout, MigratePar
 
   long clientUnblock(long clientId, UnblockType unblockType);
 
+  String clientPause(long timeout);
+
+  String clientPause(long timeout, ClientPauseMode mode);
+
   byte[] memoryDoctorBinary();
 
   Long memoryUsage(byte[] key);

diff --git a/src/main/java/redis/clients/jedis/commands/AdvancedJedisCommands.java b/src/main/java/redis/clients/jedis/commands/AdvancedJedisCommands.java
@@ -4,6 +4,7 @@
 
 import redis.clients.jedis.AccessControlLogEntry;
 import redis.clients.jedis.AccessControlUser;
+import redis.clients.jedis.args.ClientPauseMode;
 import redis.clients.jedis.args.ClientType;
 import redis.clients.jedis.args.UnblockType;
 import redis.clients.jedis.params.MigrateParams;
@@ -63,6 +64,10 @@ String migrate(String host, int port, int destinationDB, int timeout, MigratePar
 
   long clientUnblock(long clientId, UnblockType unblockType);
 
+  String clientPause(long timeout);
+
+  String clientPause(long timeout, ClientPauseMode mode);
+
   String memoryDoctor();
 
   Long memoryUsage(String key);

diff --git a/src/test/java/redis/clients/jedis/tests/commands/ControlCommandsTest.java b/src/test/java/redis/clients/jedis/tests/commands/ControlCommandsTest.java
@@ -20,6 +20,7 @@
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisMonitor;
 import redis.clients.jedis.Protocol;
+import redis.clients.jedis.args.ClientPauseMode;
 import redis.clients.jedis.exceptions.JedisDataException;
 import redis.clients.jedis.util.SafeEncoder;
 
@@ -204,6 +205,77 @@ public Long call() throws Exception {
     }
   }
 
+  @Test
+  public void clientPauseAll() throws InterruptedException, ExecutionException {
+    final int pauseMillis = 1250;
+    final int pauseMillisDelta = 100;
+
+    ExecutorService executorService = Executors.newFixedThreadPool(1);
+    try (Jedis jedisPause = createJedis()) {
+
+      jedis.clientPause(pauseMillis, ClientPauseMode.ALL);
+
+      Future<Long> latency = executorService.submit(new Callable<Long>() {
+        @Override
+        public Long call() throws Exception {
+          long startMillis = System.currentTimeMillis();
+          jedisPause.get("key");
+          return System.currentTimeMillis() - startMillis;
+        }
+      });
+
+      long latencyMillis = latency.get();
+      assertTrue(pauseMillis <= latencyMillis && latencyMillis <= pauseMillis + pauseMillisDelta);
+
+    } finally {
+      executorService.shutdown();
+      if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
+        executorService.shutdownNow();
+      }
+    }
+  }
+
+  @Test
+  public void clientPauseWrite() throws InterruptedException, ExecutionException {
+    final int pauseMillis = 1250;
+    final int pauseMillisDelta = 100;
+
+    ExecutorService executorService = Executors.newFixedThreadPool(2);
+    try (Jedis jedisRead = createJedis(); Jedis jedisWrite = createJedis();) {
+
+      jedis.clientPause(pauseMillis, ClientPauseMode.WRITE);
+
+      Future<Long> latencyRead = executorService.submit(new Callable<Long>() {
+        @Override
+        public Long call() throws Exception {
+          long startMillis = System.currentTimeMillis();
+          jedisRead.get("key");
+          return System.currentTimeMillis() - startMillis;
+        }
+      });
+      Future<Long> latencyWrite = executorService.submit(new Callable<Long>() {
+        @Override
+        public Long call() throws Exception {
+          long startMillis = System.currentTimeMillis();
+          jedisWrite.set("key", "value");
+          return System.currentTimeMillis() - startMillis;
+        }
+      });
+
+      long latencyReadMillis = latencyRead.get();
+      assertTrue(0 <= latencyReadMillis && latencyReadMillis <= pauseMillisDelta);
+
+      long latencyWriteMillis = latencyWrite.get();
+      assertTrue(pauseMillis <= latencyWriteMillis && latencyWriteMillis <= pauseMillis + pauseMillisDelta);
+
+    } finally {
+      executorService.shutdown();
+      if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
+        executorService.shutdownNow();
+      }
+    }
+  }
+
   @Test
   public void memoryDoctorString() {
     String memoryInfo = jedis.memoryDoctor();