Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
2f2987ff
Commit
2f2987ff
authored
Jul 31, 2018
by
antirez
Browse files
Streams IDs parsing refactoring.
Related to #5184.
parent
187fa786
Changes
1
Hide whitespace changes
Inline
Side-by-side
src/t_stream.c
View file @
2f2987ff
...
@@ -1059,18 +1059,19 @@ int string2ull(const char *s, unsigned long long *value) {
...
@@ -1059,18 +1059,19 @@ int string2ull(const char *s, unsigned long long *value) {
* to the client, otherwise C_OK is returned. The ID may be in incomplete
* to the client, otherwise C_OK is returned. The ID may be in incomplete
* form, just stating the milliseconds time part of the stream. In such a case
* form, just stating the milliseconds time part of the stream. In such a case
* the missing part is set according to the value of 'missing_seq' parameter.
* the missing part is set according to the value of 'missing_seq' parameter.
*
* The IDs "-" and "+" specify respectively the minimum and maximum IDs
* The IDs "-" and "+" specify respectively the minimum and maximum IDs
* that can be represented. If '
mmid_supp
' is set to
0
, "-" and "+" will be
* that can be represented. If '
strict
' is set to
1
, "-" and "+" will be
* treated as an invalid ID.
* treated as an invalid ID.
*
*
* If 'c' is set to NULL, no reply is sent to the client. */
* If 'c' is set to NULL, no reply is sent to the client. */
int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq, int
mmid_supp
) {
int stream
Generic
ParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq, int
strict
) {
char buf[128];
char buf[128];
if (sdslen(o->ptr) > sizeof(buf)-1) goto invalid;
if (sdslen(o->ptr) > sizeof(buf)-1) goto invalid;
memcpy(buf,o->ptr,sdslen(o->ptr)+1);
memcpy(buf,o->ptr,sdslen(o->ptr)+1);
if (
!mmid_supp
&& (buf[0] == '-' || buf[0] == '+') &&
if (
strict
&& (buf[0] == '-' || buf[0] == '+') &&
buf[1] == '\0')
buf[1] == '\0')
goto invalid;
goto invalid;
/* Handle the "-" and "+" special cases. */
/* Handle the "-" and "+" special cases. */
if (buf[0] == '-' && buf[1] == '\0') {
if (buf[0] == '-' && buf[1] == '\0') {
...
@@ -1100,6 +1101,20 @@ invalid:
...
@@ -1100,6 +1101,20 @@ invalid:
return C_ERR;
return C_ERR;
}
}
/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
* 0, to be used when - and + are accetable IDs. */
int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
return streamGenericParseIDOrReply(c,o,id,missing_seq,0);
}
/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
* 1, to be used when we want to return an error if the special IDs + or -
* are provided. */
int streamParseStrictIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
return streamGenericParseIDOrReply(c,o,id,missing_seq,1);
}
/* XADD key [MAXLEN <count>] <ID or *> [field value] [field value] ... */
/* XADD key [MAXLEN <count>] <ID or *> [field value] [field value] ... */
void xaddCommand(client *c) {
void xaddCommand(client *c) {
streamID id;
streamID id;
...
@@ -1137,7 +1152,7 @@ void xaddCommand(client *c) {
...
@@ -1137,7 +1152,7 @@ void xaddCommand(client *c) {
maxlen_arg_idx = i;
maxlen_arg_idx = i;
} else {
} else {
/* If we are here is a syntax error or a valid ID. */
/* If we are here is a syntax error or a valid ID. */
if (streamParseIDOrReply(c,c->argv[i],&id,0
,0
) != C_OK) return;
if (streamParse
Strict
IDOrReply(c,c->argv[i],&id,0) != C_OK) return;
id_given = 1;
id_given = 1;
break;
break;
}
}
...
@@ -1206,8 +1221,8 @@ void xrangeGenericCommand(client *c, int rev) {
...
@@ -1206,8 +1221,8 @@ void xrangeGenericCommand(client *c, int rev) {
robj *startarg = rev ? c->argv[3] : c->argv[2];
robj *startarg = rev ? c->argv[3] : c->argv[2];
robj *endarg = rev ? c->argv[2] : c->argv[3];
robj *endarg = rev ? c->argv[2] : c->argv[3];
if (streamParseIDOrReply(c,startarg,&startid,0
,1
) == C_ERR) return;
if (streamParseIDOrReply(c,startarg,&startid,0) == C_ERR) return;
if (streamParseIDOrReply(c,endarg,&endid,UINT64_MAX
,1
) == C_ERR) return;
if (streamParseIDOrReply(c,endarg,&endid,UINT64_MAX) == C_ERR) return;
/* Parse the COUNT option if any. */
/* Parse the COUNT option if any. */
if (c->argc > 4) {
if (c->argc > 4) {
...
@@ -1393,7 +1408,7 @@ void xreadCommand(client *c) {
...
@@ -1393,7 +1408,7 @@ void xreadCommand(client *c) {
ids[id_idx].seq = UINT64_MAX;
ids[id_idx].seq = UINT64_MAX;
continue;
continue;
}
}
if (streamParseIDOrReply(c,c->argv[i],ids+id_idx,0
,0
) != C_OK)
if (streamParse
Strict
IDOrReply(c,c->argv[i],ids+id_idx,0) != C_OK)
goto cleanup;
goto cleanup;
}
}
...
@@ -1667,7 +1682,7 @@ NULL
...
@@ -1667,7 +1682,7 @@ NULL
streamID id;
streamID id;
if (!strcmp(c->argv[4]->ptr,"$")) {
if (!strcmp(c->argv[4]->ptr,"$")) {
id = s->last_id;
id = s->last_id;
} else if (streamParseIDOrReply(c,c->argv[4],&id,0
,0
) != C_OK) {
} else if (streamParse
Strict
IDOrReply(c,c->argv[4],&id,0) != C_OK) {
return;
return;
}
}
streamCG *cg = streamCreateCG(s,grpname,sdslen(grpname),&id);
streamCG *cg = streamCreateCG(s,grpname,sdslen(grpname),&id);
...
@@ -1684,7 +1699,7 @@ NULL
...
@@ -1684,7 +1699,7 @@ NULL
streamID id;
streamID id;
if (!strcmp(c->argv[4]->ptr,"$")) {
if (!strcmp(c->argv[4]->ptr,"$")) {
id = s->last_id;
id = s->last_id;
} else if (streamParseIDOrReply(c,c->argv[4],&id,0
,1
) != C_OK) {
} else if (streamParseIDOrReply(c,c->argv[4],&id,0) != C_OK) {
return;
return;
}
}
cg->last_id = id;
cg->last_id = id;
...
@@ -1744,7 +1759,7 @@ void xackCommand(client *c) {
...
@@ -1744,7 +1759,7 @@ void xackCommand(client *c) {
for (int j = 3; j < c->argc; j++) {
for (int j = 3; j < c->argc; j++) {
streamID id;
streamID id;
unsigned char buf[sizeof(streamID)];
unsigned char buf[sizeof(streamID)];
if (streamParseIDOrReply(c,c->argv[j],&id,0
,0
) != C_OK) return;
if (streamParse
Strict
IDOrReply(c,c->argv[j],&id,0) != C_OK) return;
streamEncodeID(buf,&id);
streamEncodeID(buf,&id);
/* Lookup the ID in the group PEL: it will have a reference to the
/* Lookup the ID in the group PEL: it will have a reference to the
...
@@ -1791,9 +1806,9 @@ void xpendingCommand(client *c) {
...
@@ -1791,9 +1806,9 @@ void xpendingCommand(client *c) {
if (c->argc >= 6) {
if (c->argc >= 6) {
if (getLongLongFromObjectOrReply(c,c->argv[5],&count,NULL) == C_ERR)
if (getLongLongFromObjectOrReply(c,c->argv[5],&count,NULL) == C_ERR)
return;
return;
if (streamParseIDOrReply(c,c->argv[3],&startid,0
,1
) == C_ERR)
if (streamParseIDOrReply(c,c->argv[3],&startid,0) == C_ERR)
return;
return;
if (streamParseIDOrReply(c,c->argv[4],&endid,UINT64_MAX
,1
) == C_ERR)
if (streamParseIDOrReply(c,c->argv[4],&endid,UINT64_MAX) == C_ERR)
return;
return;
}
}
...
@@ -2002,7 +2017,7 @@ void xclaimCommand(client *c) {
...
@@ -2002,7 +2017,7 @@ void xclaimCommand(client *c) {
int j;
int j;
for (j = 4; j < c->argc; j++) {
for (j = 4; j < c->argc; j++) {
streamID id;
streamID id;
if (streamParseIDOrReply(NULL,c->argv[j],&id,0
,0
) != C_OK) break;
if (streamParse
Strict
IDOrReply(NULL,c->argv[j],&id,0) != C_OK) break;
}
}
int last_id_arg = j-1; /* Next time we iterate the IDs we now the range. */
int last_id_arg = j-1; /* Next time we iterate the IDs we now the range. */
...
@@ -2061,7 +2076,7 @@ void xclaimCommand(client *c) {
...
@@ -2061,7 +2076,7 @@ void xclaimCommand(client *c) {
for (int j = 5; j <= last_id_arg; j++) {
for (int j = 5; j <= last_id_arg; j++) {
streamID id;
streamID id;
unsigned char buf[sizeof(streamID)];
unsigned char buf[sizeof(streamID)];
if (streamParseIDOrReply(c,c->argv[j],&id,0
,0
) != C_OK) return;
if (streamParse
Strict
IDOrReply(c,c->argv[j],&id,0) != C_OK) return;
streamEncodeID(buf,&id);
streamEncodeID(buf,&id);
/* Lookup the ID in the group PEL. */
/* Lookup the ID in the group PEL. */
...
@@ -2144,13 +2159,13 @@ void xdelCommand(client *c) {
...
@@ -2144,13 +2159,13 @@ void xdelCommand(client *c) {
* executed because at some point an invalid ID is parsed. */
* executed because at some point an invalid ID is parsed. */
streamID id;
streamID id;
for (int j = 2; j < c->argc; j++) {
for (int j = 2; j < c->argc; j++) {
if (streamParseIDOrReply(c,c->argv[j],&id,0
,0
) != C_OK) return;
if (streamParse
Strict
IDOrReply(c,c->argv[j],&id,0) != C_OK) return;
}
}
/* Actually apply the command. */
/* Actually apply the command. */
int deleted = 0;
int deleted = 0;
for (int j = 2; j < c->argc; j++) {
for (int j = 2; j < c->argc; j++) {
streamParseIDOrReply(c,c->argv[j],&id,0
,0
); /* Retval already checked. */
streamParse
Strict
IDOrReply(c,c->argv[j],&id,0); /* Retval already checked. */
deleted += streamDeleteItem(s,&id);
deleted += streamDeleteItem(s,&id);
}
}
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment