Function processInputBuffer

src/networking.cpp:2637–2682

This function is called every time, in the client structure 'c', there is * more query buffer to process, because we read more data from the socket * or because a client was blocked and later reactivated, so there could be * pending query buffer, already representing a full command, to process. */

 * or because a client was blocked and later reactivated, so there could be
 * pending query buffer, already representing a full command, to process. */
void processInputBuffer(client *c, bool fParse, int callFlags) {
    AssertCorrectThread(c);
    
    if (fParse)
        parseClientCommandBuffer(c);

    /* Keep processing while there is something in the input buffer */
    while (!c->vecqueuedcmd.empty()) {
        /* Return if we're still parsing this command */
        auto &cmd = c->vecqueuedcmd.front();
        if (cmd.argc != cmd.argcMax) break;
        if (c->flags & CLIENT_EXECUTING_COMMAND) break;

        if (!FClientReady(c)) break;

        if ((callFlags & CMD_CALL_ASYNC) && !FAsyncCommand(cmd))
            break;

        zfree(c->argv);
        c->argc = cmd.argc;
        c->argv = cmd.argv;
        cmd.argv = nullptr;
        c->argv_len_sumActive = cmd.argv_len_sum;
        cmd.argv_len_sum = 0;
        c->reploff_cmd = cmd.reploff;
        serverAssert(c->argv != nullptr);

        c->vecqueuedcmd.erase(c->vecqueuedcmd.begin());

        /* Multibulk processing could see a <= 0 length. */
        if (c->argc == 0) {
            resetClient(c);
        } else {
            c->flags |= CLIENT_EXECUTING_COMMAND;
            /* We are finally ready to execute the command. */
            if (processCommandAndResetClient(c, callFlags) == C_ERR) {
                /* If the client is no longer valid, we avoid exiting this
                 * loop and trimming the client buffer later. So we return
                 * ASAP in that case. */
                c->flags &= ~CLIENT_EXECUTING_COMMAND;
                return;
            }
            c->flags &= ~CLIENT_EXECUTING_COMMAND;
        }
    }
}

void readQueryFromClient(connection *conn) {
    client *c = (client*)connGetPrivateData(conn);