Function replicationFeedMonitors

src/replication.cpp:748–791

}

void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv, int argc) {
    if (!(listLength(g_pserver->monitors) && !g_pserver->loading)) return;
    listNode *ln;
    listIter li;
    int j;
    sds cmdrepr = sdsnew("+");
    robj *cmdobj;
    struct timeval tv;
    serverAssert(GlobalLocksAcquired());

    gettimeofday(&tv,NULL);
    cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
    if (c->flags & CLIENT_LUA) {
        cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ",dictid);
    } else if (c->flags & CLIENT_UNIX_SOCKET) {
        cmdrepr = sdscatprintf(cmdrepr,"[%d unix:%s] ",dictid,g_pserver->unixsocket);
    } else {
        cmdrepr = sdscatprintf(cmdrepr,"[%d %s] ",dictid,getClientPeerId(c));
    }

    for (j = 0; j < argc; j++) {
        if (argv[j]->encoding == OBJ_ENCODING_INT) {
            cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)ptrFromObj(argv[j]));
        } else {
            cmdrepr = sdscatrepr(cmdrepr,(char*)ptrFromObj(argv[j]),
                        sdslen((sds)ptrFromObj(argv[j])));
        }
        if (j != argc-1)
            cmdrepr = sdscatlen(cmdrepr," ",1);
    }
    cmdrepr = sdscatlen(cmdrepr,"\r\n",2);
    cmdobj = createObject(OBJ_STRING,cmdrepr);

    listRewind(monitors,&li);
    while((ln = listNext(&li))) {
        client *monitor = (client*)ln->value;
		std::unique_lock<decltype(monitor->lock)> lock(monitor->lock, std::defer_lock);
		// When writing to clients on other threads the global lock is sufficient provided we only use AddReply*Async()
		if (FCorrectThread(c))
			lock.lock();
        addReply(monitor,cmdobj);
    }
    decrRefCount(cmdobj);
}

int prepareClientToWrite(client *c);