Function ParseSofaMessage

src/brpc/policy/sofa_pbrpc_protocol.cpp:165–206

}

ParseResult ParseSofaMessage(butil::IOBuf* source, Socket* socket,
                             bool /*read_eof*/, const void* /*arg*/) {
    char header_buf[24];
    const size_t n = source->copy_to(header_buf, sizeof(header_buf));
    if (n >= 4) {
        void* dummy = header_buf;
        if (*(const uint32_t*)dummy != *(const uint32_t*)"SOFA") {
            return MakeParseError(PARSE_ERROR_TRY_OTHERS);
        }
    } else {
        if (memcmp(header_buf, "SOFA", n) != 0) {
            return MakeParseError(PARSE_ERROR_TRY_OTHERS);
        }
    }
    if (n < sizeof(header_buf)) {
        return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
    }
    uint32_t meta_size;
    uint64_t body_size;
    uint64_t msg_size;
    SofaRawUnpacker ru(header_buf + 4);
    ru.unpack32(meta_size).unpack64(body_size).unpack64(msg_size);
    if (msg_size != meta_size + body_size) {
        LOG(ERROR) << "msg_size=" << msg_size << " != meta_size=" << meta_size
                   << " + body_size=" << body_size;
        return MakeParseError(PARSE_ERROR_TRY_OTHERS);
    }
    if (body_size > FLAGS_max_body_size) {
        // We need this log to report the body_size to give users some clues
        // which is not printed in InputMessenger.
        LOG(ERROR) << "body_size=" << body_size << " from "
                   << socket->remote_side() << " is too large";
        return MakeParseError(PARSE_ERROR_TOO_BIG_DATA);
    } else if (source->length() < sizeof(header_buf) + msg_size) {
        return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
    }
    source->pop_front(sizeof(header_buf));
    MostCommonMessage* msg = MostCommonMessage::Get();
    source->cutn(&msg->meta, meta_size);
    source->cutn(&msg->payload, body_size);
    return MakeMessage(msg);
}

// Assemble response packet using `correlation_id', `controller',
// `res', and then write this packet to `sock'