diff --git a/CHANGELOG.md b/CHANGELOG.md index 083641eb..10d5cd3c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Changes +* Add `CLIENT PAUSE` and `CLIENT UNPAUSE` commands for standalone and cluster clients ([#274](https://github.com/valkey-io/valkey-glide-php/pull/274)) * Add `RESET` command for standalone and cluster clients ([#273](https://github.com/valkey-io/valkey-glide-php/pull/273)) * Add `SAVE` command for standalone and cluster clients ([#272](https://github.com/valkey-io/valkey-glide-php/pull/272)) * Add `BGREWRITEAOF` command for standalone and cluster clients ([#271](https://github.com/valkey-io/valkey-glide-php/pull/271)) diff --git a/tests/ValkeyGlideClusterTest.php b/tests/ValkeyGlideClusterTest.php index e2262243..6d9d3e7c 100644 --- a/tests/ValkeyGlideClusterTest.php +++ b/tests/ValkeyGlideClusterTest.php @@ -637,6 +637,113 @@ public function testBgRewriteAofBatch() } } + public function testClientPause() + { + $key = '{client_pause_all}_' . uniqid(); + $this->valkey_glide->set($key, 'before'); + + $pauseMs = 2000; + $result = $this->valkey_glide->clientPause($pauseMs); + $this->assertTrue($result); + + // Verify that reads are blocked until the pause expires + $start = microtime(true); + $value = $this->valkey_glide->get($key); + $elapsed = (microtime(true) - $start) * 1000; + + $this->assertEquals('before', $value); + $this->assertGTE($pauseMs * 0.8, $elapsed); + + $this->valkey_glide->del($key); + } + + public function testClientPauseWithMode() + { + if (!$this->minVersionCheck('6.2.0')) { + $this->markTestSkipped('CLIENT PAUSE WRITE requires 6.2.0+'); + return; + } + + $key = '{client_pause_write}_' . uniqid(); + $this->valkey_glide->set($key, 'value'); + + // Pause with WRITE mode - only writes are blocked, reads still allowed + $pauseMs = 60000; + $result = $this->valkey_glide->clientPause($pauseMs, 'WRITE'); + $this->assertTrue($result); + + try { + // Reads should still work quickly during WRITE pause + $start = microtime(true); + $value = $this->valkey_glide->get($key); + $elapsed = (microtime(true) - $start) * 1000; + + $this->assertEquals('value', $value); + $this->assertLT($pauseMs, $elapsed); + } finally { + $this->valkey_glide->clientUnpause(); + } + + $this->valkey_glide->del($key); + } + + public function testClientUnpause() + { + if (!$this->minVersionCheck('6.2.0')) { + $this->markTestSkipped('CLIENT PAUSE WRITE requires 6.2.0+'); + return; + } + + $key = '{client_unpause}_' . uniqid(); + $this->valkey_glide->set($key, 'before'); + + // Pause writes for a long time + $pauseMs = 60000; + $this->valkey_glide->clientPause($pauseMs, 'WRITE'); + + // Unpause should unblock immediately + $result = $this->valkey_glide->clientUnpause(); + $this->assertTrue($result); + + // Verify writes work quickly after unpause + $start = microtime(true); + $this->valkey_glide->set($key, 'after'); + $elapsed = (microtime(true) - $start) * 1000; + + $this->assertLT($pauseMs, $elapsed); + $this->assertEquals('after', $this->valkey_glide->get($key)); + + $this->valkey_glide->del($key); + } + + public function testClientPauseWithReplyLiteral() + { + $this->withOptReplyLiteralEnabled(function () { + $result = $this->valkey_glide->clientPause(100); + $this->assertIsString($result); + $this->assertEquals('OK', $result); + + $result = $this->valkey_glide->clientUnpause(); + $this->assertIsString($result); + $this->assertEquals('OK', $result); + }); + } + + public function testClientPauseBatch() + { + if (!$this->havePipeline()) { + $this->markTestSkipped('Pipeline not supported'); + return; + } + + $this->valkey_glide->pipeline(); + $this->valkey_glide->clientPause(100); + $this->valkey_glide->clientUnpause(); + $result = $this->valkey_glide->exec(); + $this->assertTrue($result[0]); + $this->assertTrue($result[1]); + } + public function testReset() { $key = '{reset_test}_' . uniqid(); diff --git a/tests/ValkeyGlideTest.php b/tests/ValkeyGlideTest.php index 11af49ad..96b4f016 100644 --- a/tests/ValkeyGlideTest.php +++ b/tests/ValkeyGlideTest.php @@ -2721,6 +2721,113 @@ public function testBgRewriteAofBatch() $this->assertTrue($result[0]); } + public function testClientPause() + { + $key = 'client_pause_all_' . $this->createRandomString(); + $this->valkey_glide->set($key, 'before'); + + $pauseMs = 2000; + $result = $this->valkey_glide->clientPause($pauseMs); + $this->assertTrue($result); + + // Verify that reads are blocked until the pause expires + $start = microtime(true); + $value = $this->valkey_glide->get($key); + $elapsed = (microtime(true) - $start) * 1000; + + $this->assertEquals('before', $value); + $this->assertGTE($pauseMs * 0.8, $elapsed); + + $this->valkey_glide->del($key); + } + + public function testClientPauseWithMode() + { + if (!$this->minVersionCheck('6.2.0')) { + $this->markTestSkipped('CLIENT PAUSE WRITE requires 6.2.0+'); + return; + } + + $key = 'client_pause_write_' . $this->createRandomString(); + $this->valkey_glide->set($key, 'value'); + + // Pause with WRITE mode - only writes are blocked, reads still allowed + $pauseMs = 60000; + $result = $this->valkey_glide->clientPause($pauseMs, 'WRITE'); + $this->assertTrue($result); + + try { + // Reads should still work quickly during WRITE pause + $start = microtime(true); + $value = $this->valkey_glide->get($key); + $elapsed = (microtime(true) - $start) * 1000; + + $this->assertEquals('value', $value); + $this->assertLT($pauseMs, $elapsed); + } finally { + $this->valkey_glide->clientUnpause(); + } + + $this->valkey_glide->del($key); + } + + public function testClientUnpause() + { + if (!$this->minVersionCheck('6.2.0')) { + $this->markTestSkipped('CLIENT PAUSE WRITE requires 6.2.0+'); + return; + } + + $key = 'client_unpause_' . $this->createRandomString(); + $this->valkey_glide->set($key, 'before'); + + // Pause writes for a long time + $pauseMs = 60000; + $this->valkey_glide->clientPause($pauseMs, 'WRITE'); + + // Unpause should unblock immediately + $result = $this->valkey_glide->clientUnpause(); + $this->assertTrue($result); + + // Verify writes work quickly after unpause + $start = microtime(true); + $this->valkey_glide->set($key, 'after'); + $elapsed = (microtime(true) - $start) * 1000; + + $this->assertLT($pauseMs, $elapsed); + $this->assertEquals('after', $this->valkey_glide->get($key)); + + $this->valkey_glide->del($key); + } + + public function testClientPauseWithReplyLiteral() + { + $this->withOptReplyLiteralEnabled(function () { + $result = $this->valkey_glide->clientPause(100); + $this->assertIsString($result); + $this->assertEquals('OK', $result); + + $result = $this->valkey_glide->clientUnpause(); + $this->assertIsString($result); + $this->assertEquals('OK', $result); + }); + } + + public function testClientPauseBatch() + { + if (!$this->havePipeline()) { + $this->markTestSkipped('Pipeline not supported'); + return; + } + + $this->valkey_glide->pipeline(); + $this->valkey_glide->clientPause(100); + $this->valkey_glide->clientUnpause(); + $result = $this->valkey_glide->exec(); + $this->assertTrue($result[0]); + $this->assertTrue($result[1]); + } + public function testReset() { $key = '{reset_test}_' . $this->createRandomString(); diff --git a/valkey_glide.stub.php b/valkey_glide.stub.php index 325109ea..fc6e64ba 100644 --- a/valkey_glide.stub.php +++ b/valkey_glide.stub.php @@ -2360,6 +2360,35 @@ public function pfmerge(string $dst, array $srckeys): ValkeyGlide|bool; */ public function ping(?string $message = null): ValkeyGlide|string|bool; + /** + * Suspend all clients for the specified timeout. + * + * All subsequent commands from all clients will be blocked until the timeout expires + * or CLIENT UNPAUSE is called. + * + * @param int $timeout The time in milliseconds to pause clients. + * @param string|null $mode Optional pause mode: "ALL" (default) pauses all commands, + * "WRITE" pauses only write commands (reads still allowed). + * + * @return ValkeyGlide|bool|string Returns true on success, false on failure. + * With OPT_REPLY_LITERAL enabled, returns "OK" on success, + * false on failure. + * + * @see https://valkey.io/commands/client-pause + */ + public function clientPause(int $timeout, ?string $mode = null): ValkeyGlide|bool|string; + + /** + * Resume processing commands on all clients that were paused by CLIENT PAUSE. + * + * @return ValkeyGlide|bool|string Returns true on success, false on failure. + * With OPT_REPLY_LITERAL enabled, returns "OK" on success, + * false on failure. + * + * @see https://valkey.io/commands/client-unpause + */ + public function clientUnpause(): ValkeyGlide|bool|string; + /** * Reset the connection's server-side context. * diff --git a/valkey_glide_cluster.c b/valkey_glide_cluster.c index 03c633ff..37856d20 100644 --- a/valkey_glide_cluster.c +++ b/valkey_glide_cluster.c @@ -985,6 +985,14 @@ BGSAVE_METHOD_IMPL(ValkeyGlideCluster) BGREWRITEAOF_METHOD_IMPL(ValkeyGlideCluster) /* }}} */ +/* {{{ proto ValkeyGlideCluster::clientPause(int timeout [, string mode]) */ +CLIENT_PAUSE_METHOD_IMPL(ValkeyGlideCluster) +/* }}} */ + +/* {{{ proto ValkeyGlideCluster::clientUnpause() */ +CLIENT_UNPAUSE_METHOD_IMPL(ValkeyGlideCluster) +/* }}} */ + /* {{{ proto ValkeyGlideCluster::reset() */ RESET_METHOD_IMPL(ValkeyGlideCluster) /* }}} */ diff --git a/valkey_glide_cluster.stub.php b/valkey_glide_cluster.stub.php index 6684150b..4068048b 100644 --- a/valkey_glide_cluster.stub.php +++ b/valkey_glide_cluster.stub.php @@ -971,6 +971,16 @@ public function pfmerge(string $key, array $keys): ValkeyGlideCluster|bool; */ public function ping(mixed $route, ?string $message = null): mixed; + /** + * @see ValkeyGlide::clientPause + */ + public function clientPause(int $timeout, ?string $mode = null): ValkeyGlideCluster|bool|string; + + /** + * @see ValkeyGlide::clientUnpause + */ + public function clientUnpause(): ValkeyGlideCluster|bool|string; + /** * @see ValkeyGlide::reset */ diff --git a/valkey_glide_commands.c b/valkey_glide_commands.c index 8e0fd2e7..c98b545e 100644 --- a/valkey_glide_commands.c +++ b/valkey_glide_commands.c @@ -432,6 +432,84 @@ int execute_reset_command(zval* object, int argc, zval* return_value, zend_class return execute_and_handle_batch(valkey_glide, &core_args, processor, return_value, object); } +/* Execute a CLIENT PAUSE command using the Valkey Glide client - UNIFIED IMPLEMENTATION */ +int execute_client_pause_command(zval* object, int argc, zval* return_value, zend_class_entry* ce) { + valkey_glide_object* valkey_glide; + long timeout; + char* mode = NULL; + size_t mode_len = 0; + + /* Get ValkeyGlide object */ + valkey_glide = VALKEY_GLIDE_PHP_ZVAL_GET_OBJECT(valkey_glide_object, object); + if (!valkey_glide || !valkey_glide->glide_client) { + return 0; + } + + /* Parse parameters: timeout (required), mode (optional) */ + if (zend_parse_method_parameters( + argc, object, "Ol|s!", &object, ce, &timeout, &mode, &mode_len) == FAILURE) { + return 0; + } + + /* Setup core command arguments */ + core_command_args_t core_args = {0}; + core_args.glide_client = valkey_glide->glide_client; + core_args.cmd_type = ClientPause; + core_args.is_cluster = (ce == get_valkey_glide_cluster_ce()); + + /* Add timeout argument */ + core_args.args[0].type = CORE_ARG_TYPE_LONG; + core_args.args[0].data.long_arg.value = timeout; + core_args.arg_count = 1; + + /* Add optional mode argument (ALL or WRITE) */ + if (mode && mode_len > 0) { + core_args.args[1].type = CORE_ARG_TYPE_STRING; + core_args.args[1].data.string_arg.value = mode; + core_args.args[1].data.string_arg.len = mode_len; + core_args.arg_count = 2; + } + + /* Select processor based on OPT_REPLY_LITERAL */ + z_result_processor_t processor = valkey_glide->opt_reply_literal + ? process_core_status_string_result + : process_core_status_bool_result; + + return execute_and_handle_batch(valkey_glide, &core_args, processor, return_value, object); +} + +/* Execute a CLIENT UNPAUSE command using the Valkey Glide client - UNIFIED IMPLEMENTATION */ +int execute_client_unpause_command(zval* object, + int argc, + zval* return_value, + zend_class_entry* ce) { + valkey_glide_object* valkey_glide; + + /* Get ValkeyGlide object */ + valkey_glide = VALKEY_GLIDE_PHP_ZVAL_GET_OBJECT(valkey_glide_object, object); + if (!valkey_glide || !valkey_glide->glide_client) { + return 0; + } + + /* No parameters - validate object only */ + if (zend_parse_method_parameters(argc, object, "O", &object, ce) == FAILURE) { + return 0; + } + + /* Setup core command arguments */ + core_command_args_t core_args = {0}; + core_args.glide_client = valkey_glide->glide_client; + core_args.cmd_type = ClientUnpause; + core_args.is_cluster = (ce == get_valkey_glide_cluster_ce()); + + /* Select processor based on OPT_REPLY_LITERAL */ + z_result_processor_t processor = valkey_glide->opt_reply_literal + ? process_core_status_string_result + : process_core_status_bool_result; + + return execute_and_handle_batch(valkey_glide, &core_args, processor, return_value, object); +} + /* Execute a TIME command using the Valkey Glide client - UNIFIED IMPLEMENTATION */ int execute_time_command(zval* object, int argc, zval* return_value, zend_class_entry* ce) { valkey_glide_object* valkey_glide; @@ -531,7 +609,6 @@ int execute_unwatch_command(zval* object, int argc, zval* return_value, zend_cla return 0; } - /* Execute using core framework */ core_command_args_t args = {0}; args.glide_client = valkey_glide->glide_client; args.cmd_type = UnWatch; diff --git a/valkey_glide_commands_common.h b/valkey_glide_commands_common.h index 8818dfbc..1cfbbf12 100644 --- a/valkey_glide_commands_common.h +++ b/valkey_glide_commands_common.h @@ -206,6 +206,11 @@ int execute_flushdb_command(zval* object, int argc, zval* return_value, zend_cla int execute_flushall_command(zval* object, int argc, zval* return_value, zend_class_entry* ce); int execute_bgsave_command(zval* object, int argc, zval* return_value, zend_class_entry* ce); int execute_bgrewriteaof_command(zval* object, int argc, zval* return_value, zend_class_entry* ce); +int execute_client_pause_command(zval* object, int argc, zval* return_value, zend_class_entry* ce); +int execute_client_unpause_command(zval* object, + int argc, + zval* return_value, + zend_class_entry* ce); int execute_reset_command(zval* object, int argc, zval* return_value, zend_class_entry* ce); int execute_save_command(zval* object, int argc, zval* return_value, zend_class_entry* ce); int execute_time_command(zval* object, int argc, zval* return_value, zend_class_entry* ce); @@ -489,6 +494,12 @@ int execute_unlink_command(zval* object, int argc, zval* return_value, zend_clas #define RESET_METHOD_IMPL(class_name) STANDARD_METHOD_IMPL(class_name, reset, execute_reset_command) +#define CLIENT_PAUSE_METHOD_IMPL(class_name) \ + STANDARD_METHOD_IMPL(class_name, clientPause, execute_client_pause_command) + +#define CLIENT_UNPAUSE_METHOD_IMPL(class_name) \ + STANDARD_METHOD_IMPL(class_name, clientUnpause, execute_client_unpause_command) + #define TIME_METHOD_IMPL(class_name) STANDARD_METHOD_IMPL(class_name, time, execute_time_command) #define SCAN_METHOD_IMPL(class_name) STANDARD_METHOD_IMPL(class_name, scan, execute_scan_command) diff --git a/valkey_glide_core_common.c b/valkey_glide_core_common.c index 5703828d..e86f6b2a 100644 --- a/valkey_glide_core_common.c +++ b/valkey_glide_core_common.c @@ -197,6 +197,7 @@ int prepare_core_args(core_command_args_t* args, switch (args->cmd_type) { /* Zero argument operations */ case BgRewriteAof: + case ClientUnpause: case DBSize: case Discard: case Exec: @@ -311,6 +312,7 @@ int prepare_core_args(core_command_args_t* args, /* Message operations (no key, just arguments) */ case BgSave: + case ClientPause: case Echo: case FlushAll: case FlushDB: diff --git a/valkey_z_php_methods.c b/valkey_z_php_methods.c index 8d1c193e..3b50da5f 100644 --- a/valkey_z_php_methods.c +++ b/valkey_z_php_methods.c @@ -822,6 +822,14 @@ BGSAVE_METHOD_IMPL(ValkeyGlide) BGREWRITEAOF_METHOD_IMPL(ValkeyGlide) /* }}} */ +/* {{{ proto bool ValkeyGlide::clientPause(int timeout [, string mode]) */ +CLIENT_PAUSE_METHOD_IMPL(ValkeyGlide) +/* }}} */ + +/* {{{ proto bool ValkeyGlide::clientUnpause() */ +CLIENT_UNPAUSE_METHOD_IMPL(ValkeyGlide) +/* }}} */ + /* {{{ proto bool ValkeyGlide::reset() */ RESET_METHOD_IMPL(ValkeyGlide) /* }}} */