diff options
Diffstat (limited to 'examples/redis-unstable/src/pubsub.c')
| -rw-r--r-- | examples/redis-unstable/src/pubsub.c | 768 |
1 files changed, 0 insertions, 768 deletions
diff --git a/examples/redis-unstable/src/pubsub.c b/examples/redis-unstable/src/pubsub.c deleted file mode 100644 index 7199be1..0000000 --- a/examples/redis-unstable/src/pubsub.c +++ /dev/null @@ -1,768 +0,0 @@ -/* - * Copyright (c) 2009-Present, Redis Ltd. - * All rights reserved. - * - * Copyright (c) 2024-present, Valkey contributors. - * All rights reserved. - * - * Licensed under your choice of (a) the Redis Source Available License 2.0 - * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the - * GNU Affero General Public License v3 (AGPLv3). - * - * Portions of this file are available under BSD3 terms; see REDISCONTRIBUTIONS for more information. - */ - -#include "server.h" -#include "cluster.h" -#include "cluster_slot_stats.h" - -/* Structure to hold the pubsub related metadata. Currently used - * for pubsub and pubsubshard feature. */ -typedef struct pubsubtype { - int shard; - dict *(*clientPubSubChannels)(client*); - int (*subscriptionCount)(client*); - kvstore **serverPubSubChannels; - robj **subscribeMsg; - robj **unsubscribeMsg; - robj **messageBulk; -}pubsubtype; - -/* - * Get client's global Pub/Sub channels subscription count. - */ -int clientSubscriptionsCount(client *c); - -/* - * Get client's shard level Pub/Sub channels subscription count. - */ -int clientShardSubscriptionsCount(client *c); - -/* - * Get client's global Pub/Sub channels dict. - */ -dict* getClientPubSubChannels(client *c); - -/* - * Get client's shard level Pub/Sub channels dict. - */ -dict* getClientPubSubShardChannels(client *c); - -/* - * Get list of channels client is subscribed to. - * If a pattern is provided, the subset of channels is returned - * matching the pattern. - */ -void channelList(client *c, sds pat, kvstore *pubsub_channels); - -/* - * Pub/Sub type for global channels. - */ -pubsubtype pubSubType = { - .shard = 0, - .clientPubSubChannels = getClientPubSubChannels, - .subscriptionCount = clientSubscriptionsCount, - .serverPubSubChannels = &server.pubsub_channels, - .subscribeMsg = &shared.subscribebulk, - .unsubscribeMsg = &shared.unsubscribebulk, - .messageBulk = &shared.messagebulk, -}; - -/* - * Pub/Sub type for shard level channels bounded to a slot. - */ -pubsubtype pubSubShardType = { - .shard = 1, - .clientPubSubChannels = getClientPubSubShardChannels, - .subscriptionCount = clientShardSubscriptionsCount, - .serverPubSubChannels = &server.pubsubshard_channels, - .subscribeMsg = &shared.ssubscribebulk, - .unsubscribeMsg = &shared.sunsubscribebulk, - .messageBulk = &shared.smessagebulk, -}; - -/*----------------------------------------------------------------------------- - * Pubsub client replies API - *----------------------------------------------------------------------------*/ - -/* Send a pubsub message of type "message" to the client. - * Normally 'msg' is a Redis object containing the string to send as - * message. However if the caller sets 'msg' as NULL, it will be able - * to send a special message (for instance an Array type) by using the - * addReply*() API family. */ -void addReplyPubsubMessage(client *c, robj *channel, robj *msg, robj *message_bulk) { - uint64_t old_flags = c->flags; - c->flags |= CLIENT_PUSHING; - if (c->resp == 2) - addReply(c,shared.mbulkhdr[3]); - else - addReplyPushLen(c,3); - addReply(c,message_bulk); - addReplyBulk(c,channel); - if (msg) addReplyBulk(c,msg); - if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; -} - -/* Send a pubsub message of type "pmessage" to the client. The difference - * with the "message" type delivered by addReplyPubsubMessage() is that - * this message format also includes the pattern that matched the message. */ -void addReplyPubsubPatMessage(client *c, robj *pat, robj *channel, robj *msg) { - uint64_t old_flags = c->flags; - c->flags |= CLIENT_PUSHING; - if (c->resp == 2) - addReply(c,shared.mbulkhdr[4]); - else - addReplyPushLen(c,4); - addReply(c,shared.pmessagebulk); - addReplyBulk(c,pat); - addReplyBulk(c,channel); - addReplyBulk(c,msg); - if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; -} - -/* Send the pubsub subscription notification to the client. */ -void addReplyPubsubSubscribed(client *c, robj *channel, pubsubtype type) { - uint64_t old_flags = c->flags; - c->flags |= CLIENT_PUSHING; - if (c->resp == 2) - addReply(c,shared.mbulkhdr[3]); - else - addReplyPushLen(c,3); - addReply(c,*type.subscribeMsg); - addReplyBulk(c,channel); - addReplyLongLong(c,type.subscriptionCount(c)); - if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; -} - -/* Send the pubsub unsubscription notification to the client. - * Channel can be NULL: this is useful when the client sends a mass - * unsubscribe command but there are no channels to unsubscribe from: we - * still send a notification. */ -void addReplyPubsubUnsubscribed(client *c, robj *channel, pubsubtype type) { - uint64_t old_flags = c->flags; - c->flags |= CLIENT_PUSHING; - if (c->resp == 2) - addReply(c,shared.mbulkhdr[3]); - else - addReplyPushLen(c,3); - addReply(c, *type.unsubscribeMsg); - if (channel) - addReplyBulk(c,channel); - else - addReplyNull(c); - addReplyLongLong(c,type.subscriptionCount(c)); - if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; -} - -/* Send the pubsub pattern subscription notification to the client. */ -void addReplyPubsubPatSubscribed(client *c, robj *pattern) { - uint64_t old_flags = c->flags; - c->flags |= CLIENT_PUSHING; - if (c->resp == 2) - addReply(c,shared.mbulkhdr[3]); - else - addReplyPushLen(c,3); - addReply(c,shared.psubscribebulk); - addReplyBulk(c,pattern); - addReplyLongLong(c,clientSubscriptionsCount(c)); - if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; -} - -/* Send the pubsub pattern unsubscription notification to the client. - * Pattern can be NULL: this is useful when the client sends a mass - * punsubscribe command but there are no pattern to unsubscribe from: we - * still send a notification. */ -void addReplyPubsubPatUnsubscribed(client *c, robj *pattern) { - uint64_t old_flags = c->flags; - c->flags |= CLIENT_PUSHING; - if (c->resp == 2) - addReply(c,shared.mbulkhdr[3]); - else - addReplyPushLen(c,3); - addReply(c,shared.punsubscribebulk); - if (pattern) - addReplyBulk(c,pattern); - else - addReplyNull(c); - addReplyLongLong(c,clientSubscriptionsCount(c)); - if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; -} - -/*----------------------------------------------------------------------------- - * Pubsub low level API - *----------------------------------------------------------------------------*/ - -/* Return the number of pubsub channels + patterns is handled. */ -int serverPubsubSubscriptionCount(void) { - return kvstoreSize(server.pubsub_channels) + dictSize(server.pubsub_patterns); -} - -/* Return the number of pubsub shard level channels is handled. */ -int serverPubsubShardSubscriptionCount(void) { - return kvstoreSize(server.pubsubshard_channels); -} - -/* Return the number of channels + patterns a client is subscribed to. */ -int clientSubscriptionsCount(client *c) { - return dictSize(c->pubsub_channels) + dictSize(c->pubsub_patterns); -} - -/* Return the number of shard level channels a client is subscribed to. */ -int clientShardSubscriptionsCount(client *c) { - return dictSize(c->pubsubshard_channels); -} - -dict* getClientPubSubChannels(client *c) { - return c->pubsub_channels; -} - -dict* getClientPubSubShardChannels(client *c) { - return c->pubsubshard_channels; -} - -/* Return the number of pubsub + pubsub shard level channels - * a client is subscribed to. */ -int clientTotalPubSubSubscriptionCount(client *c) { - return clientSubscriptionsCount(c) + clientShardSubscriptionsCount(c); -} - -void markClientAsPubSub(client *c) { - if (!(c->flags & CLIENT_PUBSUB)) { - c->flags |= CLIENT_PUBSUB; - server.pubsub_clients++; - } -} - -void unmarkClientAsPubSub(client *c) { - if (c->flags & CLIENT_PUBSUB) { - c->flags &= ~CLIENT_PUBSUB; - server.pubsub_clients--; - } -} - -/* 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(client *c, robj *channel, pubsubtype type) { - dictEntry *de, *existing; - dict *clients = NULL; - int retval = 0; - unsigned int slot = 0; - - /* Add the channel to the client -> channels hash table */ - dictEntryLink bucket; - dictEntryLink link = dictFindLink(type.clientPubSubChannels(c),channel,&bucket); - if (link == NULL) { /* Not yet subscribed to this channel */ - retval = 1; - /* Add the client to the channel -> list of clients hash table */ - if (server.cluster_enabled && type.shard) { - slot = getKeySlot(channel->ptr); - } - - de = kvstoreDictAddRaw(*type.serverPubSubChannels, slot, channel, &existing); - - if (existing) { - clients = dictGetVal(existing); - channel = dictGetKey(existing); - } else { - clients = dictCreate(&clientDictType); - kvstoreDictSetVal(*type.serverPubSubChannels, slot, de, clients); - incrRefCount(channel); - } - - serverAssert(dictAdd(clients, c, NULL) != DICT_ERR); - dictSetKeyAtLink(type.clientPubSubChannels(c), channel, &bucket, 1); - incrRefCount(channel); - } - /* Notify the client */ - addReplyPubsubSubscribed(c,channel,type); - return retval; -} - -/* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or - * 0 if the client was not subscribed to the specified channel. */ -int pubsubUnsubscribeChannel(client *c, robj *channel, int notify, pubsubtype type) { - dictEntry *de; - dict *clients; - int retval = 0; - int slot = 0; - - /* Remove the channel from the client -> channels hash table */ - incrRefCount(channel); /* channel may be just a pointer to the same object - we have in the hash tables. Protect it... */ - if (dictDelete(type.clientPubSubChannels(c),channel) == DICT_OK) { - retval = 1; - /* Remove the client from the channel -> clients list hash table */ - if (server.cluster_enabled && type.shard) { - slot = getKeySlot(channel->ptr); - } - de = kvstoreDictFind(*type.serverPubSubChannels, slot, channel); - serverAssertWithInfo(c,NULL,de != NULL); - clients = dictGetVal(de); - serverAssertWithInfo(c, NULL, dictDelete(clients, c) == DICT_OK); - if (dictSize(clients) == 0) { - /* Free the dict and associated hash entry at all if this was - * the latest client, so that it will be possible to abuse - * Redis PUBSUB creating millions of channels. */ - kvstoreDictDelete(*type.serverPubSubChannels, slot, channel); - } - } - /* Notify the client */ - if (notify) { - addReplyPubsubUnsubscribed(c,channel,type); - } - decrRefCount(channel); /* it is finally safe to release it */ - return retval; -} - -/* Unsubscribe all shard channels in a slot. */ -void pubsubShardUnsubscribeAllChannelsInSlot(unsigned int slot) { - if (!kvstoreDictSize(server.pubsubshard_channels, slot)) - return; - - dictEntry *de; - kvstoreDictIterator kvs_di; - kvstoreInitDictSafeIterator(&kvs_di, server.pubsubshard_channels, slot); - while ((de = kvstoreDictIteratorNext(&kvs_di)) != NULL) { - robj *channel = dictGetKey(de); - dict *clients = dictGetVal(de); - /* For each client subscribed to the channel, unsubscribe it. */ - dictIterator iter; - dictEntry *entry; - - dictInitIterator(&iter, clients); - while ((entry = dictNext(&iter)) != NULL) { - client *c = dictGetKey(entry); - int retval = dictDelete(c->pubsubshard_channels, channel); - serverAssertWithInfo(c,channel,retval == DICT_OK); - addReplyPubsubUnsubscribed(c, channel, pubSubShardType); - /* If the client has no other pubsub subscription, - * move out of pubsub mode. */ - if (clientTotalPubSubSubscriptionCount(c) == 0) { - unmarkClientAsPubSub(c); - } - } - dictResetIterator(&iter); - kvstoreDictDelete(server.pubsubshard_channels, slot, channel); - } - kvstoreResetDictIterator(&kvs_di); -} - -/* Subscribe a client to a pattern. Returns 1 if the operation succeeded, or 0 if the client was already subscribed to that pattern. */ -int pubsubSubscribePattern(client *c, robj *pattern) { - dictEntry *de; - dict *clients; - int retval = 0; - - if (dictAdd(c->pubsub_patterns, pattern, NULL) == DICT_OK) { - retval = 1; - incrRefCount(pattern); - /* Add the client to the pattern -> list of clients hash table */ - de = dictFind(server.pubsub_patterns,pattern); - if (de == NULL) { - clients = dictCreate(&clientDictType); - dictAdd(server.pubsub_patterns,pattern,clients); - incrRefCount(pattern); - } else { - clients = dictGetVal(de); - } - serverAssert(dictAdd(clients, c, NULL) != DICT_ERR); - } - /* Notify the client */ - addReplyPubsubPatSubscribed(c,pattern); - return retval; -} - -/* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or - * 0 if the client was not subscribed to the specified channel. */ -int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) { - dictEntry *de; - dict *clients; - int retval = 0; - - incrRefCount(pattern); /* Protect the object. May be the same we remove */ - if (dictDelete(c->pubsub_patterns, pattern) == DICT_OK) { - retval = 1; - /* Remove the client from the pattern -> clients list hash table */ - de = dictFind(server.pubsub_patterns,pattern); - serverAssertWithInfo(c,NULL,de != NULL); - clients = dictGetVal(de); - serverAssertWithInfo(c, NULL, dictDelete(clients, c) == DICT_OK); - if (dictSize(clients) == 0) { - /* Free the dict and associated hash entry at all if this was - * the latest client. */ - dictDelete(server.pubsub_patterns,pattern); - } - } - /* Notify the client */ - if (notify) addReplyPubsubPatUnsubscribed(c,pattern); - decrRefCount(pattern); - return retval; -} - -/* Unsubscribe from all the channels. Return the number of channels the - * client was subscribed to. */ -int pubsubUnsubscribeAllChannelsInternal(client *c, int notify, pubsubtype type) { - int count = 0; - if (dictSize(type.clientPubSubChannels(c)) > 0) { - dictIterator di; - dictEntry *de; - - dictInitSafeIterator(&di, type.clientPubSubChannels(c)); - while((de = dictNext(&di)) != NULL) { - robj *channel = dictGetKey(de); - - count += pubsubUnsubscribeChannel(c,channel,notify,type); - } - dictResetIterator(&di); - } - /* We were subscribed to nothing? Still reply to the client. */ - if (notify && count == 0) { - addReplyPubsubUnsubscribed(c,NULL,type); - } - return count; -} - -/* - * Unsubscribe a client from all global channels. - */ -int pubsubUnsubscribeAllChannels(client *c, int notify) { - int count = pubsubUnsubscribeAllChannelsInternal(c,notify,pubSubType); - return count; -} - -/* - * Unsubscribe a client from all shard subscribed channels. - */ -int pubsubUnsubscribeShardAllChannels(client *c, int notify) { - int count = pubsubUnsubscribeAllChannelsInternal(c, notify, pubSubShardType); - return count; -} - -/* Unsubscribe from all the patterns. Return the number of patterns the - * client was subscribed from. */ -int pubsubUnsubscribeAllPatterns(client *c, int notify) { - int count = 0; - - if (dictSize(c->pubsub_patterns) > 0) { - dictIterator di; - dictEntry *de; - - dictInitSafeIterator(&di, c->pubsub_patterns); - while ((de = dictNext(&di)) != NULL) { - robj *pattern = dictGetKey(de); - count += pubsubUnsubscribePattern(c, pattern, notify); - } - dictResetIterator(&di); - } - - /* We were subscribed to nothing? Still reply to the client. */ - if (notify && count == 0) addReplyPubsubPatUnsubscribed(c,NULL); - return count; -} - -/* - * Publish a message to all the subscribers. - */ -int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type) { - int receivers = 0; - dictEntry *de; - dictIterator di; - unsigned int slot = 0; - - /* Send to clients listening for that channel */ - if (server.cluster_enabled && type.shard) { - slot = keyHashSlot(channel->ptr, sdslen(channel->ptr)); - } - de = kvstoreDictFind(*type.serverPubSubChannels, slot, channel); - if (de) { - dict *clients = dictGetVal(de); - dictEntry *entry; - dictIterator iter; - - dictInitIterator(&iter, clients); - while ((entry = dictNext(&iter)) != NULL) { - client *c = dictGetKey(entry); - addReplyPubsubMessage(c,channel,message,*type.messageBulk); - if (clusterSlotStatsEnabled(CLUSTER_SLOT_STATS_NET)) - clusterSlotStatsAddNetworkBytesOutForShardedPubSubInternalPropagation(c, slot); - updateClientMemUsageAndBucket(c); - receivers++; - } - dictResetIterator(&iter); - } - - if (type.shard) { - /* Shard pubsub ignores patterns. */ - return receivers; - } - - /* Send to clients listening to matching channels */ - if (dictSize(server.pubsub_patterns) > 0) { - channel = getDecodedObject(channel); - dictInitIterator(&di, server.pubsub_patterns); - while((de = dictNext(&di)) != NULL) { - robj *pattern = dictGetKey(de); - dict *clients = dictGetVal(de); - if (!stringmatchlen((char*)pattern->ptr, - sdslen(pattern->ptr), - (char*)channel->ptr, - sdslen(channel->ptr),0)) continue; - - dictEntry *entry; - dictIterator iter; - - dictInitIterator(&iter, clients); - while ((entry = dictNext(&iter)) != NULL) { - client *c = dictGetKey(entry); - addReplyPubsubPatMessage(c,pattern,channel,message); - updateClientMemUsageAndBucket(c); - receivers++; - } - dictResetIterator(&iter); - } - decrRefCount(channel); - dictResetIterator(&di); - } - return receivers; -} - -/* Publish a message to all the subscribers. */ -int pubsubPublishMessage(robj *channel, robj *message, int sharded) { - return pubsubPublishMessageInternal(channel, message, sharded? pubSubShardType : pubSubType); -} - -/*----------------------------------------------------------------------------- - * Pubsub commands implementation - *----------------------------------------------------------------------------*/ - -/* SUBSCRIBE channel [channel ...] */ -void subscribeCommand(client *c) { - int j; - if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) { - /** - * A client that has CLIENT_DENY_BLOCKING flag on - * expect a reply per command and so can not execute subscribe. - * - * Notice that we have a special treatment for multi because of - * backward compatibility - */ - addReplyError(c, "SUBSCRIBE isn't allowed for a DENY BLOCKING client"); - return; - } - for (j = 1; j < c->argc; j++) - pubsubSubscribeChannel(c,c->argv[j],pubSubType); - markClientAsPubSub(c); -} - -/* UNSUBSCRIBE [channel ...] */ -void unsubscribeCommand(client *c) { - if (c->argc == 1) { - pubsubUnsubscribeAllChannels(c,1); - } else { - int j; - - for (j = 1; j < c->argc; j++) - pubsubUnsubscribeChannel(c,c->argv[j],1,pubSubType); - } - if (clientTotalPubSubSubscriptionCount(c) == 0) { - unmarkClientAsPubSub(c); - } -} - -/* PSUBSCRIBE pattern [pattern ...] */ -void psubscribeCommand(client *c) { - int j; - if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) { - /** - * A client that has CLIENT_DENY_BLOCKING flag on - * expect a reply per command and so can not execute subscribe. - * - * Notice that we have a special treatment for multi because of - * backward compatibility - */ - addReplyError(c, "PSUBSCRIBE isn't allowed for a DENY BLOCKING client"); - return; - } - - for (j = 1; j < c->argc; j++) - pubsubSubscribePattern(c,c->argv[j]); - markClientAsPubSub(c); -} - -/* PUNSUBSCRIBE [pattern [pattern ...]] */ -void punsubscribeCommand(client *c) { - if (c->argc == 1) { - pubsubUnsubscribeAllPatterns(c,1); - } else { - int j; - - for (j = 1; j < c->argc; j++) - pubsubUnsubscribePattern(c,c->argv[j],1); - } - if (clientTotalPubSubSubscriptionCount(c) == 0) { - unmarkClientAsPubSub(c); - } -} - -/* This function wraps pubsubPublishMessage and also propagates the message to cluster. - * Used by the commands PUBLISH/SPUBLISH and their respective module APIs.*/ -int pubsubPublishMessageAndPropagateToCluster(robj *channel, robj *message, int sharded) { - int receivers = pubsubPublishMessage(channel, message, sharded); - if (server.cluster_enabled) - clusterPropagatePublish(channel, message, sharded); - return receivers; -} - -/* PUBLISH <channel> <message> */ -void publishCommand(client *c) { - if (server.sentinel_mode) { - sentinelPublishCommand(c); - return; - } - - int receivers = pubsubPublishMessageAndPropagateToCluster(c->argv[1],c->argv[2],0); - if (!server.cluster_enabled) - forceCommandPropagation(c,PROPAGATE_REPL); - addReplyLongLong(c,receivers); -} - -/* PUBSUB command for Pub/Sub introspection. */ -void pubsubCommand(client *c) { - if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) { - const char *help[] = { -"CHANNELS [<pattern>]", -" Return the currently active channels matching a <pattern> (default: '*').", -"NUMPAT", -" Return number of subscriptions to patterns.", -"NUMSUB [<channel> ...]", -" Return the number of subscribers for the specified channels, excluding", -" pattern subscriptions(default: no channels).", -"SHARDCHANNELS [<pattern>]", -" Return the currently active shard level channels matching a <pattern> (default: '*').", -"SHARDNUMSUB [<shardchannel> ...]", -" Return the number of subscribers for the specified shard level channel(s)", -NULL - }; - addReplyHelp(c, help); - } else if (!strcasecmp(c->argv[1]->ptr,"channels") && - (c->argc == 2 || c->argc == 3)) - { - /* PUBSUB CHANNELS [<pattern>] */ - sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr; - channelList(c, pat, server.pubsub_channels); - } else if (!strcasecmp(c->argv[1]->ptr,"numsub") && c->argc >= 2) { - /* PUBSUB NUMSUB [Channel_1 ... Channel_N] */ - int j; - - addReplyArrayLen(c,(c->argc-2)*2); - for (j = 2; j < c->argc; j++) { - dict *d = kvstoreDictFetchValue(server.pubsub_channels, 0, c->argv[j]); - - addReplyBulk(c,c->argv[j]); - addReplyLongLong(c, d ? dictSize(d) : 0); - } - } else if (!strcasecmp(c->argv[1]->ptr,"numpat") && c->argc == 2) { - /* PUBSUB NUMPAT */ - addReplyLongLong(c,dictSize(server.pubsub_patterns)); - } else if (!strcasecmp(c->argv[1]->ptr,"shardchannels") && - (c->argc == 2 || c->argc == 3)) - { - /* PUBSUB SHARDCHANNELS */ - sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr; - channelList(c,pat,server.pubsubshard_channels); - } else if (!strcasecmp(c->argv[1]->ptr,"shardnumsub") && c->argc >= 2) { - /* PUBSUB SHARDNUMSUB [ShardChannel_1 ... ShardChannel_N] */ - int j; - addReplyArrayLen(c, (c->argc-2)*2); - for (j = 2; j < c->argc; j++) { - unsigned int slot = calculateKeySlot(c->argv[j]->ptr); - dict *clients = kvstoreDictFetchValue(server.pubsubshard_channels, slot, c->argv[j]); - - addReplyBulk(c,c->argv[j]); - addReplyLongLong(c, clients ? dictSize(clients) : 0); - } - } else { - addReplySubcommandSyntaxError(c); - } -} - -void channelList(client *c, sds pat, kvstore *pubsub_channels) { - long mblen = 0; - void *replylen; - unsigned int slot_cnt = kvstoreNumDicts(pubsub_channels); - - replylen = addReplyDeferredLen(c); - for (unsigned int i = 0; i < slot_cnt; i++) { - if (!kvstoreDictSize(pubsub_channels, i)) - continue; - dictEntry *de; - kvstoreDictIterator kvs_di; - kvstoreInitDictIterator(&kvs_di, pubsub_channels, i); - while((de = kvstoreDictIteratorNext(&kvs_di)) != NULL) { - robj *cobj = dictGetKey(de); - sds channel = cobj->ptr; - - if (!pat || stringmatchlen(pat, sdslen(pat), - channel, sdslen(channel),0)) - { - addReplyBulk(c,cobj); - mblen++; - } - } - kvstoreResetDictIterator(&kvs_di); - } - setDeferredArrayLen(c,replylen,mblen); -} - -/* SPUBLISH <shardchannel> <message> */ -void spublishCommand(client *c) { - int receivers = pubsubPublishMessageAndPropagateToCluster(c->argv[1],c->argv[2],1); - if (!server.cluster_enabled) - forceCommandPropagation(c,PROPAGATE_REPL); - addReplyLongLong(c,receivers); -} - -/* SSUBSCRIBE shardchannel [shardchannel ...] */ -void ssubscribeCommand(client *c) { - if (c->flags & CLIENT_DENY_BLOCKING) { - /* A client that has CLIENT_DENY_BLOCKING flag on - * expect a reply per command and so can not execute subscribe. */ - addReplyError(c, "SSUBSCRIBE isn't allowed for a DENY BLOCKING client"); - return; - } - - for (int j = 1; j < c->argc; j++) { - pubsubSubscribeChannel(c, c->argv[j], pubSubShardType); - } - markClientAsPubSub(c); -} - -/* SUNSUBSCRIBE [shardchannel [shardchannel ...]] */ -void sunsubscribeCommand(client *c) { - if (c->argc == 1) { - pubsubUnsubscribeShardAllChannels(c, 1); - } else { - for (int j = 1; j < c->argc; j++) { - pubsubUnsubscribeChannel(c, c->argv[j], 1, pubSubShardType); - } - } - if (clientTotalPubSubSubscriptionCount(c) == 0) { - unmarkClientAsPubSub(c); - } -} - -size_t pubsubMemOverhead(client *c) { - /* PubSub patterns */ - size_t mem = dictMemUsage(c->pubsub_patterns); - /* Global PubSub channels */ - mem += dictMemUsage(c->pubsub_channels); - /* Sharded PubSub channels */ - mem += dictMemUsage(c->pubsubshard_channels); - return mem; -} - -int pubsubTotalSubscriptions(void) { - return dictSize(server.pubsub_patterns) + - kvstoreSize(server.pubsub_channels) + - kvstoreSize(server.pubsubshard_channels); -} |