AI: Add utest to cover gb module.

trunk/configure

@@ -384,7 +384,7 @@ 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_app12" "srs_utest_app13")
+        "srs_utest_app10" "srs_utest_app11" "srs_utest_app12" "srs_utest_app13" "srs_utest_app14")
     # Always include SRT utest
     MODULE_FILES+=("srs_utest_srt")
     if [[ $SRS_GB28181 == YES ]]; then

trunk/src/app/srs_app_caster_flv.cpp

@@ -47,7 +47,8 @@ srs_error_t SrsHttpFlvListener::initialize(SrsConfDirective *c)
         return srs_error_new(ERROR_STREAM_CASTER_PORT, "invalid port=%d", port);
     }
 
-    listener_->set_endpoint(srs_net_address_any(), port)->set_label("PUSH-FLV");
+    listener_->set_endpoint(srs_net_address_any(), port);
+    listener_->set_label("PUSH-FLV");
 
     if ((err = caster_->initialize(c)) != srs_success) {
         return srs_error_wrap(err, "init caster port=%d", port);
@@ -160,11 +161,36 @@ void SrsAppCasterFlv::add(ISrsResource *conn, bool *exists)
     manager_->add(conn, exists);
 }
 
+void SrsAppCasterFlv::add_with_id(const std::string &id, ISrsResource *conn)
+{
+    manager_->add_with_id(id, conn);
+}
+
+void SrsAppCasterFlv::add_with_fast_id(uint64_t id, ISrsResource *conn)
+{
+    manager_->add_with_fast_id(id, conn);
+}
+
 ISrsResource *SrsAppCasterFlv::at(int index)
 {
     return manager_->at(index);
 }
 
+ISrsResource *SrsAppCasterFlv::find_by_id(std::string id)
+{
+    return manager_->find_by_id(id);
+}
+
+ISrsResource *SrsAppCasterFlv::find_by_fast_id(uint64_t id)
+{
+    return manager_->find_by_fast_id(id);
+}
+
+ISrsResource *SrsAppCasterFlv::find_by_name(std::string name)
+{
+    return manager_->find_by_name(name);
+}
+
 void SrsAppCasterFlv::remove(ISrsResource *c)
 {
     ISrsConnection *conn = dynamic_cast<ISrsConnection *>(c);

trunk/src/app/srs_app_caster_flv.hpp

@@ -74,7 +74,12 @@ public:
     virtual bool empty();
     virtual size_t size();
     virtual void add(ISrsResource *conn, bool *exists = NULL);
+    virtual void add_with_id(const std::string &id, ISrsResource *conn);
+    virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
     virtual ISrsResource *at(int index);
+    virtual ISrsResource *find_by_id(std::string id);
+    virtual ISrsResource *find_by_fast_id(uint64_t id);
+    virtual ISrsResource *find_by_name(std::string name);
     virtual void remove(ISrsResource *c);
     virtual void subscribe(ISrsDisposingHandler *h);
     virtual void unsubscribe(ISrsDisposingHandler *h);

trunk/src/app/srs_app_config.hpp

@@ -359,6 +359,8 @@ public:
     virtual std::vector<SrsConfDirective *> get_stream_casters() = 0;
     virtual bool get_stream_caster_enabled(SrsConfDirective *conf) = 0;
     virtual std::string get_stream_caster_engine(SrsConfDirective *conf) = 0;
+    virtual std::string get_stream_caster_output(SrsConfDirective *conf) = 0;
+    virtual int get_stream_caster_listen(SrsConfDirective *conf) = 0;
 
 public:
     // Exporter config

trunk/src/app/srs_app_factory.cpp

@@ -22,6 +22,7 @@
 #include <srs_kernel_flv.hpp>
 #include <srs_kernel_mp4.hpp>
 #include <srs_app_dvr.hpp>
+#include <srs_app_gb28181.hpp>
 
 ISrsAppFactory::ISrsAppFactory()
 {
@@ -128,6 +129,16 @@ ISrsDvrSegmenter *SrsAppFactory::create_dvr_mp4_segmenter()
     return new SrsDvrMp4Segmenter();
 }
 
+ISrsGbMediaTcpConn *SrsAppFactory::create_gb_media_tcp_conn()
+{
+    return new SrsGbMediaTcpConn();
+}
+
+ISrsGbSession *SrsAppFactory::create_gb_session()
+{
+    return new SrsGbSession();
+}
+
 SrsFinalFactory::SrsFinalFactory()
 {
 }

trunk/src/app/srs_app_factory.hpp

@@ -29,6 +29,8 @@ class SrsRtcTrackDescription;
 class ISrsFlvTransmuxer;
 class ISrsMp4Encoder;
 class ISrsDvrSegmenter;
+class ISrsGbMediaTcpConn;
+class ISrsGbSession;
 
 // The factory to create app objects.
 class ISrsAppFactory
@@ -57,6 +59,8 @@ public:
     virtual ISrsMp4Encoder *create_mp4_encoder() = 0;
     virtual ISrsDvrSegmenter *create_dvr_flv_segmenter() = 0;
     virtual ISrsDvrSegmenter *create_dvr_mp4_segmenter() = 0;
+    virtual ISrsGbMediaTcpConn *create_gb_media_tcp_conn() = 0;
+    virtual ISrsGbSession *create_gb_session() = 0;
 };
 
 // The factory to create app objects.
@@ -86,6 +90,8 @@ public:
     virtual ISrsMp4Encoder *create_mp4_encoder();
     virtual ISrsDvrSegmenter *create_dvr_flv_segmenter();
     virtual ISrsDvrSegmenter *create_dvr_mp4_segmenter();
+    virtual ISrsGbMediaTcpConn *create_gb_media_tcp_conn();
+    virtual ISrsGbSession *create_gb_session();
 };
 
 extern ISrsAppFactory *_srs_app_factory;

trunk/src/app/srs_app_gb28181.cpp

@@ -25,6 +25,7 @@
 #include <srs_protocol_raw_avc.hpp>
 #include <srs_protocol_sdp.hpp>
 #include <srs_protocol_utility.hpp>
+#include <srs_app_factory.hpp>
 
 #include <sstream>
 using namespace std;
@@ -55,6 +56,14 @@ std::string srs_gb_state(SrsGbSessionState ostate, SrsGbSessionState state)
     return srs_fmt_sprintf("%s->%s", srs_gb_session_state(ostate).c_str(), srs_gb_session_state(state).c_str());
 }
 
+ISrsGbSession::ISrsGbSession()
+{
+}
+
+ISrsGbSession::~ISrsGbSession()
+{
+}
+
 SrsGbSession::SrsGbSession() : media_(new SrsGbMediaTcpConn())
 {
     wrapper_ = NULL;
@@ -86,23 +95,27 @@ SrsGbSession::SrsGbSession() : media_(new SrsGbMediaTcpConn())
 
     cid_ = _srs_context->generate_id();
     _srs_context->set_id(cid_); // Also change current coroutine cid as session's.
+
+    config_ = _srs_config;
 }
 
 SrsGbSession::~SrsGbSession()
 {
     srs_freep(muxer_);
     srs_freep(ppp_);
+
+    config_ = NULL;
 }
 
 void SrsGbSession::setup(SrsConfDirective *conf)
 {
-    std::string output = _srs_config->get_stream_caster_output(conf);
+    std::string output = config_->get_stream_caster_output(conf);
     muxer_->setup(output);
 
     srs_trace("Session: Start output=%s", output.c_str());
 }
 
-void SrsGbSession::setup_owner(SrsSharedResource<SrsGbSession> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid)
+void SrsGbSession::setup_owner(SrsSharedResource<ISrsGbSession> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid)
 {
     wrapper_ = wrapper;
     owner_coroutine_ = owner_coroutine;
@@ -114,7 +127,7 @@ void SrsGbSession::on_executor_done(ISrsInterruptable *executor)
     owner_coroutine_ = NULL;
 }
 
-void SrsGbSession::on_ps_pack(SrsPackContext *ctx, SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs)
+void SrsGbSession::on_ps_pack(ISrsPackContext *ctx, SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs)
 {
     // Got a new context, that is new media transport.
     if (media_id_ != ctx->media_id_) {
@@ -172,7 +185,7 @@ void SrsGbSession::on_ps_pack(SrsPackContext *ctx, SrsPsPacket *ps, const std::v
     }
 }
 
-void SrsGbSession::on_media_transport(SrsSharedResource<SrsGbMediaTcpConn> media)
+void SrsGbSession::on_media_transport(SrsSharedResource<ISrsGbMediaTcpConn> media)
 {
     media_ = media;
 
@@ -301,16 +314,34 @@ std::string SrsGbSession::desc()
     return "GBS";
 }
 
+ISrsGbListener::ISrsGbListener()
+{
+}
+
+ISrsGbListener::~ISrsGbListener()
+{
+}
+
 SrsGbListener::SrsGbListener()
 {
     conf_ = NULL;
     media_listener_ = new SrsTcpListener(this);
+
+    config_ = _srs_config;
+    api_server_owner_ = _srs_server;
+    gb_manager_ = _srs_gb_manager;
+    app_factory_ = _srs_app_factory;
 }
 
 SrsGbListener::~SrsGbListener()
 {
     srs_freep(conf_);
     srs_freep(media_listener_);
+
+    config_ = NULL;
+    api_server_owner_ = NULL;
+    gb_manager_ = NULL;
+    app_factory_ = NULL;
 }
 
 srs_error_t SrsGbListener::initialize(SrsConfDirective *conf)
@@ -322,8 +353,9 @@ srs_error_t SrsGbListener::initialize(SrsConfDirective *conf)
 
     string ip = srs_net_address_any();
     if (true) {
-        int port = _srs_config->get_stream_caster_listen(conf);
+        int port = config_->get_stream_caster_listen(conf);
-        media_listener_->set_endpoint(ip, port)->set_label("GB-TCP");
+        media_listener_->set_endpoint(ip, port);
+        media_listener_->set_label("GB-TCP");
     }
 
     return err;
@@ -348,7 +380,7 @@ srs_error_t SrsGbListener::listen_api()
 {
     srs_error_t err = srs_success;
 
-    ISrsHttpServeMux *mux = _srs_server->api_server();
+    ISrsHttpServeMux *mux = api_server_owner_->api_server();
     if ((err = mux->handle("/gb/v1/publish/", new SrsGoApiGbPublish(conf_))) != srs_success) {
         return srs_error_wrap(err, "handle publish");
     }
@@ -366,13 +398,13 @@ srs_error_t SrsGbListener::on_tcp_client(ISrsListener *listener, srs_netfd_t stf
 
     // Handle TCP connections.
     if (listener == media_listener_) {
-        SrsGbMediaTcpConn *raw_conn = new SrsGbMediaTcpConn();
+        ISrsGbMediaTcpConn *raw_conn = app_factory_->create_gb_media_tcp_conn();
         raw_conn->setup(stfd);
 
-        SrsSharedResource<SrsGbMediaTcpConn> *conn = new SrsSharedResource<SrsGbMediaTcpConn>(raw_conn);
+        SrsSharedResource<ISrsGbMediaTcpConn> *conn = new SrsSharedResource<ISrsGbMediaTcpConn>(raw_conn);
-        _srs_gb_manager->add(conn, NULL);
+        gb_manager_->add(conn, NULL);
 
-        SrsExecutorCoroutine *executor = new SrsExecutorCoroutine(_srs_gb_manager, conn, raw_conn, raw_conn);
+        SrsExecutorCoroutine *executor = new SrsExecutorCoroutine(gb_manager_, conn, raw_conn, raw_conn);
         raw_conn->setup_owner(conn, executor, executor);
 
         if ((err = executor->start()) != srs_success) {
@@ -395,6 +427,14 @@ ISrsPsPackHandler::~ISrsPsPackHandler()
 {
 }
 
+ISrsGbMediaTcpConn::ISrsGbMediaTcpConn()
+{
+}
+
+ISrsGbMediaTcpConn::~ISrsGbMediaTcpConn()
+{
+}
+
 SrsGbMediaTcpConn::SrsGbMediaTcpConn()
 {
     pack_ = new SrsPackContext(this);
@@ -409,6 +449,8 @@ SrsGbMediaTcpConn::SrsGbMediaTcpConn()
     session_ = NULL;
     connected_ = false;
     nn_rtcp_ = 0;
+
+    gb_manager_ = _srs_gb_manager;
 }
 
 SrsGbMediaTcpConn::~SrsGbMediaTcpConn()
@@ -416,6 +458,8 @@ SrsGbMediaTcpConn::~SrsGbMediaTcpConn()
     srs_freep(conn_);
     srs_freepa(buffer_);
     srs_freep(pack_);
+
+    gb_manager_ = NULL;
 }
 
 void SrsGbMediaTcpConn::setup(srs_netfd_t stfd)
@@ -424,7 +468,7 @@ void SrsGbMediaTcpConn::setup(srs_netfd_t stfd)
     conn_ = new SrsTcpConnection(stfd);
 }
 
-void SrsGbMediaTcpConn::setup_owner(SrsSharedResource<SrsGbMediaTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid)
+void SrsGbMediaTcpConn::setup_owner(SrsSharedResource<ISrsGbMediaTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid)
 {
     wrapper_ = wrapper;
     owner_coroutine_ = owner_coroutine;
@@ -511,7 +555,7 @@ srs_error_t SrsGbMediaTcpConn::do_cycle()
     SrsRecoverablePsContext context;
 
     // If bytes is not enough(defined by SRS_PS_MIN_REQUIRED), ignore.
-    context.ctx_.set_detect_ps_integrity(true);
+    context.ctx_->set_detect_ps_integrity(true);
 
     // Previous left bytes, to parse in next loop.
     uint32_t reserved = 0;
@@ -538,8 +582,8 @@ srs_error_t SrsGbMediaTcpConn::do_cycle()
         }
 
         if (length > SRS_GB_LARGE_PACKET) {
-            const SrsPsDecodeHelper &h = context.ctx_.helper_;
+            const SrsPsDecodeHelper *h = context.ctx_->helper();
-            srs_warn("PS: Large length=%u, previous-seq=%u, previous-ts=%u", length, h.rtp_seq_, h.rtp_ts_);
+            srs_warn("PS: Large length=%u, previous-seq=%u, previous-ts=%u", length, h->rtp_seq_, h->rtp_ts_);
         }
 
         // Read length of bytes of RTP packet.
@@ -629,7 +673,7 @@ srs_error_t SrsGbMediaTcpConn::on_ps_pack(SrsPsPacket *ps, const std::vector<Srs
     return err;
 }
 
-srs_error_t SrsGbMediaTcpConn::bind_session(uint32_t ssrc, SrsGbSession **psession)
+srs_error_t SrsGbMediaTcpConn::bind_session(uint32_t ssrc, ISrsGbSession **psession)
 {
     srs_error_t err = srs_success;
 
@@ -637,11 +681,11 @@ srs_error_t SrsGbMediaTcpConn::bind_session(uint32_t ssrc, SrsGbSession **psessi
         return err;
 
     // Find exists session for register, might be created by another object and still alive.
-    SrsSharedResource<SrsGbSession> *session = dynamic_cast<SrsSharedResource<SrsGbSession> *>(_srs_gb_manager->find_by_fast_id(ssrc));
+    SrsSharedResource<ISrsGbSession> *session = dynamic_cast<SrsSharedResource<ISrsGbSession> *>(gb_manager_->find_by_fast_id(ssrc));
     if (!session)
         return err;
 
-    SrsGbSession *raw_session = (*session).get();
+    ISrsGbSession *raw_session = (*session).get();
     srs_assert(raw_session);
 
     // Notice session to use current media connection.
@@ -651,6 +695,14 @@ srs_error_t SrsGbMediaTcpConn::bind_session(uint32_t ssrc, SrsGbSession **psessi
     return err;
 }
 
+ISrsMpegpsQueue::ISrsMpegpsQueue()
+{
+}
+
+ISrsMpegpsQueue::~ISrsMpegpsQueue()
+{
+}
+
 SrsMpegpsQueue::SrsMpegpsQueue()
 {
     nb_audios_ = nb_videos_ = 0;
@@ -725,7 +777,15 @@ SrsMediaPacket *SrsMpegpsQueue::dequeue()
     return NULL;
 }
 
-SrsGbMuxer::SrsGbMuxer(SrsGbSession *session)
+ISrsGbMuxer::ISrsGbMuxer()
+{
+}
+
+ISrsGbMuxer::~ISrsGbMuxer()
+{
+}
+
+SrsGbMuxer::SrsGbMuxer(ISrsGbSession *session)
 {
     sdk_ = NULL;
     session_ = session;
@@ -743,6 +803,8 @@ SrsGbMuxer::SrsGbMuxer(SrsGbSession *session)
 
     queue_ = new SrsMpegpsQueue();
     pprint_ = SrsPithyPrint::create_caster();
+
+    app_factory_ = _srs_app_factory;
 }
 
 SrsGbMuxer::~SrsGbMuxer()
@@ -754,6 +816,8 @@ SrsGbMuxer::~SrsGbMuxer()
     srs_freep(aac_);
     srs_freep(queue_);
     srs_freep(pprint_);
+
+    app_factory_ = NULL;
 }
 
 void SrsGbMuxer::setup(std::string output)
@@ -1272,7 +1336,7 @@ srs_error_t SrsGbMuxer::connect()
 
     srs_utime_t cto = SRS_CONSTS_RTMP_TIMEOUT;
     srs_utime_t sto = SRS_CONSTS_RTMP_PULSE;
-    sdk_ = new SrsSimpleRtmpClient(url, cto, sto);
+    sdk_ = app_factory_->create_rtmp_client(url, cto, sto);
 
     if ((err = sdk_->connect()) != srs_success) {
         close();
@@ -1300,7 +1364,7 @@ void SrsGbMuxer::close()
     h264_pps_ = "";
 }
 
-SrsPackContext::SrsPackContext(ISrsPsPackHandler *handler)
+ISrsPackContext::ISrsPackContext()
 {
     static uint32_t gid = 0;
     media_id_ = ++gid;
@@ -1309,7 +1373,14 @@ SrsPackContext::SrsPackContext(ISrsPsPackHandler *handler)
     media_nn_recovered_ = 0;
     media_nn_msgs_dropped_ = 0;
     media_reserved_ = 0;
+}
 
+ISrsPackContext::~ISrsPackContext()
+{
+}
+
+SrsPackContext::SrsPackContext(ISrsPsPackHandler *handler)
+{
     ps_ = new SrsPsPacket(NULL);
     handler_ = handler;
 }
@@ -1391,13 +1462,23 @@ void SrsPackContext::on_recover_mode(int nn_recover)
     }
 }
 
+ISrsRecoverablePsContext::ISrsRecoverablePsContext()
+{
+}
+
+ISrsRecoverablePsContext::~ISrsRecoverablePsContext()
+{
+}
+
 SrsRecoverablePsContext::SrsRecoverablePsContext()
 {
     recover_ = 0;
+    ctx_ = new SrsPsContext();
 }
 
 SrsRecoverablePsContext::~SrsRecoverablePsContext()
 {
+    srs_freep(ctx_);
 }
 
 srs_error_t SrsRecoverablePsContext::decode_rtp(SrsBuffer *stream, int reserved, ISrsPsMessageHandler *handler)
@@ -1434,9 +1515,9 @@ srs_error_t SrsRecoverablePsContext::decode_rtp(SrsBuffer *stream, int reserved,
     SrsBuffer b((char *)rtp_raw->payload_, rtp_raw->nn_payload_);
     // srs_trace("GB: Got RTP length=%d, payload=%d, seq=%u, ts=%d", length, rtp_raw->nn_payload, rtp.header_.get_sequence(), rtp.header_.get_timestamp());
 
-    ctx_.helper_.rtp_seq_ = rtp.header_.get_sequence();
+    ctx_->helper()->rtp_seq_ = rtp.header_.get_sequence();
-    ctx_.helper_.rtp_ts_ = rtp.header_.get_timestamp();
+    ctx_->helper()->rtp_ts_ = rtp.header_.get_timestamp();
-    ctx_.helper_.rtp_pt_ = rtp.header_.get_payload_type();
+    ctx_->helper()->rtp_pt_ = rtp.header_.get_payload_type();
     if ((err = decode(&b, handler)) != srs_success) {
         return srs_error_wrap(err, "decode");
     }
@@ -1466,7 +1547,7 @@ srs_error_t SrsRecoverablePsContext::decode(SrsBuffer *stream, ISrsPsMessageHand
     }
 
     // Got packet to decode.
-    if ((err = ctx_.decode(stream, handler)) != srs_success) {
+    if ((err = ctx_->decode(stream, handler)) != srs_success) {
         return enter_recover_mode(stream, handler, stream->pos(), srs_error_wrap(err, "decode pack"));
     }
     return err;
@@ -1482,13 +1563,13 @@ srs_error_t SrsRecoverablePsContext::enter_recover_mode(SrsBuffer *stream, ISrsP
     stream->skip(pos - stream->pos());
     string bytes = srs_strings_dumps_hex(stream->head(), stream->left(), 8);
 
-    SrsPsDecodeHelper &h = ctx_.helper_;
+    SrsPsDecodeHelper *h = ctx_->helper();
-    uint16_t pack_seq = h.pack_first_seq_;
+    uint16_t pack_seq = h->pack_first_seq_;
-    uint16_t pack_msgs = h.pack_nn_msgs_;
+    uint16_t pack_msgs = h->pack_nn_msgs_;
-    uint16_t lsopm = h.pack_pre_msg_last_seq_;
+    uint16_t lsopm = h->pack_pre_msg_last_seq_;
-    SrsTsMessage *last = ctx_.last();
+    SrsTsMessage *last = ctx_->last();
     srs_warn("PS: Enter recover=%d, seq=%u, ts=%u, pt=%u, pack=%u, msgs=%u, lsopm=%u, last=%u/%u, bytes=[%s], pos=%d, left=%d for err %s",
-             recover_, h.rtp_seq_, h.rtp_ts_, h.rtp_pt_, pack_seq, pack_msgs, lsopm, last->PES_packet_length_, last->payload_->length(),
+             recover_, h->rtp_seq_, h->rtp_ts_, h->rtp_pt_, pack_seq, pack_msgs, lsopm, last->PES_packet_length_, last->payload_->length(),
              bytes.c_str(), npos, stream->left(), srs_error_desc(err).c_str());
 
     // If RTP packet exceed SRS_GB_LARGE_PACKET, which is large packet, might be correct length and impossible to
@@ -1500,11 +1581,11 @@ srs_error_t SrsRecoverablePsContext::enter_recover_mode(SrsBuffer *stream, ISrsP
     // Sometimes, we're unable to recover it, so we limit the max retry.
     if (recover_ > SRS_GB_MAX_RECOVER) {
         return srs_error_wrap(err, "exceed max recover, pack=%u, pack-seq=%u, seq=%u",
-                              h.pack_id_, h.pack_first_seq_, h.rtp_seq_);
+                              h->pack_id_, h->pack_first_seq_, h->rtp_seq_);
     }
 
     // Reap and dispose last incomplete message.
-    SrsTsMessage *msg = ctx_.reap();
+    SrsTsMessage *msg = ctx_->reap();
     srs_freep(msg);
     // Skip all left bytes in buffer, reset error because recovered.
     stream->skip(stream->left());
@@ -1519,7 +1600,7 @@ srs_error_t SrsRecoverablePsContext::enter_recover_mode(SrsBuffer *stream, ISrsP
 void SrsRecoverablePsContext::quit_recover_mode(SrsBuffer *stream, ISrsPsMessageHandler *handler)
 {
     string bytes = srs_strings_dumps_hex(stream->head(), stream->left(), 8);
-    srs_warn("PS: Quit recover=%d, seq=%u, bytes=[%s], pos=%d, left=%d", recover_, ctx_.helper_.rtp_seq_,
+    srs_warn("PS: Quit recover=%d, seq=%u, bytes=[%s], pos=%d, left=%d", recover_, ctx_->helper()->rtp_seq_,
              bytes.c_str(), stream->pos(), stream->left());
     recover_ = 0;
 }
@@ -1550,11 +1631,19 @@ bool srs_skip_util_pack(SrsBuffer *stream)
 SrsGoApiGbPublish::SrsGoApiGbPublish(SrsConfDirective *conf)
 {
     conf_ = conf->copy();
+
+    config_ = _srs_config;
+    gb_manager_ = _srs_gb_manager;
+    app_factory_ = _srs_app_factory;
 }
 
 SrsGoApiGbPublish::~SrsGoApiGbPublish()
 {
     srs_freep(conf_);
+
+    config_ = NULL;
+    gb_manager_ = NULL;
+    app_factory_ = NULL;
 }
 
 srs_error_t SrsGoApiGbPublish::serve_http(ISrsHttpResponseWriter *w, ISrsHttpMessage *r)
@@ -1615,7 +1704,7 @@ srs_error_t SrsGoApiGbPublish::do_serve_http(ISrsHttpResponseWriter *w, ISrsHttp
     }
 
     res->set("code", SrsJsonAny::integer(ERROR_SUCCESS));
-    int port = _srs_config->get_stream_caster_listen(conf_);
+    int port = config_->get_stream_caster_listen(conf_);
     res->set("port", SrsJsonAny::integer(port));
     res->set("is_tcp", SrsJsonAny::boolean(true)); // only tcp supported
 
@@ -1628,26 +1717,26 @@ srs_error_t SrsGoApiGbPublish::bind_session(std::string id, uint64_t ssrc)
 {
     srs_error_t err = srs_success;
 
-    SrsSharedResource<SrsGbSession> *session = NULL;
+    SrsSharedResource<ISrsGbSession> *session = NULL;
-    session = dynamic_cast<SrsSharedResource<SrsGbSession> *>(_srs_gb_manager->find_by_id(id));
+    session = dynamic_cast<SrsSharedResource<ISrsGbSession> *>(gb_manager_->find_by_id(id));
     if (session) {
         return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "stream already exists");
     }
 
-    session = dynamic_cast<SrsSharedResource<SrsGbSession> *>(_srs_gb_manager->find_by_fast_id(ssrc));
+    session = dynamic_cast<SrsSharedResource<ISrsGbSession> *>(gb_manager_->find_by_fast_id(ssrc));
     if (session) {
         return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "ssrc already exists");
     }
 
     // Create new GB session.
-    SrsGbSession *raw_session = new SrsGbSession();
+    ISrsGbSession *raw_session = app_factory_->create_gb_session();
     raw_session->setup(conf_);
 
-    session = new SrsSharedResource<SrsGbSession>(raw_session);
+    session = new SrsSharedResource<ISrsGbSession>(raw_session);
-    _srs_gb_manager->add_with_id(id, session);
+    gb_manager_->add_with_id(id, session);
-    _srs_gb_manager->add_with_fast_id(ssrc, session);
+    gb_manager_->add_with_fast_id(ssrc, session);
 
-    SrsExecutorCoroutine *executor = new SrsExecutorCoroutine(_srs_gb_manager, session, raw_session, raw_session);
+    SrsExecutorCoroutine *executor = new SrsExecutorCoroutine(gb_manager_, session, raw_session, raw_session);
     raw_session->setup_owner(session, executor, executor);
     raw_session->device_id_ = id;
 

trunk/src/app/srs_app_gb28181.hpp

@@ -24,11 +24,8 @@ class SrsTcpConnection;
 class ISrsCoroutine;
 class SrsPackContext;
 class SrsBuffer;
-
 class SrsGbSession;
-
 class SrsGbMediaTcpConn;
-
 class SrsAlonePithyPrint;
 class SrsGbMuxer;
 class SrsSimpleRtmpClient;
@@ -39,6 +36,27 @@ class SrsMediaPacket;
 class SrsPithyPrint;
 class SrsRawAacStream;
 class ISrsHttpServeMux;
+class ISrsGbMuxer;
+class ISrsGbSession;
+class ISrsPackContext;
+class ISrsMpegpsQueue;
+class ISrsPsContext;
+class ISrsListener;
+class ISrsProtocolReadWriter;
+class ISrsRawH264Stream;
+class ISrsRawHEVCStream;
+class ISrsRawAacStream;
+class ISrsPithyPrint;
+class ISrsBasicRtmpClient;
+class SrsTsMessage;
+class SrsPsPacket;
+class ISrsGbSession;
+class ISrsGbMediaTcpConn;
+class ISrsAppConfig;
+class ISrsApiServerOwner;
+class ISrsResourceManager;
+class ISrsAppFactory;
+class ISrsIpListener;
 
 // The state machine for GB session.
 // init:
@@ -55,6 +73,7 @@ enum SrsGbSessionState {
     SrsGbSessionStateEstablished,
 };
 std::string srs_gb_session_state(SrsGbSessionState state);
+std::string srs_gb_state(SrsGbSessionState ostate, SrsGbSessionState state);
 
 // For external SIP server mode, where SRS acts only as a media relay server
 //     1. SIP server POST request via HTTP API with stream ID and SSRC
@@ -72,6 +91,11 @@ std::string srs_gb_session_state(SrsGbSessionState state);
 //      {"port":9000, "is_tcp": true}
 class SrsGoApiGbPublish : public ISrsHttpHandler
 {
+private:
+    ISrsAppConfig *config_;
+    ISrsResourceManager *gb_manager_;
+    ISrsAppFactory *app_factory_;
+
 private:
     SrsConfDirective *conf_;
 
@@ -87,18 +111,42 @@ private:
     srs_error_t bind_session(std::string stream, uint64_t ssrc);
 };
 
+// The interface for GB session.
+class ISrsGbSession : public ISrsResource, public ISrsCoroutineHandler, public ISrsExecutorHandler
+{
+public:
+    std::string device_id_;
+
+public:
+    ISrsGbSession();
+    virtual ~ISrsGbSession();
+
+public:
+    // Initialize the GB session.
+    virtual void setup(SrsConfDirective *conf) = 0;
+    // Setup the owner, the wrapper is the shared ptr, the interruptable object is the coroutine, and the cid is the context id.
+    virtual void setup_owner(SrsSharedResource<ISrsGbSession> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid) = 0;
+    // Notice session to use current media connection.
+    virtual void on_media_transport(SrsSharedResource<ISrsGbMediaTcpConn> media) = 0;
+public:
+    virtual void on_ps_pack(ISrsPackContext *ctx, SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs) = 0;
+};
+
 // The main logic object for GB, the session.
 // Each session contains a media object, that are managed by session. This means session always
 // lives longer than media, and session will dispose media when session disposed. In another word,
 // media objects use directly pointer to session, while session use shared ptr.
-class SrsGbSession : public ISrsResource, public ISrsCoroutineHandler, public ISrsExecutorHandler
+class SrsGbSession : public ISrsGbSession
 {
+private:
+    ISrsAppConfig *config_;
+    
 private:
     SrsContextId cid_;
 
 private:
     // The shared resource which own this object, we should never free it because it's managed by shared ptr.
-    SrsSharedResource<SrsGbSession> *wrapper_;
+    SrsSharedResource<ISrsGbSession> *wrapper_;
     // The owner coroutine, allow user to interrupt the loop.
     ISrsInterruptable *owner_coroutine_;
     ISrsContextIdSetter *owner_cid_;
@@ -106,8 +154,8 @@ private:
 private:
     SrsGbSessionState state_;
 
-    SrsSharedResource<SrsGbMediaTcpConn> media_;
+    SrsSharedResource<ISrsGbMediaTcpConn> media_;
-    SrsGbMuxer *muxer_;
+    ISrsGbMuxer *muxer_;
 
 private:
     // When wait for media connecting, timeout if exceed.
@@ -148,17 +196,17 @@ public:
     // Initialize the GB session.
     void setup(SrsConfDirective *conf);
     // Setup the owner, the wrapper is the shared ptr, the interruptable object is the coroutine, and the cid is the context id.
-    void setup_owner(SrsSharedResource<SrsGbSession> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
+    void setup_owner(SrsSharedResource<ISrsGbSession> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
     // Interface ISrsExecutorHandler
 public:
     virtual void on_executor_done(ISrsInterruptable *executor);
 
 public:
     // When got a pack of messages.
-    void on_ps_pack(SrsPackContext *ctx, SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs);
+    void on_ps_pack(ISrsPackContext *ctx, SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs);
 
     // When got available media transport.
-    void on_media_transport(SrsSharedResource<SrsGbMediaTcpConn> media);
+    void on_media_transport(SrsSharedResource<ISrsGbMediaTcpConn> media);
 
     // Interface ISrsCoroutineHandler
 public:
@@ -176,12 +224,28 @@ public:
     virtual std::string desc();
 };
 
+// The listener for GB.
+class ISrsGbListener : public ISrsListener
+{
+public:
+    ISrsGbListener();
+    virtual ~ISrsGbListener();
+
+public:
+};
+
 // The Media listener for GB.
-class SrsGbListener : public ISrsListener, public ISrsTcpHandler
+class SrsGbListener : public ISrsGbListener, public ISrsTcpHandler
 {
+private:
+    ISrsAppConfig *config_;
+    ISrsApiServerOwner *api_server_owner_;
+    ISrsResourceManager *gb_manager_;
+    ISrsAppFactory *app_factory_;
+
 private:
     SrsConfDirective *conf_;
-    SrsTcpListener *media_listener_;
+    ISrsIpListener *media_listener_;
 
 public:
     SrsGbListener();
@@ -212,26 +276,50 @@ public:
     virtual srs_error_t on_ps_pack(SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs) = 0;
 };
 
-// A GB28181 TCP media connection, for PS stream.
+// The interface for GB media transport.
-class SrsGbMediaTcpConn : public ISrsResource, public ISrsCoroutineHandler, public ISrsPsPackHandler, public ISrsExecutorHandler
+class ISrsGbMediaTcpConn : public ISrsResource, public ISrsCoroutineHandler, public ISrsExecutorHandler
 {
+public:
+    ISrsGbMediaTcpConn();
+    virtual ~ISrsGbMediaTcpConn();
+
+public:
+    // Setup object, to keep empty constructor.
+    virtual void setup(srs_netfd_t stfd) = 0;
+    // Setup the owner, the wrapper is the shared ptr, the interruptable object is the coroutine, and the cid is the context id.
+    virtual void setup_owner(SrsSharedResource<ISrsGbMediaTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid) = 0;
+    // Whether media is connected.
+    virtual bool is_connected() = 0;
+    // Interrupt transport by session.
+    virtual void interrupt() = 0;
+    // Set the cid of all coroutines.
+    virtual void set_cid(const SrsContextId &cid) = 0;
+};
+
+// A GB28181 TCP media connection, for PS stream.
+class SrsGbMediaTcpConn : public ISrsGbMediaTcpConn, // It's a resource, coroutine handler, and executor handler.
+    public ISrsPsPackHandler
+{
+private:
+    ISrsResourceManager *gb_manager_;
+
 private:
     bool connected_;
     // The owner session object, note that we use the raw pointer and should never free it.
-    SrsGbSession *session_;
+    ISrsGbSession *session_;
     uint32_t nn_rtcp_;
 
 private:
     // The shared resource which own this object, we should never free it because it's managed by shared ptr.
-    SrsSharedResource<SrsGbMediaTcpConn> *wrapper_;
+    SrsSharedResource<ISrsGbMediaTcpConn> *wrapper_;
     // The owner coroutine, allow user to interrupt the loop.
     ISrsInterruptable *owner_coroutine_;
     ISrsContextIdSetter *owner_cid_;
     SrsContextId cid_;
 
 private:
-    SrsPackContext *pack_;
+    ISrsPackContext *pack_;
-    SrsTcpConnection *conn_;
+    ISrsProtocolReadWriter *conn_;
     uint8_t *buffer_;
 
 public:
@@ -242,7 +330,7 @@ public:
     // Setup object, to keep empty constructor.
     void setup(srs_netfd_t stfd);
     // Setup the owner, the wrapper is the shared ptr, the interruptable object is the coroutine, and the cid is the context id.
-    void setup_owner(SrsSharedResource<SrsGbMediaTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
+    void setup_owner(SrsSharedResource<ISrsGbMediaTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
     // Interface ISrsExecutorHandler
 public:
     virtual void on_executor_done(ISrsInterruptable *executor);
@@ -270,13 +358,25 @@ public:
 
 private:
     // Create session if no one, or bind to an existed session.
-    srs_error_t bind_session(uint32_t ssrc, SrsGbSession **psession);
+    srs_error_t bind_session(uint32_t ssrc, ISrsGbSession **psession);
+};
+
+// The interface for mpegps queue.
+class ISrsMpegpsQueue
+{
+public:
+    ISrsMpegpsQueue();
+    virtual ~ISrsMpegpsQueue();
+
+public:
+    virtual srs_error_t push(SrsMediaPacket *msg) = 0;
+    virtual SrsMediaPacket *dequeue() = 0;
 };
 
 // The queue for mpegts over udp to send packets.
 // For the aac in mpegts contains many flv packets in a pes packet,
 // we must recalc the timestamp.
-class SrsMpegpsQueue
+class SrsMpegpsQueue : public ISrsMpegpsQueue
 {
 private:
     // The key: dts, value: msg.
@@ -293,24 +393,39 @@ public:
     virtual SrsMediaPacket *dequeue();
 };
 
+// The interface for GB muxer.
+class ISrsGbMuxer
+{
+public:
+    ISrsGbMuxer();
+    virtual ~ISrsGbMuxer();
+
+public:
+    virtual void setup(std::string output) = 0;
+    virtual srs_error_t on_ts_message(SrsTsMessage *msg) = 0;
+};
+
 // Mux GB28181 to RTMP.
-class SrsGbMuxer
+class SrsGbMuxer : public ISrsGbMuxer
 {
 private:
-    // The owner session object, note that we use the raw pointer and should never free it.
+    ISrsAppFactory *app_factory_;
-    SrsGbSession *session_;
-    std::string output_;
-    SrsSimpleRtmpClient *sdk_;
 
 private:
-    SrsRawH264Stream *avc_;
+    // The owner session object, note that we use the raw pointer and should never free it.
+    ISrsGbSession *session_;
+    std::string output_;
+    ISrsBasicRtmpClient *sdk_;
+
+private:
+    ISrsRawH264Stream *avc_;
     std::string h264_sps_;
     bool h264_sps_changed_;
     std::string h264_pps_;
     bool h264_pps_changed_;
     bool h264_sps_pps_sent_;
 
-    SrsRawHEVCStream *hevc_;
+    ISrsRawHEVCStream *hevc_;
     bool vps_sps_pps_change_;
     std::string h265_vps_;
     std::string h265_sps_;
@@ -318,15 +433,15 @@ private:
     bool vps_sps_pps_sent_;
 
 private:
-    SrsRawAacStream *aac_;
+    ISrsRawAacStream *aac_;
     std::string aac_specific_config_;
 
 private:
-    SrsMpegpsQueue *queue_;
+    ISrsMpegpsQueue *queue_;
-    SrsPithyPrint *pprint_;
+    ISrsPithyPrint *pprint_;
 
 public:
-    SrsGbMuxer(SrsGbSession *session);
+    SrsGbMuxer(ISrsGbSession *session);
     virtual ~SrsGbMuxer();
 
 public:
@@ -352,11 +467,21 @@ private:
     virtual void close();
 };
 
-// Recoverable PS context for GB28181.
+// The interface for recoverable PS context.
-class SrsRecoverablePsContext
+class ISrsRecoverablePsContext
 {
 public:
-    SrsPsContext ctx_;
+    ISrsRecoverablePsContext();
+    virtual ~ISrsRecoverablePsContext();
+
+public:
+};
+
+// Recoverable PS context for GB28181.
+class SrsRecoverablePsContext : public ISrsRecoverablePsContext
+{
+public:
+    ISrsPsContext *ctx_;
 
 private:
     // If decoding error, enter the recover mode. Drop all left bytes util next pack header.
@@ -380,13 +505,8 @@ private:
     void quit_recover_mode(SrsBuffer *stream, ISrsPsMessageHandler *handler);
 };
 
-// The PS pack context, for GB28181 to process based on PS pack, which contains a video and audios messages. For large
+// The interface for PS pack context.
-// video frame, it might be split to multiple PES packets, which must be group to one video frame.
+class ISrsPackContext : public ISrsPsMessageHandler
-// Please note that a pack might contain multiple audio frames, so size of audio PES packet should not exceed 64KB,
-// which is limited by the 16 bits PES_packet_length.
-// We also correct the timestamp, or DTS/PTS of video frames, which might be 0 if more than one video PES packets in a
-// PS pack stream.
-class SrsPackContext : public ISrsPsMessageHandler
 {
 public:
     // Each media transport only use one context, so the context id is the media id.
@@ -396,6 +516,19 @@ public:
     uint64_t media_nn_msgs_dropped_;
     uint64_t media_reserved_;
 
+public:
+    ISrsPackContext();
+    virtual ~ISrsPackContext();
+};
+
+// The PS pack context, for GB28181 to process based on PS pack, which contains a video and audios messages. For large
+// video frame, it might be split to multiple PES packets, which must be group to one video frame.
+// Please note that a pack might contain multiple audio frames, so size of audio PES packet should not exceed 64KB,
+// which is limited by the 16 bits PES_packet_length.
+// We also correct the timestamp, or DTS/PTS of video frames, which might be 0 if more than one video PES packets in a
+// PS pack stream.
+class SrsPackContext : public ISrsPackContext
+{
 private:
     // To process a pack of TS/PS messages.
     ISrsPsPackHandler *handler_;
@@ -408,8 +541,9 @@ public:
     SrsPackContext(ISrsPsPackHandler *handler);
     virtual ~SrsPackContext();
 
-private:
+public:
     void clear();
+
     // Interface ISrsPsMessageHandler
 public:
     virtual srs_error_t on_ts_message(SrsTsMessage *msg);

trunk/src/app/srs_app_listener.cpp

@@ -64,6 +64,14 @@ ISrsListener::~ISrsListener()
 {
 }
 
+ISrsIpListener::ISrsIpListener()
+{
+}
+
+ISrsIpListener::~ISrsIpListener()
+{
+}
+
 ISrsTcpHandler::ISrsTcpHandler()
 {
 }
@@ -92,13 +100,13 @@ SrsUdpListener::~SrsUdpListener()
     srs_freepa(buf_);
 }
 
-SrsUdpListener *SrsUdpListener::set_label(const std::string &label)
+ISrsListener *SrsUdpListener::set_label(const std::string &label)
 {
     label_ = label;
     return this;
 }
 
-SrsUdpListener *SrsUdpListener::set_endpoint(const std::string &i, int p)
+ISrsListener *SrsUdpListener::set_endpoint(const std::string &i, int p)
 {
     ip_ = i;
     port_ = p;
@@ -250,20 +258,20 @@ SrsTcpListener::~SrsTcpListener()
     srs_close_stfd(lfd_);
 }
 
-SrsTcpListener *SrsTcpListener::set_label(const std::string &label)
+ISrsListener *SrsTcpListener::set_label(const std::string &label)
 {
     label_ = label;
     return this;
 }
 
-SrsTcpListener *SrsTcpListener::set_endpoint(const std::string &i, int p)
+ISrsListener *SrsTcpListener::set_endpoint(const std::string &i, int p)
 {
     ip_ = i;
     port_ = p;
     return this;
 }
 
-SrsTcpListener *SrsTcpListener::set_endpoint(const std::string &endpoint)
+ISrsListener *SrsTcpListener::set_endpoint(const std::string &endpoint)
 {
     std::string ip;
     int port_;
@@ -358,7 +366,7 @@ SrsMultipleTcpListeners::~SrsMultipleTcpListeners()
     }
 }
 
-SrsMultipleTcpListeners *SrsMultipleTcpListeners::set_label(const std::string &label)
+ISrsListener *SrsMultipleTcpListeners::set_label(const std::string &label)
 {
     for (vector<SrsTcpListener *>::iterator it = listeners_.begin(); it != listeners_.end(); ++it) {
         SrsTcpListener *l = *it;
@@ -368,6 +376,16 @@ SrsMultipleTcpListeners *SrsMultipleTcpListeners::set_label(const std::string &l
     return this;
 }
 
+ISrsListener *SrsMultipleTcpListeners::set_endpoint(const std::string &i, int p)
+{
+    for (vector<SrsTcpListener *>::iterator it = listeners_.begin(); it != listeners_.end(); ++it) {
+        SrsTcpListener *l = *it;
+        l->set_endpoint(i, p);
+    }
+
+    return this;
+}
+
 SrsMultipleTcpListeners *SrsMultipleTcpListeners::add(const std::vector<std::string> &endpoints)
 {
     for (int i = 0; i < (int)endpoints.size(); i++) {
@@ -376,7 +394,8 @@ SrsMultipleTcpListeners *SrsMultipleTcpListeners::add(const std::vector<std::str
         srs_net_split_for_listener(endpoints[i], ip, port);
 
         SrsTcpListener *l = new SrsTcpListener(this);
-        listeners_.push_back(l->set_endpoint(ip, port));
+        l->set_endpoint(ip, port);
+        listeners_.push_back(l);
     }
 
     return this;

trunk/src/app/srs_app_listener.hpp

@@ -64,6 +64,18 @@ public:
     virtual srs_error_t listen() = 0;
 };
 
+// The IP layer TCP/UDP listener.
+class ISrsIpListener : public ISrsListener
+{
+public:
+    ISrsIpListener();
+    virtual ~ISrsIpListener();
+
+public:
+    virtual ISrsListener *set_endpoint(const std::string &i, int p) = 0;
+    virtual ISrsListener *set_label(const std::string &label) = 0;
+};
+
 // The tcp connection handler.
 class ISrsTcpHandler
 {
@@ -77,7 +89,7 @@ public:
 };
 
 // Bind udp port, start thread to recv packet and handler it.
-class SrsUdpListener : public ISrsCoroutineHandler
+class SrsUdpListener : public ISrsCoroutineHandler, public ISrsIpListener
 {
 protected:
     std::string label_;
@@ -98,8 +110,8 @@ public:
     virtual ~SrsUdpListener();
 
 public:
-    SrsUdpListener *set_label(const std::string &label);
+    ISrsListener *set_label(const std::string &label);
-    SrsUdpListener *set_endpoint(const std::string &i, int p);
+    ISrsListener *set_endpoint(const std::string &i, int p);
 
 private:
     virtual int fd();
@@ -120,7 +132,7 @@ private:
 };
 
 // Bind and listen tcp port, use handler to process the client.
-class SrsTcpListener : public ISrsCoroutineHandler, public ISrsListener
+class SrsTcpListener : public ISrsCoroutineHandler, public ISrsIpListener
 {
 private:
     std::string label_;
@@ -137,9 +149,9 @@ public:
     virtual ~SrsTcpListener();
 
 public:
-    SrsTcpListener *set_label(const std::string &label);
+    ISrsListener *set_label(const std::string &label);
-    SrsTcpListener *set_endpoint(const std::string &i, int p);
+    ISrsListener *set_endpoint(const std::string &i, int p);
-    SrsTcpListener *set_endpoint(const std::string &endpoint);
+    ISrsListener *set_endpoint(const std::string &endpoint);
     int port();
 
 public:
@@ -154,7 +166,7 @@ private:
 };
 
 // Bind and listen tcp port, use handler to process the client.
-class SrsMultipleTcpListeners : public ISrsListener, public ISrsTcpHandler
+class SrsMultipleTcpListeners : public ISrsIpListener, public ISrsTcpHandler
 {
 private:
     ISrsTcpHandler *handler_;
@@ -165,7 +177,8 @@ public:
     virtual ~SrsMultipleTcpListeners();
 
 public:
-    SrsMultipleTcpListeners *set_label(const std::string &label);
+    ISrsListener *set_label(const std::string &label);
+    ISrsListener *set_endpoint(const std::string &i, int p);
     SrsMultipleTcpListeners *add(const std::vector<std::string> &endpoints);
 
 public:

trunk/src/app/srs_app_mpegts_udp.cpp

@@ -52,7 +52,8 @@ srs_error_t SrsUdpCasterListener::initialize(SrsConfDirective *conf)
         return srs_error_new(ERROR_STREAM_CASTER_PORT, "invalid port=%d", port);
     }
 
-    listener_->set_endpoint(srs_net_address_any(), port)->set_label("MPEGTS");
+    listener_->set_endpoint(srs_net_address_any(), port);
+    listener_->set_label("MPEGTS");
 
     if ((err = caster_->initialize(conf)) != srs_success) {
         return srs_error_wrap(err, "init caster port=%d", port);

trunk/src/app/srs_app_rtc_conn.cpp

@@ -1821,6 +1821,14 @@ void SrsRtcPublishStream::update_send_report_time(uint32_t ssrc, const SrsNtp &n
     }
 }
 
+ISrsRtcConnection::ISrsRtcConnection()
+{
+}
+
+ISrsRtcConnection::~ISrsRtcConnection()
+{
+}
+
 ISrsRtcConnectionNackTimerHandler::ISrsRtcConnectionNackTimerHandler()
 {
 }
@@ -2411,12 +2419,12 @@ void SrsRtcConnection::alive()
     last_stun_time_ = srs_time_now_cached();
 }
 
-SrsRtcUdpNetwork *SrsRtcConnection::udp()
+ISrsRtcNetwork *SrsRtcConnection::udp()
 {
     return networks_->udp();
 }
 
-SrsRtcTcpNetwork *SrsRtcConnection::tcp()
+ISrsRtcNetwork *SrsRtcConnection::tcp()
 {
     return networks_->tcp();
 }

trunk/src/app/srs_app_rtc_conn.hpp

@@ -552,16 +552,42 @@ public:
     virtual srs_error_t exec_rtc_async_work(ISrsAsyncCallTask *t) = 0;
 };
 
-// A RTC Peer Connection, SDP level object.
+// The interface for RTC connection.
-//
+class ISrsRtcConnection : public ISrsResource, // It's a resource.
-// For performance, we use non-public from resource,
-// see https://stackoverflow.com/questions/3747066/c-cannot-convert-from-base-a-to-derived-type-b-via-virtual-base-a
-class SrsRtcConnection : public ISrsResource, // It's a resource.
                          public ISrsDisposingHandler,
                          public ISrsExpire,
                          public ISrsRtcPacketSender,
                          public ISrsRtcPacketReceiver,
                          public ISrsRtcConnectionNackTimerHandler
+{
+public:
+    ISrsRtcConnection();
+    virtual ~ISrsRtcConnection();
+
+public:
+    // DTLS callbacks.
+    virtual srs_error_t on_dtls_handshake_done() = 0;
+    virtual srs_error_t on_dtls_alert(std::string type, std::string desc) = 0;
+    // RTP/RTCP packet handling.
+    virtual srs_error_t on_rtp_cipher(char *data, int nb_data) = 0;
+    virtual srs_error_t on_rtp_plaintext(char *data, int nb_data) = 0;
+    virtual srs_error_t on_rtcp(char *data, int nb_data) = 0;
+    // STUN binding request.
+    virtual srs_error_t on_binding_request(SrsStunPacket *r, std::string &ice_pwd) = 0;
+    // Network access.
+    virtual ISrsRtcNetwork *udp() = 0;
+    virtual ISrsRtcNetwork *tcp() = 0;
+    // Keep alive.
+    virtual void alive() = 0;
+    // Context switching.
+    virtual void switch_to_context() = 0;
+};
+
+// A RTC Peer Connection, SDP level object.
+//
+// For performance, we use non-public from resource,
+// see https://stackoverflow.com/questions/3747066/c-cannot-convert-from-base-a-to-derived-type-b-via-virtual-base-a
+class SrsRtcConnection : public ISrsRtcConnection
 {
     friend class SrsSecurityTransport;
 
@@ -698,8 +724,8 @@ public:
     void alive();
 
 public:
-    SrsRtcUdpNetwork *udp();
+    ISrsRtcNetwork *udp();
-    SrsRtcTcpNetwork *tcp();
+    ISrsRtcNetwork *tcp();
 
 public:
     // send rtcp

trunk/src/app/srs_app_rtc_network.cpp

@@ -36,7 +36,15 @@ extern bool srs_is_dtls(const uint8_t *data, size_t len);
 extern bool srs_is_rtp_or_rtcp(const uint8_t *data, size_t len);
 extern bool srs_is_rtcp(const uint8_t *data, size_t len);
 
-SrsRtcNetworks::SrsRtcNetworks(SrsRtcConnection *conn)
+ISrsRtcNetworks::ISrsRtcNetworks()
+{
+}
+
+ISrsRtcNetworks::~ISrsRtcNetworks()
+{
+}
+
+SrsRtcNetworks::SrsRtcNetworks(ISrsRtcConnection *conn)
 {
     conn_ = conn;
     delta_ = new SrsEphemeralDelta();
@@ -74,12 +82,12 @@ void SrsRtcNetworks::set_state(SrsRtcNetworkState state)
     tcp_->set_state(state);
 }
 
-SrsRtcUdpNetwork *SrsRtcNetworks::udp()
+ISrsRtcNetwork *SrsRtcNetworks::udp()
 {
     return udp_;
 }
 
-SrsRtcTcpNetwork *SrsRtcNetworks::tcp()
+ISrsRtcNetwork *SrsRtcNetworks::tcp()
 {
     return tcp_;
 }
@@ -117,6 +125,15 @@ SrsRtcDummyNetwork::~SrsRtcDummyNetwork()
 {
 }
 
+srs_error_t SrsRtcDummyNetwork::initialize(SrsSessionConfig *cfg, bool dtls, bool srtp)
+{
+    return srs_success;
+}
+
+void SrsRtcDummyNetwork::set_state(SrsRtcNetworkState state)
+{
+}
+
 bool SrsRtcDummyNetwork::is_establelished()
 {
     return true;
@@ -132,6 +149,11 @@ srs_error_t SrsRtcDummyNetwork::on_dtls_alert(std::string type, std::string desc
     return srs_success;
 }
 
+srs_error_t SrsRtcDummyNetwork::on_dtls(char *data, int nb_data)
+{
+    return srs_success;
+}
+
 srs_error_t SrsRtcDummyNetwork::protect_rtp(void *packet, int *nb_cipher)
 {
     return srs_success;
@@ -142,12 +164,27 @@ srs_error_t SrsRtcDummyNetwork::protect_rtcp(void *packet, int *nb_cipher)
     return srs_success;
 }
 
+srs_error_t SrsRtcDummyNetwork::on_stun(SrsStunPacket *r, char *data, int nb_data)
+{
+    return srs_success;
+}
+
+srs_error_t SrsRtcDummyNetwork::on_rtp(char *data, int nb_data)
+{
+    return srs_success;
+}
+
+srs_error_t SrsRtcDummyNetwork::on_rtcp(char *data, int nb_data)
+{
+    return srs_success;
+}
+
 srs_error_t SrsRtcDummyNetwork::write(void *buf, size_t size, ssize_t *nwrite)
 {
     return srs_success;
 }
 
-SrsRtcUdpNetwork::SrsRtcUdpNetwork(SrsRtcConnection *conn, SrsEphemeralDelta *delta)
+SrsRtcUdpNetwork::SrsRtcUdpNetwork(ISrsRtcConnection *conn, ISrsEphemeralDelta *delta)
 {
     state_ = SrsRtcNetworkStateInit;
     conn_ = conn;
@@ -155,6 +192,8 @@ SrsRtcUdpNetwork::SrsRtcUdpNetwork(SrsRtcConnection *conn, SrsEphemeralDelta *de
     sendonly_skt_ = NULL;
     pp_address_change_ = new SrsErrorPithyPrint();
     transport_ = new SrsSecurityTransport(this);
+
+    conn_manager_ = _srs_conn_manager;
 }
 
 SrsRtcUdpNetwork::~SrsRtcUdpNetwork()
@@ -171,6 +210,8 @@ SrsRtcUdpNetwork::~SrsRtcUdpNetwork()
     }
 
     srs_freep(pp_address_change_);
+
+    conn_manager_ = NULL;
 }
 
 srs_error_t SrsRtcUdpNetwork::initialize(SrsSessionConfig *cfg, bool dtls, bool srtp)
@@ -363,11 +404,11 @@ void SrsRtcUdpNetwork::update_sendonly_socket(SrsUdpMuxSocket *skt)
     // If no cache, build cache and setup the relations in connection.
     if (!addr_cache) {
         peer_addresses_[peer_id] = addr_cache = skt->copy_sendonly();
-        _srs_conn_manager->add_with_id(peer_id, conn_);
+        conn_manager_->add_with_id(peer_id, conn_);
 
         uint64_t fast_id = skt->fast_id();
         if (fast_id) {
-            _srs_conn_manager->add_with_fast_id(fast_id, conn_);
+            conn_manager_->add_with_fast_id(fast_id, conn_);
         }
     }
 
@@ -451,7 +492,7 @@ srs_error_t SrsRtcUdpNetwork::write(void *buf, size_t size, ssize_t *nwrite)
     return sendonly_skt_->sendto(buf, size, SRS_UTIME_NO_TIMEOUT);
 }
 
-SrsRtcTcpNetwork::SrsRtcTcpNetwork(SrsRtcConnection *conn, SrsEphemeralDelta *delta) : owner_(new SrsRtcTcpConn())
+SrsRtcTcpNetwork::SrsRtcTcpNetwork(ISrsRtcConnection *conn, ISrsEphemeralDelta *delta) : owner_(new SrsRtcTcpConn())
 {
     conn_ = conn;
     delta_ = delta;
@@ -718,6 +759,17 @@ void SrsRtcTcpConn::setup()
     pkt_ = NULL;
     delta_ = NULL;
     skt_ = NULL;
+
+    conn_manager_ = _srs_conn_manager;
+    stat_ = _srs_stat;
+}
+
+ISrsRtcTcpConn::ISrsRtcTcpConn()
+{
+}
+
+ISrsRtcTcpConn::~ISrsRtcTcpConn()
+{
 }
 
 SrsRtcTcpConn::SrsRtcTcpConn()
@@ -743,9 +795,12 @@ SrsRtcTcpConn::~SrsRtcTcpConn()
     srs_freepa(pkt_);
     srs_freep(delta_);
     srs_freep(skt_);
+
+    conn_manager_ = NULL;
+    stat_ = NULL;
 }
 
-void SrsRtcTcpConn::setup_owner(SrsSharedResource<SrsRtcTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid)
+void SrsRtcTcpConn::setup_owner(SrsSharedResource<ISrsRtcTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid)
 {
     wrapper_ = wrapper;
     owner_coroutine_ = owner_coroutine;
@@ -789,8 +844,8 @@ srs_error_t SrsRtcTcpConn::cycle()
     srs_error_t err = do_cycle();
 
     // Only stat the HTTP streaming clients, ignore all API clients.
-    _srs_stat->on_disconnect(get_id().c_str(), err);
+    stat_->on_disconnect(get_id().c_str(), err);
-    _srs_stat->kbps_add_delta(get_id().c_str(), delta_);
+    stat_->kbps_add_delta(get_id().c_str(), delta_);
 
     // Only remove session when network is established, because client might use other UDP network.
     if (session_ && session_->tcp()->is_establelished()) {
@@ -888,7 +943,7 @@ srs_error_t SrsRtcTcpConn::handshake()
     }
 
     srs_assert(!session_);
-    SrsRtcConnection *session = dynamic_cast<SrsRtcConnection *>(_srs_conn_manager->find_by_name(ping.get_username()));
+    ISrsRtcConnection *session = dynamic_cast<ISrsRtcConnection *>(conn_manager_->find_by_name(ping.get_username()));
     // TODO: FIXME: For ICE trickle, we may get STUN packets before SDP answer, so maybe should response it.
     if (!session) {
         return srs_error_new(ERROR_RTC_TCP_STUN, "no session, stun username=%s", ping.get_username().c_str());
@@ -910,12 +965,12 @@ srs_error_t SrsRtcTcpConn::handshake()
 
     // For each binding request, update the TCP socket.
     if (ping.is_binding_request()) {
-        session_->tcp()->update_sendonly_socket(skt_);
+        network->update_sendonly_socket(skt_);
-        session_->tcp()->set_peer_id(ip_, port_);
+        network->set_peer_id(ip_, port_);
     }
 
     // Use the session network to handle packet.
-    return session_->tcp()->on_stun(&ping, pkt_, npkt);
+    return network->on_stun(&ping, pkt_, npkt);
 }
 
 srs_error_t SrsRtcTcpConn::read_packet(char *pkt, int *nb_pkt)

trunk/src/app/srs_app_rtc_network.hpp

@@ -20,18 +20,23 @@
 class ISrsResourceManager;
 class ISrsCoroutine;
 class SrsNetworkDelta;
+class ISrsNetworkDelta;
 class SrsTcpConnection;
+class ISrsTcpConnection;
 class ISrsKbpsDelta;
 class SrsUdpMuxSocket;
 class SrsErrorPithyPrint;
 class ISrsRtcTransport;
 class SrsEphemeralDelta;
+class ISrsEphemeralDelta;
 class ISrsKbpsDelta;
 class SrsRtcUdpNetwork;
+class ISrsRtcUdpNetwork;
 class ISrsRtcNetwork;
 class SrsRtcTcpNetwork;
 class SrsRtcDummyNetwork;
 class SrsRtcTcpConn;
+class ISrsRtcTcpConn;
 
 // The network stat.
 enum SrsRtcNetworkState {
@@ -43,25 +48,35 @@ enum SrsRtcNetworkState {
     SrsRtcNetworkStateClosed = 5,
 };
 
+// The network manager interface.
+class ISrsRtcNetworks
+{
+public:
+    ISrsRtcNetworks();
+    virtual ~ISrsRtcNetworks();
+
+public:
+};
+
 // A group of networks, each has its own DTLS and SRTP context.
-class SrsRtcNetworks
+class SrsRtcNetworks : public ISrsRtcNetworks
 {
 private:
     // Network over UDP.
-    SrsRtcUdpNetwork *udp_;
+    ISrsRtcNetwork *udp_;
     // Network over TCP
-    SrsRtcTcpNetwork *tcp_;
+    ISrsRtcNetwork *tcp_;
     // Network over dummy
-    SrsRtcDummyNetwork *dummy_;
+    ISrsRtcNetwork *dummy_;
 
 private:
     // WebRTC session object.
-    SrsRtcConnection *conn_;
+    ISrsRtcConnection *conn_;
     // Delta object for statistics.
-    SrsEphemeralDelta *delta_;
+    ISrsEphemeralDelta *delta_;
 
 public:
-    SrsRtcNetworks(SrsRtcConnection *conn);
+    SrsRtcNetworks(ISrsRtcConnection *conn);
     virtual ~SrsRtcNetworks();
     // DTLS transport functions.
 public:
@@ -71,8 +86,8 @@ public:
     // Connection level state machine, for ARQ of UDP packets.
     void set_state(SrsRtcNetworkState state);
     // Get the UDP network object.
-    SrsRtcUdpNetwork *udp();
+    ISrsRtcNetwork *udp();
-    SrsRtcTcpNetwork *tcp();
+    ISrsRtcNetwork *tcp();
     // Get an available network.
     ISrsRtcNetwork *available();
 
@@ -88,11 +103,19 @@ public:
     ISrsRtcNetwork();
     virtual ~ISrsRtcNetwork();
 
+public:
+    // Initialize the network with DTLS and SRTP configuration.
+    virtual srs_error_t initialize(SrsSessionConfig *cfg, bool dtls, bool srtp) = 0;
+    // Set the network state.
+    virtual void set_state(SrsRtcNetworkState state) = 0;
+
 public:
     // Callback when DTLS connected.
     virtual srs_error_t on_dtls_handshake_done() = 0;
     // Callback when DTLS disconnected.
     virtual srs_error_t on_dtls_alert(std::string type, std::string desc) = 0;
+    // Handle DTLS data.
+    virtual srs_error_t on_dtls(char *data, int nb_data) = 0;
 
 public:
     // Protect RTP packet by SRTP context.
@@ -100,6 +123,14 @@ public:
     // Protect RTCP packet by SRTP context.
     virtual srs_error_t protect_rtcp(void *packet, int *nb_cipher) = 0;
 
+public:
+    // Handle STUN packet.
+    virtual srs_error_t on_stun(SrsStunPacket *r, char *data, int nb_data) = 0;
+    // Handle RTP packet.
+    virtual srs_error_t on_rtp(char *data, int nb_data) = 0;
+    // Handle RTCP packet.
+    virtual srs_error_t on_rtcp(char *data, int nb_data) = 0;
+
 public:
     virtual bool is_establelished() = 0;
 };
@@ -113,12 +144,22 @@ public:
 
     // The interface of ISrsRtcNetwork
 public:
+    virtual srs_error_t initialize(SrsSessionConfig *cfg, bool dtls, bool srtp);
+    virtual void set_state(SrsRtcNetworkState state);
     virtual srs_error_t on_dtls_handshake_done();
     virtual srs_error_t on_dtls_alert(std::string type, std::string desc);
+    virtual srs_error_t on_dtls(char *data, int nb_data);
 
 public:
     virtual srs_error_t protect_rtp(void *packet, int *nb_cipher);
     virtual srs_error_t protect_rtcp(void *packet, int *nb_cipher);
+
+public:
+    virtual srs_error_t on_stun(SrsStunPacket *r, char *data, int nb_data);
+    virtual srs_error_t on_rtp(char *data, int nb_data);
+    virtual srs_error_t on_rtcp(char *data, int nb_data);
+
+public:
     virtual bool is_establelished();
     // Interface ISrsStreamWriter.
 public:
@@ -128,11 +169,14 @@ public:
 // The WebRTC over UDP network.
 class SrsRtcUdpNetwork : public ISrsRtcNetwork
 {
+private:
+    ISrsResourceManager *conn_manager_;
+
 private:
     // WebRTC session object.
-    SrsRtcConnection *conn_;
+    ISrsRtcConnection *conn_;
     // Delta object for statistics.
-    SrsEphemeralDelta *delta_;
+    ISrsEphemeralDelta *delta_;
     SrsRtcNetworkState state_;
 
 private:
@@ -146,7 +190,7 @@ private:
     ISrsRtcTransport *transport_;
 
 public:
-    SrsRtcUdpNetwork(SrsRtcConnection *conn, SrsEphemeralDelta *delta);
+    SrsRtcUdpNetwork(ISrsRtcConnection *conn, ISrsEphemeralDelta *delta);
     virtual ~SrsRtcUdpNetwork();
 
 public:
@@ -185,14 +229,14 @@ public:
 class SrsRtcTcpNetwork : public ISrsRtcNetwork
 {
 private:
-    SrsRtcConnection *conn_;
+    ISrsRtcConnection *conn_;
-    SrsEphemeralDelta *delta_;
+    ISrsEphemeralDelta *delta_;
     ISrsProtocolReadWriter *sendonly_skt_;
 
 private:
     // The DTLS transport over this network.
     ISrsRtcTransport *transport_;
-    SrsSharedResource<SrsRtcTcpConn> owner_;
+    SrsSharedResource<ISrsRtcTcpConn> owner_;
 
 private:
     std::string peer_ip_;
@@ -200,12 +244,12 @@ private:
     SrsRtcNetworkState state_;
 
 public:
-    SrsRtcTcpNetwork(SrsRtcConnection *conn, SrsEphemeralDelta *delta);
+    SrsRtcTcpNetwork(ISrsRtcConnection *conn, ISrsEphemeralDelta *delta);
     virtual ~SrsRtcTcpNetwork();
 
 public:
-    void set_owner(SrsSharedResource<SrsRtcTcpConn> v) { owner_ = v; }
+    void set_owner(SrsSharedResource<ISrsRtcTcpConn> v) { owner_ = v; }
-    SrsSharedResource<SrsRtcTcpConn> owner() { return owner_; }
+    SrsSharedResource<ISrsRtcTcpConn> owner() { return owner_; }
     void update_sendonly_socket(ISrsProtocolReadWriter *skt);
     // ISrsRtcNetwork
 public:
@@ -248,12 +292,28 @@ public:
     void dispose();
 };
 
+// The interface for TCP connection.
+class ISrsRtcTcpConn : public ISrsConnection, public ISrsCoroutineHandler, public ISrsExecutorHandler
+{
+public:
+    ISrsRtcTcpConn();
+    virtual ~ISrsRtcTcpConn();
+
+public:
+    // Interrupt the TCP connection.
+    virtual void interrupt() = 0;
+};
+
 // For WebRTC over TCP.
-class SrsRtcTcpConn : public ISrsConnection, public ISrsCoroutineHandler, public ISrsExecutorHandler
+class SrsRtcTcpConn : public ISrsRtcTcpConn
 {
+private:
+    ISrsResourceManager *conn_manager_;
+    ISrsStatistic *stat_;
+
 private:
     // Because session references to this object, so we should directly use the session ptr.
-    SrsRtcConnection *session_;
+    ISrsRtcConnection *session_;
 
 private:
     // The ip and port of client.
@@ -267,7 +327,7 @@ private:
 
 private:
     // The shared resource which own this object, we should never free it because it's managed by shared ptr.
-    SrsSharedResource<SrsRtcTcpConn> *wrapper_;
+    SrsSharedResource<ISrsRtcTcpConn> *wrapper_;
     // The owner coroutine, allow user to interrupt the loop.
     ISrsInterruptable *owner_coroutine_;
     ISrsContextIdSetter *owner_cid_;
@@ -283,7 +343,7 @@ public:
 
 public:
     // Setup the owner, the wrapper is the shared ptr, the interruptable object is the coroutine, and the cid is the context id.
-    void setup_owner(SrsSharedResource<SrsRtcTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
+    void setup_owner(SrsSharedResource<ISrsRtcTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
 
 public:
     ISrsKbpsDelta *delta();

trunk/src/app/srs_app_rtc_server.cpp

@@ -609,7 +609,9 @@ srs_error_t SrsRtcSessionManager::on_udp_packet(SrsUdpMuxSocket *skt)
 
         // For each binding request, update the UDP socket.
         if (ping.is_binding_request()) {
-            session->udp()->update_sendonly_socket(skt);
+            SrsRtcUdpNetwork *udp_network = dynamic_cast<SrsRtcUdpNetwork *>(session->udp());
+            srs_assert(udp_network);
+            udp_network->update_sendonly_socket(skt);
         }
 
         return session->udp()->on_stun(&ping, data, size);

trunk/src/app/srs_app_server.cpp

@@ -150,6 +150,14 @@ ISrsSignalHandler::~ISrsSignalHandler()
 {
 }
 
+ISrsApiServerOwner::ISrsApiServerOwner()
+{
+}
+
+ISrsApiServerOwner::~ISrsApiServerOwner()
+{
+}
+
 SrsServer::SrsServer()
 {
     signal_reload_ = false;
@@ -270,7 +278,9 @@ SrsServer::~SrsServer()
     circuit_breaker_ = NULL;
     srt_sources_ = NULL;
     rtc_sources_ = NULL;
+#ifdef SRS_RTSP
     rtsp_sources_ = NULL;
+#endif
 #ifdef SRS_GB28181
     gb_manager_ = NULL;
 #endif
@@ -682,7 +692,8 @@ srs_error_t SrsServer::listen()
 
     // Create exporter server listener.
     if (config_->get_exporter_enabled()) {
-        exporter_listener_->set_endpoint(config_->get_exporter_listen())->set_label("Exporter-Server");
+        exporter_listener_->set_endpoint(config_->get_exporter_listen());
+        exporter_listener_->set_label("Exporter-Server");
         if ((err = exporter_listener_->listen()) != srs_success) {
             return srs_error_wrap(err, "exporter server listen");
         }
@@ -1558,7 +1569,7 @@ srs_error_t SrsServer::do_on_tcp_client(ISrsListener *listener, srs_netfd_t &stf
     // For RTC TCP connection, use resource executor to manage the resource.
     SrsRtcTcpConn *raw_conn = dynamic_cast<SrsRtcTcpConn *>(resource);
     if (raw_conn) {
-        SrsSharedResource<SrsRtcTcpConn> *conn = new SrsSharedResource<SrsRtcTcpConn>(raw_conn);
+        SrsSharedResource<ISrsRtcTcpConn> *conn = new SrsSharedResource<ISrsRtcTcpConn>(raw_conn);
         SrsExecutorCoroutine *executor = new SrsExecutorCoroutine(conn_manager_, conn, raw_conn, raw_conn);
         raw_conn->setup_owner(conn, executor, executor);
         if ((err = executor->start()) != srs_success) {

trunk/src/app/srs_app_server.hpp

@@ -84,6 +84,17 @@ public:
     virtual void on_signal(int signo) = 0;
 };
 
+// The API server owner interface.
+class ISrsApiServerOwner
+{
+public:
+    ISrsApiServerOwner();
+    virtual ~ISrsApiServerOwner();
+
+public:
+    virtual ISrsHttpServeMux *api_server() = 0;
+};
+
 // SrsServer is the main server class of SRS (Simple Realtime Server) that provides comprehensive
 // streaming media server functionality. It serves as the central orchestrator for all streaming
 // protocols and services in a single-threaded, coroutine-based architecture.
@@ -93,7 +104,8 @@ class SrsServer : public ISrsReloadHandler, // Reload framework for permormance
                   public ISrsHourGlassHandler,
                   public ISrsSrtClientHandler,
                   public ISrsUdpMuxHandler,
-                  public ISrsSignalHandler
+                  public ISrsSignalHandler,
+                  public ISrsApiServerOwner
 {
 private:
     ISrsAppConfig *config_;
@@ -104,7 +116,9 @@ private:
     ISrsCircuitBreaker *circuit_breaker_;
     ISrsSrtSourceManager *srt_sources_;
     ISrsRtcSourceManager *rtc_sources_;
+#ifdef SRS_RTSP
     ISrsRtspSourceManager *rtsp_sources_;
+#endif
 #ifdef SRS_GB28181
     ISrsResourceManager *gb_manager_;
 #endif

trunk/src/app/srs_app_st.hpp

@@ -196,7 +196,8 @@ public:
 //          srs_freep(executor);
 //          return err;
 //      }
-class SrsExecutorCoroutine : public ISrsResource, public ISrsStartable, public ISrsInterruptable, public ISrsContextIdSetter, public ISrsContextIdGetter, public ISrsCoroutineHandler
+class SrsExecutorCoroutine : public ISrsResource, // It's a resource.
+    public ISrsStartable, public ISrsInterruptable, public ISrsContextIdSetter, public ISrsContextIdGetter, public ISrsCoroutineHandler
 {
 private:
     ISrsResourceManager *manager_;

trunk/src/kernel/srs_kernel_pithy_print.cpp

@@ -171,6 +171,14 @@ bool SrsAlonePithyPrint::can_print()
 // The global stage manager for pithy print, multiple stages.
 SrsStageManager *_srs_stages = NULL;
 
+ISrsPithyPrint::ISrsPithyPrint()
+{
+}
+
+ISrsPithyPrint::~ISrsPithyPrint()
+{
+}
+
 SrsPithyPrint::SrsPithyPrint(int _stage_id)
 {
     stage_id_ = _stage_id;

trunk/src/kernel/srs_kernel_pithy_print.hpp

@@ -95,6 +95,19 @@ public:
     virtual bool can_print();
 };
 
+// The interface for pithy print.
+class ISrsPithyPrint
+{
+public:
+    ISrsPithyPrint();
+    virtual ~ISrsPithyPrint();
+
+public:
+    virtual void elapse() = 0;
+    virtual bool can_print() = 0;
+    virtual srs_utime_t age() = 0;
+};
+
 // The stage is used for a collection of object to do print,
 // the print time in a stage is constant and not changed,
 // that is, we always got one message to print every specified time.
@@ -112,7 +125,7 @@ public:
 //            }
 //            // read and write RTMP messages.
 //        }
-class SrsPithyPrint
+class SrsPithyPrint : public ISrsPithyPrint
 {
 private:
     int client_id_;

trunk/src/kernel/srs_kernel_ps.cpp

@@ -43,6 +43,14 @@ ISrsPsMessageHandler::~ISrsPsMessageHandler()
 {
 }
 
+ISrsPsContext::ISrsPsContext()
+{
+}
+
+ISrsPsContext::~ISrsPsContext()
+{
+}
+
 SrsPsContext::SrsPsContext()
 {
     last_ = NULL;
@@ -81,6 +89,11 @@ SrsTsMessage *SrsPsContext::reap()
     return msg;
 }
 
+SrsPsDecodeHelper* SrsPsContext::helper()
+{
+    return &helper_;
+}
+
 srs_error_t SrsPsContext::decode(SrsBuffer *stream, ISrsPsMessageHandler *handler)
 {
     srs_error_t err = srs_success;

trunk/src/kernel/srs_kernel_ps.hpp

@@ -57,8 +57,23 @@ public:
     virtual void on_recover_mode(int nn_recover) = 0;
 };
 
+// The interface for PS context.
+class ISrsPsContext
+{
+public:
+    ISrsPsContext();
+    virtual ~ISrsPsContext();
+
+public:
+    virtual SrsPsDecodeHelper* helper() = 0;
+    virtual void set_detect_ps_integrity(bool v) = 0;
+    virtual srs_error_t decode(SrsBuffer *stream, ISrsPsMessageHandler *handler) = 0;
+    virtual SrsTsMessage *last() = 0;
+    virtual SrsTsMessage *reap() = 0;
+};
+
 // The PS context, to process PS PES stream.
-class SrsPsContext
+class SrsPsContext : public ISrsPsContext
 {
 public:
     SrsPsDecodeHelper helper_;
@@ -87,6 +102,7 @@ public:
     SrsTsMessage *last();
     // Reap the last message and create a fresh one.
     SrsTsMessage *reap();
+    virtual SrsPsDecodeHelper* helper();
 
 public:
     // Feed with ts packets, decode as ts message, callback handler if got one ts message.

trunk/src/kernel/srs_kernel_resource.hpp

@@ -93,8 +93,18 @@ public:
 public:
     // Add a resource to the manager.
     virtual void add(ISrsResource *conn, bool *exists = NULL) = 0;
+    // Add a resource with string id to the manager.
+    virtual void add_with_id(const std::string &id, ISrsResource *conn) = 0;
+    // Add a resource with fast(int) id to the manager.
+    virtual void add_with_fast_id(uint64_t id, ISrsResource *conn) = 0;
     // Get resource at specified index.
     virtual ISrsResource *at(int index) = 0;
+    // Find resource by string id.
+    virtual ISrsResource *find_by_id(std::string id) = 0;
+    // Find resource by fast(int) id.
+    virtual ISrsResource *find_by_fast_id(uint64_t id) = 0;
+    // Find resource by name.
+    virtual ISrsResource *find_by_name(std::string name) = 0;
 
 public:
     // Remove then free the specified connection. Note that the manager always free c resource,

trunk/src/protocol/srs_protocol_raw_avc.cpp

@@ -39,6 +39,14 @@ bool srs_aac_startswith_adts(SrsBuffer *stream)
     return true;
 }
 
+ISrsRawH264Stream::ISrsRawH264Stream()
+{
+}
+
+ISrsRawH264Stream::~ISrsRawH264Stream()
+{
+}
+
 SrsRawH264Stream::SrsRawH264Stream()
 {
 }
@@ -284,6 +292,14 @@ srs_error_t SrsRawH264Stream::mux_avc2flv(string video, int8_t frame_type, int8_
     return err;
 }
 
+ISrsRawHEVCStream::ISrsRawHEVCStream()
+{
+}
+
+ISrsRawHEVCStream::~ISrsRawHEVCStream()
+{
+}
+
 SrsRawHEVCStream::SrsRawHEVCStream()
 {
 }
@@ -641,6 +657,14 @@ srs_error_t SrsRawHEVCStream::mux_hevc2flv_enhanced(std::string video, int8_t fr
     return err;
 }
 
+ISrsRawAacStream::ISrsRawAacStream()
+{
+}
+
+ISrsRawAacStream::~ISrsRawAacStream()
+{
+}
+
 SrsRawAacStream::SrsRawAacStream()
 {
 }

trunk/src/protocol/srs_protocol_raw_avc.hpp

@@ -15,8 +15,26 @@
 
 class SrsBuffer;
 
+// The interface for raw h.264 stream.
+class ISrsRawH264Stream
+{
+public:
+    ISrsRawH264Stream();
+    virtual ~ISrsRawH264Stream();
+
+public:
+    virtual srs_error_t annexb_demux(SrsBuffer *stream, char **pframe, int *pnb_frame) = 0;
+    virtual bool is_sps(char *frame, int nb_frame) = 0;
+    virtual bool is_pps(char *frame, int nb_frame) = 0;
+    virtual srs_error_t sps_demux(char *frame, int nb_frame, std::string &sps) = 0;
+    virtual srs_error_t pps_demux(char *frame, int nb_frame, std::string &pps) = 0;
+    virtual srs_error_t mux_sequence_header(std::string sps, std::string pps, std::string &sh) = 0;
+    virtual srs_error_t mux_ipb_frame(char *frame, int nb_frame, std::string &ibp) = 0;
+    virtual srs_error_t mux_avc2flv(std::string video, int8_t frame_type, int8_t avc_packet_type, uint32_t dts, uint32_t pts, char **flv, int *nb_flv) = 0;
+};
+
 // The raw h.264 stream, in annexb.
-class SrsRawH264Stream
+class SrsRawH264Stream : public ISrsRawH264Stream
 {
 public:
     SrsRawH264Stream();
@@ -55,8 +73,28 @@ public:
     virtual srs_error_t mux_avc2flv(std::string video, int8_t frame_type, int8_t avc_packet_type, uint32_t dts, uint32_t pts, char **flv, int *nb_flv);
 };
 
+// The interface for raw h.265 stream.
+class ISrsRawHEVCStream
+{
+public:
+    ISrsRawHEVCStream();
+    virtual ~ISrsRawHEVCStream();
+
+public:
+    virtual srs_error_t annexb_demux(SrsBuffer *stream, char **pframe, int *pnb_frame) = 0;
+    virtual bool is_sps(char *frame, int nb_frame) = 0;
+    virtual bool is_pps(char *frame, int nb_frame) = 0;
+    virtual bool is_vps(char *frame, int nb_frame) = 0;
+    virtual srs_error_t sps_demux(char *frame, int nb_frame, std::string &sps) = 0;
+    virtual srs_error_t pps_demux(char *frame, int nb_frame, std::string &pps) = 0;
+    virtual srs_error_t vps_demux(char *frame, int nb_frame, std::string &vps) = 0;
+    virtual srs_error_t mux_sequence_header(std::string vps, std::string sps, std::vector<std::string> &pps, std::string &sh) = 0;
+    virtual srs_error_t mux_ipb_frame(char *frame, int nb_frame, std::string &ibp) = 0;
+    virtual srs_error_t mux_hevc2flv(std::string video, int8_t frame_type, int8_t avc_packet_type, uint32_t dts, uint32_t pts, char **flv, int *nb_flv) = 0;
+};
+
 // The raw h.265 stream, in annexb.
-class SrsRawHEVCStream
+class SrsRawHEVCStream : public ISrsRawHEVCStream
 {
 public:
     SrsRawHEVCStream();
@@ -131,8 +169,21 @@ struct SrsRawAacStreamCodec {
     int8_t aac_packet_type_;
 };
 
+// The interface for raw aac stream.
+class ISrsRawAacStream
+{
+public:
+    ISrsRawAacStream();
+    virtual ~ISrsRawAacStream();
+
+public:
+    virtual srs_error_t adts_demux(SrsBuffer *stream, char **pframe, int *pnb_frame, SrsRawAacStreamCodec &codec) = 0;
+    virtual srs_error_t mux_sequence_header(SrsRawAacStreamCodec *codec, std::string &sh) = 0;
+    virtual srs_error_t mux_aac2flv(char *frame, int nb_frame, SrsRawAacStreamCodec *codec, uint32_t dts, char **flv, int *nb_flv) = 0;
+};
+
 // The raw aac stream, in adts.
-class SrsRawAacStream
+class SrsRawAacStream : public ISrsRawAacStream
 {
 public:
     SrsRawAacStream();

trunk/src/utest/srs_utest_app10.cpp

@@ -699,6 +699,14 @@ void MockConnectionManagerForResampleKbps::add(ISrsResource *conn, bool *exists)
     connections_.push_back(conn);
 }
 
+void MockConnectionManagerForResampleKbps::add_with_id(const std::string & /*id*/, ISrsResource * /*conn*/)
+{
+}
+
+void MockConnectionManagerForResampleKbps::add_with_fast_id(uint64_t /*id*/, ISrsResource * /*conn*/)
+{
+}
+
 ISrsResource *MockConnectionManagerForResampleKbps::at(int index)
 {
     if (index < 0 || index >= (int)connections_.size()) {
@@ -707,6 +715,21 @@ ISrsResource *MockConnectionManagerForResampleKbps::at(int index)
     return connections_[index];
 }
 
+ISrsResource *MockConnectionManagerForResampleKbps::find_by_id(std::string /*id*/)
+{
+    return NULL;
+}
+
+ISrsResource *MockConnectionManagerForResampleKbps::find_by_fast_id(uint64_t /*id*/)
+{
+    return NULL;
+}
+
+ISrsResource *MockConnectionManagerForResampleKbps::find_by_name(std::string /*name*/)
+{
+    return NULL;
+}
+
 void MockConnectionManagerForResampleKbps::remove(ISrsResource *c)
 {
 }
@@ -966,11 +989,34 @@ void MockConnectionManagerForConnectionLimit::add(ISrsResource *conn, bool *exis
 {
 }
 
+void MockConnectionManagerForConnectionLimit::add_with_id(const std::string & /*id*/, ISrsResource * /*conn*/)
+{
+}
+
+void MockConnectionManagerForConnectionLimit::add_with_fast_id(uint64_t /*id*/, ISrsResource * /*conn*/)
+{
+}
+
 ISrsResource *MockConnectionManagerForConnectionLimit::at(int index)
 {
     return NULL;
 }
 
+ISrsResource *MockConnectionManagerForConnectionLimit::find_by_id(std::string /*id*/)
+{
+    return NULL;
+}
+
+ISrsResource *MockConnectionManagerForConnectionLimit::find_by_fast_id(uint64_t /*id*/)
+{
+    return NULL;
+}
+
+ISrsResource *MockConnectionManagerForConnectionLimit::find_by_name(std::string /*name*/)
+{
+    return NULL;
+}
+
 void MockConnectionManagerForConnectionLimit::remove(ISrsResource *c)
 {
 }

trunk/src/utest/srs_utest_app10.hpp

@@ -231,7 +231,12 @@ public:
     virtual bool empty();
     virtual size_t size();
     virtual void add(ISrsResource *conn, bool *exists = NULL);
+    virtual void add_with_id(const std::string &id, ISrsResource *conn);
+    virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
     virtual ISrsResource *at(int index);
+    virtual ISrsResource *find_by_id(std::string id);
+    virtual ISrsResource *find_by_fast_id(uint64_t id);
+    virtual ISrsResource *find_by_name(std::string name);
     virtual void remove(ISrsResource *c);
     virtual void subscribe(ISrsDisposingHandler *h);
     virtual void unsubscribe(ISrsDisposingHandler *h);
@@ -300,7 +305,12 @@ public:
     virtual bool empty();
     virtual size_t size();
     virtual void add(ISrsResource *conn, bool *exists = NULL);
+    virtual void add_with_id(const std::string &id, ISrsResource *conn);
+    virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
     virtual ISrsResource *at(int index);
+    virtual ISrsResource *find_by_id(std::string id);
+    virtual ISrsResource *find_by_fast_id(uint64_t id);
+    virtual ISrsResource *find_by_name(std::string name);
     virtual void remove(ISrsResource *c);
     virtual void subscribe(ISrsDisposingHandler *h);
     virtual void unsubscribe(ISrsDisposingHandler *h);

trunk/src/utest/srs_utest_app13.cpp

@@ -3209,6 +3209,16 @@ ISrsDvrSegmenter *MockDvrAppFactory::create_dvr_mp4_segmenter()
     return segmenter;
 }
 
+ISrsGbMediaTcpConn *MockDvrAppFactory::create_gb_media_tcp_conn()
+{
+    return NULL;
+}
+
+ISrsGbSession *MockDvrAppFactory::create_gb_session()
+{
+    return NULL;
+}
+
 VOID TEST(DvrSegmenterTest, OpenTypicalScenario)
 {
     srs_error_t err;

trunk/src/utest/srs_utest_app13.hpp

@@ -635,6 +635,8 @@ public:
     virtual ISrsMp4Encoder *create_mp4_encoder();
     virtual ISrsDvrSegmenter *create_dvr_flv_segmenter();
     virtual ISrsDvrSegmenter *create_dvr_mp4_segmenter();
+    virtual ISrsGbMediaTcpConn *create_gb_media_tcp_conn();
+    virtual ISrsGbSession *create_gb_session();
 };
 
 // Mock ISrsDvrSegmenter for testing SrsDvrPlan

trunk/src/utest/srs_utest_app14.hpp

@@ -0,0 +1,458 @@
+//
+// Copyright (c) 2013-2025 The SRS Authors
+//
+// SPDX-License-Identifier: MIT
+//
+
+#ifndef SRS_UTEST_APP14_HPP
+#define SRS_UTEST_APP14_HPP
+
+/*
+#include <srs_utest_app14.hpp>
+*/
+#include <srs_utest.hpp>
+
+#include <srs_app_gb28181.hpp>
+#include <srs_app_config.hpp>
+#include <srs_app_server.hpp>
+#include <srs_app_factory.hpp>
+#include <srs_app_dvr.hpp>
+#include <srs_protocol_http_stack.hpp>
+#include <srs_protocol_rtmp_conn.hpp>
+#include <srs_protocol_raw_avc.hpp>
+#include <srs_protocol_http_client.hpp>
+#include <srs_utest_app6.hpp>
+#include <srs_utest_app11.hpp>
+
+#ifdef SRS_RTSP
+#include <srs_app_rtsp_conn.hpp>
+#endif
+
+// Mock ISrsGbMuxer for testing SrsGbSession
+class MockGbMuxer : public ISrsGbMuxer
+{
+public:
+    bool setup_called_;
+    std::string setup_output_;
+    bool on_ts_message_called_;
+    srs_error_t on_ts_message_error_;
+
+public:
+    MockGbMuxer();
+    virtual ~MockGbMuxer();
+
+public:
+    virtual void setup(std::string output);
+    virtual srs_error_t on_ts_message(SrsTsMessage *msg);
+    void reset();
+};
+
+// Mock ISrsAppConfig for testing SrsGbSession
+class MockAppConfigForGbSession : public MockAppConfig
+{
+public:
+    std::string stream_caster_output_;
+
+public:
+    MockAppConfigForGbSession();
+    virtual ~MockAppConfigForGbSession();
+
+public:
+    virtual std::string get_stream_caster_output(SrsConfDirective *conf);
+    void set_stream_caster_output(const std::string &output);
+};
+
+// Mock ISrsPackContext for testing SrsGbSession::on_ps_pack
+class MockPackContext : public ISrsPackContext
+{
+public:
+    MockPackContext();
+    virtual ~MockPackContext();
+
+public:
+    virtual srs_error_t on_ts_message(SrsTsMessage *msg);
+    virtual void on_recover_mode(int nn_recover);
+};
+
+// Mock ISrsGbMediaTcpConn for testing SrsGbSession::on_media_transport
+class MockGbMediaTcpConn : public ISrsGbMediaTcpConn
+{
+public:
+    bool set_cid_called_;
+    SrsContextId received_cid_;
+    bool is_connected_;
+
+public:
+    MockGbMediaTcpConn();
+    virtual ~MockGbMediaTcpConn();
+
+public:
+    virtual void setup(srs_netfd_t stfd);
+    virtual void setup_owner(SrsSharedResource<ISrsGbMediaTcpConn> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
+    virtual bool is_connected();
+    virtual void interrupt();
+    virtual void set_cid(const SrsContextId &cid);
+    virtual const SrsContextId &get_id();
+    virtual std::string desc();
+    virtual srs_error_t cycle();
+    virtual void on_executor_done(ISrsInterruptable *executor);
+    void reset();
+};
+
+// Mock ISrsIpListener for testing SrsGbListener::initialize
+class MockIpListener : public ISrsIpListener
+{
+public:
+    std::string endpoint_ip_;
+    int endpoint_port_;
+    std::string label_;
+    bool set_endpoint_called_;
+    bool set_label_called_;
+
+public:
+    MockIpListener();
+    virtual ~MockIpListener();
+
+public:
+    virtual ISrsListener *set_endpoint(const std::string &i, int p);
+    virtual ISrsListener *set_label(const std::string &label);
+    virtual srs_error_t listen();
+    void reset();
+};
+
+// Mock ISrsAppConfig for testing SrsGbListener::initialize
+class MockAppConfigForGbListener : public MockAppConfig
+{
+public:
+    int stream_caster_listen_port_;
+
+public:
+    MockAppConfigForGbListener();
+    virtual ~MockAppConfigForGbListener();
+
+public:
+    virtual int get_stream_caster_listen(SrsConfDirective *conf);
+};
+
+// Mock ISrsHttpServeMux for testing SrsGbListener::listen_api
+class MockHttpServeMuxForGbListener : public ISrsHttpServeMux
+{
+public:
+    bool handle_called_;
+    std::string handle_pattern_;
+    ISrsHttpHandler *handle_handler_;
+
+public:
+    MockHttpServeMuxForGbListener();
+    virtual ~MockHttpServeMuxForGbListener();
+
+public:
+    virtual srs_error_t handle(std::string pattern, ISrsHttpHandler *handler);
+    virtual srs_error_t serve_http(ISrsHttpResponseWriter *w, ISrsHttpMessage *r);
+    void reset();
+};
+
+// Mock ISrsApiServerOwner for testing SrsGbListener::listen_api
+class MockApiServerOwnerForGbListener : public ISrsApiServerOwner
+{
+public:
+    ISrsHttpServeMux *mux_;
+
+public:
+    MockApiServerOwnerForGbListener();
+    virtual ~MockApiServerOwnerForGbListener();
+
+public:
+    virtual ISrsHttpServeMux *api_server();
+};
+
+// Mock ISrsIpListener for testing SrsGbListener::listen
+class MockIpListenerForGbListen : public ISrsIpListener
+{
+public:
+    bool listen_called_;
+
+public:
+    MockIpListenerForGbListen();
+    virtual ~MockIpListenerForGbListen();
+
+public:
+    virtual ISrsListener *set_endpoint(const std::string &i, int p);
+    virtual ISrsListener *set_label(const std::string &label);
+    virtual srs_error_t listen();
+    void reset();
+};
+
+// Mock ISrsGbSession for testing SrsGbMediaTcpConn::on_ps_pack
+class MockGbSessionForMediaConn : public ISrsGbSession
+{
+public:
+    bool on_ps_pack_called_;
+    ISrsPackContext *received_pack_;
+    SrsPsPacket *received_ps_;
+    std::vector<SrsTsMessage *> received_msgs_;
+    bool on_media_transport_called_;
+    SrsSharedResource<ISrsGbMediaTcpConn> received_media_;
+
+public:
+    MockGbSessionForMediaConn();
+    virtual ~MockGbSessionForMediaConn();
+
+public:
+    virtual void setup(SrsConfDirective *conf);
+    virtual void setup_owner(SrsSharedResource<ISrsGbSession> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
+    virtual void on_media_transport(SrsSharedResource<ISrsGbMediaTcpConn> media);
+    virtual void on_ps_pack(ISrsPackContext *ctx, SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs);
+    virtual const SrsContextId &get_id();
+    virtual std::string desc();
+    virtual srs_error_t cycle();
+    virtual void on_executor_done(ISrsInterruptable *executor);
+    void reset();
+};
+
+// Mock ISrsResourceManager for testing SrsGbMediaTcpConn::bind_session
+class MockResourceManagerForBindSession : public ISrsResourceManager
+{
+public:
+    ISrsResource *session_to_return_;
+
+public:
+    MockResourceManagerForBindSession();
+    virtual ~MockResourceManagerForBindSession();
+
+public:
+    virtual srs_error_t start();
+    virtual bool empty();
+    virtual size_t size();
+    virtual void add(ISrsResource *conn, bool *exists = NULL);
+    virtual void add_with_id(const std::string &id, ISrsResource *conn);
+    virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
+    virtual ISrsResource *at(int index);
+    virtual ISrsResource *find_by_id(std::string id);
+    virtual ISrsResource *find_by_fast_id(uint64_t id);
+    virtual ISrsResource *find_by_name(std::string name);
+    virtual void remove(ISrsResource *c);
+    virtual void subscribe(ISrsDisposingHandler *h);
+    virtual void unsubscribe(ISrsDisposingHandler *h);
+    void reset();
+};
+
+// Mock ISrsBasicRtmpClient for testing SrsGbMuxer
+class MockGbRtmpClient : public ISrsBasicRtmpClient
+{
+public:
+    bool connect_called_;
+    bool publish_called_;
+    bool close_called_;
+    srs_error_t connect_error_;
+    srs_error_t publish_error_;
+    int stream_id_;
+
+public:
+    MockGbRtmpClient();
+    virtual ~MockGbRtmpClient();
+
+public:
+    virtual srs_error_t connect();
+    virtual void close();
+    virtual srs_error_t publish(int chunk_size, bool with_vhost = true, std::string *pstream = NULL);
+    virtual srs_error_t play(int chunk_size, bool with_vhost = true, std::string *pstream = NULL);
+    virtual void kbps_sample(const char *label, srs_utime_t age);
+    virtual int sid();
+    virtual srs_error_t recv_message(SrsRtmpCommonMessage **pmsg);
+    virtual srs_error_t decode_message(SrsRtmpCommonMessage *msg, SrsRtmpCommand **ppacket);
+    virtual srs_error_t send_and_free_messages(SrsMediaPacket **msgs, int nb_msgs);
+    virtual srs_error_t send_and_free_message(SrsMediaPacket *msg);
+    virtual void set_recv_timeout(srs_utime_t timeout);
+    void reset();
+};
+
+// Mock ISrsRawAacStream for testing SrsGbMuxer
+class MockGbRawAacStream : public ISrsRawAacStream
+{
+public:
+    bool adts_demux_called_;
+    bool mux_sequence_header_called_;
+    bool mux_aac2flv_called_;
+    srs_error_t adts_demux_error_;
+    srs_error_t mux_sequence_header_error_;
+    srs_error_t mux_aac2flv_error_;
+    std::string sequence_header_output_;
+    int demux_frame_size_;
+
+public:
+    MockGbRawAacStream();
+    virtual ~MockGbRawAacStream();
+
+public:
+    virtual srs_error_t adts_demux(SrsBuffer *stream, char **pframe, int *pnb_frame, SrsRawAacStreamCodec &codec);
+    virtual srs_error_t mux_sequence_header(SrsRawAacStreamCodec *codec, std::string &sh);
+    virtual srs_error_t mux_aac2flv(char *frame, int nb_frame, SrsRawAacStreamCodec *codec, uint32_t dts, char **flv, int *nb_flv);
+    void reset();
+};
+
+// Mock ISrsMpegpsQueue for testing SrsGbMuxer
+class MockGbMpegpsQueue : public ISrsMpegpsQueue
+{
+public:
+    bool push_called_;
+    srs_error_t push_error_;
+    int push_count_;
+
+public:
+    MockGbMpegpsQueue();
+    virtual ~MockGbMpegpsQueue();
+
+public:
+    virtual srs_error_t push(SrsMediaPacket *msg);
+    virtual SrsMediaPacket *dequeue();
+    void reset();
+};
+
+// Mock ISrsGbSession for testing SrsGbMuxer
+class MockGbSessionForMuxer : public ISrsGbSession
+{
+public:
+    std::string device_id_;
+
+public:
+    MockGbSessionForMuxer();
+    virtual ~MockGbSessionForMuxer();
+
+public:
+    virtual void setup(SrsConfDirective *conf);
+    virtual void setup_owner(SrsSharedResource<ISrsGbSession> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
+    virtual void on_media_transport(SrsSharedResource<ISrsGbMediaTcpConn> media);
+    virtual void on_ps_pack(ISrsPackContext *ctx, SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs);
+    virtual const SrsContextId &get_id();
+    virtual std::string desc();
+    virtual srs_error_t cycle();
+    virtual void on_executor_done(ISrsInterruptable *executor);
+};
+
+// Mock ISrsPsPackHandler for testing SrsPackContext
+class MockPsPackHandler : public ISrsPsPackHandler
+{
+public:
+    bool on_ps_pack_called_;
+    int on_ps_pack_count_;
+    uint32_t last_pack_id_;
+    int last_msgs_count_;
+    srs_error_t on_ps_pack_error_;
+
+public:
+    MockPsPackHandler();
+    virtual ~MockPsPackHandler();
+
+public:
+    virtual srs_error_t on_ps_pack(SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs);
+    void reset();
+};
+
+// Mock ISrsHttpMessage for testing SrsGoApiGbPublish
+class MockHttpMessageForGbPublish : public SrsHttpMessage
+{
+public:
+    std::string body_content_;
+    MockHttpConn *mock_conn_;
+
+public:
+    MockHttpMessageForGbPublish();
+    virtual ~MockHttpMessageForGbPublish();
+
+public:
+    virtual srs_error_t body_read_all(std::string &body);
+};
+
+// Mock ISrsResourceManager for testing SrsGoApiGbPublish
+class MockResourceManagerForGbPublish : public ISrsResourceManager
+{
+public:
+    std::map<std::string, ISrsResource*> id_map_;
+    std::map<uint64_t, ISrsResource*> fast_id_map_;
+
+public:
+    MockResourceManagerForGbPublish();
+    virtual ~MockResourceManagerForGbPublish();
+
+public:
+    virtual srs_error_t start();
+    virtual bool empty();
+    virtual size_t size();
+    virtual void add(ISrsResource *conn, bool *exists = NULL);
+    virtual void add_with_id(const std::string &id, ISrsResource *conn);
+    virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
+    virtual ISrsResource *at(int index);
+    virtual ISrsResource *find_by_id(std::string id);
+    virtual ISrsResource *find_by_fast_id(uint64_t id);
+    virtual ISrsResource *find_by_name(std::string name);
+    virtual void remove(ISrsResource *c);
+    virtual void subscribe(ISrsDisposingHandler *h);
+    virtual void unsubscribe(ISrsDisposingHandler *h);
+    void reset();
+};
+
+// Mock ISrsGbSession for testing SrsGoApiGbPublish
+class MockGbSessionForApiPublish : public ISrsGbSession
+{
+public:
+    bool setup_called_;
+    bool setup_owner_called_;
+    SrsConfDirective *setup_conf_;
+    ISrsInterruptable *owner_coroutine_;
+
+public:
+    MockGbSessionForApiPublish();
+    virtual ~MockGbSessionForApiPublish();
+
+public:
+    virtual void setup(SrsConfDirective *conf);
+    virtual void setup_owner(SrsSharedResource<ISrsGbSession> *wrapper, ISrsInterruptable *owner_coroutine, ISrsContextIdSetter *owner_cid);
+    virtual void on_media_transport(SrsSharedResource<ISrsGbMediaTcpConn> media);
+    virtual void on_ps_pack(ISrsPackContext *ctx, SrsPsPacket *ps, const std::vector<SrsTsMessage *> &msgs);
+    virtual const SrsContextId &get_id();
+    virtual std::string desc();
+    virtual srs_error_t cycle();
+    virtual void on_executor_done(ISrsInterruptable *executor);
+    void reset();
+};
+
+// Mock ISrsAppFactory for testing SrsGoApiGbPublish
+class MockAppFactoryForGbPublish : public ISrsAppFactory
+{
+public:
+    MockGbSessionForApiPublish *mock_gb_session_;
+
+public:
+    MockAppFactoryForGbPublish();
+    virtual ~MockAppFactoryForGbPublish();
+
+public:
+    virtual ISrsFileWriter *create_file_writer();
+    virtual ISrsFileWriter *create_enc_file_writer();
+    virtual ISrsFileReader *create_file_reader();
+    virtual SrsPath *create_path();
+    virtual SrsLiveSource *create_live_source();
+    virtual ISrsOriginHub *create_origin_hub();
+    virtual ISrsHourGlass *create_hourglass(const std::string &name, ISrsHourGlassHandler *handler, srs_utime_t interval);
+    virtual ISrsBasicRtmpClient *create_rtmp_client(std::string url, srs_utime_t cto, srs_utime_t sto);
+    virtual SrsHttpClient *create_http_client();
+    virtual ISrsHttpResponseReader *create_http_response_reader(ISrsHttpResponseReader *r);
+    virtual ISrsFileReader *create_http_file_reader(ISrsHttpResponseReader *r);
+    virtual ISrsFlvDecoder *create_flv_decoder();
+    virtual ISrsBasicRtmpClient *create_basic_rtmp_client(std::string url, srs_utime_t ctm, srs_utime_t stm);
+#ifdef SRS_RTSP
+    virtual ISrsRtspSendTrack *create_rtsp_audio_send_track(ISrsRtspConnection *session, SrsRtcTrackDescription *track_desc);
+    virtual ISrsRtspSendTrack *create_rtsp_video_send_track(ISrsRtspConnection *session, SrsRtcTrackDescription *track_desc);
+#endif
+    virtual ISrsFlvTransmuxer *create_flv_transmuxer();
+    virtual ISrsMp4Encoder *create_mp4_encoder();
+    virtual SrsDvrFlvSegmenter *create_dvr_flv_segmenter();
+    virtual SrsDvrMp4Segmenter *create_dvr_mp4_segmenter();
+    virtual ISrsGbMediaTcpConn *create_gb_media_tcp_conn();
+    virtual ISrsGbSession *create_gb_session();
+    void reset();
+};
+
+#endif
+

trunk/src/utest/srs_utest_app6.cpp

@@ -166,6 +166,15 @@ void MockRtcNetwork::reset()
     is_established_ = true;
 }
 
+srs_error_t MockRtcNetwork::initialize(SrsSessionConfig *cfg, bool dtls, bool srtp)
+{
+    return srs_success;
+}
+
+void MockRtcNetwork::set_state(SrsRtcNetworkState state)
+{
+}
+
 srs_error_t MockRtcNetwork::on_dtls_handshake_done()
 {
     on_dtls_handshake_done_count_++;
@@ -180,6 +189,11 @@ srs_error_t MockRtcNetwork::on_dtls_alert(std::string type, std::string desc)
     return srs_error_copy(on_dtls_alert_error_);
 }
 
+srs_error_t MockRtcNetwork::on_dtls(char *data, int nb_data)
+{
+    return srs_success;
+}
+
 srs_error_t MockRtcNetwork::protect_rtp(void *packet, int *nb_cipher)
 {
     protect_rtp_count_++;
@@ -192,6 +206,21 @@ srs_error_t MockRtcNetwork::protect_rtcp(void *packet, int *nb_cipher)
     return srs_error_copy(protect_rtcp_error_);
 }
 
+srs_error_t MockRtcNetwork::on_stun(SrsStunPacket *r, char *data, int nb_data)
+{
+    return srs_success;
+}
+
+srs_error_t MockRtcNetwork::on_rtp(char *data, int nb_data)
+{
+    return srs_success;
+}
+
+srs_error_t MockRtcNetwork::on_rtcp(char *data, int nb_data)
+{
+    return srs_success;
+}
+
 bool MockRtcNetwork::is_establelished()
 {
     return is_established_;

trunk/src/utest/srs_utest_app6.hpp

@@ -84,10 +84,16 @@ public:
     virtual ~MockRtcNetwork();
 
 public:
+    virtual srs_error_t initialize(SrsSessionConfig *cfg, bool dtls, bool srtp);
+    virtual void set_state(SrsRtcNetworkState state);
     virtual srs_error_t on_dtls_handshake_done();
     virtual srs_error_t on_dtls_alert(std::string type, std::string desc);
+    virtual srs_error_t on_dtls(char *data, int nb_data);
     virtual srs_error_t protect_rtp(void *packet, int *nb_cipher);
     virtual srs_error_t protect_rtcp(void *packet, int *nb_cipher);
+    virtual srs_error_t on_stun(SrsStunPacket *r, char *data, int nb_data);
+    virtual srs_error_t on_rtp(char *data, int nb_data);
+    virtual srs_error_t on_rtcp(char *data, int nb_data);
     virtual bool is_establelished();
     virtual srs_error_t write(void *buf, size_t size, ssize_t *nwrite);
 
@@ -291,6 +297,8 @@ public:
     virtual std::vector<SrsConfDirective *> get_stream_casters() { return std::vector<SrsConfDirective *>(); }
     virtual bool get_stream_caster_enabled(SrsConfDirective *conf) { return false; }
     virtual std::string get_stream_caster_engine(SrsConfDirective *conf) { return ""; }
+    virtual std::string get_stream_caster_output(SrsConfDirective *conf) { return ""; }
+    virtual int get_stream_caster_listen(SrsConfDirective *conf) { return 0; }
     virtual bool get_exporter_enabled() { return false; }
     virtual std::string get_exporter_listen() { return ""; }
     virtual std::string get_exporter_label() { return ""; }

trunk/src/utest/srs_utest_app7.cpp

@@ -1021,11 +1021,34 @@ void MockConnectionManagerForExpire::add(ISrsResource * /*conn*/, bool * /*exist
 {
 }
 
+void MockConnectionManagerForExpire::add_with_id(const std::string & /*id*/, ISrsResource * /*conn*/)
+{
+}
+
+void MockConnectionManagerForExpire::add_with_fast_id(uint64_t /*id*/, ISrsResource * /*conn*/)
+{
+}
+
 ISrsResource *MockConnectionManagerForExpire::at(int /*index*/)
 {
     return NULL;
 }
 
+ISrsResource *MockConnectionManagerForExpire::find_by_id(std::string /*id*/)
+{
+    return NULL;
+}
+
+ISrsResource *MockConnectionManagerForExpire::find_by_fast_id(uint64_t /*id*/)
+{
+    return NULL;
+}
+
+ISrsResource *MockConnectionManagerForExpire::find_by_name(std::string /*name*/)
+{
+    return NULL;
+}
+
 void MockConnectionManagerForExpire::remove(ISrsResource *c)
 {
     removed_resource_ = c;

trunk/src/utest/srs_utest_app7.hpp

@@ -105,7 +105,12 @@ public:
     virtual bool empty();
     virtual size_t size();
     virtual void add(ISrsResource *conn, bool *exists = NULL);
+    virtual void add_with_id(const std::string &id, ISrsResource *conn);
+    virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
     virtual ISrsResource *at(int index);
+    virtual ISrsResource *find_by_id(std::string id);
+    virtual ISrsResource *find_by_fast_id(uint64_t id);
+    virtual ISrsResource *find_by_name(std::string name);
     virtual void remove(ISrsResource *c);
     virtual void subscribe(ISrsDisposingHandler *h);
     virtual void unsubscribe(ISrsDisposingHandler *h);

trunk/src/utest/srs_utest_gb28181.cpp

@@ -47,7 +47,7 @@ VOID TEST(KernelPSTest, PsPacketDecodePartialPesHeader2)
     SrsRecoverablePsContext context;
 
     // Ignore if PS header is not integrity.
-    context.ctx_.set_detect_ps_integrity(true);
+    context.ctx_->set_detect_ps_integrity(true);
 
     // A PES packet with complete header, but without enough data.
     string raw = string("\x00\x00\x01\xc0\x00\x82\x8c\x80", 8);
@@ -445,7 +445,7 @@ VOID TEST(KernelPSTest, PsPacketDecodeInvalidStartCode)
         ASSERT_EQ((size_t)3, handler.msgs_.size());
         EXPECT_EQ(0, context.recover_);
 
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(65472, last->PES_packet_length_);
         ASSERT_EQ(1156, last->payload_->length());
     }
@@ -459,12 +459,12 @@ VOID TEST(KernelPSTest, PsPacketDecodeInvalidStartCode)
         ASSERT_EQ((size_t)3, handler.msgs_.size()); // We don't clear handler, so there must be 3 messages.
         EXPECT_EQ(0, context.recover_);
 
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(65472, last->PES_packet_length_);
         ASSERT_EQ(1156 + 1400 * (i + 1), last->payload_->length());
     }
     if (true) {
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(65472, last->PES_packet_length_);
         ASSERT_EQ(64156, last->payload_->length());
     }
@@ -487,7 +487,7 @@ VOID TEST(KernelPSTest, PsPacketDecodeInvalidStartCode)
         ASSERT_EQ((size_t)4, handler.msgs_.size());
         EXPECT_EQ(0, context.recover_);
 
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(65472, last->PES_packet_length_);
         ASSERT_EQ(72, last->payload_->length());
     }
@@ -501,12 +501,12 @@ VOID TEST(KernelPSTest, PsPacketDecodeInvalidStartCode)
         ASSERT_EQ((size_t)4, handler.msgs_.size()); // We don't clear handler, so there must be 4 messages.
         EXPECT_EQ(0, context.recover_);
 
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(65472, last->PES_packet_length_);
         ASSERT_EQ(72 + 1400 * (i + 1), last->payload_->length());
     }
     if (true) {
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(65472, last->PES_packet_length_);
         ASSERT_EQ(64472, last->payload_->length());
     }
@@ -526,7 +526,7 @@ VOID TEST(KernelPSTest, PsPacketDecodeInvalidStartCode)
         ASSERT_EQ((size_t)5, handler.msgs_.size());
         EXPECT_EQ(0, context.recover_);
 
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(10172, last->PES_packet_length_);
         ASSERT_EQ(388, last->payload_->length());
     }
@@ -540,12 +540,12 @@ VOID TEST(KernelPSTest, PsPacketDecodeInvalidStartCode)
         ASSERT_EQ((size_t)5, handler.msgs_.size()); // We don't clear handler, so there must be 5 messages.
         EXPECT_EQ(0, context.recover_);
 
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(10172, last->PES_packet_length_);
         ASSERT_EQ(388 + 1400 * (i + 1), last->payload_->length());
     }
     if (true) {
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(10172, last->PES_packet_length_);
         ASSERT_EQ(8788, last->payload_->length());
     }
@@ -564,7 +564,7 @@ VOID TEST(KernelPSTest, PsPacketDecodeInvalidStartCode)
         ASSERT_EQ((size_t)6, handler.msgs_.size());
         EXPECT_EQ(0, context.recover_);
 
-        SrsTsMessage *last = context.ctx_.last_;
+        SrsTsMessage *last = context.ctx_->last();
         ASSERT_EQ(96, last->PES_packet_length_);
         ASSERT_EQ(0, last->payload_->length());
     }

trunk/src/utest/srs_utest_service.cpp

@@ -1371,11 +1371,34 @@ void MockConnectionManager::add(ISrsResource * /*conn*/, bool * /*exists*/)
 {
 }
 
+void MockConnectionManager::add_with_id(const std::string & /*id*/, ISrsResource * /*conn*/)
+{
+}
+
+void MockConnectionManager::add_with_fast_id(uint64_t /*id*/, ISrsResource * /*conn*/)
+{
+}
+
 ISrsResource *MockConnectionManager::at(int /*index*/)
 {
     return NULL;
 }
 
+ISrsResource *MockConnectionManager::find_by_id(std::string /*id*/)
+{
+    return NULL;
+}
+
+ISrsResource *MockConnectionManager::find_by_fast_id(uint64_t /*id*/)
+{
+    return NULL;
+}
+
+ISrsResource *MockConnectionManager::find_by_name(std::string /*name*/)
+{
+    return NULL;
+}
+
 void MockConnectionManager::remove(ISrsResource * /*c*/)
 {
 }

trunk/src/utest/srs_utest_service.hpp

@@ -92,7 +92,12 @@ public:
     virtual bool empty();
     virtual size_t size();
     virtual void add(ISrsResource *conn, bool *exists = NULL);
+    virtual void add_with_id(const std::string &id, ISrsResource *conn);
+    virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
     virtual ISrsResource *at(int index);
+    virtual ISrsResource *find_by_id(std::string id);
+    virtual ISrsResource *find_by_fast_id(uint64_t id);
+    virtual ISrsResource *find_by_name(std::string name);
     virtual void remove(ISrsResource *c);
     virtual void subscribe(ISrsDisposingHandler *h);
     virtual void unsubscribe(ISrsDisposingHandler *h);