Function PackRtmpRequest

src/brpc/policy/rtmp_protocol.cpp:3626–3669

}

void PackRtmpRequest(butil::IOBuf* /*buf*/,
                     SocketMessage** user_message,
                     uint64_t /*correlation_id*/,
                     const google::protobuf::MethodDescriptor* /*NULL*/,
                     Controller* cntl,
                     const butil::IOBuf& /*request*/,
                     const Authenticator*) {
    // Send createStream command
    ControllerPrivateAccessor accessor(cntl);
    Socket* s = accessor.get_sending_socket();
    RtmpContext* ctx = static_cast<RtmpContext*>(s->parsing_context());
    if (ctx == NULL) {
        cntl->SetFailed(EINVAL, "RtmpContext of %s is not created",
                        s->description().c_str());
        return;
    }
    // Hack: we pass stream as response in RtmpClientStream::Create
    RtmpClientStream* stream = (RtmpClientStream*)cntl->response();

    // Hack: save last transaction_id into log_id(useless here) so that we
    // can get it back and cancel the transaction before creating new one
    // (for retrying).
    CHECK_LT(cntl->log_id(), (uint64_t)std::numeric_limits<uint32_t>::max());
    uint32_t transaction_id = cntl->log_id();
    if (transaction_id != 0) {
        RtmpTransactionHandler* last_handler =
            ctx->RemoveTransaction(transaction_id);
        if (last_handler) {
            last_handler->Cancel();
        }
    }
    OnServerStreamCreated* cb = new OnServerStreamCreated(stream, cntl->call_id());
    if (!ctx->AddTransaction(&transaction_id, cb)) {
        cntl->SetFailed(EINVAL, "Fail to add transaction");
        delete cb;
        return;
    }
    cntl->set_log_id(transaction_id);
    RtmpCreateStreamMessage* msg = new RtmpCreateStreamMessage;
    s->ReAddress(&msg->socket);
    msg->transaction_id = transaction_id;
    msg->options = stream->options();
    *user_message = msg;
}

void SerializeRtmpRequest(butil::IOBuf* /*buf*/,
                          Controller* /*cntl*/,