Unverified Commit d3384260 authored by Bjorn Svensson's avatar Bjorn Svensson Committed by GitHub
Browse files

Support PING while subscribing (RESP2) (#1027)

* Handle PING during pubsub in RESP2

* Rename invalid callback list

Some commands are valid to send during a subscribe in RESP2, and
most in RESP3. Renaming the callback list from `invalid` to `replies`
to detail this fact.

* Fix review comment
parent e3a479e4
...@@ -144,8 +144,8 @@ static redisAsyncContext *redisAsyncInitialize(redisContext *c) { ...@@ -144,8 +144,8 @@ static redisAsyncContext *redisAsyncInitialize(redisContext *c) {
ac->replies.head = NULL; ac->replies.head = NULL;
ac->replies.tail = NULL; ac->replies.tail = NULL;
ac->sub.invalid.head = NULL; ac->sub.replies.head = NULL;
ac->sub.invalid.tail = NULL; ac->sub.replies.tail = NULL;
ac->sub.channels = channels; ac->sub.channels = channels;
ac->sub.patterns = patterns; ac->sub.patterns = patterns;
...@@ -312,9 +312,7 @@ static void __redisAsyncFree(redisAsyncContext *ac) { ...@@ -312,9 +312,7 @@ static void __redisAsyncFree(redisAsyncContext *ac) {
/* Execute pending callbacks with NULL reply. */ /* Execute pending callbacks with NULL reply. */
while (__redisShiftCallback(&ac->replies,&cb) == REDIS_OK) while (__redisShiftCallback(&ac->replies,&cb) == REDIS_OK)
__redisRunCallback(ac,&cb,NULL); __redisRunCallback(ac,&cb,NULL);
while (__redisShiftCallback(&ac->sub.replies,&cb) == REDIS_OK)
/* Execute callbacks for invalid commands */
while (__redisShiftCallback(&ac->sub.invalid,&cb) == REDIS_OK)
__redisRunCallback(ac,&cb,NULL); __redisRunCallback(ac,&cb,NULL);
/* Run subscription callbacks with NULL reply */ /* Run subscription callbacks with NULL reply */
...@@ -419,10 +417,11 @@ static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, ...@@ -419,10 +417,11 @@ static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply,
char *stype; char *stype;
sds sname; sds sname;
/* Custom reply functions are not supported for pub/sub. This will fail /* Match reply with the expected format of a pushed message.
* very hard when they are used... */ * The type and number of elements (3 to 4) are specified at:
if (reply->type == REDIS_REPLY_ARRAY || reply->type == REDIS_REPLY_PUSH) { * https://redis.io/topics/pubsub#format-of-pushed-messages */
assert(reply->elements >= 2); if ((reply->type == REDIS_REPLY_ARRAY && reply->elements >= 3) ||
reply->type == REDIS_REPLY_PUSH) {
assert(reply->element[0]->type == REDIS_REPLY_STRING); assert(reply->element[0]->type == REDIS_REPLY_STRING);
stype = reply->element[0]->str; stype = reply->element[0]->str;
pvariant = (tolower(stype[0]) == 'p') ? 1 : 0; pvariant = (tolower(stype[0]) == 'p') ? 1 : 0;
...@@ -467,8 +466,8 @@ static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, ...@@ -467,8 +466,8 @@ static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply,
} }
sdsfree(sname); sdsfree(sname);
} else { } else {
/* Shift callback for invalid commands. */ /* Shift callback for pending command in subscribed context. */
__redisShiftCallback(&ac->sub.invalid,dstcb); __redisShiftCallback(&ac->sub.replies,dstcb);
} }
return REDIS_OK; return REDIS_OK;
oom: oom:
...@@ -814,9 +813,7 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void ...@@ -814,9 +813,7 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void
goto oom; goto oom;
} else { } else {
if (c->flags & REDIS_SUBSCRIBED) { if (c->flags & REDIS_SUBSCRIBED) {
/* This will likely result in an error reply, but it needs to be if (__redisPushCallback(&ac->sub.replies,&cb) != REDIS_OK)
* received and passed to the callback. */
if (__redisPushCallback(&ac->sub.invalid,&cb) != REDIS_OK)
goto oom; goto oom;
} else { } else {
if (__redisPushCallback(&ac->replies,&cb) != REDIS_OK) if (__redisPushCallback(&ac->replies,&cb) != REDIS_OK)
......
...@@ -102,7 +102,7 @@ typedef struct redisAsyncContext { ...@@ -102,7 +102,7 @@ typedef struct redisAsyncContext {
/* Subscription callbacks */ /* Subscription callbacks */
struct { struct {
redisCallbackList invalid; redisCallbackList replies;
struct dict *channels; struct dict *channels;
struct dict *patterns; struct dict *patterns;
} sub; } sub;
......
...@@ -1518,6 +1518,15 @@ void subscribe_cb(redisAsyncContext *ac, void *r, void *privdata) { ...@@ -1518,6 +1518,15 @@ void subscribe_cb(redisAsyncContext *ac, void *r, void *privdata) {
} }
} }
/* Expect a reply of type ARRAY */
void array_cb(redisAsyncContext *ac, void *r, void *privdata) {
(void) ac;
redisReply *reply = r;
TestState *state = privdata;
assert(reply != NULL && reply->type == REDIS_REPLY_ARRAY);
state->checkpoint++;
}
static void test_pubsub_handling(struct config config) { static void test_pubsub_handling(struct config config) {
test("Subscribe, handle published message and unsubscribe: "); test("Subscribe, handle published message and unsubscribe: ");
/* Setup event dispatcher with a testcase timeout */ /* Setup event dispatcher with a testcase timeout */
...@@ -1539,13 +1548,16 @@ static void test_pubsub_handling(struct config config) { ...@@ -1539,13 +1548,16 @@ static void test_pubsub_handling(struct config config) {
TestState state = {.options = &options}; TestState state = {.options = &options};
redisAsyncCommand(ac,subscribe_cb,&state,"subscribe mychannel"); redisAsyncCommand(ac,subscribe_cb,&state,"subscribe mychannel");
/* Make sure non-subscribe commands are handled */
redisAsyncCommand(ac,array_cb,&state,"PING");
/* Start event dispatching loop */ /* Start event dispatching loop */
test_cond(event_base_dispatch(base) == 0); test_cond(event_base_dispatch(base) == 0);
event_free(timeout); event_free(timeout);
event_base_free(base); event_base_free(base);
/* Verify test checkpoints */ /* Verify test checkpoints */
assert(state.checkpoint == 1); assert(state.checkpoint == 2);
} }
/* Unexpected push message, will trigger a failure */ /* Unexpected push message, will trigger a failure */
......
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