PubSub clients refactoring and new PUBSUB flag.

The code tested many times if a client had active Pub/Sub subscriptions
by checking the length of a list and dictionary where the patterns and
channels are stored. This was substituted with a client flag called
REDIS_PUBSUB that is simpler to test for. Moreover in order to manage
this flag some code was refactored.

This commit is believed to have no effects in the behavior of the
server.

1 files changed, 13 insertions, 3 deletions

diff --git a/src/pubsub.c b/src/pubsub.c
index f83c3a7a1..720cd5185 100644
--- a/src/pubsub.c
+++ b/src/pubsub.c
@@ -47,6 +47,12 @@ int listMatchPubsubPattern(void *a, void *b) {
            (equalStringObjects(pa->pattern,pb->pattern));
 }
 
+/* Return the number of channels + patterns a client is subscribed to. */
+int clientSubscriptionsCount(redisClient *c) {
+    return dictSize(c->pubsub_channels)+
+           listLength(c->pubsub_patterns);
+}
+
 /* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
  * 0 if the client was already subscribed to that channel. */
 int pubsubSubscribeChannel(redisClient *c, robj *channel) {
@@ -73,7 +79,7 @@ int pubsubSubscribeChannel(redisClient *c, robj *channel) {
     addReply(c,shared.mbulkhdr[3]);
     addReply(c,shared.subscribebulk);
     addReplyBulk(c,channel);
-    addReplyLongLong(c,dictSize(c->pubsub_channels)+listLength(c->pubsub_patterns));
+    addReplyLongLong(c,clientSubscriptionsCount(c));
     return retval;
 }
 
@@ -135,7 +141,7 @@ int pubsubSubscribePattern(redisClient *c, robj *pattern) {
     addReply(c,shared.mbulkhdr[3]);
     addReply(c,shared.psubscribebulk);
     addReplyBulk(c,pattern);
-    addReplyLongLong(c,dictSize(c->pubsub_channels)+listLength(c->pubsub_patterns));
+    addReplyLongLong(c,clientSubscriptionsCount(c));
     return retval;
 }
 
@@ -168,7 +174,7 @@ int pubsubUnsubscribePattern(redisClient *c, robj *pattern, int notify) {
 }
 
 /* Unsubscribe from all the channels. Return the number of channels the
- * client was subscribed from. */
+ * client was subscribed to. */
 int pubsubUnsubscribeAllChannels(redisClient *c, int notify) {
     dictIterator *di = dictGetSafeIterator(c->pubsub_channels);
     dictEntry *de;
@@ -273,6 +279,7 @@ void subscribeCommand(redisClient *c) {
 
     for (j = 1; j < c->argc; j++)
         pubsubSubscribeChannel(c,c->argv[j]);
+    c->flags |= REDIS_PUBSUB;
 }
 
 void unsubscribeCommand(redisClient *c) {
@@ -284,6 +291,7 @@ void unsubscribeCommand(redisClient *c) {
         for (j = 1; j < c->argc; j++)
             pubsubUnsubscribeChannel(c,c->argv[j],1);
     }
+    if (clientSubscriptionsCount(c) == 0) c->flags &= ~REDIS_PUBSUB;
 }
 
 void psubscribeCommand(redisClient *c) {
@@ -291,6 +299,7 @@ void psubscribeCommand(redisClient *c) {
 
     for (j = 1; j < c->argc; j++)
         pubsubSubscribePattern(c,c->argv[j]);
+    c->flags |= REDIS_PUBSUB;
 }
 
 void punsubscribeCommand(redisClient *c) {
@@ -302,6 +311,7 @@ void punsubscribeCommand(redisClient *c) {
         for (j = 1; j < c->argc; j++)
             pubsubUnsubscribePattern(c,c->argv[j],1);
     }
+    if (clientSubscriptionsCount(c) == 0) c->flags &= ~REDIS_PUBSUB;
 }
 
 void publishCommand(redisClient *c) {