| 412 | } |
| 413 | |
| 414 | static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, redisCallback *dstcb) { |
| 415 | redisContext *c = &(ac->c); |
| 416 | dict *callbacks; |
| 417 | redisCallback *cb; |
| 418 | dictEntry *de; |
| 419 | int pvariant; |
| 420 | char *stype; |
| 421 | hisds sname; |
| 422 | |
| 423 | /* Custom reply functions are not supported for pub/sub. This will fail |
| 424 | * very hard when they are used... */ |
| 425 | if (reply->type == REDIS_REPLY_ARRAY || reply->type == REDIS_REPLY_PUSH) { |
| 426 | assert(reply->elements >= 2); |
| 427 | assert(reply->element[0]->type == REDIS_REPLY_STRING); |
| 428 | stype = reply->element[0]->str; |
| 429 | pvariant = (tolower(stype[0]) == 'p') ? 1 : 0; |
| 430 | |
| 431 | if (pvariant) |
| 432 | callbacks = ac->sub.patterns; |
| 433 | else |
| 434 | callbacks = ac->sub.channels; |
| 435 | |
| 436 | /* Locate the right callback */ |
| 437 | assert(reply->element[1]->type == REDIS_REPLY_STRING); |
| 438 | sname = hi_sdsnewlen(reply->element[1]->str,reply->element[1]->len); |
| 439 | if (sname == NULL) |
| 440 | goto oom; |
| 441 | |
| 442 | de = dictFind(callbacks,sname); |
| 443 | if (de != NULL) { |
| 444 | cb = dictGetEntryVal(de); |
| 445 | |
| 446 | /* If this is an subscribe reply decrease pending counter. */ |
| 447 | if (strcasecmp(stype+pvariant,"subscribe") == 0) { |
| 448 | cb->pending_subs -= 1; |
| 449 | } |
| 450 | |
| 451 | memcpy(dstcb,cb,sizeof(*dstcb)); |
| 452 | |
| 453 | /* If this is an unsubscribe message, remove it. */ |
| 454 | if (strcasecmp(stype+pvariant,"unsubscribe") == 0) { |
| 455 | if (cb->pending_subs == 0) |
| 456 | dictDelete(callbacks,sname); |
| 457 | |
| 458 | /* If this was the last unsubscribe message, revert to |
| 459 | * non-subscribe mode. */ |
| 460 | assert(reply->element[2]->type == REDIS_REPLY_INTEGER); |
| 461 | |
| 462 | /* Unset subscribed flag only when no pipelined pending subscribe. */ |
| 463 | if (reply->element[2]->integer == 0 |
| 464 | && dictSize(ac->sub.channels) == 0 |
| 465 | && dictSize(ac->sub.patterns) == 0) |
| 466 | c->flags &= ~REDIS_SUBSCRIBED; |
| 467 | } |
| 468 | } |
| 469 | hi_sdsfree(sname); |
| 470 | } else { |
| 471 | /* Shift callback for invalid commands. */ |
no test coverage detected