• SRS流媒体服务器：服务器读取RTMP推流数据

    目录

    1. 处理RTMP推流video message
    2. 处理RTMP推流audio message
    3. 处理RTMP推流onMetaData message

    1. 服务器读取RTMP推流数据

    RTMP推流、拉流创建连接说到，SrsRtmpConn::publishing会创建SrsPublishRecvThread协程来接收数据，然后在SrsRtmpConn::do_publishing开启接收协程。

    而最终会执行到SrsRecvThread::do_cycle，负责读取RTMP推流数据。

    srs_error_t SrsRecvThread::do_cycle() { srs_error_t err = srs_success; while (true) { if ((err = trd->pull()) != srs_success) { return srs_error_wrap(err, "recv thread"); } // When the pumper is interrupted, wait then retry. if (pumper->interrupted()) { srs_usleep(timeout); continue; } SrsCommonMessage *msg = NULL; // Process the received message. 读取msg并进行消费 if ((err = rtmp->recv_message(&msg)) == srs_success) { err = pumper->consume(msg); } if (err != srs_success) { // Interrupt the receive thread for any error. trd->interrupt(); // Notify the pumper to quit for error. pumper->interrupt(err); return srs_error_wrap(err, "recv thread"); } } return err; }

    3.SrsRtmpServer::recv_message会获取message，如果获取的是协议控制消息或者用户控制消息会进行解码成packet然后进行响应处理，比如设置ackowledgement_window_size或者chunk_size。

    如果是Aduio message，Video message或者Data message，不进行解码成packet，仅获取message。

    TODO：后续写一篇关于如何接收message的分析，详细代码见SrsProtocol::recv_interlaced_message

    1. srs_error_t SrsProtocol::recv_message(SrsCommonMessage** pmsg)
    2. {
    3.     *pmsg = NULL;
    4.   
    5.     srs_error_t err = srs_success;
    6.   
    7.     while (true) {
    8.         SrsCommonMessage* msg = NULL;
    9.       
    10.         if ((err = recv_interlaced_message(&msg)) != srs_success) { //获取message
    11.             srs_freep(msg);
    12.             return srs_error_wrap(err, "recv interlaced message");
    13.         }
    14.       
    15.         if (!msg) {
    16.             continue;
    17.         }
    18.       
    19.         if (msg->size <= 0 || msg->header.payload_length <= 0) {
    20.             srs_trace("ignore empty message(type=%d, size=%d, time=%" PRId64 ", sid=%d).",
    21.                       msg->header.message_type, msg->header.payload_length,
    22.                       msg->header.timestamp, msg->header.stream_id);
    23.             srs_freep(msg);
    24.             continue;
    25.         }
    26.       
    27.         if ((err = on_recv_message(msg)) != srs_success) { //对获取的message进行判断类型并进行相应处理
    28.             srs_freep(msg);
    29.             return srs_error_wrap(err, "on recv message");
    30.         }
    31.       
    32.         *pmsg = msg;
    33.         break;
    34.     }
    35.   
    36.     return err;
    37. }
    38.

    4.SrsRtmpServer::recv_message会获取message后，在SrsPublishRecvThread::consume中进行处理，最后交由SrsRtmpConn::handle_publish_message进行处理。

    1. srs_error_t SrsPublishRecvThread::consume(SrsCommonMessage *msg) {
    2.     srs_error_t err = srs_success;
    3.  
    4.     // when cid changed, change it.
    5.     if (ncid.compare(cid)) {
    6.         _srs_context->set_id(ncid);
    7.         cid = ncid;
    8.     }
    9.  
    10.     _nb_msgs++;
    11.  
    12.     if (msg->header.is_video()) {
    13.         video_frames++;
    14.     }
    15.     if (msg->header.is_audio()) {
    16.         audio_frames++;
    17.     }
    18.  
    19.     // log to show the time of recv thread.
    20.     srs_verbose("recv thread now=%" PRId64 "us, got msg time=%" PRId64 "ms, size=%d",
    21.                 srs_update_system_time(), msg->header.timestamp, msg->size);
    22.  
    23.     // the rtmp connection will handle this message 由RTMP连接处理message
    24.     err = _conn->handle_publish_message(_source, msg);
    25.  
    26.     // must always free it,
    27.     // the source will copy it if need to use.
    28.     srs_freep(msg);
    29.  
    30.     if (err != srs_success) {
    31.         return srs_error_wrap(err, "handle publish message");
    32.     }
    33.  
    34.     // Yield to another coroutines.
    35.     // @see https://github.com/ossrs/srs/issues/2194#issuecomment-777463768
    36.     if (++nn_msgs_for_yield_ >= 15) {
    37.         nn_msgs_for_yield_ = 0;
    38.         srs_thread_yield();
    39.     }
    40.  
    41.     return err;
    42. }
    43.

    5.SrsRtmpConn::handle_publish_message会先判断message type是否是命令消息，如果是进行解码并进行处理，推流不需要这些事件数据。

    6.audio、video以及data数据由SrsRtmpConn::process_publish_message进行处理。

    1. srs_error_t SrsRtmpConn::handle_publish_message(SrsLiveSource* source, SrsCommonMessage* msg)
    2. {
    3.     srs_error_t err = srs_success;
    4.   
    5.     // process publish event. 处理推流事件，因为推流不需要这些事件数据，audio、video以及data数据不会进入此逻辑
    6.     if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) {
    7.         SrsPacket* pkt = NULL;
    8.         if ((err = rtmp->decode_message(msg, &pkt)) != srs_success) {
    9.             return srs_error_wrap(err, "rtmp: decode message");
    10.         }
    11.         SrsAutoFree(SrsPacket, pkt);
    12.       
    13.         // for flash, any packet is republish.
    14.         if (info->type == SrsRtmpConnFlashPublish) {
    15.             // flash unpublish.
    16.             // TODO: maybe need to support republish.
    17.             srs_trace("flash flash publish finished.");
    18.             return srs_error_new(ERROR_CONTROL_REPUBLISH, "rtmp: republish");
    19.         }
    20.       
    21.         // for fmle, drop others except the fmle start packet.
    22.         if (dynamic_cast<SrsFMLEStartPacket*>(pkt)) {
    23.             SrsFMLEStartPacket* unpublish = dynamic_cast<SrsFMLEStartPacket*>(pkt);
    24.             if ((err = rtmp->fmle_unpublish(info->res->stream_id, unpublish->transaction_id)) != srs_success) {
    25.                 return srs_error_wrap(err, "rtmp: republish");
    26.             }
    27.             return srs_error_new(ERROR_CONTROL_REPUBLISH, "rtmp: republish");
    28.         }
    29.       
    30.         srs_trace("fmle ignore AMF0/AMF3 command message.");
    31.         return err;
    32.     }
    33.   
    34.     // video, audio, data message
    35.     if ((err = process_publish_message(source, msg)) != srs_success) {
    36.         return srs_error_wrap(err, "rtmp: consume message");
    37.     }
    38.   
    39.     return err;
    40. }
    41.

    7.SrsRtmpConn::process_publish_message会判断message_type来区别audio、video和onMetaData来进行不同处理。

    SrsLiveSource::on_audio负责处理音频message。

    SrsLiveSource::on_video负责处理视频message。

    SrsLiveSource::on_meta_data负责处理onMetaData message。

    1. srs_error_t SrsRtmpConn::process_publish_message(SrsLiveSource* source, SrsCommonMessage* msg)
    2. {
    3.     srs_error_t err = srs_success;
    4.   
    5.     // for edge, directly proxy message to origin.
    6.     if (info->edge) {
    7.         if ((err = source->on_edge_proxy_publish(msg)) != srs_success) {
    8.             return srs_error_wrap(err, "rtmp: proxy publish");
    9.         }
    10.         return err;
    11.     }
    12.   
    13.     // process audio packet
    14.     if (msg->header.is_audio()) {
    15.         if ((err = source->on_audio(msg)) != srs_success) {
    16.             return srs_error_wrap(err, "rtmp: consume audio");
    17.         }
    18.         return err;
    19.     }
    20.     // process video packet
    21.     if (msg->header.is_video()) {
    22.         if ((err = source->on_video(msg)) != srs_success) {
    23.             return srs_error_wrap(err, "rtmp: consume video");
    24.         }
    25.         return err;
    26.     }
    27.   
    28.     // process aggregate packet
    29.     if (msg->header.is_aggregate()) {
    30.         if ((err = source->on_aggregate(msg)) != srs_success) {
    31.             return srs_error_wrap(err, "rtmp: consume aggregate");
    32.         }
    33.         return err;
    34.     }
    35.   
    36.     // process onMetaData
    37.     if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
    38.         SrsPacket* pkt = NULL;
    39.         if ((err = rtmp->decode_message(msg, &pkt)) != srs_success) {
    40.             return srs_error_wrap(err, "rtmp: decode message");
    41.         }
    42.         SrsAutoFree(SrsPacket, pkt);
    43.       
    44.         if (dynamic_cast<SrsOnMetaDataPacket*>(pkt)) {
    45.             SrsOnMetaDataPacket* metadata = dynamic_cast<SrsOnMetaDataPacket*>(pkt);
    46.             if ((err = source->on_meta_data(msg, metadata)) != srs_success) {
    47.                 return srs_error_wrap(err, "rtmp: consume metadata");
    48.             }
    49.             return err;
    50.         }
    51.         return err;
    52.     }
    53.   
    54.     return err;
    55. }
    56.
    1. 处理RTMP推流video message
      SrsLiveSource::on_video负责处理视频message。主要的功能为：
      将SrsCommonMessage转换成SrsSharedPtrMessage，用于更少内存拷贝的音频/视频/数据message，SrsSharedPtrMessage使用共享计数方法。
      将SrsSharedPtrMessage交给SrsLiveSource::on_video_imp处理。
    1. srs_error_t SrsLiveSource::on_video(SrsCommonMessage* shared_video)
    2. {
    3.     srs_error_t err = srs_success;
    4.   
    5.     // monotically increase detect.
    6.     if (!mix_correct && is_monotonically_increase) {
    7.         if (last_packet_time > 0 && shared_video->header.timestamp < last_packet_time) {
    8.             is_monotonically_increase = false;
    9.             srs_warn("VIDEO: stream not monotonically increase, please open mix_correct.");
    10.         }
    11.     }
    12.     last_packet_time = shared_video->header.timestamp;
    13.   
    14.     // drop any unknown header video.
    15.     // @see https://github.com/ossrs/srs/issues/421
    16.     if (!SrsFlvVideo::acceptable(shared_video->payload, shared_video->size)) {
    17.         char b0 = 0x00;
    18.         if (shared_video->size > 0) {
    19.             b0 = shared_video->payload[0];
    20.         }
    21.       
    22.         srs_warn("drop unknown header video, size=%d, bytes[0]=%#x", shared_video->size, b0);
    23.         return err;
    24.     }
    25.   
    26.     // convert shared_video to msg, user should not use shared_video again.
    27.     // the payload is transfer to msg, and set to NULL in shared_video.
    28.     SrsSharedPtrMessage msg; //创建共享message ptr，将SrsCommonMessage转换成SrsSharedPtrMessage，用于更少内存拷贝的音频/视频/数据message
    29.     if ((err = msg.create(shared_video)) != srs_success) { //SrsSharedPtrMessage使用引用计数
    30.         return srs_error_wrap(err, "create message");
    31.     }
    32.   
    33.     // directly process the video message. mix_correct为false，直接处理video message
    34.     if (!mix_correct) {
    35.         return on_video_imp(&msg);
    36.     }
    37.   
    38.     // insert msg to the queue. 如果mix_correct为true，先将msg加入按时间戳排序map，然后从中取出第一个msg进行处理
    39.     mix_queue->push(msg.copy());
    40.   
    41.     // fetch someone from mix queue. 取出的msg可能是音频或视频，所以需要进行判断
    42.     SrsSharedPtrMessage* m = mix_queue->pop(); 
    43.     if (!m) {
    44.         return err;
    45.     }
    46.   
    47.     // consume the monotonically increase message. 
    48.     if (m->is_audio()) {
    49.         err = on_audio_imp(m);
    50.     } else {
    51.         err = on_video_imp(m);
    52.     }
    53.     srs_freep(m);
    54.   
    55.     return err;
    56. }
    57.

    2.SrsLiveSource::on_video_imp处理SrsSharedPtrMessage主要过程包括：

    1. 遍历对应source拉流客户端，将msg发送给每个客户端。
    2. 如果不是sequence header，加入gop缓存中，gop会缓存最后一个gop的packet。
    1. srs_error_t SrsLiveSource::on_video_imp(SrsSharedPtrMessage* msg)
    2. {
    3.     srs_error_t err = srs_success;
    4.   
    5.     bool is_sequence_header = SrsFlvVideo::sh(msg->payload, msg->size);
    6.   
    7.     // whether consumer should drop for the duplicated sequence header.
    8.     bool drop_for_reduce = false;
    9.     if (is_sequence_header && meta->previous_vsh() && _srs_config->get_reduce_sequence_header(req->vhost)) {
    10.         if (meta->previous_vsh()->size == msg->size) {
    11.             drop_for_reduce = srs_bytes_equals(meta->previous_vsh()->payload, msg->payload, msg->size);
    12.             srs_warn("drop for reduce sh video, size=%d", msg->size);
    13.         }
    14.     }
    15.   
    16.     // cache the sequence header if h264
    17.     // donot cache the sequence header to gop_cache, return here.
    18.     if (is_sequence_header && (err = meta->update_vsh(msg)) != srs_success) {
    19.         return srs_error_wrap(err, "meta update video");
    20.     }
    21.   
    22.     // Copy to hub to all utilities.
    23.     if ((err = hub->on_video(msg, is_sequence_header)) != srs_success) {
    24.         return srs_error_wrap(err, "hub consume video");
    25.     }
    26.  
    27.     // For bridger to consume the message.
    28.     if (bridger_ && (err = bridger_->on_video(msg)) != srs_success) {
    29.         return srs_error_wrap(err, "bridger consume video");
    30.     }
    31.  
    32.     // copy to all consumer 遍历对应source拉流客户端，将msg发送给每个客户端
    33.     if (!drop_for_reduce) {
    34.         for (int i = 0; i < (int)consumers.size(); i++) {
    35.             SrsLiveConsumer* consumer = consumers.at(i);
    36.             if ((err = consumer->enqueue(msg, atc, jitter_algorithm)) != srs_success) {
    37.                 return srs_error_wrap(err, "consume video");
    38.             }
    39.         }
    40.     }
    41.   
    42.     // when sequence header, donot push to gop cache and adjust the timestamp.
    43.     if (is_sequence_header) {
    44.         return err;
    45.     }
    46.   
    47.     // cache the last gop packets
    48.     if ((err = gop_cache->cache(msg)) != srs_success) {
    49.         return srs_error_wrap(err, "gop cache consume vdieo");
    50.     }
    51.   
    52.     // if atc, update the sequence header to abs time.
    53.     if (atc) {
    54.         if (meta->vsh()) {
    55.             meta->vsh()->timestamp = msg->timestamp;
    56.         }
    57.         if (meta->data()) {
    58.             meta->data()->timestamp = msg->timestamp;
    59.         }
    60.     }
    61.   
    62.     return err;
    63. }
    64.

    3.每个拉流客户端是一个SrsLiveConsumer，会将msg加入SrsMessageQueue对象中，实际保存的是SrsMessageQueue对象中的SrsFastVector对象中的SrsSharedPtrMessage数组。

    1. srs_error_t SrsLiveConsumer::enqueue(SrsSharedPtrMessage* shared_msg, bool atc, SrsRtmpJitterAlgorithm ag)
    2. {
    3.     srs_error_t err = srs_success;
    4.   
    5.     SrsSharedPtrMessage* msg = shared_msg->copy();
    6.  
    7.     if (!atc) {
    8.         if ((err = jitter->correct(msg, ag)) != srs_success) {
    9.             return srs_error_wrap(err, "consume message");
    10.         }
    11.     }
    12.  
    13.     if ((err = queue->enqueue(msg, NULL)) != srs_success) {
    14.         return srs_error_wrap(err, "enqueue message");
    15.     }
    16.   
    17. 	...
    18.   
    19.     return err;
    20. }
    21.  
    1. srs_error_t SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg, bool* is_overflow)
    2. {
    3.     srs_error_t err = srs_success;
    4.  
    5.     msgs.push_back(msg);
    6.  
    7.     // If jitter is off, the timestamp of first sequence header is zero, which wll cause SRS to shrink and drop the
    8.     // keyframes even if there is not overflow packets in queue, so we must ignore the zero timestamps, please
    9.     // @see https://github.com/ossrs/srs/pull/2186#issuecomment-953383063
    10.     if (msg->is_av() && msg->timestamp != 0) {
    11.         if (av_start_time == -1) {
    12.             av_start_time = srs_utime_t(msg->timestamp * SRS_UTIME_MILLISECONDS);
    13.         }
    14.       
    15.         av_end_time = srs_utime_t(msg->timestamp * SRS_UTIME_MILLISECONDS);
    16.     }
    17.  
    18.     if (max_queue_size <= 0) {
    19.         return err;
    20.     }
    21.  
    22.     while (av_end_time - av_start_time > max_queue_size) {
    23.         // notice the caller queue already overflow and shrinked.
    24.         if (is_overflow) {
    25.             *is_overflow = true;
    26.         }
    27.       
    28.         shrink();
    29.     }
    30.   
    31.     return err;
    32. }
    33.  
    1. void SrsFastVector::push_back(SrsSharedPtrMessage* msg)
    2. {
    3.     // increase vector.
    4.     if (count >= nb_msgs) {
    5.         int size = srs_max(SRS_PERF_MW_MSGS * 8, nb_msgs * 2);
    6.         SrsSharedPtrMessage** buf = new SrsSharedPtrMessage*[size];
    7.         for (int i = 0; i < nb_msgs; i++) {
    8.             buf[i] = msgs[i];
    9.         }
    10.         srs_info("fast vector incrase %d=>%d", nb_msgs, size);
    11.       
    12.         // use new array.
    13.         srs_freepa(msgs);
    14.         msgs = buf;
    15.         nb_msgs = size;
    16.     }
    17.   
    18.     msgs[count++] = msg;
    19. }
    20.

    2. 处理RTMP推流audio message

    SrsLiveSource::on_audio负责处理音频message。主要的功能跟处理video message一样，为：
    将SrsCommonMessage转换成SrsSharedPtrMessage，用于更少内存拷贝的音频/视频/数据message，SrsSharedPtrMessage使用共享计数方法。
    将SrsSharedPtrMessage交给SrsLiveSource::on_audio_imp处理。

    1. srs_error_t SrsLiveSource::on_audio(SrsCommonMessage* shared_audio)
    2. {
    3.     srs_error_t err = srs_success;
    4.   
    5.     // monotically increase detect.
    6.     if (!mix_correct && is_monotonically_increase) {
    7.         if (last_packet_time > 0 && shared_audio->header.timestamp < last_packet_time) {
    8.             is_monotonically_increase = false;
    9.             srs_warn("AUDIO: stream not monotonically increase, please open mix_correct.");
    10.         }
    11.     }
    12.     last_packet_time = shared_audio->header.timestamp;
    13.   
    14.     // convert shared_audio to msg, user should not use shared_audio again.
    15.     // the payload is transfer to msg, and set to NULL in shared_audio.
    16.     SrsSharedPtrMessage msg;
    17.     if ((err = msg.create(shared_audio)) != srs_success) {
    18.         return srs_error_wrap(err, "create message");
    19.     }
    20.   
    21.     // directly process the audio message.
    22.     if (!mix_correct) {
    23.         return on_audio_imp(&msg);
    24.     }
    25.   
    26.     // insert msg to the queue.
    27.     mix_queue->push(msg.copy());
    28.   
    29.     // fetch someone from mix queue.
    30.     SrsSharedPtrMessage* m = mix_queue->pop();
    31.     if (!m) {
    32.         return err;
    33.     }
    34.   
    35.     // consume the monotonically increase message.
    36.     if (m->is_audio()) {
    37.         err = on_audio_imp(m);
    38.     } else {
    39.         err = on_video_imp(m);
    40.     }
    41.     srs_freep(m);
    42.   
    43.     return err;
    44. }
    45.

    2.SrsLiveSource::on_audio_imp处理SrsSharedPtrMessage主要过程包括：

    1. 遍历对应source拉流客户端，将msg发送给每个客户端。
    2. 如果不是sequence header，加入gop缓存中，gop会缓存最后一个gop的packet。
    1. srs_error_t SrsLiveSource::on_audio_imp(SrsSharedPtrMessage* msg)
    2. {
    3.     srs_error_t err = srs_success;
    4.   
    5.     bool is_aac_sequence_header = SrsFlvAudio::sh(msg->payload, msg->size);
    6.     bool is_sequence_header = is_aac_sequence_header;
    7.   
    8.     // whether consumer should drop for the duplicated sequence header.
    9.     bool drop_for_reduce = false;
    10.     if (is_sequence_header && meta->previous_ash() && _srs_config->get_reduce_sequence_header(req->vhost)) {
    11.         if (meta->previous_ash()->size == msg->size) {
    12.             drop_for_reduce = srs_bytes_equals(meta->previous_ash()->payload, msg->payload, msg->size);
    13.             srs_warn("drop for reduce sh audio, size=%d", msg->size);
    14.         }
    15.     }
    16.   
    17.     // Copy to hub to all utilities.
    18.     if ((err = hub->on_audio(msg)) != srs_success) {
    19.         return srs_error_wrap(err, "consume audio");
    20.     }
    21.  
    22.     // For bridger to consume the message.
    23.     if (bridger_ && (err = bridger_->on_audio(msg)) != srs_success) {
    24.         return srs_error_wrap(err, "bridger consume audio");
    25.     }
    26.  
    27.     // copy to all consumer 遍历对应source拉流客户端，将msg发送给每个客户端
    28.     if (!drop_for_reduce) {
    29.         for (int i = 0; i < (int)consumers.size(); i++) {
    30.             SrsLiveConsumer* consumer = consumers.at(i);
    31.             if ((err = consumer->enqueue(msg, atc, jitter_algorithm)) != srs_success) {
    32.                 return srs_error_wrap(err, "consume message");
    33.             }
    34.         }
    35.     }
    36.   
    37.     // cache the sequence header of aac, or first packet of mp3.
    38.     // for example, the mp3 is used for hls to write the "right" audio codec.
    39.     // TODO: FIXME: to refine the stream info system.
    40.     if (is_aac_sequence_header || !meta->ash()) {
    41.         if ((err = meta->update_ash(msg)) != srs_success) {
    42.             return srs_error_wrap(err, "meta consume audio");
    43.         }
    44.     }
    45.   
    46.     // when sequence header, donot push to gop cache and adjust the timestamp.
    47.     if (is_sequence_header) {
    48.         return err;
    49.     }
    50.   
    51.     // cache the last gop packets
    52.     if ((err = gop_cache->cache(msg)) != srs_success) {
    53.         return srs_error_wrap(err, "gop cache consume audio");
    54.     }
    55.  
    56.     // if atc, update the sequence header to abs time.
    57.     if (atc) {
    58.         if (meta->ash()) {
    59.             meta->ash()->timestamp = msg->timestamp;
    60.         }
    61.         if (meta->data()) {
    62.             meta->data()->timestamp = msg->timestamp;
    63.         }
    64.     }
    65.   
    66.     return err;
    67. }
    68.

    3.每个拉流客户端是一个SrsLiveConsumer，会将msg加入SrsMessageQueue对象中，实际保存的是SrsMessageQueue对象中的SrsFastVector对象中的SrsSharedPtrMessage数组。跟video存储逻辑一样。

    3. 处理RTMP推流onMetaData message

    对于onMetaData message，SrsRtmpConn::process_publish_message会先解码成SrsOnMetaDataPacket，然后和onMetaData message一起作为参数传入SrsLiveSource::on_meta_data。主要的功能有：

    更新onMetaData message。即更新SrsMetaCache的meta（SrsSharedPtrMessage）

    遍历对应source拉流客户端，将共享meta发送给每个客户端。

    1. srs_error_t SrsLiveSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata)
    2. {
    3.     srs_error_t err = srs_success;
    4.   
    5.     // if allow atc_auto and bravo-atc detected, open atc for vhost.
    6.     SrsAmf0Any* prop = NULL;
    7.     atc = _srs_config->get_atc(req->vhost);
    8.     if (_srs_config->get_atc_auto(req->vhost)) {
    9.         if ((prop = metadata->metadata->get_property("bravo_atc")) != NULL) {
    10.             if (prop->is_string() && prop->to_str() == "true") {
    11.                 atc = true;
    12.             }
    13.         }
    14.     }
    15.   
    16.     // Update the meta cache. 如果需要更新MetaData，会更新共享SrsSharedPtrMessage* meta
    17.     bool updated = false;
    18.     if ((err = meta->update_data(&msg->header, metadata, updated)) != srs_success) {
    19.         return srs_error_wrap(err, "update metadata");
    20.     }
    21.     if (!updated) {
    22.         return err;
    23.     }
    24.   
    25.     // when already got metadata, drop when reduce sequence header.
    26.     bool drop_for_reduce = false;
    27.     if (meta->data() && _srs_config->get_reduce_sequence_header(req->vhost)) {
    28.         drop_for_reduce = true;
    29.         srs_warn("drop for reduce sh metadata, size=%d", msg->size);
    30.     }
    31.   
    32.     // copy to all consumer 遍历对应source拉流客户端，将共享meta发送给每个客户端。
    33.     if (!drop_for_reduce) {
    34.         std::vector<SrsLiveConsumer*>::iterator it;
    35.         for (it = consumers.begin(); it != consumers.end(); ++it) {
    36.             SrsLiveConsumer* consumer = *it;
    37.             if ((err = consumer->enqueue(meta->data(), atc, jitter_algorithm)) != srs_success) {
    38.                 return srs_error_wrap(err, "consume metadata");
    39.             }
    40.         }
    41.     }
    42.   
    43.     // Copy to hub to all utilities.
    44.     return hub->on_meta_data(meta->data(), metadata);
    45. }
    46.

    2.每个拉流客户端是一个SrsLiveConsumer，会将msg加入SrsMessageQueue对象中，实际保存的是SrsMessageQueue对象中的SrsFastVector对象中的SrsSharedPtrMessage数组。跟video和audio的存储逻辑一样。

  • 相关阅读:
    某今日头条_signature解析
    解析Collection接口中的常用的被实现子类重写的方法
    Dubbo—Admin 整体架构与安装步骤
    【我的机器学习之旅】打造个性化手写汉字识别系统
    Docker指定容器使用内存
    RabbitMQ有什么优缺点
    thinkphp使用dompdf导出pdf（html转pdf）
    软件评测师之逻辑运算
    HTML网页期末作业：基于Html+Css+javascript的网页制作（化妆品企业官网设计20页）
    使用 DFS 轻松求解数独难题（C++ 的一个简单实现）
  • 原文地址：https://blog.csdn.net/irainsa/article/details/127941373