rtp发送启停支持触发观看人数统计事件

api/source/mk_media.cpp

@@ -266,7 +266,7 @@ API_EXPORT void API_CALL mk_media_start_send_rtp(mk_media ctx, const char *dst_u
     args.is_udp = is_udp;
 
     //sender参数无用
-    (*obj)->getChannel()->startSendRtp(*MediaSource::NullMediaSource, args, [cb, user_data](uint16_t local_port, const SockException &ex){
+    (*obj)->getChannel()->startSendRtp(MediaSource::NullMediaSource, args, [cb, user_data](uint16_t local_port, const SockException &ex){
         if (cb) {
             cb(user_data, local_port, ex.getErrCode(), ex.what());
         }
@@ -277,5 +277,5 @@ API_EXPORT int API_CALL mk_media_stop_send_rtp(mk_media ctx, const char *ssrc){
     assert(ctx);
     MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
     //sender参数无用
-    return (*obj)->getChannel()->stopSendRtp(*MediaSource::NullMediaSource, ssrc ? ssrc : "");
+    return (*obj)->getChannel()->stopSendRtp(MediaSource::NullMediaSource, ssrc ? ssrc : "");
 }

src/Common/MediaSource.cpp

@@ -44,15 +44,17 @@ string getOriginTypeString(MediaOriginType type){
 }
 
 static string getOriginUrl_l(const MediaSource *thiz) {
-    if (thiz == MediaSource::NullMediaSource) {
-        return "";
-    }
     return thiz->getSchema() + "://" + thiz->getVhost() + "/" + thiz->getApp() + "/" + thiz->getId();
 }
 
 //////////////////////////////////////////////////////////////////////////////////////////////////////////////
+struct MediaSourceNull : public MediaSource {
+    MediaSourceNull() : MediaSource("schema", "vhost", "app", "stream") {};
+    int readerCount() override { return 0; }
+};
 
-MediaSource * const MediaSource::NullMediaSource = nullptr;
+static std::shared_ptr<MediaSource> s_null = std::make_shared<MediaSourceNull>();
+MediaSource &MediaSource::NullMediaSource = *s_null;
 
 MediaSource::MediaSource(const string &schema, const string &vhost, const string &app, const string &stream_id){
     GET_CONFIG(bool, enableVhost, General::kEnableVhost);

src/Common/MediaSource.h

@@ -235,7 +235,7 @@ private:
  */
 class MediaSource: public TrackSource, public std::enable_shared_from_this<MediaSource> {
 public:
-    static MediaSource * const NullMediaSource;
+    static MediaSource& NullMediaSource;
     using Ptr = std::shared_ptr<MediaSource>;
     using StreamMap = std::unordered_map<std::string/*stream_id*/, std::weak_ptr<MediaSource> >;
     using AppStreamMap = std::unordered_map<std::string/*app*/, StreamMap>;

src/Common/MultiMediaSourceMuxer.cpp

@@ -78,7 +78,7 @@ static string getTrackInfoStr(const TrackSource *track_src){
 MultiMediaSourceMuxer::MultiMediaSourceMuxer(const string &vhost, const string &app, const string &stream, float dur_sec, const ProtocolOption &option) {
     _option = option;
     _get_origin_url = [this, vhost, app, stream]() {
-        auto ret = getOriginUrl(*MediaSource::NullMediaSource);
+        auto ret = getOriginUrl(MediaSource::NullMediaSource);
         if (!ret.empty()) {
             return ret;
         }
@@ -237,11 +237,14 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &, const MediaSourceEvent::
         auto ssrc = args.ssrc;
         rtp_sender->setOnClose([weak_self, ssrc]() {
             if (auto strong_self = weak_self.lock()) {
-                strong_self->_rtp_sender.erase(ssrc);
                 WarnL << "stream:" << strong_self->_get_origin_url() << " stop send rtp:" << ssrc;
+                strong_self->_rtp_sender.erase(ssrc);
+                //触发观看人数统计
+                strong_self->onReaderChanged(MediaSource::NullMediaSource, strong_self->totalReaderCount());
             }
         });
         strong_self->_rtp_sender[args.ssrc] = std::move(rtp_sender);
+        strong_self->onReaderChanged(MediaSource::NullMediaSource, strong_self->totalReaderCount());
     });
 #else
     cb(0, SockException(Err_other, "该功能未启用，编译时请打开ENABLE_RTPPROXY宏"));
@@ -250,13 +253,10 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &, const MediaSourceEvent::
 
 bool MultiMediaSourceMuxer::stopSendRtp(MediaSource &sender, const string &ssrc) {
 #if defined(ENABLE_RTPPROXY)
-    std::unique_ptr<onceToken> token;
+    onceToken token(nullptr, [&]() {
-    if (&sender != MediaSource::NullMediaSource) {
+        //关闭rtp推流，可能触发无人观看事件
-        token.reset(new onceToken(nullptr, [&]() {
+        onReaderChanged(sender, totalReaderCount());
-            //关闭rtp推流，可能触发无人观看事件
+    });
-            MediaSourceEventInterceptor::onReaderChanged(sender, totalReaderCount());
-        }));
-    }
     if (ssrc.empty()) {
         //关闭全部
         auto size = _rtp_sender.size();