Function pubsubPublishMessage

src/pubsub.cpp:294–350

Publish a message */

/* Publish a message */
int pubsubPublishMessage(robj *channel, robj *message) {
    serverAssert(GlobalLocksAcquired());
    int receivers = 0;
    dictEntry *de;
    dictIterator *di;
    listNode *ln;
    listIter li;

    /* Send to clients listening for that channel */
    de = dictFind(g_pserver->pubsub_channels,channel);
    if (de) {
        list *list = reinterpret_cast<::list*>(dictGetVal(de));
        listNode *ln;
        listIter li;

        listRewind(list,&li);
        while ((ln = listNext(&li)) != NULL) {
            client *c = reinterpret_cast<client*>(ln->value);
            if (c->flags & CLIENT_CLOSE_ASAP)   // avoid blocking if the write will be ignored
                continue;
            if (FCorrectThread(c))
                fastlock_lock(&c->lock);
            addReplyPubsubMessage(c,channel,message);
            if (FCorrectThread(c))
                fastlock_unlock(&c->lock);
            receivers++;
        }
    }
    /* Send to clients listening to matching channels */
    di = dictGetIterator(g_pserver->pubsub_patterns);
    if (di) {
        channel = getDecodedObject(channel);
        while((de = dictNext(di)) != NULL) {
            robj *pattern = (robj*)dictGetKey(de);
            list *clients = (list*)dictGetVal(de);
            if (!stringmatchlen(szFromObj(pattern),
                                sdslen(szFromObj(pattern)),
                                szFromObj(channel),
                                sdslen(szFromObj(channel)),0)) continue;

            listRewind(clients,&li);
            while ((ln = listNext(&li)) != NULL) {
                client *c = (client*)listNodeValue(ln);
                if (c->flags & CLIENT_CLOSE_ASAP)
                    continue;
                std::unique_lock<fastlock> l(c->lock, std::defer_lock);
                if (FCorrectThread(c))
                    l.lock();
                addReplyPubsubPatMessage(c,pattern,channel,message);
                receivers++;
            }
        }
        decrRefCount(channel);
        dictReleaseIterator(di);
    }
    return receivers;
}