AI: Add workflow utest for rtc publisher

trunk/configure

@@ -384,8 +384,8 @@ if [[ $SRS_UTEST == YES ]]; then
         "srs_utest_coworkers" "srs_utest_pithy_print" "srs_utest_kernel3" "srs_utest_protocol4"
         "srs_utest_protocol3" "srs_utest_app" "srs_utest_app2" "srs_utest_app3" "srs_utest_app4"
         "srs_utest_app5" "srs_utest_app6" "srs_utest_app7" "srs_utest_app8" "srs_utest_app9"
-        "srs_utest_app10" "srs_utest_app11" "srs_utest_app15" "srs_utest_app16"
+        "srs_utest_app10" "srs_utest_app11" "srs_utest_app15" "srs_utest_app16" "srs_utest_app17" 
-        "srs_utest_app17" "srs_utest_mock" "srs_utest_rtc_playstream")
+        "srs_utest_mock" "srs_utest_rtc_playstream" "srs_utest_rtc_publishstream")
     # Always include SRT utest
     MODULE_FILES+=("srs_utest_srt")
     if [[ $SRS_GB28181 == YES ]]; then

trunk/src/app/srs_app_rtc_conn.cpp

@@ -346,6 +346,15 @@ void SrsRtcPliWorker::request_keyframe(uint32_t ssrc, SrsContextId cid)
     wait_->signal();
 }
 
+void SrsRtcPliWorker::stop()
+{
+    wait_->signal();
+    
+    if (trd_) {
+        trd_->stop();
+    }
+}
+
 srs_error_t SrsRtcPliWorker::cycle()
 {
     srs_error_t err = srs_success;
@@ -963,19 +972,37 @@ ISrsRtcRtcpSender::~ISrsRtcRtcpSender()
 {
 }
 
+ISrsRtcPublishRtcpTimer::ISrsRtcPublishRtcpTimer()
+{
+}
+
+ISrsRtcPublishRtcpTimer::~ISrsRtcPublishRtcpTimer()
+{
+}
+
 SrsRtcPublishRtcpTimer::SrsRtcPublishRtcpTimer(ISrsRtcRtcpSender *sender) : sender_(sender)
 {
     lock_ = srs_mutex_new();
-    _srs_shared_timer->timer1s()->subscribe(this);
+
+    shared_timer_ = _srs_shared_timer;
 }
 
 SrsRtcPublishRtcpTimer::~SrsRtcPublishRtcpTimer()
 {
-    if (true) {
+    if (shared_timer_) {
         SrsLocker(&lock_);
-        _srs_shared_timer->timer1s()->unsubscribe(this);
+        shared_timer_->timer1s()->unsubscribe(this);
     }
     srs_mutex_destroy(lock_);
+
+    shared_timer_ = NULL;
+}
+
+srs_error_t SrsRtcPublishRtcpTimer::initialize()
+{
+    shared_timer_->timer1s()->subscribe(this);
+
+    return srs_success;
 }
 
 srs_error_t SrsRtcPublishRtcpTimer::on_timer(srs_utime_t interval)
@@ -1010,23 +1037,39 @@ srs_error_t SrsRtcPublishRtcpTimer::on_timer(srs_utime_t interval)
     return err;
 }
 
+ISrsRtcPublishTwccTimer::ISrsRtcPublishTwccTimer()
+{
+}
+
+ISrsRtcPublishTwccTimer::~ISrsRtcPublishTwccTimer()
+{
+}
+
 SrsRtcPublishTwccTimer::SrsRtcPublishTwccTimer(ISrsRtcRtcpSender *sender) : sender_(sender)
 {
     lock_ = srs_mutex_new();
-    _srs_shared_timer->timer100ms()->subscribe(this);
 
     circuit_breaker_ = _srs_circuit_breaker;
+    shared_timer_ = _srs_shared_timer;
 }
 
 SrsRtcPublishTwccTimer::~SrsRtcPublishTwccTimer()
 {
-    if (true) {
+    if (shared_timer_) {
         SrsLocker(&lock_);
-        _srs_shared_timer->timer100ms()->unsubscribe(this);
+        shared_timer_->timer100ms()->unsubscribe(this);
     }
     srs_mutex_destroy(lock_);
 
     circuit_breaker_ = NULL;
+    shared_timer_ = NULL;
+}
+
+srs_error_t SrsRtcPublishTwccTimer::initialize()
+{
+    shared_timer_->timer100ms()->subscribe(this);
+
+    return srs_success;
 }
 
 srs_error_t SrsRtcPublishTwccTimer::on_timer(srs_utime_t interval)
@@ -1152,6 +1195,7 @@ SrsRtcPublishStream::SrsRtcPublishStream(ISrsExecRtcAsyncTask *exec, ISrsExpire
 
     timer_rtcp_ = new SrsRtcPublishRtcpTimer(this);
     timer_twcc_ = new SrsRtcPublishTwccTimer(this);
+    rtcp_twcc_ = new SrsRtcpTWCC();
 
     stat_ = _srs_stat;
     config_ = _srs_config;
@@ -1169,6 +1213,7 @@ SrsRtcPublishStream::~SrsRtcPublishStream()
 
     srs_freep(timer_rtcp_);
     srs_freep(timer_twcc_);
+    srs_freep(rtcp_twcc_);
 
     source_->set_publish_stream(NULL);
     source_->on_unpublish();
@@ -1192,7 +1237,9 @@ SrsRtcPublishStream::~SrsRtcPublishStream()
 
     // update the statistic when client coveried.
     // TODO: FIXME: Should finger out the err.
-    stat_->on_disconnect(cid_.c_str(), srs_success);
+    if (stat_) {
+        stat_->on_disconnect(cid_.c_str(), srs_success);
+    }
 
     // Optional but just to make it clear.
     stat_ = NULL;
@@ -1209,6 +1256,14 @@ srs_error_t SrsRtcPublishStream::initialize(ISrsRequest *r, SrsRtcSourceDescript
 
     req_ = r->copy();
 
+    if ((err = timer_rtcp_->initialize()) != srs_success) {
+        return srs_error_wrap(err, "initialize timer rtcp");
+    }
+
+    if ((err = timer_twcc_->initialize()) != srs_success) {
+        return srs_error_wrap(err, "initialize timer twcc");
+    }
+
     // We must do stat the client before hooks, because hooks depends on it.
     if ((err = stat_->on_client(cid_.c_str(), req_, expire_, SrsRtcConnPublish)) != srs_success) {
         return srs_error_wrap(err, "rtc: stat client");
@@ -1236,7 +1291,7 @@ srs_error_t SrsRtcPublishStream::initialize(ISrsRequest *r, SrsRtcSourceDescript
     if (twcc_id > 0) {
         twcc_id_ = twcc_id;
         extension_types_.register_by_uri(twcc_id_, kTWCCExt);
-        rtcp_twcc_.set_media_ssrc(media_ssrc);
+        rtcp_twcc_->set_media_ssrc(media_ssrc);
     }
 
     nack_enabled_ = config_->get_rtc_nack_enabled(req_->vhost_);
@@ -1346,6 +1401,11 @@ srs_error_t SrsRtcPublishStream::start()
     return err;
 }
 
+void SrsRtcPublishStream::stop()
+{
+    pli_worker_->stop();
+}
+
 void SrsRtcPublishStream::set_all_tracks_status(bool status)
 {
     std::ostringstream merged_log;
@@ -1437,7 +1497,7 @@ srs_error_t SrsRtcPublishStream::on_twcc(uint16_t sn)
     srs_error_t err = srs_success;
 
     srs_utime_t now = srs_time_now_cached();
-    err = rtcp_twcc_.recv_packet(sn, now);
+    err = rtcp_twcc_->recv_packet(sn, now);
 
     return err;
 }
@@ -1621,21 +1681,21 @@ srs_error_t SrsRtcPublishStream::send_periodic_twcc()
     }
     last_time_send_twcc_ = srs_time_now_cached();
 
-    if (!rtcp_twcc_.need_feedback()) {
+    if (!rtcp_twcc_->need_feedback()) {
         return err;
     }
 
     ++_srs_pps_srtcps->sugar_;
 
     // limit the max count=1024 to avoid dead loop.
-    for (int i = 0; i < 1024 && rtcp_twcc_.need_feedback(); ++i) {
+    for (int i = 0; i < 1024 && rtcp_twcc_->need_feedback(); ++i) {
         char pkt[kMaxUDPDataSize];
         SrsUniquePtr<SrsBuffer> buffer(new SrsBuffer(pkt, sizeof(pkt)));
 
-        rtcp_twcc_.set_feedback_count(twcc_fb_count_);
+        rtcp_twcc_->set_feedback_count(twcc_fb_count_);
         twcc_fb_count_++;
 
-        if ((err = rtcp_twcc_.encode(buffer.get())) != srs_success) {
+        if ((err = rtcp_twcc_->encode(buffer.get())) != srs_success) {
             return srs_error_wrap(err, "encode, count=%u", twcc_fb_count_);
         }
 

trunk/src/app/srs_app_rtc_conn.hpp

@@ -204,6 +204,7 @@ public:
 public:
     virtual srs_error_t start() = 0;
     virtual void request_keyframe(uint32_t ssrc, SrsContextId cid) = 0;
+    virtual void stop() = 0;
 };
 
 // A worker coroutine to request the PLI.
@@ -228,6 +229,7 @@ public:
 public:
     virtual srs_error_t start();
     virtual void request_keyframe(uint32_t ssrc, SrsContextId cid);
+    virtual void stop();
     // interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
@@ -369,9 +371,24 @@ public:
     virtual srs_error_t send_periodic_twcc() = 0;
 };
 
-// A fast timer for publish stream, for RTCP feedback.
+// The RTC publish RTCP timer interface.
-class SrsRtcPublishRtcpTimer : public ISrsFastTimerHandler
+class ISrsRtcPublishRtcpTimer: public ISrsFastTimerHandler
 {
+public:
+    ISrsRtcPublishRtcpTimer();
+    virtual ~ISrsRtcPublishRtcpTimer();
+
+public:
+    virtual srs_error_t initialize() = 0;
+};
+
+// A fast timer for publish stream, for RTCP feedback.
+class SrsRtcPublishRtcpTimer : public ISrsRtcPublishRtcpTimer
+{
+// clang-format off
+SRS_DECLARE_PRIVATE: // clang-format on
+    ISrsSharedTimer *shared_timer_;
+
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on
     ISrsRtcRtcpSender *sender_;
@@ -380,17 +397,33 @@ SRS_DECLARE_PRIVATE: // clang-format on
 public:
     SrsRtcPublishRtcpTimer(ISrsRtcRtcpSender *sender);
     virtual ~SrsRtcPublishRtcpTimer();
+
+public:
+    virtual srs_error_t initialize();
+
     // interface ISrsFastTimerHandler
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on
     srs_error_t on_timer(srs_utime_t interval);
 };
 
+// The RTC publish TWCC timer interface.
+class ISrsRtcPublishTwccTimer: public ISrsFastTimerHandler
+{
+public:
+    ISrsRtcPublishTwccTimer();
+    virtual ~ISrsRtcPublishTwccTimer();
+
+public:
+    virtual srs_error_t initialize() = 0;
+};
+
 // A fast timer for publish stream, for TWCC feedback.
-class SrsRtcPublishTwccTimer : public ISrsFastTimerHandler
+class SrsRtcPublishTwccTimer : public ISrsRtcPublishTwccTimer
 {
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on
+    ISrsSharedTimer *shared_timer_;
     ISrsCircuitBreaker *circuit_breaker_;
 
 // clang-format off
@@ -401,6 +434,10 @@ SRS_DECLARE_PRIVATE: // clang-format on
 public:
     SrsRtcPublishTwccTimer(ISrsRtcRtcpSender *sender);
     virtual ~SrsRtcPublishTwccTimer();
+
+public:
+    virtual srs_error_t initialize();
+
     // interface ISrsFastTimerHandler
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on
@@ -451,14 +488,14 @@ SRS_DECLARE_PRIVATE: // clang-format on
 SRS_DECLARE_PRIVATE: // clang-format on
     friend class SrsRtcPublishRtcpTimer;
     friend class SrsRtcPublishTwccTimer;
-    SrsRtcPublishRtcpTimer *timer_rtcp_;
+    ISrsRtcPublishRtcpTimer *timer_rtcp_;
-    SrsRtcPublishTwccTimer *timer_twcc_;
+    ISrsRtcPublishTwccTimer *timer_twcc_;
 
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on
     SrsContextId cid_;
     uint64_t nn_audio_frames_;
-    SrsRtcPliWorker *pli_worker_;
+    ISrsRtcPliWorker *pli_worker_;
     SrsErrorPithyPrint *twcc_epp_;
 
 // clang-format off
@@ -492,7 +529,7 @@ SRS_DECLARE_PRIVATE: // clang-format on
 SRS_DECLARE_PRIVATE: // clang-format on
     int twcc_id_;
     uint8_t twcc_fb_count_;
-    SrsRtcpTWCC rtcp_twcc_;
+    SrsRtcpTWCC *rtcp_twcc_;
     SrsRtpExtensionTypes extension_types_;
     bool is_sender_started_;
     srs_utime_t last_time_send_twcc_;
@@ -504,6 +541,7 @@ public:
 public:
     srs_error_t initialize(ISrsRequest *req, SrsRtcSourceDescription *stream_desc);
     srs_error_t start();
+    void stop();
     // Directly set the status of track, generally for init to set the default value.
     void set_all_tracks_status(bool status);
     virtual const SrsContextId &context_id();

trunk/src/utest/srs_utest_app6.cpp

@@ -3439,88 +3439,6 @@ void MockRtcRtcpSender::reset()
     send_periodic_twcc_count_ = 0;
 }
 
-// Mock RTC packet receiver implementation
-MockRtcPacketReceiver::MockRtcPacketReceiver()
-{
-    send_rtcp_rr_error_ = srs_success;
-    send_rtcp_xr_rrtr_error_ = srs_success;
-    send_rtcp_error_ = srs_success;
-    send_rtcp_fb_pli_error_ = srs_success;
-    send_rtcp_rr_count_ = 0;
-    send_rtcp_xr_rrtr_count_ = 0;
-    send_rtcp_count_ = 0;
-    send_rtcp_fb_pli_count_ = 0;
-    check_send_nacks_count_ = 0;
-}
-
-MockRtcPacketReceiver::~MockRtcPacketReceiver()
-{
-}
-
-srs_error_t MockRtcPacketReceiver::send_rtcp_rr(uint32_t ssrc, SrsRtpRingBuffer *rtp_queue, const uint64_t &last_send_systime, const SrsNtp &last_send_ntp)
-{
-    send_rtcp_rr_count_++;
-    return send_rtcp_rr_error_;
-}
-
-srs_error_t MockRtcPacketReceiver::send_rtcp_xr_rrtr(uint32_t ssrc)
-{
-    send_rtcp_xr_rrtr_count_++;
-    return send_rtcp_xr_rrtr_error_;
-}
-
-void MockRtcPacketReceiver::check_send_nacks(SrsRtpNackForReceiver *nack, uint32_t ssrc, uint32_t &sent_nacks, uint32_t &timeout_nacks)
-{
-    check_send_nacks_count_++;
-    sent_nacks = 0;
-    timeout_nacks = 0;
-}
-
-srs_error_t MockRtcPacketReceiver::send_rtcp(char *data, int nb_data)
-{
-    send_rtcp_count_++;
-    return send_rtcp_error_;
-}
-
-srs_error_t MockRtcPacketReceiver::send_rtcp_fb_pli(uint32_t ssrc, const SrsContextId &cid_of_subscriber)
-{
-    send_rtcp_fb_pli_count_++;
-    return send_rtcp_fb_pli_error_;
-}
-
-void MockRtcPacketReceiver::set_send_rtcp_rr_error(srs_error_t err)
-{
-    send_rtcp_rr_error_ = err;
-}
-
-void MockRtcPacketReceiver::set_send_rtcp_xr_rrtr_error(srs_error_t err)
-{
-    send_rtcp_xr_rrtr_error_ = err;
-}
-
-void MockRtcPacketReceiver::set_send_rtcp_error(srs_error_t err)
-{
-    send_rtcp_error_ = err;
-}
-
-void MockRtcPacketReceiver::set_send_rtcp_fb_pli_error(srs_error_t err)
-{
-    send_rtcp_fb_pli_error_ = err;
-}
-
-void MockRtcPacketReceiver::reset()
-{
-    send_rtcp_rr_error_ = srs_success;
-    send_rtcp_xr_rrtr_error_ = srs_success;
-    send_rtcp_error_ = srs_success;
-    send_rtcp_fb_pli_error_ = srs_success;
-    send_rtcp_rr_count_ = 0;
-    send_rtcp_xr_rrtr_count_ = 0;
-    send_rtcp_count_ = 0;
-    send_rtcp_fb_pli_count_ = 0;
-    check_send_nacks_count_ = 0;
-}
-
 VOID TEST(SrsRtcPublishRtcpTimerTest, OnTimer)
 {
     srs_error_t err;

trunk/src/utest/srs_utest_app6.hpp

@@ -438,39 +438,6 @@ public:
     void reset();
 };
 
-// Mock RTC packet receiver for testing SrsRtcPublishStream
-class MockRtcPacketReceiver : public ISrsRtcPacketReceiver
-{
-public:
-    srs_error_t send_rtcp_rr_error_;
-    srs_error_t send_rtcp_xr_rrtr_error_;
-    srs_error_t send_rtcp_error_;
-    srs_error_t send_rtcp_fb_pli_error_;
-    int send_rtcp_rr_count_;
-    int send_rtcp_xr_rrtr_count_;
-    int send_rtcp_count_;
-    int send_rtcp_fb_pli_count_;
-    int check_send_nacks_count_;
-
-public:
-    MockRtcPacketReceiver();
-    virtual ~MockRtcPacketReceiver();
-
-public:
-    virtual srs_error_t send_rtcp_rr(uint32_t ssrc, SrsRtpRingBuffer *rtp_queue, const uint64_t &last_send_systime, const SrsNtp &last_send_ntp);
-    virtual srs_error_t send_rtcp_xr_rrtr(uint32_t ssrc);
-    virtual void check_send_nacks(SrsRtpNackForReceiver *nack, uint32_t ssrc, uint32_t &sent_nacks, uint32_t &timeout_nacks);
-    virtual srs_error_t send_rtcp(char *data, int nb_data);
-    virtual srs_error_t send_rtcp_fb_pli(uint32_t ssrc, const SrsContextId &cid_of_subscriber);
-
-public:
-    void set_send_rtcp_rr_error(srs_error_t err);
-    void set_send_rtcp_xr_rrtr_error(srs_error_t err);
-    void set_send_rtcp_error(srs_error_t err);
-    void set_send_rtcp_fb_pli_error(srs_error_t err);
-    void reset();
-};
-
 // Mock expire for testing SrsRtcPublishStream
 class MockRtcExpire : public ISrsExpire
 {

trunk/src/utest/srs_utest_mock.cpp

@@ -42,6 +42,19 @@ std::map<uint32_t, SrsRtcTrackDescription *> MockRtcTrackDescriptionFactory::cre
     return sub_relations;
 }
 
+SrsRtcSourceDescription *MockRtcTrackDescriptionFactory::create_stream_description()
+{
+    SrsRtcSourceDescription *stream_desc = new SrsRtcSourceDescription();
+
+    // Create audio track
+    stream_desc->audio_track_desc_ = create_audio_track(audio_ssrc_, "audio-track-1", "0");
+
+    // Create video track
+    stream_desc->video_track_descs_.push_back(create_video_track(video_ssrc_, "video-track-1", "1"));
+
+    return stream_desc;
+}
+
 SrsRtcTrackDescription *MockRtcTrackDescriptionFactory::create_audio_track(uint32_t ssrc, std::string id, std::string mid)
 {
     SrsRtcTrackDescription *audio_desc = new SrsRtcTrackDescription();
@@ -147,7 +160,8 @@ srs_error_t MockRtcSourceManager::fetch_or_create(ISrsRequest *r, SrsSharedPtr<S
         return srs_error_copy(fetch_or_create_error_);
     }
     pps = mock_source_;
-    return srs_success;
+
+    return mock_source_->initialize(r);
 }
 
 SrsSharedPtr<SrsRtcSource> MockRtcSourceManager::fetch(ISrsRequest *r)
@@ -805,3 +819,85 @@ void MockAppConfig::set_keep_api_domain(bool enabled)
 {
     keep_api_domain_ = enabled;
 }
+
+// Mock RTC packet receiver implementation
+MockRtcPacketReceiver::MockRtcPacketReceiver()
+{
+    send_rtcp_rr_error_ = srs_success;
+    send_rtcp_xr_rrtr_error_ = srs_success;
+    send_rtcp_error_ = srs_success;
+    send_rtcp_fb_pli_error_ = srs_success;
+    send_rtcp_rr_count_ = 0;
+    send_rtcp_xr_rrtr_count_ = 0;
+    send_rtcp_count_ = 0;
+    send_rtcp_fb_pli_count_ = 0;
+    check_send_nacks_count_ = 0;
+}
+
+MockRtcPacketReceiver::~MockRtcPacketReceiver()
+{
+}
+
+srs_error_t MockRtcPacketReceiver::send_rtcp_rr(uint32_t ssrc, SrsRtpRingBuffer *rtp_queue, const uint64_t &last_send_systime, const SrsNtp &last_send_ntp)
+{
+    send_rtcp_rr_count_++;
+    return send_rtcp_rr_error_;
+}
+
+srs_error_t MockRtcPacketReceiver::send_rtcp_xr_rrtr(uint32_t ssrc)
+{
+    send_rtcp_xr_rrtr_count_++;
+    return send_rtcp_xr_rrtr_error_;
+}
+
+void MockRtcPacketReceiver::check_send_nacks(SrsRtpNackForReceiver *nack, uint32_t ssrc, uint32_t &sent_nacks, uint32_t &timeout_nacks)
+{
+    check_send_nacks_count_++;
+    sent_nacks = 0;
+    timeout_nacks = 0;
+}
+
+srs_error_t MockRtcPacketReceiver::send_rtcp(char *data, int nb_data)
+{
+    send_rtcp_count_++;
+    return send_rtcp_error_;
+}
+
+srs_error_t MockRtcPacketReceiver::send_rtcp_fb_pli(uint32_t ssrc, const SrsContextId &cid_of_subscriber)
+{
+    send_rtcp_fb_pli_count_++;
+    return send_rtcp_fb_pli_error_;
+}
+
+void MockRtcPacketReceiver::set_send_rtcp_rr_error(srs_error_t err)
+{
+    send_rtcp_rr_error_ = err;
+}
+
+void MockRtcPacketReceiver::set_send_rtcp_xr_rrtr_error(srs_error_t err)
+{
+    send_rtcp_xr_rrtr_error_ = err;
+}
+
+void MockRtcPacketReceiver::set_send_rtcp_error(srs_error_t err)
+{
+    send_rtcp_error_ = err;
+}
+
+void MockRtcPacketReceiver::set_send_rtcp_fb_pli_error(srs_error_t err)
+{
+    send_rtcp_fb_pli_error_ = err;
+}
+
+void MockRtcPacketReceiver::reset()
+{
+    send_rtcp_rr_error_ = srs_success;
+    send_rtcp_xr_rrtr_error_ = srs_success;
+    send_rtcp_error_ = srs_success;
+    send_rtcp_fb_pli_error_ = srs_success;
+    send_rtcp_rr_count_ = 0;
+    send_rtcp_xr_rrtr_count_ = 0;
+    send_rtcp_count_ = 0;
+    send_rtcp_fb_pli_count_ = 0;
+    check_send_nacks_count_ = 0;
+}

trunk/src/utest/srs_utest_mock.hpp

@@ -38,9 +38,12 @@ public:
     uint32_t screen_ssrc_;
 
 public:
-    // Create a map of track descriptions with audio and video tracks
+    // Create a map of track descriptions with audio and video tracks (for play stream)
     std::map<uint32_t, SrsRtcTrackDescription *> create_audio_video_tracks();
 
+    // Create a stream description with audio and video tracks (for publish stream)
+    SrsRtcSourceDescription *create_stream_description();
+
     // Create a single audio track description
     SrsRtcTrackDescription *create_audio_track(uint32_t ssrc, std::string id, std::string mid);
 
@@ -467,4 +470,37 @@ public:
     void set_keep_api_domain(bool enabled);
 };
 
+// Mock RTC packet receiver for testing SrsRtcPublishStream
+class MockRtcPacketReceiver : public ISrsRtcPacketReceiver
+{
+public:
+    srs_error_t send_rtcp_rr_error_;
+    srs_error_t send_rtcp_xr_rrtr_error_;
+    srs_error_t send_rtcp_error_;
+    srs_error_t send_rtcp_fb_pli_error_;
+    int send_rtcp_rr_count_;
+    int send_rtcp_xr_rrtr_count_;
+    int send_rtcp_count_;
+    int send_rtcp_fb_pli_count_;
+    int check_send_nacks_count_;
+
+public:
+    MockRtcPacketReceiver();
+    virtual ~MockRtcPacketReceiver();
+
+public:
+    virtual srs_error_t send_rtcp_rr(uint32_t ssrc, SrsRtpRingBuffer *rtp_queue, const uint64_t &last_send_systime, const SrsNtp &last_send_ntp);
+    virtual srs_error_t send_rtcp_xr_rrtr(uint32_t ssrc);
+    virtual void check_send_nacks(SrsRtpNackForReceiver *nack, uint32_t ssrc, uint32_t &sent_nacks, uint32_t &timeout_nacks);
+    virtual srs_error_t send_rtcp(char *data, int nb_data);
+    virtual srs_error_t send_rtcp_fb_pli(uint32_t ssrc, const SrsContextId &cid_of_subscriber);
+
+public:
+    void set_send_rtcp_rr_error(srs_error_t err);
+    void set_send_rtcp_xr_rrtr_error(srs_error_t err);
+    void set_send_rtcp_error(srs_error_t err);
+    void set_send_rtcp_fb_pli_error(srs_error_t err);
+    void reset();
+};
+
 #endif

trunk/src/utest/srs_utest_rtc_publishstream.cpp

@@ -0,0 +1,105 @@
+/**
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2013-2025 Winlin
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to
+ * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ * the Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+#include <srs_utest_rtc_publishstream.hpp>
+
+#include <srs_kernel_error.hpp>
+#include <srs_app_rtc_conn.hpp>
+#include <srs_app_rtc_source.hpp>
+#include <srs_utest_mock.hpp>
+#include <srs_utest_app6.hpp>
+
+// This test is used to verify the basic workflow of the RTC publish stream.
+// It's finished with the help of AI, but each step is manually designed
+// and verified. So this is not dominated by AI, but by humanbeing.
+VOID TEST(RtcPublishStreamTest, ManuallyVerifyBasicWorkflow)
+{
+    srs_error_t err;
+
+    // Create mock objects for dependencies
+    MockAppConfig mock_config;
+    MockRtcSourceManager mock_rtc_sources;
+    MockRtcStatistic mock_stat;
+    MockRtcAsyncCallRequest mock_request("test.vhost", "live", "stream1");
+    MockRtcAsyncTaskExecutor mock_exec;
+    MockExpire mock_expire;
+    MockRtcPacketReceiver mock_receiver;
+    MockRtcTrackDescriptionFactory track_factory;
+    SrsContextId cid;
+    cid.set_value("test-publish-stream-cid");
+
+    // Create RTC publish stream - use real pli_worker_
+    SrsUniquePtr<SrsRtcPublishStream> publish_stream(new SrsRtcPublishStream(&mock_exec, &mock_expire, &mock_receiver, cid));
+
+    // Mock the publish stream object
+    if (true) {
+        // Inject mock dependencies
+        publish_stream->config_ = &mock_config;
+        publish_stream->rtc_sources_ = &mock_rtc_sources;
+        publish_stream->stat_ = &mock_stat;
+    }
+
+    // Create stream description with audio and video tracks
+    if (true) {
+        SrsUniquePtr<SrsRtcSourceDescription> stream_desc(track_factory.create_stream_description());
+
+        // Initialize the publish stream (it will take ownership of track descriptions)
+        HELPER_EXPECT_SUCCESS(publish_stream->initialize(&mock_request, stream_desc.get()));
+
+        // Check the tracks, should be one audio track
+        EXPECT_EQ(publish_stream->audio_tracks_.size(), 1);
+        // Check the tracks, should be one video track
+        EXPECT_EQ(publish_stream->video_tracks_.size(), 1);
+
+        // Test: First call to start() should succeed
+        HELPER_EXPECT_SUCCESS(publish_stream->start());
+
+        // Verify is_sender_started_ flag is set
+        EXPECT_TRUE(publish_stream->is_sender_started_);
+
+        // Wait for coroutine to start. Normally it should be ready and stopped at wait 
+        // for PLI requests.
+        srs_usleep(1 * SRS_UTIME_MILLISECONDS);
+    }
+
+    // Request a PLI about the video ssrc to the publisher.
+    if (true) {
+        uint32_t video_ssrc = track_factory.video_ssrc_;
+        publish_stream->request_keyframe(video_ssrc, cid);
+
+        // Wait for coroutine to process the request
+        srs_usleep(1 * SRS_UTIME_MILLISECONDS);
+
+        // Verify the PLI is sent out
+        EXPECT_EQ(mock_receiver.send_rtcp_fb_pli_count_, 1);
+    }
+
+    // Stop the publish stream
+    publish_stream->stop();
+
+    // Clean up - set injected fields to NULL to avoid double-free
+    publish_stream->config_ = NULL;
+    publish_stream->rtc_sources_ = NULL;
+    publish_stream->stat_ = NULL;
+}
+

trunk/src/utest/srs_utest_rtc_publishstream.hpp

@@ -0,0 +1,30 @@
+/**
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2013-2025 Winlin
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to
+ * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ * the Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef SRS_UTEST_RTC_PUBLISHSTREAM_HPP
+#define SRS_UTEST_RTC_PUBLISHSTREAM_HPP
+
+#include <srs_utest.hpp>
+
+#endif
+