深入理解rtmp(四)之协议实现分析

------------------------------------>
+--------+--------+--------+--------+
|   11   |   66   |    85  |   27   |
+--------+--------+--------+--------+

0                   1                   2                   3
  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 | Message Type |               Payload length                   |
 |   (1 byte)   |                   (3 bytes)                    |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |                          Timestamp                            |
 |                          (4 bytes)                            |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |                          Stream ID            |
 |                          (3 bytes)            |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                         Message Header

+-------------+                                     +----------+    
         | Play Client |                  |                  |  Server  |
         +-------------+                  |                  +----------+
                |          |Handshaking and Application|          |
                |          |         connect done      |          |
                |                         |                       |
       ---+---- |---------Command Message(createStream) --------->|
       Create   |                                                 |
       Stream   |                                                 |
       ---+---- |<-------------- Command Message -----------------|
                |       (_result- createStream response)          |
                |                                                 |
       ---+---- |------------ Command Message (play) ------------>|
         play   |                                                 |
          |     |<---------------- SetChunkSize ------------------|
          |     |                                                 |
          |     |<----- User Control (StreamIsRecorded) ----------|
          |     |                                                 |
          |     |<-------- UserControl (StreamBegin) -------------|
          |     |                                                 |
          |     |<---- Command Message(onStatus-play reset) ------|
          |     |                                                 |
          |     |<---- Command Message(onStatus-play start) ------|
          |     |                                                 |
          |     |------------------ Audio Message---------------->|
          |     |                                                 |
          |     |------------------ Video Message---------------->|
          |     |                                                 |
                                          |
                                          |
                 Keep receiving audio and video stream till finishes

+-------------+                                     +----------+    
         |  Client     |                  |                  |  Server  |
         +-------------+                  |                  +----------+
                |          |      Handshaking  Done    |          |
                |                         |                       |
                |                         |                       |
       ---+---- |---------   Command Message(connect)   --------->|
          |     |                                                 |
       Connect  |<---------- Window Acknowledge Size -------------|
          |     |                                                 |
          |     |<------------- Set Peer BandWidth ---------------|
          |     |                                                 |
          |     |----------- Window Acknowledge Size ------------>|
          |     |                                                 |
          |     |<--------- User Control(StreamBegin) ------------|
          |     |                                                 |
       ---+---- |--------------- Command Message ---------------->|
                |          (_result- connect response)            |
                |                                                 |
       ---+---- |---------Command Message(createStream) --------->|
       Create   |                                                 |
       Stream   |                                                 |
       ---+---- |<-------------- Command Message -----------------|
                |       (_result- createStream response)          |
                |                                                 |
       ---+---- |--------- Command Message (publish) ------------>|
          |     |                                                 |
       publish  |<-------- UserControl (StreamBegin) -------------|
          |     |                                                 |
          |     |---------- Data Message (Metadata) ------------->|
          |     |                                                 |
          |     |------------------ Audio Message---------------->|
          |     |                                                 |
          |     |----------------- SetChunkSize ----------------->|
          |     |                                                 |
          |     |<--------------- Command Message ----------------|
          |     |            (_result- publish result)            |
          |     |------------------ Video Message---------------->|
                                         |
                                         |
                           Until the stream is complete

+---------+-------------------------+  
  | Header  | Aggregate Message body  |  
  +---------+-------------------------+  
          聚合消息的格式
  +--------+--------------+--------------+--------+-------------+---------------+ - - - -
  |Header 0|Message Data 0|Back Pointer 0|Header 1|Message Data 1|Back Pointer 1|
  +--------+--------------+--------------+--------+--------------+--------------+ - - - -
            聚合消息的body  

#define RTMPConnectSuccess      @"NetConnection.Connect.Success"
#define RTMPPublishStart        @"NetStream.Publish.Start"
#define RTMPPublishBadName      @"NetStream.Publish.BadName"
#define RTMPPlayStart           @"NetStream.Play.Start"
#define RTMPPlayReset           @"NetStream.Play.Reset"
#define RTMPPlayStreamNotFound  @"NetStream.Play.StreamNotFound"
typedef enum : char {
    RTMPResponseCommand_Result       = 0x1, //_Result命令
    RTMPResponseCommandOnBWDone      = 0x2, //OnBWDone命令
    RTMPResponseCommandOnFCPublish   = 0x3, //OnFCPublish命令
    RTMPResponseCommandOnStatus      = 0x4, //OnStatus命令
    RTMPResponseCommandOnFCUnpublish = 0x5, //OnFCUnpublish命令
    RTMPResponseCommandOnMetaData    = 0x6, //OnMetaData命令
    RTMPResponseCommandUnkonwn       = 0x7f,//未知类型
} RTMPResponseCommandType;

+-------+     +--------------+----------------+  
  | Chunk |  =  | Chunk Header |   Chunk Data   |  
  +-------+     +--------------+----------------+

+--------------+     +-------------+----------------+-------------------+ 
  | Chunk Header |  =  | Basic header| Message Header |Extended Timestamp |  
  +--------------+     +-------------+----------------+-------------------+

+-+-+-+-+-+-+-+-+
  |fmt|   cs id   |
  +-+-+-+-+-+-+-+-+

    0                   1            
    0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 
     +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
     |fmt|0 0 0 0 0 0|the second byte|
     +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

    0                   1                   2       
    0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 
   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
     |fmt|1 1 1 1 1 1|the second byte| the third byte|
     +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

    0               
    0 1 2 3 4 5 6 7 
   +-+-+-+-+-+-+-+-+
     |fmt|0<csid<0x3f|
     +-+-+-+-+-+-+-+-+

0                   1                   2                   3   
   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 
  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+  
  |                     timestamp                 | message length:           
  +-------------------------------+---------------+---------------+  
  :                               |message type id|               : 
  +-------------------------------+---------------+---------------+ 
  :                  message stream id            |  
  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

0                   1                   2                   3   
   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 
  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ 
  |                           timestamp                           |   
  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

+-----------+
  |Chunk Data |
  +-----------+

class SrsChunkStream
{
public:
    // Represents the basic header fmt,
    // which used to identify the variant message header type.
    char fmt;
    // Represents the basic header cid,
    // which is the chunk stream id.
    int cid;
    // Cached message header
    SrsMessageHeader header;
    // Whether the chunk message header has extended timestamp.
    bool extended_timestamp;
    // The partially read message.
    SrsCommonMessage* msg;
    // Decoded msg count, to identify whether the chunk stream is fresh.
    int64_t msg_count;
public:
    SrsChunkStream(int _cid);
    virtual ~SrsChunkStream();
};

struct Context
{
    // The original RTMP url.
    std::string url;
    // Parse from url.
    std::string tcUrl;
    std::string host;
    std::string vhost;
    std::string app;
    std::string stream;
    std::string param;
    // Parse ip:port from host.
    std::string ip;
    int port;
    // The URL schema, about vhost/app/stream?param
    srs_url_schema schema;
    // The server information, response by connect app.
    SrsServerInfo si;
    // The extra request object for connect to server, NULL to ignore.
    SrsRequest* req;
    // the message received cache,
    // for example, when got aggregate message,
    // the context will parse to videos/audios,
    // and return one by one.
    std::vector<SrsCommonMessage*> msgs;
    SrsRtmpClient* rtmp;
    SimpleSocketStream* skt;
    int stream_id;
    // the remux raw codec.
    SrsRawH264Stream avc_raw;
    SrsRawAacStream aac_raw;
    // about SPS, @see: 7.3.2.1.1, ISO_IEC_14496-10-AVC-2012.pdf, page 62
    std::string h264_sps;
    std::string h264_pps;
    // whether the sps and pps sent,
    // @see https://github.com/ossrs/srs/issues/203
    bool h264_sps_pps_sent;
    // only send the ssp and pps when both changed.
    // @see https://github.com/ossrs/srs/issues/204
    bool h264_sps_changed;
    bool h264_pps_changed;
    // the aac sequence header.
    std::string aac_specific_config;
    // user set timeout, in ms.
    int64_t stimeout;
    int64_t rtimeout;
    // The RTMP handler level buffer, can used to format packet.
    char buffer[1024];
    Context() : port(0) {
        rtmp = NULL;
        skt = NULL;
        req = NULL;
        stream_id = 0;
        h264_sps_pps_sent = false;
        h264_sps_changed = false;
        h264_pps_changed = false;
        rtimeout = stimeout = SRS_UTIME_NO_TIMEOUT;
        schema = srs_url_schema_normal;
    }
    virtual ~Context() {
        srs_freep(req);
        srs_freep(rtmp);
        srs_freep(skt);
        std::vector<SrsCommonMessage*>::iterator it;
        for (it = msgs.begin(); it != msgs.end(); ++it) {
            SrsCommonMessage* msg = *it;
            srs_freep(msg);
        }
        msgs.clear();
    }
};

struct SrsServerInfo
{
    std::string ip;
    std::string sig;
    int pid;
    int cid;
    int major;
    int minor;
    int revision;
    int build;
    SrsServerInfo();
};

class SrsRequest
{
public:
    // The client ip.
    std::string ip;
public:
    // The tcUrl: rtmp://request_vhost:port/app/stream
    // support pass vhost in query string, such as:
    //    rtmp://ip:port/app?vhost=request_vhost/stream
    //    rtmp://ip:port/app...vhost...request_vhost/stream
    std::string tcUrl;
    std::string pageUrl;
    std::string swfUrl;
    double objectEncoding;
// The data discovery from request.
public:
    // Discovery from tcUrl and play/publish.
    std::string schema;
    // The vhost in tcUrl.
    std::string vhost;
    // The host in tcUrl.
    std::string host;
    // The port in tcUrl.
    int port;
    // The app in tcUrl, without param.
    std::string app;
    // The param in tcUrl(app).
    std::string param;
    // The stream in play/publish
    std::string stream;
    // For play live stream,
    // used to specified the stop when exceed the duration.
    // @see https://github.com/ossrs/srs/issues/45
    // in srs_utime_t.
    srs_utime_t duration;
    // The token in the connect request,
    // used for edge traverse to origin authentication,
    // @see https://github.com/ossrs/srs/issues/104
    SrsAmf0Object* args;
public:
    SrsRequest();
    virtual ~SrsRequest();
public:
    // Deep copy the request, for source to use it to support reload,
    // For when initialize the source, the request is valid,
    // When reload it, the request maybe invalid, so need to copy it.
    virtual SrsRequest* copy();
    // update the auth info of request,
    // To keep the current request ptr is ok,
    // For many components use the ptr of request.
    virtual void update_auth(SrsRequest* req);
    // Get the stream identify, vhost/app/stream.
    virtual std::string get_stream_url();
    // To strip url, user must strip when update the url.
    virtual void strip();
public:
    // Transform it as HTTP request.
    virtual SrsRequest* as_http();
};

// implements the client role protocol.
class SrsRtmpClient
{
private:
    SrsHandshakeBytes* hs_bytes;//跟我们之前封装的握手接口类似
protected:
    SrsProtocol* protocol;//rtmp协议封装,就是封装了我们的connect, createStream,publish等接口
    ISrsProtocolReadWriter* io;//就是SimpleSocketStream
public:
    SrsRtmpClient(ISrsProtocolReadWriter* skt);
    virtual ~SrsRtmpClient();
// Protocol methods proxy
public:
    virtual void set_recv_timeout(srs_utime_t tm);
    virtual void set_send_timeout(srs_utime_t tm);
    virtual int64_t get_recv_bytes();
    virtual int64_t get_send_bytes();
    virtual srs_error_t recv_message(SrsCommonMessage** pmsg);
    virtual srs_error_t decode_message(SrsCommonMessage* msg, SrsPacket** ppacket);
    virtual srs_error_t send_and_free_message(SrsSharedPtrMessage* msg, int stream_id);
    virtual srs_error_t send_and_free_messages(SrsSharedPtrMessage** msgs, int nb_msgs, int stream_id);
    virtual srs_error_t send_and_free_packet(SrsPacket* packet, int stream_id);
public:
    // handshake with server, try complex, then simple handshake.
    virtual srs_error_t handshake();
    // only use simple handshake
    virtual srs_error_t simple_handshake();
    // only use complex handshake
    virtual srs_error_t complex_handshake();
    // Connect to RTMP tcUrl and app, get the server info.
    //
    // @param app, The app to connect at, for example, live.
    // @param tcUrl, The tcUrl to connect at, for example, rtmp://ossrs.net/live.
    // @param req, the optional req object, use the swfUrl/pageUrl if specified. NULL to ignore.
    // @param dsu, Whether debug SRS upnode. For edge, set to true to send its info to upnode.
    // @param si, The server information, retrieve from response of connect app request. NULL to ignore.
    virtual srs_error_t connect_app(std::string app, std::string tcUrl, SrsRequest* r, bool dsu, SrsServerInfo* si);
    // Create a stream, then play/publish data over this stream.
    virtual srs_error_t create_stream(int& stream_id);
    // start play stream.
    virtual srs_error_t play(std::string stream, int stream_id, int chunk_size);
    // start publish stream. use flash publish workflow:
    //       connect-app => create-stream => flash-publish
    virtual srs_error_t publish(std::string stream, int stream_id, int chunk_size);
    // start publish stream. use FMLE publish workflow:
    //       connect-app => FMLE publish
    virtual srs_error_t fmle_publish(std::string stream, int& stream_id);
public:
    template<class T>
    srs_error_t expect_message(SrsCommonMessage** pmsg, T** ppacket)
    {
        return protocol->expect_message<T>(pmsg, ppacket);
    }
};

/**
 * The system io reader/writer architecture:
 *                                         +---------------+  +---------------+
 *                                         | IStreamWriter |  | IVectorWriter |
 *                                         +---------------+  +---------------+
 *                                         | + write()     |  | + writev()    |
 *                                         +-------------+-+  ++--------------+
 * +----------+     +--------------------+               /\   /\
 * | IReader  |     |    IStatistic      |                 \ /
 * +----------+     +--------------------+                  V
 * | + read() |     | + get_recv_bytes() |           +------+----+
 * +------+---+     | + get_send_bytes() |           |  IWriter  |
 *       / \        +---+--------------+-+           +-------+---+
 *        |            / \            / \                   / \
 *        |             |              |                     |
 * +------+-------------+------+      ++---------------------+--+
 * | IProtocolReader           |      | IProtocolWriter         |
 * +---------------------------+      +-------------------------+
 * | + readfully()             |      | + set_send_timeout()    |
 * | + set_recv_timeout()      |      +-------+-----------------+
 * +------------+--------------+             / \
 *             / \                            |
 *              |                             |
 *           +--+-----------------------------+-+
 *           |       IProtocolReadWriter        |
 *           +----------------------------------+
 */

// The protocol provides the rtmp-message-protocol services,
// To recv RTMP message from RTMP chunk stream,
// and to send out RTMP message over RTMP chunk stream.
class SrsProtocol
{
private:
    class AckWindowSize
    {
    public:
        uint32_t window;
        // number of received bytes.
        int64_t nb_recv_bytes;
        // previous responsed sequence number.
        uint32_t sequence_number;
        AckWindowSize();
    };
// For peer in/out
private:
    // The underlayer socket object, send/recv bytes.
    ISrsProtocolReadWriter* skt;
    // The requests sent out, used to build the response.
    // key: transactionId
    // value: the request command name
    std::map<double, std::string> requests;
// For peer in
private:
    // The chunk stream to decode RTMP messages.
    std::map<int, SrsChunkStream*> chunk_streams;
    // Cache some frequently used chunk header.
    // cs_cache, the chunk stream cache.
    SrsChunkStream** cs_cache;
    // The bytes buffer cache, recv from skt, provide services for stream.
    SrsFastStream* in_buffer;
    // The input chunk size, default to 128, set by peer packet.
    int32_t in_chunk_size;
    // The input ack window, to response acknowledge to peer,
    // For example, to respose the encoder, for server got lots of packets.
    AckWindowSize in_ack_size;
    // The output ack window, to require peer to response the ack.
    AckWindowSize out_ack_size;
    // The buffer length set by peer.
    int32_t in_buffer_length;
    // Whether print the protocol level debug info.
    // Generally we print the debug info when got or send first A/V packet.
    bool show_debug_info;
    // Whether auto response when recv messages.
    bool auto_response_when_recv;
    // When not auto response message, manual flush the messages in queue.
    std::vector<SrsPacket*> manual_response_queue;
// For peer out
private:
    iovec* out_iovs;
    int nb_out_iovs;
    char* out_c0c3_caches;
    // Whether warned user to increase the c0c3 header cache.
    bool warned_c0c3_cache_dry;
    // The output chunk size, default to 128, set by config.
    int32_t out_chunk_size;
public:
    SrsProtocol(ISrsProtocolReadWriter* io);
    virtual ~SrsProtocol();
public:
    // Set the auto response message when recv for protocol stack.
    virtual void set_auto_response(bool v);
    virtual srs_error_t manual_response_flush();
public:
public:
    virtual void set_recv_timeout(srs_utime_t tm);
    virtual srs_utime_t get_recv_timeout();
    virtual void set_send_timeout(srs_utime_t tm);
    virtual srs_utime_t get_send_timeout();
    // Get recv/send bytes.
    virtual int64_t get_recv_bytes();
    virtual int64_t get_send_bytes();
public:
    virtual srs_error_t set_in_window_ack_size(int ack_size);
public:
    virtual srs_error_t recv_message(SrsCommonMessage** pmsg);
    virtual srs_error_t decode_message(SrsCommonMessage* msg, SrsPacket** ppacket);
    virtual srs_error_t send_and_free_message(SrsSharedPtrMessage* msg, int stream_id);
    virtual srs_error_t send_and_free_messages(SrsSharedPtrMessage** msgs, int nb_msgs, int stream_id);
    virtual srs_error_t send_and_free_packet(SrsPacket* packet, int stream_id);
private:
    virtual srs_error_t do_send_messages(SrsSharedPtrMessage** msgs, int nb_msgs);
    virtual srs_error_t do_iovs_send(iovec* iovs, int size);
    virtual srs_error_t do_send_and_free_packet(SrsPacket* packet, int stream_id);
    virtual srs_error_t do_decode_message(SrsMessageHeader& header, SrsBuffer* stream, SrsPacket** ppacket);
    virtual srs_error_t recv_interlaced_message(SrsCommonMessage** pmsg);
    virtual srs_error_t read_basic_header(char& fmt, int& cid);
    virtual srs_error_t read_message_header(SrsChunkStream* chunk, char fmt);
    virtual srs_error_t read_message_payload(SrsChunkStream* chunk, SrsCommonMessage** pmsg);
    virtual srs_error_t on_recv_message(SrsCommonMessage* msg);
    virtual srs_error_t on_send_packet(SrsMessageHeader* mh, SrsPacket* packet);
private:
    virtual srs_error_t response_acknowledgement_message();
    virtual srs_error_t response_ping_message(int32_t timestamp);
private:
    virtual void print_debug_info();
};

context->rtmp->connect_app(c->app, tcUrl, c->req, true, &c->si))

SrsConnectAppPacket* pkt = new SrsConnectAppPacket();
        pkt->command_object->set("app", SrsAmf0Any::str(app.c_str()));
        pkt->command_object->set("flashVer", SrsAmf0Any::str("WIN 15,0,0,239"));
        if (r) {
            pkt->command_object->set("swfUrl", SrsAmf0Any::str(r->swfUrl.c_str()));
        } else {
            pkt->command_object->set("swfUrl", SrsAmf0Any::str());
        }
        if (r && r->tcUrl != "") {
            pkt->command_object->set("tcUrl", SrsAmf0Any::str(r->tcUrl.c_str()));
        } else {
            pkt->command_object->set("tcUrl", SrsAmf0Any::str(tcUrl.c_str()));
        }
        pkt->command_object->set("fpad", SrsAmf0Any::boolean(false));
        pkt->command_object->set("capabilities", SrsAmf0Any::number(239));
        pkt->command_object->set("audioCodecs", SrsAmf0Any::number(3575));
        pkt->command_object->set("videoCodecs", SrsAmf0Any::number(252));
        pkt->command_object->set("videoFunction", SrsAmf0Any::number(1));
        if (r) {
            pkt->command_object->set("pageUrl", SrsAmf0Any::str(r->pageUrl.c_str()));
        } else {
            pkt->command_object->set("pageUrl", SrsAmf0Any::str());
        }
        pkt->command_object->set("objectEncoding", SrsAmf0Any::number(0));
        if (dsu && r && r->args) {
            srs_freep(pkt->args);
            pkt->args = r->args->copy()->to_object();
        }
        if ((err = protocol->send_and_free_packet(pkt, 0)) != srs_success) {//发送packet
            return srs_error_wrap(err, "send packet");
        }

class SrsConnectAppPacket : public SrsPacket
{
public:
    // Name of the command. Set to "connect".
    std::string command_name;
    // Always set to 1.
    double transaction_id;
    SrsAmf0Object* command_object;
    SrsAmf0Object* args;
public:
    SrsConnectAppPacket();
    virtual ~SrsConnectAppPacket();
// Decode functions for concrete packet to override.
public:
    virtual srs_error_t decode(SrsBuffer* stream);
// Encode functions for concrete packet to override.
public:
    virtual int get_prefer_cid();
    virtual int get_message_type();
protected:
    virtual int get_size();
    virtual srs_error_t encode_packet(SrsBuffer* stream);
};

srs_error_t SrsProtocol::do_send_and_free_packet(SrsPacket* packet, int stream_id)
{
    srs_error_t err = srs_success;
    srs_assert(packet);
    SrsAutoFree(SrsPacket, packet);
    SrsCommonMessage* msg = new SrsCommonMessage();
    SrsAutoFree(SrsCommonMessage, msg);
    if ((err = packet->to_msg(msg, stream_id)) != srs_success) {
        return srs_error_wrap(err, "to message");
    }
    SrsSharedPtrMessage* shared_msg = new SrsSharedPtrMessage();
    if ((err = shared_msg->create(msg)) != srs_success) {
        srs_freep(shared_msg);
        return srs_error_wrap(err, "create message");
    }
    if ((err = send_and_free_message(shared_msg, stream_id)) != srs_success) {
        return srs_error_wrap(err, "send packet");
    }
    if ((err = on_send_packet(&msg->header, packet)) != srs_success) {
        return srs_error_wrap(err, "on send packet");
    }
    return err;
}

srs_error_t SrsProtocol::do_send_messages(SrsSharedPtrMessage** msgs, int nb_msgs)
{
    srs_error_t err = srs_success;
#ifdef SRS_PERF_COMPLEX_SEND
    int iov_index = 0;
    iovec* iovs = out_iovs + iov_index;
    int c0c3_cache_index = 0;
    char* c0c3_cache = out_c0c3_caches + c0c3_cache_index;
    // try to send use the c0c3 header cache,
    // if cache is consumed, try another loop.
    for (int i = 0; i < nb_msgs; i++) {
        SrsSharedPtrMessage* msg = msgs[i];
        if (!msg) {
            continue;
        }
        // ignore empty message.
        if (!msg->payload || msg->size <= 0) {
            continue;
        }
        // p set to current write position,
        char* p = msg->payload;
        char* pend = msg->payload + msg->size;
        // always write the header event payload is empty.
        while (p < pend) {
            // always has header
            int nb_cache = SRS_CONSTS_C0C3_HEADERS_MAX - c0c3_cache_index;
            int nbh = msg->chunk_header(c0c3_cache, nb_cache, p == msg->payload);
            srs_assert(nbh > 0);
            // header iov
            iovs[0].iov_base = c0c3_cache;
            iovs[0].iov_len = nbh;
            // payload iov
            int payload_size = srs_min(out_chunk_size, (int)(pend - p));
            iovs[1].iov_base = p;
            iovs[1].iov_len = payload_size;
            // consume sendout bytes.
            p += payload_size;
            if (iov_index >= nb_out_iovs - 2) {
                int ov = nb_out_iovs;
                nb_out_iovs = 2 * nb_out_iovs;
                int realloc_size = sizeof(iovec) * nb_out_iovs;
                out_iovs = (iovec*)realloc(out_iovs, realloc_size);
                srs_warn("resize iovs %d => %d, max_msgs=%d", ov, nb_out_iovs, SRS_PERF_MW_MSGS);
            }
            // to next pair of iovs
            iov_index += 2;
            iovs = out_iovs + iov_index;
            // to next c0c3 header cache
            c0c3_cache_index += nbh;
            c0c3_cache = out_c0c3_caches + c0c3_cache_index;
            // the cache header should never be realloc again,
            // for the ptr is set to iovs, so we just warn user to set larger
            // and use another loop to send again.
            int c0c3_left = SRS_CONSTS_C0C3_HEADERS_MAX - c0c3_cache_index;
            if (c0c3_left < SRS_CONSTS_RTMP_MAX_FMT0_HEADER_SIZE) {
                // only warn once for a connection.
                if (!warned_c0c3_cache_dry) {
                    srs_warn("c0c3 cache header too small, recoment to %d", SRS_CONSTS_C0C3_HEADERS_MAX + SRS_CONSTS_RTMP_MAX_FMT0_HEADER_SIZE);
                    warned_c0c3_cache_dry = true;
                }
                // when c0c3 cache dry,
                // sendout all messages and reset the cache, then send again.
                if ((err = do_iovs_send(out_iovs, iov_index)) != srs_success) {
                    return srs_error_wrap(err, "send iovs");
                }
                // reset caches, while these cache ensure
                // atleast we can sendout a chunk.
                iov_index = 0;
                iovs = out_iovs + iov_index;
                c0c3_cache_index = 0;
                c0c3_cache = out_c0c3_caches + c0c3_cache_index;
            }
        }
    }
    // maybe the iovs already sendout when c0c3 cache dry,
    // so just ignore when no iovs to send.
    if (iov_index <= 0) {
        return err;
    }
    return do_iovs_send(out_iovs, iov_index);
#else
    // try to send use the c0c3 header cache,
    // if cache is consumed, try another loop.
    for (int i = 0; i < nb_msgs; i++) {
        SrsSharedPtrMessage* msg = msgs[i];
        if (!msg) {
            continue;
        }
        // ignore empty message.
        if (!msg->payload || msg->size <= 0) {
            continue;
        }
        // p set to current write position,
        char* p = msg->payload;
        char* pend = msg->payload + msg->size;
        // always write the header event payload is empty.
        while (p < pend) {
            // for simple send, send each chunk one by one
            iovec* iovs = out_iovs;
            char* c0c3_cache = out_c0c3_caches;
            int nb_cache = SRS_CONSTS_C0C3_HEADERS_MAX;
            // always has header
            int nbh = msg->chunk_header(c0c3_cache, nb_cache, p == msg->payload);
            srs_assert(nbh > 0);
            // header iov
            iovs[0].iov_base = c0c3_cache;
            iovs[0].iov_len = nbh;
            // payload iov
            int payload_size = srs_min(out_chunk_size, pend - p);
            iovs[1].iov_base = p;
            iovs[1].iov_len = payload_size;
            // consume sendout bytes.
            p += payload_size;
            if ((er = skt->writev(iovs, 2, NULL)) != srs_success) {
                return srs_error_wrap(err, "writev");
            }
        }
    }
    return err;
#endif
}