51 int clientSubscriptionsCount(client *c) {  in clientSubscriptionsCount()  argument
52     return dictSize(c->pubsub_channels)+  in clientSubscriptionsCount()
53            listLength(c->pubsub_patterns);  in clientSubscriptionsCount()
58 int pubsubSubscribeChannel(client *c, robj *channel) {  in pubsubSubscribeChannel()  argument
64     if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {  in pubsubSubscribeChannel()
76         listAddNodeTail(clients,c);  in pubsubSubscribeChannel()
79     addReply(c,shared.mbulkhdr[3]);  in pubsubSubscribeChannel()
80     addReply(c,shared.subscribebulk);  in pubsubSubscribeChannel()
81     addReplyBulk(c,channel);  in pubsubSubscribeChannel()
82     addReplyLongLong(c,clientSubscriptionsCount(c));  in pubsubSubscribeChannel()
88 int pubsubUnsubscribeChannel(client *c, robj *channel, int notify) {  in pubsubUnsubscribeChannel()  argument
97     if (dictDelete(c->pubsub_channels,channel) == DICT_OK) {  in pubsubUnsubscribeChannel()
101         serverAssertWithInfo(c,NULL,de != NULL);  in pubsubUnsubscribeChannel()
103         ln = listSearchKey(clients,c);  in pubsubUnsubscribeChannel()
104         serverAssertWithInfo(c,NULL,ln != NULL);  in pubsubUnsubscribeChannel()
115         addReply(c,shared.mbulkhdr[3]);  in pubsubUnsubscribeChannel()
116         addReply(c,shared.unsubscribebulk);  in pubsubUnsubscribeChannel()
117         addReplyBulk(c,channel);  in pubsubUnsubscribeChannel()
118         addReplyLongLong(c,dictSize(c->pubsub_channels)+  in pubsubUnsubscribeChannel()
119                        listLength(c->pubsub_patterns));  in pubsubUnsubscribeChannel()
127 int pubsubSubscribePattern(client *c, robj *pattern) {  in pubsubSubscribePattern()  argument
130     if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {  in pubsubSubscribePattern()
133         listAddNodeTail(c->pubsub_patterns,pattern);  in pubsubSubscribePattern()
137         pat->client = c;  in pubsubSubscribePattern()
141     addReply(c,shared.mbulkhdr[3]);  in pubsubSubscribePattern()
142     addReply(c,shared.psubscribebulk);  in pubsubSubscribePattern()
143     addReplyBulk(c,pattern);  in pubsubSubscribePattern()
144     addReplyLongLong(c,clientSubscriptionsCount(c));  in pubsubSubscribePattern()
150 int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) {  in pubsubUnsubscribePattern()  argument
156     if ((ln = listSearchKey(c->pubsub_patterns,pattern)) != NULL) {  in pubsubUnsubscribePattern()
158         listDelNode(c->pubsub_patterns,ln);  in pubsubUnsubscribePattern()
159         pat.client = c;  in pubsubUnsubscribePattern()
166         addReply(c,shared.mbulkhdr[3]);  in pubsubUnsubscribePattern()
167         addReply(c,shared.punsubscribebulk);  in pubsubUnsubscribePattern()
168         addReplyBulk(c,pattern);  in pubsubUnsubscribePattern()
169         addReplyLongLong(c,dictSize(c->pubsub_channels)+  in pubsubUnsubscribePattern()
170                        listLength(c->pubsub_patterns));  in pubsubUnsubscribePattern()
178 int pubsubUnsubscribeAllChannels(client *c, int notify) {  in pubsubUnsubscribeAllChannels()  argument
179     dictIterator *di = dictGetSafeIterator(c->pubsub_channels);  in pubsubUnsubscribeAllChannels()
186         count += pubsubUnsubscribeChannel(c,channel,notify);  in pubsubUnsubscribeAllChannels()
190         addReply(c,shared.mbulkhdr[3]);  in pubsubUnsubscribeAllChannels()
191         addReply(c,shared.unsubscribebulk);  in pubsubUnsubscribeAllChannels()
192         addReply(c,shared.nullbulk);  in pubsubUnsubscribeAllChannels()
193         addReplyLongLong(c,dictSize(c->pubsub_channels)+  in pubsubUnsubscribeAllChannels()
194                        listLength(c->pubsub_patterns));  in pubsubUnsubscribeAllChannels()
202 int pubsubUnsubscribeAllPatterns(client *c, int notify) {  in pubsubUnsubscribeAllPatterns()  argument
207     listRewind(c->pubsub_patterns,&li);  in pubsubUnsubscribeAllPatterns()
211         count += pubsubUnsubscribePattern(c,pattern,notify);  in pubsubUnsubscribeAllPatterns()
215         addReply(c,shared.mbulkhdr[3]);  in pubsubUnsubscribeAllPatterns()
216         addReply(c,shared.punsubscribebulk);  in pubsubUnsubscribeAllPatterns()
217         addReply(c,shared.nullbulk);  in pubsubUnsubscribeAllPatterns()
218         addReplyLongLong(c,dictSize(c->pubsub_channels)+  in pubsubUnsubscribeAllPatterns()
219                        listLength(c->pubsub_patterns));  in pubsubUnsubscribeAllPatterns()
240             client *c = ln->value;  in pubsubPublishMessage()  local
242             addReply(c,shared.mbulkhdr[3]);  in pubsubPublishMessage()
243             addReply(c,shared.messagebulk);  in pubsubPublishMessage()
244             addReplyBulk(c,channel);  in pubsubPublishMessage()
245             addReplyBulk(c,message);  in pubsubPublishMessage()
277 void subscribeCommand(client *c) {  in subscribeCommand()  argument
280     for (j = 1; j < c->argc; j++)  in subscribeCommand()
281         pubsubSubscribeChannel(c,c->argv[j]);  in subscribeCommand()
282     c->flags |= CLIENT_PUBSUB;  in subscribeCommand()
285 void unsubscribeCommand(client *c) {  in unsubscribeCommand()  argument
286     if (c->argc == 1) {  in unsubscribeCommand()
287         pubsubUnsubscribeAllChannels(c,1);  in unsubscribeCommand()
291         for (j = 1; j < c->argc; j++)  in unsubscribeCommand()
292             pubsubUnsubscribeChannel(c,c->argv[j],1);  in unsubscribeCommand()
294     if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;  in unsubscribeCommand()
297 void psubscribeCommand(client *c) {  in psubscribeCommand()  argument
300     for (j = 1; j < c->argc; j++)  in psubscribeCommand()
301         pubsubSubscribePattern(c,c->argv[j]);  in psubscribeCommand()
302     c->flags |= CLIENT_PUBSUB;  in psubscribeCommand()
305 void punsubscribeCommand(client *c) {  in punsubscribeCommand()  argument
306     if (c->argc == 1) {  in punsubscribeCommand()
307         pubsubUnsubscribeAllPatterns(c,1);  in punsubscribeCommand()
311         for (j = 1; j < c->argc; j++)  in punsubscribeCommand()
312             pubsubUnsubscribePattern(c,c->argv[j],1);  in punsubscribeCommand()
314     if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;  in punsubscribeCommand()
317 void publishCommand(client *c) {  in publishCommand()  argument
318     int receivers = pubsubPublishMessage(c->argv[1],c->argv[2]);  in publishCommand()
320         clusterPropagatePublish(c->argv[1],c->argv[2]);  in publishCommand()
322         forceCommandPropagation(c,PROPAGATE_REPL);  in publishCommand()
323     addReplyLongLong(c,receivers);  in publishCommand()
327 void pubsubCommand(client *c) {  in pubsubCommand()  argument
328     if (!strcasecmp(c->argv[1]->ptr,"channels") &&  in pubsubCommand()
329         (c->argc == 2 || c->argc ==3))  in pubsubCommand()
332         sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr;  in pubsubCommand()
338         replylen = addDeferredMultiBulkLength(c);  in pubsubCommand()
346                 addReplyBulk(c,cobj);  in pubsubCommand()
351         setDeferredMultiBulkLength(c,replylen,mblen);  in pubsubCommand()
352     } else if (!strcasecmp(c->argv[1]->ptr,"numsub") && c->argc >= 2) {  in pubsubCommand()
356         addReplyMultiBulkLen(c,(c->argc-2)*2);  in pubsubCommand()
357         for (j = 2; j < c->argc; j++) {  in pubsubCommand()
358             list *l = dictFetchValue(server.pubsub_channels,c->argv[j]);  in pubsubCommand()
360             addReplyBulk(c,c->argv[j]);  in pubsubCommand()
361             addReplyLongLong(c,l ? listLength(l) : 0);  in pubsubCommand()
363     } else if (!strcasecmp(c->argv[1]->ptr,"numpat") && c->argc == 2) {  in pubsubCommand()
365         addReplyLongLong(c,listLength(server.pubsub_patterns));  in pubsubCommand()
367         addReplyErrorFormat(c,  in pubsubCommand()
369             (char*)c->argv[1]->ptr);  in pubsubCommand()