diff --git a/trunk/src/app/srs_app_rtc_conn.cpp b/trunk/src/app/srs_app_rtc_conn.cpp index 9653b314d..eaf7266ce 100644 --- a/trunk/src/app/srs_app_rtc_conn.cpp +++ b/trunk/src/app/srs_app_rtc_conn.cpp @@ -697,13 +697,6 @@ srs_error_t SrsRtcPlayer::cycle() return srs_error_wrap(err, "rtc create consumer, source url=%s", req->get_stream_url().c_str()); } - // For RTC, we enable pass-timestamp mode, ignore the timestamp in queue, never depends on the duration, - // because RTC allows the audio and video has its own timebase, that is the audio timestamp and video timestamp - // maybe not monotonically increase. - // In this mode, we use mw_msgs to set the delay. We never shrink the consumer queue, instead, we dumps the - // messages and drop them if the shared sender queue is full. - consumer->enable_pass_timestamp(); - // TODO: FIXME: Dumps the SPS/PPS from gop cache, without other frames. if ((err = source->consumer_dumps(consumer)) != srs_success) { return srs_error_wrap(err, "dumps consumer, source url=%s", req->get_stream_url().c_str()); diff --git a/trunk/src/app/srs_app_rtc_source.cpp b/trunk/src/app/srs_app_rtc_source.cpp index eaaab472f..d94ba9da4 100644 --- a/trunk/src/app/srs_app_rtc_source.cpp +++ b/trunk/src/app/srs_app_rtc_source.cpp @@ -62,14 +62,6 @@ SrsRtcConsumer::~SrsRtcConsumer() #endif } -void SrsRtcConsumer::enable_pass_timestamp() -{ -} - -void SrsRtcConsumer::set_queue_size(srs_utime_t queue_size) -{ -} - void SrsRtcConsumer::update_source_id() { should_update_source_id = true; @@ -81,7 +73,7 @@ srs_error_t SrsRtcConsumer::enqueue(SrsSharedPtrMessage* shared_msg, bool atc, S SrsSharedPtrMessage* msg = shared_msg->copy(); - if ((err = queue->enqueue(msg, NULL, true)) != srs_success) { + if ((err = queue->enqueue(msg, NULL)) != srs_success) { return srs_error_wrap(err, "enqueue message"); } @@ -120,7 +112,7 @@ srs_error_t SrsRtcConsumer::dump_packets(SrsMessageArray* msgs, int& count) } // pump msgs from queue. - if ((err = queue->dump_packets(max, msgs->msgs, count, true)) != srs_success) { + if ((err = queue->dump_packets(max, msgs->msgs, count)) != srs_success) { return srs_error_wrap(err, "dump packets"); } @@ -249,9 +241,6 @@ srs_error_t SrsRtcSource::consumer_dumps(SrsRtcConsumer* consumer, bool ds, bool { srs_error_t err = srs_success; - srs_utime_t queue_size = _srs_config->get_queue_length(req->vhost); - consumer->set_queue_size(queue_size); - // Copy metadata and sequence header to consumer. // TODO: FIXME: Maybe should not do this for RTC? if ((err = meta->dumps(consumer, true, SrsRtmpJitterAlgorithmOFF, dm, dg)) != srs_success) { @@ -259,11 +248,7 @@ srs_error_t SrsRtcSource::consumer_dumps(SrsRtcConsumer* consumer, bool ds, bool } // print status. - if (dg) { - srs_trace("create consumer, queue_size=%.2f", queue_size); - } else { - srs_trace("create consumer, ignore gop cache"); - } + srs_trace("create consumer, no gop cache"); return err; } diff --git a/trunk/src/app/srs_app_rtc_source.hpp b/trunk/src/app/srs_app_rtc_source.hpp index 8cbaf115e..8c5bd1b48 100644 --- a/trunk/src/app/srs_app_rtc_source.hpp +++ b/trunk/src/app/srs_app_rtc_source.hpp @@ -62,12 +62,6 @@ public: SrsRtcConsumer(SrsRtcSource* s, SrsConnection* c); virtual ~SrsRtcConsumer(); public: - // Use pass timestamp mode. - // TODO: FIXME: Remove it. - void enable_pass_timestamp(); - // Set the size of queue. - // TODO: FIXME: Remove it. - virtual void set_queue_size(srs_utime_t queue_size); // when source id changed, notice client to print. virtual void update_source_id(); // Enqueue an shared ptr message. diff --git a/trunk/src/app/srs_app_source.cpp b/trunk/src/app/srs_app_source.cpp index 5c6b47fb2..caa58a140 100755 --- a/trunk/src/app/srs_app_source.cpp +++ b/trunk/src/app/srs_app_source.cpp @@ -266,17 +266,11 @@ void SrsMessageQueue::set_queue_size(srs_utime_t queue_size) max_queue_size = queue_size; } -srs_error_t SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg, bool* is_overflow, bool pass_timestamp) +srs_error_t SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg, bool* is_overflow) { srs_error_t err = srs_success; msgs.push_back(msg); - - // For RTC, we never care about the timestamp and duration, so we never shrink queue here, - // but we will drop messages in each consumer coroutine. - if (pass_timestamp) { - return err; - } if (msg->is_av()) { if (av_start_time == -1) { @@ -285,6 +279,10 @@ srs_error_t SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg, bool* is_overflow av_end_time = srs_utime_t(msg->timestamp * SRS_UTIME_MILLISECONDS); } + + if (max_queue_size <= 0) { + return err; + } while (av_end_time - av_start_time > max_queue_size) { // notice the caller queue already overflow and shrinked. @@ -298,7 +296,7 @@ srs_error_t SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg, bool* is_overflow return err; } -srs_error_t SrsMessageQueue::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count, bool pass_timestamp) +srs_error_t SrsMessageQueue::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count) { srs_error_t err = srs_success; @@ -313,13 +311,9 @@ srs_error_t SrsMessageQueue::dump_packets(int max_count, SrsSharedPtrMessage** p SrsSharedPtrMessage** omsgs = msgs.data(); memcpy(pmsgs, omsgs, count * sizeof(SrsSharedPtrMessage*)); - // For RTC, we enable pass_timestamp mode, which never care about the timestamp and duration, - // so we do not have to update the start time here. - if (!pass_timestamp) { - SrsSharedPtrMessage* last = omsgs[count - 1]; - av_start_time = srs_utime_t(last->timestamp * SRS_UTIME_MILLISECONDS); - } - + SrsSharedPtrMessage* last = omsgs[count - 1]; + av_start_time = srs_utime_t(last->timestamp * SRS_UTIME_MILLISECONDS); + if (count >= nb_msgs) { // the pmsgs is big enough and clear msgs at most time. msgs.clear(); @@ -446,8 +440,6 @@ SrsConsumer::SrsConsumer(SrsSource* s, SrsConnection* c) mw_duration = 0; mw_waiting = false; #endif - - pass_timestamp = false; } SrsConsumer::~SrsConsumer() @@ -461,11 +453,6 @@ SrsConsumer::~SrsConsumer() #endif } -void SrsConsumer::enable_pass_timestamp() -{ - pass_timestamp = true; -} - void SrsConsumer::set_queue_size(srs_utime_t queue_size) { queue->set_queue_size(queue_size); @@ -487,33 +474,19 @@ srs_error_t SrsConsumer::enqueue(SrsSharedPtrMessage* shared_msg, bool atc, SrsR SrsSharedPtrMessage* msg = shared_msg->copy(); - // For RTC, we enable pass_timestamp mode, which never correct or depends on monotonic increasing of - // timestamp. And in RTC, the audio and video timebase can be different, so we ignore time_jitter here. - if (!pass_timestamp && !atc) { + if (!atc) { if ((err = jitter->correct(msg, ag)) != srs_success) { return srs_error_wrap(err, "consume message"); } } - // Put message in queue, here we may enable pass_timestamp mode. - if ((err = queue->enqueue(msg, NULL, pass_timestamp)) != srs_success) { + if ((err = queue->enqueue(msg, NULL)) != srs_success) { return srs_error_wrap(err, "enqueue message"); } #ifdef SRS_PERF_QUEUE_COND_WAIT // fire the mw when msgs is enough. if (mw_waiting) { - // For RTC, we use pass_timestamp mode, we don't care about the timestamp in queue, - // so we only check the messages in queue. - if (pass_timestamp) { - if (queue->size() > mw_min_msgs) { - srs_cond_signal(mw_wait); - mw_waiting = false; - return err; - } - return err; - } - // For RTMP, we wait for messages and duration. srs_utime_t duration = queue->duration(); bool match_min_msgs = queue->size() > mw_min_msgs; @@ -564,7 +537,7 @@ srs_error_t SrsConsumer::dump_packets(SrsMessageArray* msgs, int& count) } // pump msgs from queue. - if ((err = queue->dump_packets(max, msgs->msgs, count, pass_timestamp)) != srs_success) { + if ((err = queue->dump_packets(max, msgs->msgs, count)) != srs_success) { return srs_error_wrap(err, "dump packets"); } diff --git a/trunk/src/app/srs_app_source.hpp b/trunk/src/app/srs_app_source.hpp index c41bbfc47..4e54666ad 100644 --- a/trunk/src/app/srs_app_source.hpp +++ b/trunk/src/app/srs_app_source.hpp @@ -150,13 +150,12 @@ public: // Enqueue the message, the timestamp always monotonically. // @param msg, the msg to enqueue, user never free it whatever the return code. // @param is_overflow, whether overflow and shrinked. NULL to ignore. - // @remark If pass_timestamp, we never shrink and never care about the timestamp or duration. - virtual srs_error_t enqueue(SrsSharedPtrMessage* msg, bool* is_overflow = NULL, bool pass_timestamp = false); + virtual srs_error_t enqueue(SrsSharedPtrMessage* msg, bool* is_overflow = NULL); // Get packets in consumer queue. // @pmsgs SrsSharedPtrMessage*[], used to store the msgs, user must alloc it. // @count the count in array, output param. // @max_count the max count to dequeue, must be positive. - virtual srs_error_t dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count, bool pass_timestamp = false); + virtual srs_error_t dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count); // Dumps packets to consumer, use specified args. // @remark the atc/tba/tbv/ag are same to SrsConsumer.enqueue(). virtual srs_error_t dump_packets(SrsConsumer* consumer, bool atc, SrsRtmpJitterAlgorithm ag); @@ -213,18 +212,10 @@ private: int mw_min_msgs; srs_utime_t mw_duration; #endif -private: - // TODO: FIXME: Move to RTC consumer. - // For RTC, we never use jitter to correct timestamp. - // But we should not change the atc or time_jitter for source or RTMP. - // @remark In this mode, we also never check the queue by timstamp, but only by count. - bool pass_timestamp; public: SrsConsumer(SrsSource* s, SrsConnection* c); virtual ~SrsConsumer(); public: - // Use pass timestamp mode. - void enable_pass_timestamp(); // Set the size of queue. virtual void set_queue_size(srs_utime_t queue_size); // when source id changed, notice client to print.