Commit 4086dff4 authored by antirez's avatar antirez
Browse files

Streams: augment client.bpop with XREAD specific fields.

parent f80dfbf4
...@@ -384,6 +384,10 @@ void unblockClientWaitingData(client *c) { ...@@ -384,6 +384,10 @@ void unblockClientWaitingData(client *c) {
decrRefCount(c->bpop.target); decrRefCount(c->bpop.target);
c->bpop.target = NULL; c->bpop.target = NULL;
} }
if (c->bpop.xread_group) {
decrRefCount(c->bpop.xread_group);
c->bpop.xread_group = NULL;
}
} }
/* If the specified key has clients blocked waiting for list pushes, this /* If the specified key has clients blocked waiting for list pushes, this
......
...@@ -126,6 +126,7 @@ client *createClient(int fd) { ...@@ -126,6 +126,7 @@ client *createClient(int fd) {
c->bpop.timeout = 0; c->bpop.timeout = 0;
c->bpop.keys = dictCreate(&objectKeyHeapPointerValueDictType,NULL); c->bpop.keys = dictCreate(&objectKeyHeapPointerValueDictType,NULL);
c->bpop.target = NULL; c->bpop.target = NULL;
c->bpop.xread_group = NULL;
c->bpop.numreplicas = 0; c->bpop.numreplicas = 0;
c->bpop.reploffset = 0; c->bpop.reploffset = 0;
c->woff = 0; c->woff = 0;
......
...@@ -648,6 +648,11 @@ typedef struct blockingState { ...@@ -648,6 +648,11 @@ typedef struct blockingState {
robj *target; /* The key that should receive the element, robj *target; /* The key that should receive the element,
* for BRPOPLPUSH. */ * for BRPOPLPUSH. */
/* BLOCK_STREAM */
size_t xread_count; /* XREAD COUNT option. */
robj *xread_group; /* XREAD group name. */
mstime_t xread_retry_time, xread_retry_ttl;
/* BLOCKED_WAIT */ /* BLOCKED_WAIT */
int numreplicas; /* Number of replicas we are waiting for ACK. */ int numreplicas; /* Number of replicas we are waiting for ACK. */
long long reploffset; /* Replication offset to reach. */ long long reploffset; /* Replication offset to reach. */
......
...@@ -392,3 +392,11 @@ void xlenCommand(client *c) { ...@@ -392,3 +392,11 @@ void xlenCommand(client *c) {
stream *s = o->ptr; stream *s = o->ptr;
addReplyLongLong(c,s->length); addReplyLongLong(c,s->length);
} }
/* XREAD [BLOCK <milliseconds>] [COUNT <count>] [GROUP <groupname> <ttl>]
* [RETRY <milliseconds> <ttl>] STREAMS key_1 ID_1 key_2 ID_2 ...
* key_N ID_N */
void xreadCommand(client *c) {
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment