Sitelet https://github.com/phpredis/phpredis/commit/1343f500
Skip to content

Commit 1343f50

Browse files
michael-grundermichael-grunder
committed
XGROUP DELCONSUMER and ENTRIESREAD
Refactor XGROUP and implement the new DELCONSUMER (Redis 6.2.0) and ENTRIESREAD (Redis 7.0.0) options. Additionally, add a proper phpdoc block to the stub file. See #2068
1 parent 71bcbcb commit 1343f50

5 files changed

Lines changed: 137 additions & 44 deletions

File tree

‎redis.stub.php‎

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -660,7 +660,44 @@ public function xclaim(string $key, string $group, string $consumer, int $min_id
660660

661661
public function xdel(string $key, array $ids): Redis|int|false;
662662

663-
public function xgroup(string $operation, string $key = null, string $arg1 = null, string $arg2 = null, bool $arg3 = false): mixed;
663+
/**
664+
* XGROUP
665+
*
666+
* Perform various operation on consumer groups for a particular Redis STREAM.
667+
* What the command does is primarily based on which operation is passed.
668+
*
669+
* @see https://redis.io/commands/xgroup/
670+
*
671+
* @param string $operation The subcommand you intend to execute. Valid options are as follows
672+
* 'HELP' - Redis will return information about the command
673+
* Requires: none
674+
* 'CREATE' - Create a consumer group.
675+
* Requires: Key, group, consumer.
676+
* 'SETID' - Set the ID of an existing consumer group for the stream.
677+
* Requires: Key, group, id.
678+
* 'CREATECONSUMER - Create a new consumer group for the stream. You must
679+
* also pass key, group, and the consumer name you wish to
680+
* create.
681+
* Requires: Key, group, consumer.
682+
* 'DELCONSUMER' - Delete a consumer from group attached to the stream.
683+
* Requires: Key, group, consumer.
684+
* 'DESTROY' - Delete a consumer group from a stream.
685+
* Requires: Key, group.
686+
* @param string $key The STREAM we're operating on.
687+
* @param string $group The consumer group wse want to create/modify/delete.
688+
* @param string $id_or_consumer The STREAM id (e.g. '$') or consumer group. See the operation section
689+
* for information about which to send.
690+
* @param bool $mkstream This flag may be sent in combination with the 'CREATE' operation, and
691+
* cause Redis to also create the STREAM if it doesn't currently exist.
692+
*
693+
* @param bool $entriesread Allows you to set Redis's 'entries-read' STREAM value. This argument is
694+
* only relevant to the 'CREATE' and 'SETID' operations.
695+
* Note: Requires Redis >= 7.0.0.
696+
*
697+
* @return mixed This command return various results depending on the operation performed.
698+
*/
699+
public function xgroup(string $operation, string $key = null, string $group = null, string $id_or_consumer = null,
700+
bool $mkstream = false, int $entries_read = -2): mixed;
664701

665702
public function xinfo(string $operation, ?string $arg1 = null, ?string $arg2 = null, int $count = -1): mixed;
666703

‎redis_arginfo.h‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/* This is a generated file, edit the .stub.php file instead.
2-
* Stub hash: 73bbd79b67c155a90acfa2b5ca1be49effdaf8ba */
2+
* Stub hash: c04531e86379ab5c0de12e8e82868b7c7f024068 */
33

44
ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Redis___construct, 0, 0, 0)
55
ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, options, IS_ARRAY, 0, "null")
@@ -909,9 +909,10 @@ ZEND_END_ARG_INFO()
909909
ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_Redis_xgroup, 0, 1, IS_MIXED, 0)
910910
ZEND_ARG_TYPE_INFO(0, operation, IS_STRING, 0)
911911
ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, key, IS_STRING, 0, "null")
912-
ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, arg1, IS_STRING, 0, "null")
913-
ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, arg2, IS_STRING, 0, "null")
914-
ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, arg3, _IS_BOOL, 0, "false")
912+
ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, group, IS_STRING, 0, "null")
913+
ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, id_or_consumer, IS_STRING, 0, "null")
914+
ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, mkstream, _IS_BOOL, 0, "false")
915+
ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, entries_read, IS_LONG, 0, "-2")
915916
ZEND_END_ARG_INFO()
916917

917918
ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_Redis_xinfo, 0, 1, IS_MIXED, 0)

‎redis_commands.c‎

Lines changed: 55 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -5817,52 +5817,72 @@ int redis_xclaim_cmd(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
58175817
}
58185818

58195819
/* XGROUP HELP
5820-
* XGROUP CREATE key groupname id [MKSTREAM]
5821-
* XGROUP SETID key group id
5822-
* XGROUP DESTROY key groupname
5823-
* XGROUP DELCONSUMER key groupname consumername */
5820+
* XGROUP CREATE key group id [MKSTREAM] [ENTRIESREAD <n>]
5821+
* XGROUP SETID key group id [ENTRIESREAD <n>]
5822+
* XGROUP CREATECONSUMER key group consumer
5823+
* XGROUP DELCONSUMER key group consumer
5824+
* XGROUP DESTROY key group
5825+
*/
58245826
int redis_xgroup_cmd(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
58255827
char **cmd, int *cmd_len, short *slot, void **ctx)
58265828
{
5827-
char *op, *key = NULL, *arg1 = NULL, *arg2 = NULL;
5828-
size_t oplen, keylen, arg1len, arg2len;
5829+
zend_string *op = NULL, *key = NULL, *group = NULL, *id_or_consumer = NULL;
5830+
int nargs, is_create = 0, is_setid = 0;
5831+
zend_long entries_read = -2;
5832+
smart_string cmdstr = {0};
58295833
zend_bool mkstream = 0;
5830-
int argc = ZEND_NUM_ARGS();
58315834

5832-
if (zend_parse_parameters(ZEND_NUM_ARGS(), "s|sssb", &op, &oplen,
5833-
&key, &keylen, &arg1, &arg1len, &arg2, &arg2len,
5834-
&mkstream) == FAILURE)
5835+
ZEND_PARSE_PARAMETERS_START(1, 6)
5836+
Z_PARAM_STR(op)
5837+
Z_PARAM_OPTIONAL
5838+
Z_PARAM_STR(key)
5839+
Z_PARAM_STR(group)
5840+
Z_PARAM_STR(id_or_consumer)
5841+
Z_PARAM_BOOL(mkstream)
5842+
Z_PARAM_LONG(entries_read)
5843+
ZEND_PARSE_PARAMETERS_END_EX(return FAILURE);
5844+
5845+
if (zend_string_equals_literal_ci(op, "HELP")) {
5846+
nargs = 0;
5847+
} else if ((is_create = zend_string_equals_literal_ci(op, "CREATE")) ||
5848+
(is_setid = zend_string_equals_literal_ci(op, "SETID")) ||
5849+
zend_string_equals_literal_ci(op, "CREATECONSUMER") ||
5850+
zend_string_equals_literal_ci(op, "DELCONSUMER"))
58355851
{
5852+
nargs = 3;
5853+
} else if (zend_string_equals_literal_ci(op, "DESTROY")) {
5854+
nargs = 2;
5855+
} else {
5856+
php_error_docref(NULL, E_WARNING, "Unknown XGROUP operation '%s'", ZSTR_VAL(op));
58365857
return FAILURE;
58375858
}
58385859

5839-
if (argc == 1 && oplen == 4 && !strncasecmp(op, "HELP", 4)) {
5840-
*cmd_len = REDIS_CMD_SPPRINTF(cmd, "XGROUP", "s", "HELP", 4);
5841-
return SUCCESS;
5842-
} else if (argc >= 4 && (oplen == 6 && !strncasecmp(op, "CREATE", 6))) {
5843-
if (mkstream) {
5844-
*cmd_len = REDIS_CMD_SPPRINTF(cmd, "XGROUP", "sksss", op, oplen, key, keylen,
5845-
arg1, arg1len, arg2, arg2len, "MKSTREAM",
5846-
sizeof("MKSTREAM") - 1);
5847-
} else {
5848-
*cmd_len = REDIS_CMD_SPPRINTF(cmd, "XGROUP", "skss", op, oplen, key, keylen,
5849-
arg1, arg1len, arg2, arg2len);
5850-
}
5851-
return SUCCESS;
5852-
} else if (argc == 4 && ((oplen == 5 && !strncasecmp(op, "SETID", 5)) ||
5853-
(oplen == 11 && !strncasecmp(op, "DELCONSUMER", 11))))
5854-
{
5855-
*cmd_len = REDIS_CMD_SPPRINTF(cmd, "XGROUP", "skss", op, oplen, key, keylen,
5856-
arg1, arg1len, arg2, arg2len);
5857-
return SUCCESS;
5858-
} else if (argc == 3 && ((oplen == 7 && !strncasecmp(op, "DESTROY", 7)))) {
5859-
*cmd_len = REDIS_CMD_SPPRINTF(cmd, "XGROUP", "sks", op, oplen, key,
5860-
keylen, arg1, arg1len);
5861-
return SUCCESS;
5860+
if (ZEND_NUM_ARGS() < nargs) {
5861+
php_error_docref(NULL, E_WARNING, "Operation '%s' requires %d arguments", ZSTR_VAL(op), nargs);
5862+
return FAILURE;
58625863
}
58635864

5864-
/* Didn't detect any valid XGROUP command pattern */
5865-
return FAILURE;
5865+
mkstream &= is_create;
5866+
if (!(is_create || is_setid))
5867+
entries_read = -2;
5868+
5869+
REDIS_CMD_INIT_SSTR_STATIC(&cmdstr, 1 + nargs + !!mkstream + (entries_read != -2 ? 2 : 0), "XGROUP");
5870+
redis_cmd_append_sstr_zstr(&cmdstr, op);
5871+
5872+
if (nargs-- > 0) redis_cmd_append_sstr_key_zstr(&cmdstr, key, redis_sock, slot);
5873+
if (nargs-- > 0) redis_cmd_append_sstr_zstr(&cmdstr, group);
5874+
if (nargs-- > 0) redis_cmd_append_sstr_zstr(&cmdstr, id_or_consumer);
5875+
5876+
REDIS_CMD_APPEND_SSTR_OPT_STATIC(&cmdstr, !!mkstream, "MKSTREAM");
5877+
if (entries_read != -2) {
5878+
REDIS_CMD_APPEND_SSTR_STATIC(&cmdstr, "ENTRIESREAD");
5879+
redis_cmd_append_sstr_long(&cmdstr, entries_read);
5880+
}
5881+
5882+
*cmd = cmdstr.c;
5883+
*cmd_len = cmdstr.len;
5884+
5885+
return SUCCESS;
58665886
}
58675887

58685888
/* XINFO CONSUMERS key group

‎redis_legacy_arginfo.h‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/* This is a generated file, edit the .stub.php file instead.
2-
* Stub hash: 73bbd79b67c155a90acfa2b5ca1be49effdaf8ba */
2+
* Stub hash: c04531e86379ab5c0de12e8e82868b7c7f024068 */
33

44
ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Redis___construct, 0, 0, 0)
55
ZEND_ARG_INFO(0, options)
@@ -769,9 +769,10 @@ ZEND_END_ARG_INFO()
769769
ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Redis_xgroup, 0, 0, 1)
770770
ZEND_ARG_INFO(0, operation)
771771
ZEND_ARG_INFO(0, key)
772-
ZEND_ARG_INFO(0, arg1)
773-
ZEND_ARG_INFO(0, arg2)
774-
ZEND_ARG_INFO(0, arg3)
772+
ZEND_ARG_INFO(0, group)
773+
ZEND_ARG_INFO(0, id_or_consumer)
774+
ZEND_ARG_INFO(0, mkstream)
775+
ZEND_ARG_INFO(0, entries_read)
775776
ZEND_END_ARG_INFO()
776777

777778
ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Redis_xinfo, 0, 0, 1)

‎tests/RedisTest.php‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6521,6 +6521,40 @@ public function testXGroup() {
65216521
$this->assertFalse($this->redis->xGroup('SETID', 's', 'mygroup', 'BAD_ID'));
65226522

65236523
$this->assertEquals($this->redis->xGroup('DELCONSUMER', 's', 'mygroup', 'myconsumer'),0);
6524+
6525+
if (!$this->minVersionCheck('6.2.0'))
6526+
return;
6527+
6528+
/* CREATECONSUMER */
6529+
$this->assertTrue($this->redis->del('s'));
6530+
$this->assertTrue($this->redis->xgroup('create', 's', 'mygroup', '$', true));
6531+
for ($i = 0; $i < 3; $i++) {
6532+
$this->assertTrue($this->redis->xgroup('createconsumer', 's', 'mygroup', "c:$i"));
6533+
$info = $this->redis->xinfo('consumers', 's', 'mygroup');
6534+
$this->assertTrue(is_array($info) && count($info) == $i + 1);
6535+
for ($j = 0; $j <= $i; $j++) {
6536+
$this->assertTrue(isset($info[$j]) && isset($info[$j]['name']) && $info[$j]['name'] == "c:$j");
6537+
}
6538+
}
6539+
6540+
/* Make sure we don't erroneously send options that don't belong to the operation */
6541+
$this->assertTrue($this->redis->xGroup('CREATECONSUMER', 's', 'mygroup', 'fake-consumer', true, 1337));
6542+
6543+
/* Make sure we handle the case where the user doesn't send enough arguments */
6544+
$this->redis->clearLastError();
6545+
$this->assertFalse(@$this->redis->xGroup('CREATECONSUMER'));
6546+
$this->assertEquals(NULL, $this->redis->getLastError());
6547+
$this->assertFalse(@$this->redis->xGroup('create'));
6548+
$this->assertEquals(NULL, $this->redis->getLastError());
6549+
6550+
if (!$this->minVersionCheck('7.0.0'))
6551+
return;
6552+
6553+
/* ENTRIESREAD */
6554+
$this->assertTrue($this->redis->del('s'));
6555+
$this->assertTrue($this->redis->xGroup('create', 's', 'mygroup', '$', true, 1337));
6556+
$info = $this->redis->xinfo('groups', 's');
6557+
$this->assertTrue(isset($info[0]['entries-read']) && 1337 == (int)$info[0]['entries-read']);
65246558
}
65256559

65266560
public function testXAck() {

0 commit comments

Comments
 (0)