Function process_user_msg

net/net_evbuffer.c:1066–1103

}

static int process_user_msg(struct event_info *e)
{
    net_send_message_header *msg = &e->msg;
    if (e->state == 0) {
        ++e->state;
        net_send_message_header_get(msg, e->rd_buf, e->rd_buf + sizeof(*msg));
        if (msg->usertype <= USER_TYPE_MIN || msg->usertype >= USER_TYPE_MAX) {
            hprintf("BAD USER MSG TYPE:%d (htonl:%d)\n", msg->usertype, htonl(msg->usertype));
            return -1;
        }
        if (msg->datalen) {
            e->need = msg->datalen;
            return 0;
        }
    }
    netinfo_type *netinfo_ptr = e->net_info->netinfo_ptr;
    NETFP *func = netinfo_ptr->userfuncs[msg->usertype].func;
    if (func == NULL) {
        /* Startup race without accept-on-child-net: Replication net is
         * ready and accepts offload connection but offload callbacks have not
         * registered yet. */
        hprintf("NO USERFUNC FOR USERTYPE:%d\n", msg->usertype);
        return -1;
    }
    char *host = e->host;
    struct interned_string *host_interned = e->host_interned;

    void *usrptr = netinfo_ptr->usrptr;
    ack_state_type ack = {
        .seqnum = msg->seqnum,
        .needack = msg->waitforack,
        .fromhost = host,
        .netinfo = netinfo_ptr,
    };
    func(&ack, usrptr, host, host_interned, msg->usertype, e->rd_buf, msg->datalen, 1);
    message_done(e);
    return 0;
}

static void stop_base(struct event_base *b)
{