AI: Add workflow utest to verify rtc2rtmp audio async.

trunk/configure

@@ -381,12 +381,14 @@ if [[ $SRS_UTEST == YES ]]; then
         "srs_utest_manual_protocol" "srs_utest_manual_protocol2" "srs_utest_manual_kernel2" "srs_utest_manual_st" "srs_utest_manual_fmp4"
         "srs_utest_manual_source_lock" "srs_utest_manual_stream_token" "srs_utest_manual_rtc_recv_track" "srs_utest_manual_st2"
         "srs_utest_manual_hevc_structs" "srs_utest_manual_coworkers" "srs_utest_manual_pithy_print" "srs_utest_manual_protocol3"
-        "srs_utest_manual_app" "srs_utest_manual_mock" "srs_utest_workflow_rtc_playstream" "srs_utest_workflow_rtc_publishstream"
+        "srs_utest_manual_app" "srs_utest_manual_mock")
-        "srs_utest_workflow_rtc_conn" "srs_utest_workflow_rtmp_conn" "srs_utest_workflow_srt_conn" "srs_utest_workflow_http_conn")
+    MODULE_FILES+=("srs_utest_workflow_rtc_playstream" "srs_utest_workflow_rtc_publishstream"
+        "srs_utest_workflow_rtc_conn" "srs_utest_workflow_rtmp_conn" "srs_utest_workflow_srt_conn" "srs_utest_workflow_http_conn"
+        "srs_utest_workflow_forward" "srs_utest_workflow_rtc2rtmp")
     MODULE_FILES+=("srs_utest_ai01" "srs_utest_ai02" "srs_utest_ai03" "srs_utest_ai04" "srs_utest_ai05"
         "srs_utest_ai06" "srs_utest_ai07" "srs_utest_ai08" "srs_utest_ai09" "srs_utest_ai10" "srs_utest_ai11"
         "srs_utest_ai12" "srs_utest_ai13" "srs_utest_ai14" "srs_utest_ai15" "srs_utest_ai16" "srs_utest_ai17"
-        "srs_utest_ai18" "srs_utest_ai19" "srs_utest_ai20" "srs_utest_ai24" "srs_utest_workflow_forward")
+        "srs_utest_ai18" "srs_utest_ai19" "srs_utest_ai20" "srs_utest_ai24")
     if [[ $SRS_GB28181 == YES ]]; then
         MODULE_FILES+=("srs_utest_manual_gb28181" "srs_utest_ai23")
     fi

trunk/src/app/srs_app_rtc_source.cpp

@@ -1377,6 +1377,14 @@ srs_error_t SrsRtcRtpBuilder::consume_packets(vector<SrsRtpPacket *> &pkts)
     return err;
 }
 
+ISrsRtcFrameBuilderVideoPacketCache::ISrsRtcFrameBuilderVideoPacketCache()
+{
+}
+
+ISrsRtcFrameBuilderVideoPacketCache::~ISrsRtcFrameBuilderVideoPacketCache()
+{
+}
+
 SrsRtcFrameBuilderVideoPacketCache::SrsRtcFrameBuilderVideoPacketCache()
 {
     memset(cache_pkts_, 0, sizeof(cache_pkts_));
@@ -1536,7 +1544,15 @@ bool SrsRtcFrameBuilderVideoPacketCache::check_frame_complete(const uint16_t sta
     return nn_fu_start == nn_fu_end;
 }
 
-SrsRtcFrameBuilderVideoFrameDetector::SrsRtcFrameBuilderVideoFrameDetector(SrsRtcFrameBuilderVideoPacketCache *cache)
+ISrsRtcFrameBuilderVideoFrameDetector::ISrsRtcFrameBuilderVideoFrameDetector()
+{
+}
+
+ISrsRtcFrameBuilderVideoFrameDetector::~ISrsRtcFrameBuilderVideoFrameDetector()
+{
+}
+
+SrsRtcFrameBuilderVideoFrameDetector::SrsRtcFrameBuilderVideoFrameDetector(ISrsRtcFrameBuilderVideoPacketCache *cache)
 {
     video_cache_ = cache;
     header_sn_ = 0;
@@ -1640,6 +1656,14 @@ bool SrsRtcFrameBuilderVideoFrameDetector::is_lost_sn(uint16_t received)
     return lost_sn_ == received;
 }
 
+ISrsRtcFrameBuilderAudioPacketCache::ISrsRtcFrameBuilderAudioPacketCache()
+{
+}
+
+ISrsRtcFrameBuilderAudioPacketCache::~ISrsRtcFrameBuilderAudioPacketCache()
+{
+}
+
 SrsRtcFrameBuilderAudioPacketCache::SrsRtcFrameBuilderAudioPacketCache()
 {
     last_audio_seq_num_ = 0;
@@ -3115,7 +3139,7 @@ SrsRtcRecvTrack::SrsRtcRecvTrack(ISrsRtcPacketReceiver *receiver, SrsRtcTrackDes
     if (track_desc_->media_) {
         rate_ = static_cast<double>(track_desc_->media_->sample_) / 1000.0;
         srs_trace("RTC: Init %s track, ssrc=%u, rate from SDP=%.0f (RTP units per ms, will be updated after 2nd SR)",
-                 track_desc_->type_.c_str(), track_desc_->ssrc_, rate_);
+                  track_desc_->type_.c_str(), track_desc_->ssrc_, rate_);
     }
 
     last_sender_report_sys_time_ = 0;
@@ -3177,11 +3201,11 @@ void SrsRtcRecvTrack::update_send_report_time(const SrsNtp &ntp, uint32_t rtp_ti
         if (rate > 0) {
             if (rate_ != rate) {
                 srs_warn("RTC: SR update %s, ssrc=%u, ntp_ms=%u->%u (delta=%.0fms), rtp_time=%u->%u (delta=%.0f), rate %.0f->%.0f",
-                        track_desc_->type_.c_str(), track_desc_->ssrc_,
+                         track_desc_->type_.c_str(), track_desc_->ssrc_,
-                        last_sender_report_ntp1_.system_ms_, last_sender_report_ntp_.system_ms_, sys_time_elapsed,
+                         last_sender_report_ntp1_.system_ms_, last_sender_report_ntp_.system_ms_, sys_time_elapsed,
-                        (uint32_t)last_sender_report_rtp_time1_, (uint32_t)last_sender_report_rtp_time_, rtp_time_elpased,
+                         (uint32_t)last_sender_report_rtp_time1_, (uint32_t)last_sender_report_rtp_time_, rtp_time_elpased,
-                        rate_, rate);
+                         rate_, rate);
-                }
+            }
             rate_ = rate;
         }
     }

trunk/src/app/srs_app_rtc_source.hpp

@@ -419,9 +419,25 @@ SRS_DECLARE_PRIVATE: // clang-format on
     srs_error_t consume_packets(std::vector<SrsRtpPacket *> &pkts);
 };
 
+// Video packet cache interface
+class ISrsRtcFrameBuilderVideoPacketCache
+{
+public:
+    ISrsRtcFrameBuilderVideoPacketCache();
+    virtual ~ISrsRtcFrameBuilderVideoPacketCache();
+
+public:
+    virtual SrsRtpPacket *get_packet(uint16_t sequence_number) = 0;
+    virtual void store_packet(SrsRtpPacket *pkt) = 0;
+    virtual void clear_all() = 0;
+    virtual SrsRtpPacket *take_packet(uint16_t sequence_number) = 0;
+    virtual int32_t find_next_lost_sn(uint16_t current_sn, uint16_t header_sn, uint16_t &end_sn) = 0;
+    virtual bool check_frame_complete(const uint16_t start, const uint16_t end) = 0;
+};
+
 // Video packet cache for RTP packet management
 // TODO: Maybe should use SrsRtpRingBuffer?
-class SrsRtcFrameBuilderVideoPacketCache
+class SrsRtcFrameBuilderVideoPacketCache : public ISrsRtcFrameBuilderVideoPacketCache
 {
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on
@@ -462,18 +478,33 @@ SRS_DECLARE_PRIVATE: // clang-format on
     }
 };
 
+// Video frame detector interface
+class ISrsRtcFrameBuilderVideoFrameDetector
+{
+public:
+    ISrsRtcFrameBuilderVideoFrameDetector();
+    virtual ~ISrsRtcFrameBuilderVideoFrameDetector();
+
+public:
+    virtual void on_keyframe_start(SrsRtpPacket *pkt) = 0;
+    virtual srs_error_t detect_frame(uint16_t received, uint16_t &frame_start, uint16_t &frame_end, bool &frame_ready) = 0;
+    virtual srs_error_t detect_next_frame(uint16_t next_head, uint16_t &next_start, uint16_t &next_end, bool &next_ready) = 0;
+    virtual void on_keyframe_detached() = 0;
+    virtual bool is_lost_sn(uint16_t received) = 0;
+};
+
 // Video frame detector for managing frame boundaries and packet loss detection
-class SrsRtcFrameBuilderVideoFrameDetector
+class SrsRtcFrameBuilderVideoFrameDetector : public ISrsRtcFrameBuilderVideoFrameDetector
 {
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on
-    SrsRtcFrameBuilderVideoPacketCache *video_cache_;
+    ISrsRtcFrameBuilderVideoPacketCache *video_cache_;
     uint16_t header_sn_;
     uint16_t lost_sn_;
     int64_t rtp_key_frame_ts_;
 
 public:
-    SrsRtcFrameBuilderVideoFrameDetector(SrsRtcFrameBuilderVideoPacketCache *cache);
+    SrsRtcFrameBuilderVideoFrameDetector(ISrsRtcFrameBuilderVideoPacketCache *cache);
     virtual ~SrsRtcFrameBuilderVideoFrameDetector();
 
 public:
@@ -484,8 +515,20 @@ public:
     bool is_lost_sn(uint16_t received);
 };
 
+// Audio packet cache interface
+class ISrsRtcFrameBuilderAudioPacketCache
+{
+public:
+    ISrsRtcFrameBuilderAudioPacketCache();
+    virtual ~ISrsRtcFrameBuilderAudioPacketCache();
+
+public:
+    virtual srs_error_t process_packet(SrsRtpPacket *src, std::vector<SrsRtpPacket *> &ready_packets) = 0;
+    virtual void clear_all() = 0;
+};
+
 // Audio packet cache for RTP packet jitter buffer management
-class SrsRtcFrameBuilderAudioPacketCache
+class SrsRtcFrameBuilderAudioPacketCache : public ISrsRtcFrameBuilderAudioPacketCache
 {
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on
@@ -530,9 +573,9 @@ SRS_DECLARE_PRIVATE: // clang-format on
 
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on
-    SrsRtcFrameBuilderAudioPacketCache *audio_cache_;
+    ISrsRtcFrameBuilderAudioPacketCache *audio_cache_;
-    SrsRtcFrameBuilderVideoPacketCache *video_cache_;
+    ISrsRtcFrameBuilderVideoPacketCache *video_cache_;
-    SrsRtcFrameBuilderVideoFrameDetector *frame_detector_;
+    ISrsRtcFrameBuilderVideoFrameDetector *frame_detector_;
 
 // clang-format off
 SRS_DECLARE_PRIVATE: // clang-format on

trunk/src/kernel/srs_kernel_packet.cpp

@@ -142,6 +142,25 @@ srs_error_t SrsParsedPacket::add_sample(char *bytes, int size)
     return err;
 }
 
+SrsParsedPacket *SrsParsedPacket::copy()
+{
+    SrsParsedPacket *p = new SrsParsedPacket();
+    do_copy(p);
+    return p;
+}
+
+void SrsParsedPacket::do_copy(SrsParsedPacket *p)
+{
+    p->codec_ = codec_;
+    p->nb_samples_ = nb_samples_;
+    for (int i = 0; i < nb_samples_; i++) {
+        p->samples_[i].size_ = samples_[i].size_;
+        p->samples_[i].bytes_ = samples_[i].bytes_;
+    }
+    p->dts_ = dts_;
+    p->cts_ = cts_;
+}
+
 SrsParsedAudioPacket::SrsParsedAudioPacket()
 {
     aac_packet_type_ = SrsAudioAacFrameTraitForbidden;
@@ -156,6 +175,16 @@ SrsAudioCodecConfig *SrsParsedAudioPacket::acodec()
     return (SrsAudioCodecConfig *)codec_;
 }
 
+SrsParsedAudioPacket *SrsParsedAudioPacket::copy()
+{
+    SrsParsedAudioPacket *p = new SrsParsedAudioPacket();
+
+    do_copy(p);
+    p->aac_packet_type_ = aac_packet_type_;
+
+    return p;
+}
+
 SrsParsedVideoPacket::SrsParsedVideoPacket()
 {
     frame_type_ = SrsVideoAvcFrameTypeForbidden;
@@ -213,6 +242,21 @@ srs_error_t SrsParsedVideoPacket::add_sample(char *bytes, int size)
     return err;
 }
 
+SrsParsedVideoPacket *SrsParsedVideoPacket::copy()
+{
+    SrsParsedVideoPacket *p = new SrsParsedVideoPacket();
+
+    do_copy(p);
+    p->frame_type_ = frame_type_;
+    p->avc_packet_type_ = avc_packet_type_;
+    p->has_idr_ = has_idr_;
+    p->has_aud_ = has_aud_;
+    p->has_sps_pps_ = has_sps_pps_;
+    p->first_nalu_type_ = first_nalu_type_;
+
+    return p;
+}
+
 SrsVideoCodecConfig *SrsParsedVideoPacket::vcodec()
 {
     return (SrsVideoCodecConfig *)codec_;

trunk/src/kernel/srs_kernel_packet.hpp

@@ -109,6 +109,12 @@ public:
     virtual srs_error_t initialize(SrsCodecConfig *c);
     // Add a sample to frame.
     virtual srs_error_t add_sample(char *bytes, int size);
+
+// clang-format off
+SRS_DECLARE_PRIVATE: // clang-format on
+    // Copy the packet.
+    virtual SrsParsedPacket *copy();
+    virtual void do_copy(SrsParsedPacket *p);
 };
 
 // A parsed audio packet, besides a frame, contains the audio frame info, such as frame type.
@@ -123,6 +129,10 @@ public:
 
 public:
     virtual SrsAudioCodecConfig *acodec();
+
+// clang-format off
+SRS_DECLARE_PRIVATE: // clang-format on
+    virtual SrsParsedAudioPacket *copy();
 };
 
 // A parsed video packet, besides a frame, contains the video frame info, such as frame type.
@@ -151,6 +161,10 @@ public:
     // Add the sample without ANNEXB or IBMF header, or RAW AAC or MP3 data.
     virtual srs_error_t add_sample(char *bytes, int size);
 
+// clang-format off
+SRS_DECLARE_PRIVATE: // clang-format on
+    virtual SrsParsedVideoPacket *copy();
+
 public:
     virtual SrsVideoCodecConfig *vcodec();
 

trunk/src/utest/srs_utest_ai24.cpp

@@ -15,10 +15,10 @@ using namespace std;
 // Mock class to access protected members of SrsRtcRecvTrack
 class MockSrsRtcRecvTrackForAVSync : public SrsRtcRecvTrack
 {
-SRS_DECLARE_PRIVATE:
+    SRS_DECLARE_PRIVATE:
     static SrsRtcTrackDescription* create_track_desc(const string& type, uint32_t ssrc, int sample_rate)
     {
-        SrsRtcTrackDescription* desc = new SrsRtcTrackDescription();
+        SrsRtcTrackDescription *desc = new SrsRtcTrackDescription();
         desc->type_ = type;
         desc->id_ = "test_track";
         desc->ssrc_ = ssrc;
@@ -32,7 +32,7 @@ SRS_DECLARE_PRIVATE:
     }
 
 public:
-    MockSrsRtcRecvTrackForAVSync(const string& type, uint32_t ssrc, int sample_rate, bool is_audio)
+    MockSrsRtcRecvTrackForAVSync(const string &type, uint32_t ssrc, int sample_rate, bool is_audio)
         : SrsRtcRecvTrack(NULL, create_track_desc(type, ssrc, sample_rate), is_audio)
     {
     }
@@ -42,20 +42,24 @@ public:
 
     void set_rate(double rate) { rate_ = rate; }
 
-    int64_t test_cal_avsync_time(uint32_t rtp_time) {
+    int64_t test_cal_avsync_time(uint32_t rtp_time)
+    {
         return cal_avsync_time(rtp_time);
     }
 
-    void test_update_send_report_time(const SrsNtp& ntp, uint32_t rtp_time) {
+    void test_update_send_report_time(const SrsNtp &ntp, uint32_t rtp_time)
+    {
         update_send_report_time(ntp, rtp_time);
     }
 
     // Implement pure virtual methods
-    virtual srs_error_t on_rtp(SrsSharedPtr<SrsRtcSource>& source, SrsRtpPacket* pkt) {
+    virtual srs_error_t on_rtp(SrsSharedPtr<SrsRtcSource> &source, SrsRtpPacket *pkt)
+    {
         return srs_success;
     }
 
-    virtual srs_error_t check_send_nacks() {
+    virtual srs_error_t check_send_nacks()
+    {
         return srs_success;
     }
 };
@@ -85,8 +89,8 @@ VOID TEST(RtcAVSyncTest, CalAVSyncTimeWithSDPRate)
 
     // Simulate first SR received
     SrsNtp ntp;
-    ntp.system_ms_ = 1000;  // 1000 ms
+    ntp.system_ms_ = 1000;     // 1000 ms
-    uint32_t rtp_time = 48000;  // 48000 RTP units
+    uint32_t rtp_time = 48000; // 48000 RTP units
     track.test_update_send_report_time(ntp, rtp_time);
 
     // Calculate avsync time for a later RTP packet
@@ -128,8 +132,8 @@ VOID TEST(RtcAVSyncTest, AudioRateUpdateAfter2ndSR)
 
     // Second SR (20ms later, RTP increased by 960)
     SrsNtp ntp2;
-    ntp2.system_ms_ = 1020;  // 20ms later
+    ntp2.system_ms_ = 1020;     // 20ms later
-    uint32_t rtp_time2 = 48960;  // 960 RTP units later (48 * 20)
+    uint32_t rtp_time2 = 48960; // 960 RTP units later (48 * 20)
     track.test_update_send_report_time(ntp2, rtp_time2);
 
     // Rate should be updated to calculated value: 960 / 20 = 48
@@ -155,8 +159,8 @@ VOID TEST(RtcAVSyncTest, VideoRateUpdateAfter2ndSR)
 
     // Second SR (100ms later, RTP increased by 9000)
     SrsNtp ntp2;
-    ntp2.system_ms_ = 2100;  // 100ms later
+    ntp2.system_ms_ = 2100;      // 100ms later
-    uint32_t rtp_time2 = 189000;  // 9000 RTP units later (90 * 100)
+    uint32_t rtp_time2 = 189000; // 9000 RTP units later (90 * 100)
     track.test_update_send_report_time(ntp2, rtp_time2);
 
     // Rate should be updated to calculated value: 9000 / 100 = 90
@@ -182,7 +186,7 @@ VOID TEST(RtcAVSyncTest, RateUpdateWithClockDrift)
     // Actual: 100ms -> 9010 RTP units (slight drift)
     SrsNtp ntp2;
     ntp2.system_ms_ = 1100;
-    uint32_t rtp_time2 = 99010;  // Slightly more than expected
+    uint32_t rtp_time2 = 99010; // Slightly more than expected
     track.test_update_send_report_time(ntp2, rtp_time2);
 
     // Rate should be updated to: round(9010 / 100) = 90
@@ -203,7 +207,7 @@ VOID TEST(RtcAVSyncTest, RateUpdateWithLargeInterval)
     // Second SR (1000ms later)
     SrsNtp ntp2;
     ntp2.system_ms_ = 6000;
-    uint32_t rtp_time2 = 288000;  // 48000 RTP units later (48 * 1000)
+    uint32_t rtp_time2 = 288000; // 48000 RTP units later (48 * 1000)
     track.test_update_send_report_time(ntp2, rtp_time2);
 
     // Rate should be: 48000 / 1000 = 48
@@ -249,9 +253,9 @@ VOID TEST(RtcAVSyncTest, ImmediateAVSyncAvailability)
     track.test_update_send_report_time(ntp1, rtp_time1);
 
     // Should be able to calculate avsync_time immediately (not -1)
-    int64_t avsync_time = track.test_cal_avsync_time(48480);  // 10ms later
+    int64_t avsync_time = track.test_cal_avsync_time(48480); // 10ms later
-    EXPECT_GT(avsync_time, 0);  // Should be > 0, not -1
+    EXPECT_GT(avsync_time, 0);                               // Should be > 0, not -1
-    EXPECT_EQ(1010, avsync_time);  // Should be 1000 + 10 = 1010
+    EXPECT_EQ(1010, avsync_time);                            // Should be 1000 + 10 = 1010
 }
 
 // Test: RTP timestamp wraparound handling
@@ -262,13 +266,13 @@ VOID TEST(RtcAVSyncTest, RTPTimestampWraparound)
     // First SR near wraparound
     SrsNtp ntp1;
     ntp1.system_ms_ = 1000;
-    uint32_t rtp_time1 = 0xFFFFF000;  // Near max uint32_t
+    uint32_t rtp_time1 = 0xFFFFF000; // Near max uint32_t
     track.test_update_send_report_time(ntp1, rtp_time1);
 
     // Second SR after wraparound
     SrsNtp ntp2;
-    ntp2.system_ms_ = 1020;  // 20ms later
+    ntp2.system_ms_ = 1020;          // 20ms later
-    uint32_t rtp_time2 = 0x000003C0;  // Wrapped around, 960 units after wraparound
+    uint32_t rtp_time2 = 0x000003C0; // Wrapped around, 960 units after wraparound
     track.test_update_send_report_time(ntp2, rtp_time2);
 
     // Note: Current implementation may not handle wraparound correctly
@@ -292,11 +296,10 @@ VOID TEST(RtcAVSyncTest, ZeroTimeElapsedBetweenSRs)
 
     // Second SR with same timestamp (0ms elapsed)
     SrsNtp ntp2;
-    ntp2.system_ms_ = 1000;  // Same time
+    ntp2.system_ms_ = 1000;     // Same time
-    uint32_t rtp_time2 = 48000;  // Same RTP time
+    uint32_t rtp_time2 = 48000; // Same RTP time
     track.test_update_send_report_time(ntp2, rtp_time2);
 
     // Rate should remain unchanged (SDP rate)
     EXPECT_DOUBLE_EQ(rate_before, track.get_rate());
 }
-

trunk/src/utest/srs_utest_ai24.hpp

@@ -9,4 +9,3 @@
 #include <srs_utest.hpp>
 
 #endif
-

trunk/src/utest/srs_utest_manual_mock.cpp

@@ -180,6 +180,10 @@ MockRtcTrackDescriptionFactory::MockRtcTrackDescriptionFactory()
     audio_ssrc_ = 12345;
     video_ssrc_ = 67890;
     screen_ssrc_ = 98765;
+
+    audio_pt_ = 111;
+    video_pt_ = 96;
+    screen_pt_ = 97;
 }
 
 MockRtcTrackDescriptionFactory::~MockRtcTrackDescriptionFactory()
@@ -227,12 +231,17 @@ SrsRtcTrackDescription *MockRtcTrackDescriptionFactory::create_audio_track(uint3
     audio_desc->is_active_ = true;
     audio_desc->direction_ = "sendrecv";
     audio_desc->mid_ = mid;
-    audio_desc->media_ = new SrsAudioPayload(111, "opus", 48000, 2);
+    audio_desc->media_ = new SrsAudioPayload(audio_pt_, "opus", 48000, 2);
     return audio_desc;
 }
 
 SrsRtcTrackDescription *MockRtcTrackDescriptionFactory::create_video_track(uint32_t ssrc, std::string id, std::string mid)
 {
+    uint8_t pt = video_pt_;
+    if (ssrc == screen_ssrc_) {
+        pt = screen_pt_;
+    }
+
     SrsRtcTrackDescription *video_desc = new SrsRtcTrackDescription();
     video_desc->type_ = "video";
     video_desc->ssrc_ = ssrc;
@@ -240,7 +249,7 @@ SrsRtcTrackDescription *MockRtcTrackDescriptionFactory::create_video_track(uint3
     video_desc->is_active_ = true;
     video_desc->direction_ = "sendrecv";
     video_desc->mid_ = mid;
-    video_desc->media_ = new SrsVideoPayload(96, "H264", 90000);
+    video_desc->media_ = new SrsVideoPayload(pt, "H264", 90000);
     return video_desc;
 }
 
@@ -1160,6 +1169,7 @@ MockLiveSource::MockLiveSource()
     on_audio_count_ = 0;
     on_video_count_ = 0;
     on_dump_packets_count_ = 0;
+    on_frame_count_ = 0;
 }
 
 MockLiveSource::~MockLiveSource()
@@ -1196,16 +1206,25 @@ srs_error_t MockLiveSource::consumer_dumps(ISrsLiveConsumer *consumer, bool ds,
 
 srs_error_t MockLiveSource::on_audio(SrsRtmpCommonMessage *audio)
 {
-    on_audio_count_++;
     return SrsLiveSource::on_audio(audio);
 }
 
 srs_error_t MockLiveSource::on_video(SrsRtmpCommonMessage *video)
 {
-    on_video_count_++;
     return SrsLiveSource::on_video(video);
 }
 
+srs_error_t MockLiveSource::on_frame(SrsMediaPacket *msg)
+{
+    on_frame_count_++;
+    if (msg->is_audio()) {
+        on_audio_count_++;
+    } else if (msg->is_video()) {
+        on_video_count_++;
+    }
+    return SrsLiveSource::on_frame(msg);
+}
+
 // Mock SRT source implementation
 MockSrtSource::MockSrtSource()
 {

trunk/src/utest/srs_utest_manual_mock.hpp

@@ -110,6 +110,11 @@ public:
     uint32_t video_ssrc_;
     uint32_t screen_ssrc_;
 
+public:
+    uint8_t audio_pt_;
+    uint8_t video_pt_;
+    uint8_t screen_pt_;
+
 public:
     // Create a map of track descriptions with audio and video tracks (for play stream)
     std::map<uint32_t, SrsRtcTrackDescription *> create_audio_video_tracks();
@@ -634,6 +639,7 @@ public:
     int on_audio_count_;
     int on_video_count_;
     int on_dump_packets_count_;
+    int on_frame_count_;
 
 public:
     MockLiveSource();
@@ -647,6 +653,7 @@ public:
 public:
     virtual srs_error_t on_audio(SrsRtmpCommonMessage *audio);
     virtual srs_error_t on_video(SrsRtmpCommonMessage *video);
+    virtual srs_error_t on_frame(SrsMediaPacket *msg);
 };
 
 // Mock SRT source for testing SrsRtcPublishStream

trunk/src/utest/srs_utest_workflow_rtc2rtmp.cpp

@@ -0,0 +1,240 @@
+/**
+ * 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_workflow_rtc2rtmp.hpp>
+
+#include <srs_app_factory.hpp>
+#include <srs_app_rtc_codec.hpp>
+#include <srs_app_rtc_conn.hpp>
+#include <srs_app_rtc_server.hpp>
+#include <srs_app_rtc_source.hpp>
+#include <srs_app_stream_token.hpp>
+#include <srs_kernel_error.hpp>
+#include <srs_protocol_rtmp_stack.hpp>
+#include <srs_utest_ai11.hpp>
+#include <srs_utest_manual_mock.hpp>
+#include <srs_utest_manual_service.hpp>
+#include <srs_utest_workflow_rtc_conn.hpp>
+
+// Create a mock audio cache ISrsRtcFrameBuilderAudioPacketCache
+class MockAudioCache : public ISrsRtcFrameBuilderAudioPacketCache
+{
+public:
+    int process_packet_count_;
+
+public:
+    MockAudioCache();
+    virtual ~MockAudioCache();
+
+public:
+    virtual srs_error_t process_packet(SrsRtpPacket *src, std::vector<SrsRtpPacket *> &ready_packets);
+    virtual void clear_all();
+};
+
+MockAudioCache::MockAudioCache()
+{
+    process_packet_count_ = 0;
+}
+
+MockAudioCache::~MockAudioCache()
+{
+}
+
+srs_error_t MockAudioCache::process_packet(SrsRtpPacket *src, std::vector<SrsRtpPacket *> &ready_packets)
+{
+    process_packet_count_++;
+
+    // Copy the packet.
+    SrsRtpPacket *copy = src->copy();
+    ready_packets.push_back(copy);
+
+    return srs_success;
+}
+
+void MockAudioCache::clear_all()
+{
+}
+
+// Mock the audio transcoder ISrsAudioTranscoder.
+class MockAudioTranscoderForRtc2Rtmp : public ISrsAudioTranscoder
+{
+public:
+    int transcode_count_;
+    std::vector<SrsParsedAudioPacket *> output_packets_;
+    std::string aac_header_;
+
+public:
+    MockAudioTranscoderForRtc2Rtmp();
+    virtual ~MockAudioTranscoderForRtc2Rtmp();
+
+public:
+    virtual srs_error_t initialize(SrsAudioCodecId from, SrsAudioCodecId to, int channels, int sample_rate, int bit_rate);
+    virtual srs_error_t transcode(SrsParsedAudioPacket *in, std::vector<SrsParsedAudioPacket *> &outs);
+    virtual void free_frames(std::vector<SrsParsedAudioPacket *> &frames);
+    virtual void aac_codec_header(uint8_t **data, int *len);
+};
+
+MockAudioTranscoderForRtc2Rtmp::MockAudioTranscoderForRtc2Rtmp()
+{
+    transcode_count_ = 0;
+}
+
+MockAudioTranscoderForRtc2Rtmp::~MockAudioTranscoderForRtc2Rtmp()
+{
+}
+
+srs_error_t MockAudioTranscoderForRtc2Rtmp::initialize(SrsAudioCodecId from, SrsAudioCodecId to, int channels, int sample_rate, int bit_rate)
+{
+    return srs_success;
+}
+
+srs_error_t MockAudioTranscoderForRtc2Rtmp::transcode(SrsParsedAudioPacket *in, std::vector<SrsParsedAudioPacket *> &outs)
+{
+    transcode_count_++;
+
+    SrsParsedAudioPacket *out = in->copy();
+    output_packets_.push_back(out);
+    outs.push_back(out);
+
+    return srs_success;
+}
+
+void MockAudioTranscoderForRtc2Rtmp::free_frames(std::vector<SrsParsedAudioPacket *> &frames)
+{
+}
+
+void MockAudioTranscoderForRtc2Rtmp::aac_codec_header(uint8_t **data, int *len)
+{
+    int size = aac_header_.size();
+    uint8_t *copy = new uint8_t[size];
+    memcpy(copy, aac_header_.data(), size);
+    *data = copy;
+    *len = size;
+}
+
+// This test is used to verify the basic workflow of the RTC connection.
+// 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(BasicWorkflowRtc2RtmpTest, ManuallyVerifyTypicalScenario)
+{
+    srs_error_t err;
+
+    // Create mock objects for dependencies
+    SrsUniquePtr<MockAppConfig> mock_config(new MockAppConfig());
+    SrsUniquePtr<MockRtcSourceManager> mock_rtc_sources(new MockRtcSourceManager());
+    SrsUniquePtr<MockAppStatistic> mock_stat(new MockAppStatistic());
+    SrsUniquePtr<MockRtcAsyncCallRequest> mock_request(new MockRtcAsyncCallRequest("test.vhost", "live", "stream1"));
+    SrsUniquePtr<MockRtcAsyncTaskExecutor> mock_exec(new MockRtcAsyncTaskExecutor());
+    SrsUniquePtr<MockExpire> mock_expire(new MockExpire());
+    SrsUniquePtr<MockRtcPacketReceiver> mock_receiver(new MockRtcPacketReceiver());
+    SrsUniquePtr<MockRtcTrackDescriptionFactory> track_factory(new MockRtcTrackDescriptionFactory());
+    SrsUniquePtr<MockLiveSourceManager> mock_sources(new MockLiveSourceManager());
+    MockAudioCache *mock_audio_cache = new MockAudioCache();
+    MockAudioTranscoderForRtc2Rtmp *mock_audio_transcoder = new MockAudioTranscoderForRtc2Rtmp();
+
+    mock_audio_transcoder->aac_header_ = std::string("\xAF\x00\x12\x10", 4); // AAC sequence header.
+    mock_config->rtc_to_rtmp_ = true;
+
+    // Create RTC publish stream - use real pli_worker_
+    SrsContextId cid;
+    cid.set_value("test-rtc2rtmp-workflow-typical-scenario");
+    SrsUniquePtr<SrsRtcPublishStream> publish_stream(new SrsRtcPublishStream(mock_exec.get(), mock_expire.get(), mock_receiver.get(), cid));
+
+    // Mock the publish stream object
+    if (true) {
+        // Inject mock dependencies
+        publish_stream->config_ = mock_config.get();
+        publish_stream->rtc_sources_ = mock_rtc_sources.get();
+        publish_stream->live_sources_ = mock_sources.get();
+        publish_stream->stat_ = mock_stat.get();
+    }
+
+    // Initialize publish stream, rtc2rtmp bridge should be created
+    SrsRtcBridge *bridge = NULL;
+    SrsLiveSource *live_source = NULL;
+    SrsRtcFrameBuilder *frame_builder = NULL;
+    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.get(), 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);
+
+        // source bridge should be created
+        bridge = dynamic_cast<SrsRtcBridge *>(publish_stream->source_->rtc_bridge_);
+        EXPECT_TRUE(bridge != NULL);
+
+        live_source = bridge->rtmp_target_.get();
+        EXPECT_TRUE(live_source != NULL);
+
+        frame_builder = bridge->frame_builder_;
+        EXPECT_TRUE(frame_builder != NULL);
+    }
+
+    // Start the publish stream.
+    if (true) {
+        // 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_);
+
+        // When starting the publish stream, the frame builder should be recreated
+        EXPECT_TRUE(frame_builder != bridge->frame_builder_);
+        frame_builder = bridge->frame_builder_;
+        EXPECT_TRUE(frame_builder != NULL);
+
+        // Mock the frame builder object
+        srs_freep(frame_builder->audio_cache_);
+        frame_builder->audio_cache_ = mock_audio_cache;
+        srs_freep(frame_builder->audio_transcoder_);
+        frame_builder->audio_transcoder_ = mock_audio_transcoder;
+    }
+
+    // Got a RTP audio packet.
+    MockLiveSource *mock_source = dynamic_cast<MockLiveSource *>(mock_sources->mock_source_.get());
+    if (true) {
+        SrsRtpPacket pkt;
+        pkt.header_.set_ssrc(track_factory->audio_ssrc_);
+        pkt.header_.set_sequence(100);
+        pkt.header_.set_timestamp(1000);
+        pkt.header_.set_payload_type(track_factory->audio_pt_);
+
+        SrsUniquePtr<char[]> data(new char[1500]);
+        SrsBuffer buf(data.get(), 1500);
+        HELPER_EXPECT_SUCCESS(pkt.encode(&buf));
+
+        HELPER_EXPECT_SUCCESS(publish_stream->on_rtp_plaintext(data.get(), buf.pos()));
+
+        // The live source should got 2 audio packets, one is sequence header, another is audio data.
+        EXPECT_EQ(mock_source->on_audio_count_, 2);
+        EXPECT_EQ(mock_source->on_frame_count_, 2);
+        EXPECT_EQ(mock_source->on_video_count_, 0);
+    }
+
+    publish_stream->stop();
+}

trunk/src/utest/srs_utest_workflow_rtc2rtmp.hpp

@@ -0,0 +1,32 @@
+/**
+ * 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_WORKFLOW_RTC2RTMP_HPP
+#define SRS_UTEST_WORKFLOW_RTC2RTMP_HPP
+
+/*
+#include <srs_utest_workflow_rtc2rtmp.hpp>
+*/
+#include <srs_utest.hpp>
+
+#endif

trunk/src/utest/srs_utest_workflow_rtc_conn.hpp

@@ -21,11 +21,11 @@
  * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  */
 
-#ifndef SRS_UTEST_RTC_CONN_HPP
+#ifndef SRS_UTEST_WORKFLOW_RTC_CONN_HPP
-#define SRS_UTEST_RTC_CONN_HPP
+#define SRS_UTEST_WORKFLOW_RTC_CONN_HPP
 
 /*
-#include <srs_utest_rtc_conn.hpp>
+#include <srs_utest_workflow_rtc_conn.hpp>
 */
 #include <srs_utest.hpp>
 

trunk/src/utest/srs_utest_workflow_rtc_playstream.cpp

@@ -221,9 +221,4 @@ VOID TEST(BasicWorkflowRtcPlayStreamTest, ManuallyVerify)
 
     // Stop the play stream
     play_stream->stop();
-
-    // Clean up - set to NULL to avoid double-free
-    play_stream->config_ = NULL;
-    play_stream->rtc_sources_ = NULL;
-    play_stream->stat_ = NULL;
 }

trunk/src/utest/srs_utest_workflow_rtc_publishstream.cpp

@@ -96,9 +96,4 @@ VOID TEST(BasicWorkflowRtcPublishStreamTest, ManuallyVerify)
 
     // 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;
 }