Merge pull request #1 from xiongziliang/master

update

3rdpart/media-server

@@ -1 +1 @@
-Subproject commit 8d40dad3dbdce171756691d4511aca49fcf2a231
+Subproject commit 737b8d852eeb1a36bc77854f327fbbef7cfb81be

README.md

@@ -67,7 +67,7 @@ git submodule update --init
   - 支持http文件访问鉴权
 
 - GB28181
-  - 支持UDP/TCP国标RTP推流，可以转换成RTSP/RTMP/HLS等协议
+  - 支持UDP/TCP国标RTP(PS或TS)推流，可以转换成RTSP/RTMP/HLS等协议
  
 - 点播
   - 支持录制为FLV/HLS/MP4
@@ -363,10 +363,8 @@ docker build -t zlmediakit .
  - [IOS摄像头实时录制,生成rtsp/rtmp/hls/http-flv](https://gitee.com/xiahcu/IOSMedia)
  - [IOS rtmp/rtsp播放器，视频推流器](https://gitee.com/xiahcu/IOSPlayer)
  - [支持linux、windows、mac的rtmp/rtsp播放器](https://github.com/xiongziliang/ZLMediaPlayer)
-
+ - [配套的管理WEB网站](https://github.com/chenxiaolei/ZLMediaKit_NVR_UI)
-   上述工程可能在最新的代码的情况下编译不过，请手动修改
+ 
-
-
 ## 授权协议
 
 本项目自有代码使用宽松的MIT协议，在保留版权信息的情况下可以自由应用于各自商用、非商业的项目。

README_en.md

@@ -51,7 +51,7 @@
   - Auto close stream when nobody played.  
   - Play and push authentication.
   - Pull stream on Demand.
-
+  - Support TS / PS streaming push through RTP,and it can be converted to RTSP / RTMP / HLS / FLV.
 
 
 - Protocol conversion:

api/include/mk_events_objects.h

@@ -109,7 +109,20 @@ API_EXPORT const char* API_CALL mk_media_source_get_stream(const mk_media_source
 API_EXPORT int API_CALL mk_media_source_get_reader_count(const mk_media_source ctx);
 //MediaSource::totalReaderCount()
 API_EXPORT int API_CALL mk_media_source_get_total_reader_count(const mk_media_source ctx);
-//MediaSource::close()
+/**
+ * 直播源在ZLMediaKit中被称作为MediaSource，
+ * 目前支持3种，分别是RtmpMediaSource、RtspMediaSource、HlsMediaSource
+ * 源的产生有被动和主动方式:
+ * 被动方式分别是rtsp/rtmp/rtp推流、mp4点播
+ * 主动方式包括mk_media_create创建的对象(DevChannel)、mk_proxy_player_create创建的对象(PlayerProxy)
+ * 被动方式你不用做任何处理，ZLMediaKit已经默认适配了MediaSource::close()事件，都会关闭直播流
+ * 主动方式你要设置这个事件的回调，你要自己选择删除对象
+ * 通过mk_proxy_player_set_on_close、mk_media_set_on_close函数可以设置回调,
+ * 请在回调中删除对象来完成媒体的关闭，否则又为什么要调用mk_media_source_close函数？
+ * @param ctx 对象
+ * @param force 是否强制关闭，如果强制关闭，在有人观看的情况下也会关闭
+ * @return 0代表失败，1代表成功
+ */
 API_EXPORT int API_CALL mk_media_source_close(const mk_media_source ctx,int force);
 //MediaSource::seekTo()
 API_EXPORT int API_CALL mk_media_source_seek_to(const mk_media_source ctx,uint32_t stamp);
@@ -315,7 +328,7 @@ API_EXPORT mk_auth_invoker API_CALL mk_auth_invoker_clone(const mk_auth_invoker
 /**
  * 销毁堆上的克隆对象
  */
-API_EXPORT void API_CALL mk_auth_invoker_clone_relase(const mk_auth_invoker ctx);
+API_EXPORT void API_CALL mk_auth_invoker_clone_release(const mk_auth_invoker ctx);
 
 #ifdef __cplusplus
 }

api/include/mk_media.h

@@ -81,6 +81,14 @@ API_EXPORT void API_CALL mk_media_init_h265(mk_media ctx, int width, int height,
  */
 API_EXPORT void API_CALL mk_media_init_aac(mk_media ctx, int channel, int sample_bit, int sample_rate, int profile);
 
+/**
+ * 初始化h264/h265/aac完毕后调用此函数，
+ * 在单track(只有音频或视频)时，因为ZLMediaKit不知道后续是否还要添加track，所以会多等待3秒钟
+ * 如果产生的流是单Track类型，请调用此函数以便加快流生成速度，当然不调用该函数，影响也不大(会多等待3秒)
+ * @param ctx 对象指针
+ */
+API_EXPORT void API_CALL mk_media_init_complete(mk_media ctx);
+
 /**
  * 输入单帧H264视频，帧起始字节00 00 01,00 00 00 01均可
  * @param ctx 对象指针
@@ -121,6 +129,32 @@ API_EXPORT void API_CALL mk_media_input_aac(mk_media ctx, void *data, int len, u
  */
 API_EXPORT void API_CALL mk_media_input_aac1(mk_media ctx, void *data, int len, uint32_t dts, void *adts);
 
+/**
+ * MediaSource.close()回调事件
+ * 在选择关闭一个关联的MediaSource时，将会最终触发到该回调
+ * 你应该通过该事件调用mk_media_release函数并且释放其他资源
+ * 如果你不调用mk_media_release函数，那么MediaSource.close()操作将无效
+ * @param user_data 用户数据指针，通过mk_media_set_on_close函数设置
+ */
+typedef void(API_CALL *on_mk_media_close)(void *user_data);
+
+/**
+ * 监听MediaSource.close()事件
+ * 在选择关闭一个关联的MediaSource时，将会最终触发到该回调
+ * 你应该通过该事件调用mk_media_release函数并且释放其他资源
+ * @param ctx 对象指针
+ * @param cb 回调指针
+ * @param user_data 用户数据指针
+ */
+API_EXPORT void API_CALL mk_media_set_on_close(mk_media ctx, on_mk_media_close cb, void *user_data);
+
+/**
+ * 获取总的观看人数
+ * @param ctx 对象指针
+ * @return 观看人数
+ */
+API_EXPORT int API_CALL mk_media_total_reader_count(mk_media ctx);
+
 #ifdef __cplusplus
 }
 #endif

api/include/mk_player.h

@@ -114,6 +114,7 @@ API_EXPORT void API_CALL mk_player_set_on_shutdown(mk_player ctx, on_mk_play_eve
 
 /**
  * 设置音视频数据回调函数
+ * 该接口只能在播放成功事件触发后才能调用
  * @param ctx 播放器指针
  * @param cb 回调函数指针,不得为null
  * @param user_data 用户数据指针

api/include/mk_proxyplayer.h

@@ -68,6 +68,32 @@ API_EXPORT void API_CALL mk_proxy_player_set_option(mk_proxy_player ctx, const c
  */
 API_EXPORT void API_CALL mk_proxy_player_play(mk_proxy_player ctx, const char *url);
 
+/**
+ * MediaSource.close()回调事件
+ * 在选择关闭一个关联的MediaSource时，将会最终触发到该回调
+ * 你应该通过该事件调用mk_proxy_player_release函数并且释放其他资源
+ * 如果你不调用mk_proxy_player_release函数，那么MediaSource.close()操作将无效
+ * @param user_data 用户数据指针，通过mk_proxy_player_set_on_close函数设置
+ */
+typedef void(API_CALL *on_mk_proxy_player_close)(void *user_data);
+
+/**
+ * 监听MediaSource.close()事件
+ * 在选择关闭一个关联的MediaSource时，将会最终触发到该回调
+ * 你应该通过该事件调用mk_proxy_player_release函数并且释放其他资源
+ * @param ctx 对象指针
+ * @param cb 回调指针
+ * @param user_data 用户数据指针
+ */
+API_EXPORT void API_CALL mk_proxy_player_set_on_close(mk_proxy_player ctx, on_mk_proxy_player_close cb, void *user_data);
+
+/**
+ * 获取总的观看人数
+ * @param ctx 对象指针
+ * @return 观看人数
+ */
+API_EXPORT int API_CALL mk_proxy_player_total_reader_count(mk_proxy_player ctx);
+
 #ifdef __cplusplus
 }
 #endif

api/include/mk_tcp.h

@@ -120,7 +120,7 @@ API_EXPORT void* API_CALL mk_tcp_session_get_user_data(mk_tcp_session session);
  * @param port 监听端口号，0则为随机
  * @param type 服务器类型
  */
-API_EXPORT uint16_t API_CALL mk_tcp_server_server_start(uint16_t port, mk_tcp_type type);
+API_EXPORT uint16_t API_CALL mk_tcp_server_start(uint16_t port, mk_tcp_type type);
 
 /**
  * 监听tcp服务器事件

api/source/mk_events_objects.cpp

@@ -417,7 +417,7 @@ API_EXPORT mk_auth_invoker API_CALL mk_auth_invoker_clone(const mk_auth_invoker
     return new Broadcast::AuthInvoker(*invoker);
 }
 
-API_EXPORT void API_CALL mk_auth_invoker_clone_relase(const mk_auth_invoker ctx){
+API_EXPORT void API_CALL mk_auth_invoker_clone_release(const mk_auth_invoker ctx){
     assert(ctx);
     Broadcast::AuthInvoker *invoker = (Broadcast::AuthInvoker *)ctx;
     delete invoker;

api/source/mk_media.cpp

@@ -32,71 +32,148 @@ using namespace std;
 using namespace toolkit;
 using namespace mediakit;
 
+class MediaHelper : public MediaSourceEvent , public std::enable_shared_from_this<MediaHelper> {
+public:
+    typedef std::shared_ptr<MediaHelper> Ptr;
+    template<typename ...ArgsType>
+    MediaHelper(ArgsType &&...args){
+        _channel = std::make_shared<DevChannel>(std::forward<ArgsType>(args)...);
+    }
+    ~MediaHelper(){}
+
+    void attachEvent(){
+        _channel->setListener(shared_from_this());
+    }
+
+    DevChannel::Ptr &getChannel(){
+        return _channel;
+    }
+
+    void setCallBack(on_mk_media_close cb, void *user_data){
+        _cb = cb;
+        _user_data = user_data;
+    }
+protected:
+    // 通知其停止推流
+    bool close(MediaSource &sender,bool force) override{
+        if(!force && _channel->totalReaderCount()){
+            //非强制关闭且正有人在观看该视频
+            return false;
+        }
+        if(!_cb){
+            //未设置回调，没法关闭
+            WarnL << "请使用mk_media_set_on_close函数设置回调函数!";
+            return false;
+        }
+        //请在回调中调用mk_media_release函数释放资源,否则MediaSource::close()操作不会生效
+        _cb(_user_data);
+        WarnL << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" << sender.getApp() << "/" << sender.getId() << " " << force;
+        return true;
+    }
+
+    // 通知无人观看
+    void onNoneReader(MediaSource &sender) override{
+        if(_channel->totalReaderCount()){
+            //统计有误，还有人在看
+            return;
+        }
+        MediaSourceEvent::onNoneReader(sender);
+    }
+
+    // 观看总人数
+    int totalReaderCount(MediaSource &sender) override{
+        return _channel->totalReaderCount();
+    }
+private:
+    DevChannel::Ptr _channel;
+    on_mk_media_close _cb;
+    void *_user_data;
+};
+
+API_EXPORT void API_CALL mk_media_set_on_close(mk_media ctx, on_mk_media_close cb, void *user_data){
+    assert(ctx);
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
+    (*obj)->setCallBack(cb,user_data);
+}
+
+API_EXPORT int API_CALL mk_media_total_reader_count(mk_media ctx){
+    assert(ctx);
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
+    return (*obj)->getChannel()->totalReaderCount();
+}
+
 API_EXPORT mk_media API_CALL mk_media_create(const char *vhost, const char *app, const char *stream, float duration, int hls_enabled, int mp4_enabled) {
     assert(vhost && app && stream);
-    DevChannel::Ptr *obj(new DevChannel::Ptr(new DevChannel(vhost, app, stream, duration, true, true, hls_enabled, mp4_enabled)));
+    MediaHelper::Ptr *obj(new MediaHelper::Ptr(new MediaHelper(vhost, app, stream, duration, true, true, hls_enabled, mp4_enabled)));
+    (*obj)->attachEvent();
     return (mk_media) obj;
 }
 
 API_EXPORT void API_CALL mk_media_release(mk_media ctx) {
     assert(ctx);
-    DevChannel::Ptr *obj = (DevChannel::Ptr *) ctx;
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
     delete obj;
 }
 
 API_EXPORT void API_CALL mk_media_init_h264(mk_media ctx, int width, int height, int frameRate) {
     assert(ctx);
-    DevChannel::Ptr *obj = (DevChannel::Ptr *) ctx;
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
     VideoInfo info;
     info.iFrameRate = frameRate;
     info.iWidth = width;
     info.iHeight = height;
-    (*obj)->initVideo(info);
+    (*obj)->getChannel()->initVideo(info);
 }
 
 API_EXPORT void API_CALL mk_media_init_h265(mk_media ctx, int width, int height, int frameRate) {
     assert(ctx);
-    DevChannel::Ptr *obj = (DevChannel::Ptr *) ctx;
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
     VideoInfo info;
     info.iFrameRate = frameRate;
     info.iWidth = width;
     info.iHeight = height;
-    (*obj)->initH265Video(info);
+    (*obj)->getChannel()->initH265Video(info);
 }
 
 API_EXPORT void API_CALL mk_media_init_aac(mk_media ctx, int channel, int sample_bit, int sample_rate, int profile) {
     assert(ctx);
-    DevChannel::Ptr *obj = (DevChannel::Ptr *) ctx;
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
     AudioInfo info;
     info.iSampleRate = sample_rate;
     info.iChannel = channel;
     info.iSampleBit = sample_bit;
     info.iProfile = profile;
-    (*obj)->initAudio(info);
+    (*obj)->getChannel()->initAudio(info);
+}
+
+API_EXPORT void API_CALL mk_media_init_complete(mk_media ctx){
+    assert(ctx);
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
+    (*obj)->getChannel()->addTrackCompleted();
 }
 
 API_EXPORT void API_CALL mk_media_input_h264(mk_media ctx, void *data, int len, uint32_t dts, uint32_t pts) {
     assert(ctx && data && len > 0);
-    DevChannel::Ptr *obj = (DevChannel::Ptr *) ctx;
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
-    (*obj)->inputH264((char *) data, len, dts, pts);
+    (*obj)->getChannel()->inputH264((char *) data, len, dts, pts);
 }
 
 API_EXPORT void API_CALL mk_media_input_h265(mk_media ctx, void *data, int len, uint32_t dts, uint32_t pts) {
     assert(ctx && data && len > 0);
-    DevChannel::Ptr *obj = (DevChannel::Ptr *) ctx;
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
-    (*obj)->inputH265((char *) data, len, dts, pts);
+    (*obj)->getChannel()->inputH265((char *) data, len, dts, pts);
 }
 
 API_EXPORT void API_CALL mk_media_input_aac(mk_media ctx, void *data, int len, uint32_t dts, int with_adts_header) {
     assert(ctx && data && len > 0);
-    DevChannel::Ptr *obj = (DevChannel::Ptr *) ctx;
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
-    (*obj)->inputAAC((char *) data, len, dts, with_adts_header);
+    (*obj)->getChannel()->inputAAC((char *) data, len, dts, with_adts_header);
 }
 
 API_EXPORT void API_CALL mk_media_input_aac1(mk_media ctx, void *data, int len, uint32_t dts, void *adts) {
     assert(ctx && data && len > 0 && adts);
-    DevChannel::Ptr *obj = (DevChannel::Ptr *) ctx;
+    MediaHelper::Ptr *obj = (MediaHelper::Ptr *) ctx;
-    (*obj)->inputAAC((char *) data, len, dts, (char *) adts);
+    (*obj)->getChannel()->inputAAC((char *) data, len, dts, (char *) adts);
 }
 
 

api/source/mk_proxyplayer.cpp

@@ -61,3 +61,22 @@ API_EXPORT void API_CALL mk_proxy_player_play(mk_proxy_player ctx, const char *u
         obj->play(url_str);
     });
 }
+
+API_EXPORT void API_CALL mk_proxy_player_set_on_close(mk_proxy_player ctx, on_mk_proxy_player_close cb, void *user_data){
+    assert(ctx);
+    PlayerProxy::Ptr &obj = *((PlayerProxy::Ptr *) ctx);
+    obj->getPoller()->async([obj,cb,user_data](){
+        //切换线程再操作
+        obj->setOnClose([cb,user_data](){
+            if(cb){
+                cb(user_data);
+            }
+        });
+    });
+}
+
+API_EXPORT int API_CALL mk_proxy_player_total_reader_count(mk_proxy_player ctx){
+    assert(ctx);
+    PlayerProxy::Ptr &obj = *((PlayerProxy::Ptr *) ctx);
+    return obj->totalReaderCount();
+}

api/source/mk_tcp.cpp

@@ -137,7 +137,7 @@ API_EXPORT void API_CALL mk_tcp_server_events_listen(const mk_tcp_session_events
     }
 }
 
-API_EXPORT uint16_t API_CALL mk_tcp_server_server_start(uint16_t port, mk_tcp_type type){
+API_EXPORT uint16_t API_CALL mk_tcp_server_start(uint16_t port, mk_tcp_type type){
     type = MAX(mk_type_tcp, MIN(type, mk_type_wss));
     try {
         s_tcp_server[type] = std::make_shared<TcpServer>();

api/tests/websocket.c

@@ -172,7 +172,7 @@ void test_server(){
     };
 
     mk_tcp_server_events_listen(&events_server);
-    mk_tcp_server_server_start(80,TCP_TYPE);
+    mk_tcp_server_start(80, TCP_TYPE);
 }
 
 void test_client(){

conf/config.ini

@@ -168,8 +168,6 @@ checkSource=1
 dumpDir=
 #udp和tcp代理服务器，支持rtp(必须是ts或ps类型)代理
 port=10000
-#rtp如果是ts/ps类型则选择MP2P，还可以设置为MP4V-ES
-rtp_type=MP2P
 #rtp超时时间，单位秒
 timeoutSec=15
 

src/Common/MediaSink.cpp

@@ -64,19 +64,31 @@ void MediaSink::resetTracks() {
 
 void MediaSink::inputFrame(const Frame::Ptr &frame) {
     lock_guard<recursive_mutex> lck(_mtx);
-    auto codec_id = frame->getCodecId();
+    auto it = _track_map.find(frame->getCodecId());
-    auto it = _track_map.find(codec_id);
     if (it == _track_map.end()) {
         return;
     }
     it->second->inputFrame(frame);
+    checkTrackIfReady(it->second);
+}
 
-    if(!_allTrackReady && !_trackReadyCallback.empty() && it->second->ready()){
+void MediaSink::checkTrackIfReady_l(const Track::Ptr &track){
-        //Track由未就绪状态转换成就绪状态，我们就触发onTrackReady回调
+    //Track由未就绪状态转换成就绪状态，我们就触发onTrackReady回调
-        auto it_callback = _trackReadyCallback.find(codec_id);
+    auto it_callback = _trackReadyCallback.find(track->getCodecId());
-        if(it_callback != _trackReadyCallback.end()){
+    if (it_callback != _trackReadyCallback.end() && track->ready()) {
-            it_callback->second();
+        it_callback->second();
-            _trackReadyCallback.erase(it_callback);
+        _trackReadyCallback.erase(it_callback);
+    }
+}
+
+void MediaSink::checkTrackIfReady(const Track::Ptr &track){
+    if (!_allTrackReady && !_trackReadyCallback.empty()) {
+        if (track) {
+            checkTrackIfReady_l(track);
+        } else {
+            for (auto &pr : _track_map) {
+                checkTrackIfReady_l(pr.second);
+            }
         }
     }
 
@@ -107,8 +119,11 @@ void MediaSink::inputFrame(const Frame::Ptr &frame) {
 }
 
 void MediaSink::addTrackCompleted(){
-    lock_guard<recursive_mutex> lck(_mtx);
+    {
-    _max_track_size = _track_map.size();
+        lock_guard<recursive_mutex> lck(_mtx);
+        _max_track_size = _track_map.size();
+    }
+    checkTrackIfReady(nullptr);
 }
 
 void MediaSink::emitAllTrackReady() {
@@ -122,6 +137,7 @@ void MediaSink::emitAllTrackReady() {
         //移除未准备好的Track
         for (auto it = _track_map.begin(); it != _track_map.end();) {
             if (!it->second->ready()) {
+                WarnL << "该track长时间未被初始化,已忽略:" << it->second->getCodecName();
                 it = _track_map.erase(it);
                 continue;
             }

src/Common/MediaSink.h

@@ -121,6 +121,12 @@ private:
      * 触发onAllTrackReady事件
      */
     void emitAllTrackReady();
+
+    /**
+     * 检查track是否准备完毕
+     */
+    void checkTrackIfReady(const Track::Ptr &track);
+    void checkTrackIfReady_l(const Track::Ptr &track);
 private:
     mutable recursive_mutex _mtx;
     map<int,Track::Ptr> _track_map;

src/Common/MediaSource.cpp

@@ -253,22 +253,25 @@ MediaSource::Ptr MediaSource::find(const string &schema, const string &vhost_tmp
         vhost = DEFAULT_VHOST;
     }
 
-    lock_guard<recursive_mutex> lock(g_mtxMediaSrc);
     MediaSource::Ptr ret;
-	//查找某一媒体源，找到后返回
+    {
-    searchMedia(g_mapMediaSrc, schema, vhost, app, id, [&](SchemaVhostAppStreamMap::iterator &it0,
+        lock_guard<recursive_mutex> lock(g_mtxMediaSrc);
-                                                           VhostAppStreamMap::iterator &it1,
+        //查找某一媒体源，找到后返回
-                                                           AppStreamMap::iterator &it2,
+        searchMedia(g_mapMediaSrc, schema, vhost, app, id, [&](SchemaVhostAppStreamMap::iterator &it0,
-                                                           StreamMap::iterator &it3) {
+                                                               VhostAppStreamMap::iterator &it1,
-        ret = it3->second.lock();
+                                                               AppStreamMap::iterator &it2,
-        if (!ret) {
+                                                               StreamMap::iterator &it3) {
-            //该对象已经销毁
+            ret = it3->second.lock();
-            it2->second.erase(it3);
+            if (!ret) {
-            eraseIfEmpty(g_mapMediaSrc,it0, it1, it2);
+                //该对象已经销毁
-            return false;
+                it2->second.erase(it3);
-        }
+                eraseIfEmpty(g_mapMediaSrc, it0, it1, it2);
-        return true;
+                return false;
-    });
+            }
+            return true;
+        });
+    }
+
     if(!ret && bMake){
         //未查找媒体源，则创建一个
         ret = MP4Reader::onMakeMediaSource(schema, vhost,app,id);
@@ -288,29 +291,34 @@ void MediaSource::regist() {
     InfoL << _strSchema << " " << _strVhost << " " << _strApp << " " << _strId;
     NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, true, *this);
 }
-bool MediaSource::unregist() {
-    //反注册该源
-    lock_guard<recursive_mutex> lock(g_mtxMediaSrc);
-    return searchMedia(g_mapMediaSrc, _strSchema, _strVhost, _strApp, _strId,[&](SchemaVhostAppStreamMap::iterator &it0,
-                                                                                 VhostAppStreamMap::iterator &it1,
-                                                                                 AppStreamMap::iterator &it2,
-                                                                                 StreamMap::iterator &it3) {
-        auto strongMedia = it3->second.lock();
-        if (strongMedia && this != strongMedia.get()) {
-            //不是自己,不允许反注册
-            return false;
-        }
-        it2->second.erase(it3);
-        eraseIfEmpty(g_mapMediaSrc, it0, it1, it2);
-        unregisted();
-        return true;
-    });
-}
-void MediaSource::unregisted(){
-    InfoL <<  "" <<  _strSchema << " " << _strVhost << " " << _strApp << " " << _strId;
-    NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, false, *this);
-}
 
+//反注册该源
+bool MediaSource::unregist() {
+    bool ret;
+    {
+        lock_guard<recursive_mutex> lock(g_mtxMediaSrc);
+        ret = searchMedia(g_mapMediaSrc, _strSchema, _strVhost, _strApp, _strId,
+                          [&](SchemaVhostAppStreamMap::iterator &it0,
+                              VhostAppStreamMap::iterator &it1,
+                              AppStreamMap::iterator &it2,
+                              StreamMap::iterator &it3) {
+                              auto strongMedia = it3->second.lock();
+                              if (strongMedia && this != strongMedia.get()) {
+                                  //不是自己,不允许反注册
+                                  return false;
+                              }
+                              it2->second.erase(it3);
+                              eraseIfEmpty(g_mapMediaSrc, it0, it1, it2);
+                              return true;
+                          });
+    }
+
+    if(ret){
+        InfoL <<  "" <<  _strSchema << " " << _strVhost << " " << _strApp << " " << _strId;
+        NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, false, *this);
+    }
+    return ret;
+}
 
 /////////////////////////////////////MediaInfo//////////////////////////////////////
 
@@ -332,6 +340,12 @@ void MediaInfo::parse(const string &url){
         } else{
             _host = _vhost = vhost;
         }
+
+        if(_vhost == "localhost" || INADDR_NONE != inet_addr(_vhost.data())){
+            //如果访问的是localhost或ip，那么则为默认虚拟主机
+            _vhost = DEFAULT_VHOST;
+        }
+
     }
     if(split_vec.size() > 1){
         _app = split_vec[1];
@@ -358,7 +372,8 @@ void MediaInfo::parse(const string &url){
     }
 
     GET_CONFIG(bool,enableVhost,General::kEnableVhost);
-    if(!enableVhost || _vhost.empty() || _vhost == "localhost" || INADDR_NONE != inet_addr(_vhost.data())){
+    if(!enableVhost || _vhost.empty()){
+        //如果关闭虚拟主机或者虚拟主机为空，则设置虚拟主机为默认
         _vhost = DEFAULT_VHOST;
     }
 }

src/Common/MediaSource.h

@@ -151,7 +151,6 @@ public:
 protected:
     void regist() ;
     bool unregist() ;
-    void unregisted();
 private:
     string _strSchema;
     string _strVhost;

src/Common/MultiMediaSourceMuxer.h

@@ -33,6 +33,10 @@
 #include "Record/HlsMediaSource.h"
 #include "Record/HlsRecorder.h"
 
+/**
+ * 使用该对象时，应该使用setListener方法来绑定MediaSource相关的事件
+ * 否则多种不同类型的MediaSource(rtsp/rtmp/hls)将无法产生关联
+ */
 class MultiMediaSourceMuxer : public MediaSink , public std::enable_shared_from_this<MultiMediaSourceMuxer>{
 public:
     class Listener{

src/Common/config.cpp

@@ -282,15 +282,12 @@ namespace RtpProxy {
 const string kDumpDir = RTP_PROXY_FIELD"dumpDir";
 //是否限制udp数据来源ip和端口
 const string kCheckSource = RTP_PROXY_FIELD"checkSource";
-//rtp类型，支持MP2P/MP4V-ES
-const string kRtpType = RTP_PROXY_FIELD"rtp_type";
 //rtp接收超时时间
 const string kTimeoutSec = RTP_PROXY_FIELD"timeoutSec";
 
 onceToken token([](){
 	mINI::Instance()[kDumpDir] = "";
 	mINI::Instance()[kCheckSource] = 1;
-	mINI::Instance()[kRtpType] = "MP2P";
 	mINI::Instance()[kTimeoutSec] = 15;
 },nullptr);
 } //namespace RtpProxy

src/Common/config.h

@@ -306,8 +306,6 @@ namespace RtpProxy {
 extern const string kDumpDir;
 //是否限制udp数据来源ip和端口
 extern const string kCheckSource;
-//rtp类型，支持MP2P/MP4V-ES
-extern const string kRtpType;
 //rtp接收超时时间
 extern const string kTimeoutSec;
 } //namespace RtpProxy

src/Extension/AACRtp.cpp

@@ -24,6 +24,7 @@
  * SOFTWARE.
  */
 #include "AACRtp.h"
+#define ADTS_HEADER_LEN 7
 
 namespace mediakit{
 
@@ -91,8 +92,8 @@ AACRtpDecoder::AACRtpDecoder() {
 AACFrame::Ptr AACRtpDecoder::obtainFrame() {
     //从缓存池重新申请对象，防止覆盖已经写入环形缓存的对象
     auto frame = ResourcePoolHelper<AACFrame>::obtainObj();
-    frame->aac_frame_length = 7;
+    frame->aac_frame_length = ADTS_HEADER_LEN;
-    frame->iPrefixSize = 7;
+    frame->iPrefixSize = ADTS_HEADER_LEN;
     if(frame->syncword == 0 && !_aac_cfg.empty()) {
         makeAdtsHeader(_aac_cfg,*frame);
     }
@@ -100,70 +101,48 @@ AACFrame::Ptr AACRtpDecoder::obtainFrame() {
 }
 
 bool AACRtpDecoder::inputRtp(const RtpPacket::Ptr &rtppack, bool key_pos) {
-	// 获取rtp数据长度
+    //rtp数据开始部分
-    int length = rtppack->size() - rtppack->offset;
+    uint8_t *ptr = (uint8_t *) rtppack->data() + rtppack->offset;
+    //rtp数据末尾
+    const uint8_t *end = (uint8_t *) rtppack->data() + rtppack->size();
 
-	// 获取rtp数据
+    //首2字节表示Au-Header的个数，单位bit，所以除以16得到Au-Header个数
-	const uint8_t *rtp_packet_buf = (uint8_t *)rtppack->data() + rtppack->offset;
+    const uint16_t au_header_count = ((ptr[0] << 8) | ptr[1]) >> 4;
-	
+    //忽略Au-Header区
-	do
+    ptr += 2 + au_header_count * 2;
-	{
-		// 查询头部的偏移，每次2字节
-		uint32_t au_header_offset = 0;
-		//首2字节表示Au-Header的长度，单位bit，所以除以16得到Au-Header字节数
-		const uint16_t au_header_length = (((rtp_packet_buf[au_header_offset] << 8) | rtp_packet_buf[au_header_offset + 1]) >> 4);
-		au_header_offset += 2;
-		
-		//assert(length > (2 + au_header_length * 2));
-		if (length < (2 + au_header_length * 2))
-			break;
-		
-		// 存放每一个aac帧长度
-		std::vector<uint32_t > vec_aac_len;
-		for (int i = 0; i < au_header_length; ++i)
-		{
-			// 之后的2字节是AU_HEADER
-			const uint16_t au_header = ((rtp_packet_buf[au_header_offset] << 8) | rtp_packet_buf[au_header_offset + 1]);
-			// 其中高13位表示一帧AAC负载的字节长度，低3位无用
-			uint32_t nAac = (au_header >> 3);
-			vec_aac_len.push_back(nAac);
-			au_header_offset += 2;
-		}
 
-		// 真正aac负载开始处
+    static const uint32_t max_size = sizeof(AACFrame::buffer) - ADTS_HEADER_LEN;
-		const uint8_t *rtp_packet_payload = rtp_packet_buf + au_header_offset;
+    while (ptr < end) {
-		// 载荷查找
+        auto size = (uint32_t) (end - ptr);
-		uint32_t next_aac_payload_offset = 0;
+        if(size > max_size){
-		for (int j = 0; j < au_header_length; ++j)
+            size = max_size;
-		{
+        }
-			// 当前aac包长度
+        if (_adts->aac_frame_length + size > sizeof(AACFrame::buffer)) {
-			const uint32_t cur_aac_payload_len = vec_aac_len.at(j);
+            //数据太多了，先清空
+            flushData();
+        }
+        //追加aac数据
+        memcpy(_adts->buffer + _adts->aac_frame_length, ptr, size);
+        _adts->aac_frame_length += size;
+        _adts->timeStamp = rtppack->timeStamp;
+        ptr += size;
+    }
 
-			if (_adts->aac_frame_length + cur_aac_payload_len > sizeof(AACFrame::buffer)) {
+    if (rtppack->mark) {
-				_adts->aac_frame_length = 7;
+        //最后一个rtp分片
-				WarnL << "aac负载数据太长";
+        flushData();
-				return false;
+    }
-			}
-			
-			// 提取每一包aac载荷数据
-			memcpy(_adts->buffer + _adts->aac_frame_length, rtp_packet_payload + next_aac_payload_offset, cur_aac_payload_len);
-			_adts->aac_frame_length += (cur_aac_payload_len);
-			if (rtppack->mark == true) {
-				_adts->timeStamp = rtppack->timeStamp;
-				writeAdtsHeader(*_adts, _adts->buffer);
-				onGetAAC(_adts);
-			}
-
-			next_aac_payload_offset += cur_aac_payload_len;
-		}
-	} while (0);
-	
     return false;
 }
 
-void AACRtpDecoder::onGetAAC(const AACFrame::Ptr &frame) {
+
-    //写入环形缓存
+void AACRtpDecoder::flushData() {
-    RtpCodec::inputFrame(frame);
+    if(_adts->aac_frame_length == ADTS_HEADER_LEN){
+        //没有有效数据
+        return;
+    }
+    writeAdtsHeader(*_adts, _adts->buffer);
+    RtpCodec::inputFrame(_adts);
     _adts = obtainFrame();
 }
 

src/Extension/AACRtp.h

@@ -56,8 +56,8 @@ public:
 protected:
     AACRtpDecoder();
 private:
-    void onGetAAC(const AACFrame::Ptr &frame);
     AACFrame::Ptr obtainFrame();
+    void flushData();
 private:
     AACFrame::Ptr _adts;
     string _aac_cfg;

src/Extension/Factory.cpp

@@ -155,7 +155,7 @@ RtpCodec::Ptr Factory::getRtpDecoderByTrack(const Track::Ptr &track) {
         case CodecAAC:
             return std::make_shared<AACRtpDecoder>(track->clone());
         default:
-            WarnL << "暂不支持该CodecId:" << track->getCodecId();
+            WarnL << "暂不支持该CodecId:" << track->getCodecName();
             return nullptr;
     }
 }
@@ -212,7 +212,7 @@ RtmpCodec::Ptr Factory::getRtmpCodecByTrack(const Track::Ptr &track) {
         case CodecAAC:
             return std::make_shared<AACRtmpEncoder>(track);
         default:
-            WarnL << "暂不支持该CodecId:" << track->getCodecId();
+            WarnL << "暂不支持该CodecId:" << track->getCodecName();
             return nullptr;
     }
 }

src/Extension/Frame.cpp

@@ -38,5 +38,16 @@ Frame::Ptr Frame::getCacheAbleFrame(const Frame::Ptr &frame){
     return std::make_shared<FrameCacheAble>(frame);
 }
 
+#define SWITCH_CASE(codec_id) case codec_id : return #codec_id
+const char *CodecInfo::getCodecName() {
+    switch (getCodecId()) {
+        SWITCH_CASE(CodecH264);
+        SWITCH_CASE(CodecH265);
+        SWITCH_CASE(CodecAAC);
+        default:
+            return "unknown codec";
+    }
+}
+
 }//namespace mediakit
 

src/Extension/Frame.h

@@ -72,6 +72,12 @@ public:
      * 获取编解码器类型
      */
     virtual CodecId getCodecId() const = 0;
+
+    /**
+     * 获取编码器名称
+     * @return 编码器名称
+     */
+    const char *getCodecName();
 };
 
 /**

src/Http/HttpSession.cpp

@@ -53,20 +53,29 @@ HttpSession::~HttpSession() {
     TraceP(this);
 }
 
+void HttpSession::Handle_Req_HEAD(int64_t &content_len){
+    //暂时全部返回200 OK，因为HTTP GET存在按需生成流的操作，所以不能按照HTTP GET的流程返回
+    //如果直接返回404，那么又会导致按需生成流的逻辑失效，所以HTTP HEAD在静态文件或者已存在资源时才有效
+    //对于按需生成流的直播场景并不适用
+    sendResponse("200 OK", true);
+}
+
 int64_t HttpSession::onRecvHeader(const char *header,uint64_t len) {
 	typedef void (HttpSession::*HttpCMDHandle)(int64_t &);
 	static unordered_map<string, HttpCMDHandle> s_func_map;
 	static onceToken token([]() {
 		s_func_map.emplace("GET",&HttpSession::Handle_Req_GET);
 		s_func_map.emplace("POST",&HttpSession::Handle_Req_POST);
-	}, nullptr);
+        s_func_map.emplace("HEAD",&HttpSession::Handle_Req_HEAD);
+    }, nullptr);
 
 	_parser.Parse(header);
 	urlDecode(_parser);
 	string cmd = _parser.Method();
 	auto it = s_func_map.find(cmd);
 	if (it == s_func_map.end()) {
-		sendResponse("403 Forbidden", true);
+	    WarnL << "不支持该命令:" << cmd;
+		sendResponse("405 Not Allowed", true);
         return 0;
 	}
 
@@ -256,8 +265,11 @@ bool HttpSession::checkLiveFlvStream(const function<void()> &cb){
     return true;
 }
 
-
 void HttpSession::Handle_Req_GET(int64_t &content_len) {
+    Handle_Req_GET_l(content_len, true);
+}
+
+void HttpSession::Handle_Req_GET_l(int64_t &content_len, bool sendBody) {
 	//先看看是否为WebSocket请求
 	if(checkWebSocket()){
 		content_len = -1;

src/Http/HttpSession.h

@@ -107,8 +107,11 @@ protected:
 	void onWebSocketEncodeData(const Buffer::Ptr &buffer) override;
 private:
 	void Handle_Req_GET(int64_t &content_len);
-	void Handle_Req_POST(int64_t &content_len);
+    void Handle_Req_GET_l(int64_t &content_len, bool sendBody);
-	bool checkLiveFlvStream(const function<void()> &cb = nullptr);
+    void Handle_Req_POST(int64_t &content_len);
+    void Handle_Req_HEAD(int64_t &content_len);
+
+    bool checkLiveFlvStream(const function<void()> &cb = nullptr);
 	bool checkWebSocket();
 	bool emitHttpEvent(bool doInvoke);
 	void urlDecode(Parser &parser);

src/Http/strCoding.cpp

@@ -69,24 +69,19 @@ char StrToBin(const char *str)
 }
 
 string strCoding::UrlEncode(const string &str) {
-	string dd;
+	string out;
-	size_t len = str.size();
+    size_t len = str.size();
-	for (size_t i = 0; i < len; i++) {
+	for (size_t i = 0; i < len; ++i) {
-		if (isalnum((uint8_t)str[i])) {
+        char ch = str[i];
-			char tempbuff[2];
+		if (isalnum((uint8_t)ch)) {
-			sprintf(tempbuff, "%c", str[i]);
+            out.push_back(ch);
-			dd.append(tempbuff);
+		}else {
-		}
+			char buf[4];
-		else if (isspace((uint8_t)str[i])) {
+			sprintf(buf, "%%%X%X", (uint8_t)ch >> 4,(uint8_t)ch & 0x0F);
-			dd.append("+");
+            out.append(buf);
-		}
-		else {
-			char tempbuff[4];
-			sprintf(tempbuff, "%%%X%X", (uint8_t)str[i] >> 4,(uint8_t)str[i] % 16);
-			dd.append(tempbuff);
 		}
 	}
-	return dd;
+	return out;
 }
 string strCoding::UrlDecode(const string &str) {
 	string output = "";
@@ -94,16 +89,18 @@ string strCoding::UrlDecode(const string &str) {
 	int i = 0, len = str.length();
 	while (i < len) {
 		if (str[i] == '%') {
+		    if(i > len - 3){
+		        //防止内存溢出
+                break;
+		    }
 			tmp[0] = str[i + 1];
 			tmp[1] = str[i + 2];
 			output += StrToBin(tmp);
 			i = i + 3;
-		}
+		} else if (str[i] == '+') {
-		else if (str[i] == '+') {
 			output += ' ';
 			i++;
-		}
+		} else {
-		else {
 			output += str[i];
 			i++;
 		}

src/Player/PlayerProxy.h

@@ -75,13 +75,16 @@ public:
      * @param strUrl
      */
     void play(const string &strUrl) override;
+
+    /**
+     * 获取观看总人数
+     */
+    int totalReaderCount() ;
 private:
 	//MediaSourceEvent override
 	bool close(MediaSource &sender,bool force) override;
     void onNoneReader(MediaSource &sender) override;
 	int totalReaderCount(MediaSource &sender) override;
-	int totalReaderCount() ;
-
 	void rePlay(const string &strUrl,int iFailedCnt);
 	void onPlaySuccess();
 private:

src/Record/HlsMediaSource.h

@@ -1,4 +1,4 @@
-/*
+/*
  * MIT License
  *
  * Copyright (c) 2016-2019 xiongziliang <771730766@qq.com>

src/Record/MP4Muxer.cpp

@@ -258,7 +258,7 @@ void MP4Muxer::addTrack(const Track::Ptr &track) {
         }
             break;
         default:
-            WarnL << "MP4录制不支持该编码格式:" << track->getCodecId();
+            WarnL << "MP4录制不支持该编码格式:" << track->getCodecName();
             break;
     }
 }

src/Rtp/Decoder.cpp

@@ -0,0 +1,41 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 xiongziliang <771730766@qq.com>
+ *
+ * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+#if defined(ENABLE_RTPPROXY)
+#include "Decoder.h"
+#include "PSDecoder.h"
+#include "TSDecoder.h"
+namespace mediakit {
+Decoder::Ptr Decoder::createDecoder(Decoder::Type type) {
+    switch (type){
+        case decoder_ps : return std::make_shared<PSDecoder>();
+        case decoder_ts : return std::make_shared<TSDecoder>();
+        default : return nullptr;
+    }
+}
+
+}//namespace mediakit
+#endif//defined(ENABLE_RTPPROXY)

src/Rtp/Decoder.h

@@ -0,0 +1,57 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 xiongziliang <771730766@qq.com>
+ *
+ * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+#ifndef ZLMEDIAKIT_DECODER_H
+#define ZLMEDIAKIT_DECODER_H
+
+#if defined(ENABLE_RTPPROXY)
+#include <stdint.h>
+#include <memory>
+#include <functional>
+#include "Decoder.h"
+using namespace std;
+namespace mediakit {
+
+class Decoder {
+public:
+    typedef std::shared_ptr<Decoder> Ptr;
+    typedef enum {
+        decoder_ts = 0,
+        decoder_ps
+    }Type;
+
+    typedef std::function<void(int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,int bytes)> onDecode;
+    virtual int input(const uint8_t *data, int bytes) = 0;
+    virtual void setOnDecode(const onDecode &decode) = 0;
+    static Ptr createDecoder(Type type);
+protected:
+    Decoder() = default;
+    virtual ~Decoder() = default;
+};
+
+}//namespace mediakit
+#endif//defined(ENABLE_RTPPROXY)
+#endif //ZLMEDIAKIT_DECODER_H

src/Rtp/PSDecoder.cpp

@@ -27,7 +27,6 @@
 #if defined(ENABLE_RTPPROXY)
 #include "PSDecoder.h"
 #include "mpeg-ps.h"
-
 namespace mediakit{
 
 PSDecoder::PSDecoder() {
@@ -40,7 +39,9 @@ PSDecoder::PSDecoder() {
                                        const void* data,
                                        size_t bytes){
         PSDecoder *thiz = (PSDecoder *)param;
-        thiz->onPSDecode(stream, codecid, flags, pts, dts, data, bytes);
+        if(thiz->_on_decode){
+            thiz->_on_decode(stream, codecid, flags, pts, dts, data, bytes);
+        }
     },this);
 }
 
@@ -48,10 +49,13 @@ PSDecoder::~PSDecoder() {
     ps_demuxer_destroy((struct ps_demuxer_t*)_ps_demuxer);
 }
 
-int PSDecoder::decodePS(const uint8_t *data, int bytes) {
+int PSDecoder::input(const uint8_t *data, int bytes) {
     return ps_demuxer_input((struct ps_demuxer_t*)_ps_demuxer,data,bytes);
 }
 
-}//namespace mediakit
+void PSDecoder::setOnDecode(const Decoder::onDecode &decode) {
+    _on_decode = decode;
+}
 
+}//namespace mediakit
 #endif//#if defined(ENABLE_RTPPROXY)

src/Rtp/PSDecoder.h

@@ -29,27 +29,21 @@
 
 #if defined(ENABLE_RTPPROXY)
 #include <stdint.h>
-
+#include "Decoder.h"
 namespace mediakit{
 
-class PSDecoder {
+//ps解析器
+class PSDecoder : public Decoder {
 public:
     PSDecoder();
-    virtual ~PSDecoder();
+    ~PSDecoder();
-    int decodePS(const uint8_t *data, int bytes);
+    int input(const uint8_t* data, int bytes) override;
-protected:
+    void setOnDecode(const onDecode &decode) override;
-    virtual void onPSDecode(int stream,
-                            int codecid,
-                            int flags,
-                            int64_t pts,
-                            int64_t dts,
-                            const void *data,
-                            int bytes) = 0;
 private:
     void *_ps_demuxer = nullptr;
+    onDecode _on_decode;
 };
 
 }//namespace mediakit
-
 #endif//defined(ENABLE_RTPPROXY)
 #endif //ZLMEDIAKIT_PSDECODER_H

src/Rtp/RtpDecoder.cpp

@@ -25,11 +25,9 @@
  */
 
 #if defined(ENABLE_RTPPROXY)
-#include <assert.h>
 #include "Util/logger.h"
 #include "RtpDecoder.h"
 #include "rtp-payload.h"
-
 using namespace toolkit;
 
 namespace mediakit{
@@ -45,7 +43,7 @@ RtpDecoder::~RtpDecoder() {
     }
 }
 
-void RtpDecoder::decodeRtp(const void *data, int bytes,const char *type_name) {
+void RtpDecoder::decodeRtp(const void *data, int bytes) {
     if(!_rtp_decoder){
         static rtp_payload_t s_func= {
                 [](void* param, int bytes){
@@ -58,17 +56,18 @@ void RtpDecoder::decodeRtp(const void *data, int bytes,const char *type_name) {
                 },
                 [](void* param, const void *packet, int bytes, uint32_t timestamp, int flags){
                     RtpDecoder *obj = (RtpDecoder *)param;
-                    obj->onRtpDecode(packet, bytes, timestamp, flags);
+                    obj->onRtpDecode((uint8_t *)packet, bytes, timestamp, flags);
                 }
         };
 
         uint8_t rtp_type = 0x7F & ((uint8_t *) data)[1];
         InfoL << "rtp type:" << (int) rtp_type;
-        _rtp_decoder = rtp_payload_decode_create(rtp_type, type_name, &s_func, this);
+        _rtp_decoder = rtp_payload_decode_create(rtp_type, "MP2P", &s_func, this);
         if (!_rtp_decoder) {
             WarnL << "unsupported rtp type:" << (int) rtp_type << ",size:" << bytes << ",hexdump" << hexdump(data, bytes > 16 ? 16 : bytes);
         }
     }
+
     if(_rtp_decoder){
         rtp_payload_decode_input(_rtp_decoder,data,bytes);
     }

src/Rtp/RtpDecoder.h

@@ -38,8 +38,8 @@ public:
     RtpDecoder();
     virtual ~RtpDecoder();
 protected:
-    void decodeRtp(const void *data, int bytes,const char *type_name);
+    void decodeRtp(const void *data, int bytes);
-    virtual void onRtpDecode(const void *packet, int bytes, uint32_t timestamp, int flags) = 0;
+    virtual void onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp, int flags) = 0;
 private:
     void *_rtp_decoder = nullptr;
     BufferRaw::Ptr _buffer;

src/Rtp/RtpProcess.cpp

@@ -33,7 +33,6 @@
 
 namespace mediakit{
 
-
 /**
 * 合并一些时间戳相同的frame
 */
@@ -153,29 +152,47 @@ bool RtpProcess::inputRtp(const char *data, int data_len,const struct sockaddr *
     return ret;
 }
 
+//判断是否为ts负载
+static inline bool checkTS(const uint8_t *packet, int bytes){
+    return bytes % 188 == 0 && packet[0] == 0x47;
+}
+
 void RtpProcess::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
     if(rtp->sequence != _sequence + 1){
         WarnL << rtp->sequence << " != " << _sequence << "+1";
     }
     _sequence = rtp->sequence;
-
     if(_save_file_rtp){
         uint16_t  size = rtp->size() - 4;
         size = htons(size);
         fwrite((uint8_t *) &size, 2, 1, _save_file_rtp.get());
         fwrite((uint8_t *) rtp->data() + 4, rtp->size() - 4, 1, _save_file_rtp.get());
     }
-
+    decodeRtp(rtp->data() + 4 ,rtp->size() - 4);
-    GET_CONFIG(string,rtp_type,::RtpProxy::kRtpType);
-    decodeRtp(rtp->data() + 4 ,rtp->size() - 4,rtp_type.data());
 }
 
-void RtpProcess::onRtpDecode(const void *packet, int bytes, uint32_t, int flags) {
+void RtpProcess::onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp, int flags) {
     if(_save_file_ps){
         fwrite((uint8_t *)packet,bytes, 1, _save_file_ps.get());
     }
 
-    auto ret = decodePS((uint8_t *)packet,bytes);
+    if(!_decoder){
+        //创建解码器
+        if(checkTS(packet, bytes)){
+            //猜测是ts负载
+            InfoL << "judged to be TS: " << printSSRC(_ssrc);
+            _decoder = Decoder::createDecoder(Decoder::decoder_ts);
+        }else{
+            //猜测是ps负载
+            InfoL << "judged to be PS: " << printSSRC(_ssrc);
+            _decoder = Decoder::createDecoder(Decoder::decoder_ps);
+        }
+        _decoder->setOnDecode([this](int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,int bytes){
+            onDecode(stream,codecid,flags,pts,dts,data,bytes);
+        });
+    }
+
+    auto ret = _decoder->input((uint8_t *)packet,bytes);
     if(ret != bytes){
         WarnL << ret << " != " << bytes << " " << flags;
     }
@@ -200,13 +217,7 @@ static const char *getCodecName(int codec_id) {
     }
 }
 
-void RtpProcess::onPSDecode(int stream,
+void RtpProcess::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts,const void *data,int bytes) {
-                       int codecid,
-                       int flags,
-                       int64_t pts,
-                       int64_t dts,
-                       const void *data,
-                       int bytes) {
     pts /= 90;
     dts /= 90;
     _stamps[codecid].revise(dts,pts,dts,pts,false);
@@ -299,5 +310,14 @@ uint16_t RtpProcess::get_peer_port() {
     return ntohs(((struct sockaddr_in *) _addr)->sin_port);
 }
 
+int RtpProcess::totalReaderCount(){
+    return _muxer->totalReaderCount();
+}
+
+void RtpProcess::setListener(const std::weak_ptr<MediaSourceEvent> &listener){
+    _muxer->setListener(listener);
+}
+
+
 }//namespace mediakit
 #endif//defined(ENABLE_RTPPROXY)

src/Rtp/RtpProcess.h

@@ -31,7 +31,7 @@
 
 #include "Rtsp/RtpReceiver.h"
 #include "RtpDecoder.h"
-#include "PSDecoder.h"
+#include "Decoder.h"
 #include "Common/Device.h"
 #include "Common/Stamp.h"
 using namespace mediakit;
@@ -40,7 +40,7 @@ namespace mediakit{
 
 string printSSRC(uint32_t ui32Ssrc);
 class FrameMerger;
-class RtpProcess : public RtpReceiver , public RtpDecoder , public PSDecoder {
+class RtpProcess : public RtpReceiver , public RtpDecoder{
 public:
     typedef std::shared_ptr<RtpProcess> Ptr;
     RtpProcess(uint32_t ssrc);
@@ -49,16 +49,14 @@ public:
     bool alive();
     string get_peer_ip();
     uint16_t get_peer_port();
+    int totalReaderCount();
+    void setListener(const std::weak_ptr<MediaSourceEvent> &listener);
 protected:
     void onRtpSorted(const RtpPacket::Ptr &rtp, int track_index) override ;
-    void onRtpDecode(const void *packet, int bytes, uint32_t timestamp, int flags) override;
+    void onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp, int flags) override;
-    void onPSDecode(int stream,
+    void onDecode(int stream,int codecid,int flags,int64_t pts,int64_t dts, const void *data,int bytes);
-                    int codecid,
+private:
-                    int flags,
+    void getNextRtpType();
-                    int64_t pts,
-                    int64_t dts,
-                    const void *data,
-                    int bytes) override ;
 private:
     std::shared_ptr<FILE> _save_file_rtp;
     std::shared_ptr<FILE> _save_file_ps;
@@ -74,6 +72,7 @@ private:
     Ticker _last_rtp_time;
     map<int,Stamp> _stamps;
     uint32_t _dts = 0;
+    Decoder::Ptr _decoder;
 };
 
 }//namespace mediakit

src/Rtp/RtpSelector.cpp

@@ -32,10 +32,6 @@ namespace mediakit{
 INSTANCE_IMP(RtpSelector);
 
 bool RtpSelector::inputRtp(const char *data, int data_len,const struct sockaddr *addr,uint32_t *dts_out) {
-    if(_last_rtp_time.elapsedTime() > 3000){
-        _last_rtp_time.resetTime();
-        onManager();
-    }
     uint32_t ssrc = 0;
     if(!getSSRC(data,data_len,ssrc)){
         WarnL << "get ssrc from rtp failed:" << data_len;
@@ -63,11 +59,28 @@ RtpProcess::Ptr RtpSelector::getProcess(uint32_t ssrc,bool makeNew) {
     if(it == _map_rtp_process.end() && !makeNew){
         return nullptr;
     }
-    RtpProcess::Ptr &ref = _map_rtp_process[ssrc];
+    RtpProcessHelper::Ptr &ref = _map_rtp_process[ssrc];
     if(!ref){
-        ref = std::make_shared<RtpProcess>(ssrc);
+        ref = std::make_shared<RtpProcessHelper>(ssrc,shared_from_this());
+        ref->attachEvent();
+        createTimer();
+    }
+    return ref->getProcess();
+}
+
+void RtpSelector::createTimer() {
+    if (!_timer) {
+        //创建超时管理定时器
+        weak_ptr<RtpSelector> weakSelf = shared_from_this();
+        _timer = std::make_shared<Timer>(3.0, [weakSelf] {
+            auto strongSelf = weakSelf.lock();
+            if (!strongSelf) {
+                return false;
+            }
+            strongSelf->onManager();
+            return true;
+        }, EventPollerPool::Instance().getPoller());
     }
-    return ref;
 }
 
 void RtpSelector::delProcess(uint32_t ssrc,const RtpProcess *ptr) {
@@ -77,7 +90,7 @@ void RtpSelector::delProcess(uint32_t ssrc,const RtpProcess *ptr) {
         return;
     }
 
-    if(it->second.get() != ptr){
+    if(it->second->getProcess().get() != ptr){
         return;
     }
 
@@ -87,7 +100,7 @@ void RtpSelector::delProcess(uint32_t ssrc,const RtpProcess *ptr) {
 void RtpSelector::onManager() {
     lock_guard<decltype(_mtx_map)> lck(_mtx_map);
     for (auto it = _map_rtp_process.begin(); it != _map_rtp_process.end();) {
-        if (it->second->alive()) {
+        if (it->second->getProcess()->alive()) {
             ++it;
             continue;
         }
@@ -102,5 +115,47 @@ RtpSelector::RtpSelector() {
 RtpSelector::~RtpSelector() {
 }
 
+RtpProcessHelper::RtpProcessHelper(uint32_t ssrc, const weak_ptr<RtpSelector> &parent) {
+    _ssrc = ssrc;
+    _parent = parent;
+    _process = std::make_shared<RtpProcess>(_ssrc);
+}
+
+RtpProcessHelper::~RtpProcessHelper() {
+}
+
+void RtpProcessHelper::attachEvent() {
+    _process->setListener(shared_from_this());
+}
+
+bool RtpProcessHelper::close(MediaSource &sender, bool force) {
+    //此回调在其他线程触发
+    if(!_process || (!force && _process->totalReaderCount())){
+        return false;
+    }
+    auto parent = _parent.lock();
+    if(!parent){
+        return false;
+    }
+    parent->delProcess(_ssrc,_process.get());
+    WarnL << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" << sender.getApp() << "/" << sender.getId() << " " << force;
+    return true;
+}
+
+void RtpProcessHelper::onNoneReader(MediaSource &sender) {
+    if(!_process || _process->totalReaderCount()){
+        return;
+    }
+    MediaSourceEvent::onNoneReader(sender);
+}
+
+int RtpProcessHelper::totalReaderCount(MediaSource &sender) {
+    return _process ? _process->totalReaderCount() : sender.totalReaderCount();
+}
+
+RtpProcess::Ptr &RtpProcessHelper::getProcess() {
+    return _process;
+}
+
 }//namespace mediakit
 #endif//defined(ENABLE_RTPPROXY)

src/Rtp/RtpSelector.h

@@ -32,9 +32,31 @@
 #include <mutex>
 #include <unordered_map>
 #include "RtpProcess.h"
+#include "Common/MediaSource.h"
 
 namespace mediakit{
 
+class RtpSelector;
+class RtpProcessHelper : public MediaSourceEvent , public std::enable_shared_from_this<RtpProcessHelper> {
+public:
+    typedef std::shared_ptr<RtpProcessHelper> Ptr;
+    RtpProcessHelper(uint32_t ssrc,const weak_ptr<RtpSelector > &parent);
+    ~RtpProcessHelper();
+    void attachEvent();
+    RtpProcess::Ptr & getProcess();
+protected:
+    // 通知其停止推流
+    bool close(MediaSource &sender,bool force) override;
+    // 通知无人观看
+    void onNoneReader(MediaSource &sender) override;
+    // 观看总人数
+    int totalReaderCount(MediaSource &sender) override;
+private:
+    weak_ptr<RtpSelector > _parent;
+    RtpProcess::Ptr _process;
+    uint32_t _ssrc = 0;
+};
+
 class RtpSelector : public std::enable_shared_from_this<RtpSelector>{
 public:
     RtpSelector();
@@ -47,10 +69,11 @@ public:
     void delProcess(uint32_t ssrc,const RtpProcess *ptr);
 private:
     void onManager();
+    void createTimer();
 private:
-    unordered_map<uint32_t,RtpProcess::Ptr> _map_rtp_process;
+    unordered_map<uint32_t,RtpProcessHelper::Ptr> _map_rtp_process;
     recursive_mutex _mtx_map;
-    Ticker _last_rtp_time;
+    Timer::Ptr _timer;
 };
 
 }//namespace mediakit

src/Rtp/RtpSession.cpp

@@ -71,10 +71,34 @@ void RtpSession::onRtpPacket(const char *data, uint64_t len) {
             return;
         }
         _process = RtpSelector::Instance().getProcess(_ssrc, true);
+        _process->setListener(dynamic_pointer_cast<RtpSession>(shared_from_this()));
     }
     _process->inputRtp(data + 2, len - 2, &addr);
     _ticker.resetTime();
 }
 
+bool RtpSession::close(MediaSource &sender, bool force) {
+    //此回调在其他线程触发
+    if(!_process || (!force && _process->totalReaderCount())){
+        return false;
+    }
+    string err = StrPrinter << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" << sender.getApp() << "/" << sender.getId() << " " << force;
+    safeShutdown(SockException(Err_shutdown,err));
+    return true;
+}
+
+void RtpSession::onNoneReader(MediaSource &sender) {
+    //此回调在其他线程触发
+    if(!_process || _process->totalReaderCount()){
+        return;
+    }
+    MediaSourceEvent::onNoneReader(sender);
+}
+
+int RtpSession::totalReaderCount(MediaSource &sender) {
+    //此回调在其他线程触发
+    return _process ? _process->totalReaderCount() : sender.totalReaderCount();
+}
+
 }//namespace mediakit
 #endif//defined(ENABLE_RTPPROXY)

src/Rtp/RtpSession.h

@@ -36,14 +36,20 @@ using namespace toolkit;
 
 namespace mediakit{
 
-class RtpSession : public TcpSession , public RtpSplitter{
+class RtpSession : public TcpSession , public RtpSplitter , public MediaSourceEvent{
 public:
     RtpSession(const Socket::Ptr &sock);
     ~RtpSession() override;
     void onRecv(const Buffer::Ptr &) override;
     void onError(const SockException &err) override;
     void onManager() override;
-private:
+protected:
+    // 通知其停止推流
+    bool close(MediaSource &sender,bool force) override;
+    // 通知无人观看
+    void onNoneReader(MediaSource &sender) override;
+    // 观看总人数
+    int totalReaderCount(MediaSource &sender) override;
     void onRtpPacket(const char *data,uint64_t len) override;
 private:
     uint32_t _ssrc = 0;

src/Rtp/TSDecoder.cpp

@@ -0,0 +1,96 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 xiongziliang <771730766@qq.com>
+ *
+ * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+#if defined(ENABLE_RTPPROXY)
+#include "mpeg-ts.h"
+#include "TSDecoder.h"
+#define TS_PACKET_SIZE 188
+namespace mediakit {
+
+void TSSegment::setOnSegment(const TSSegment::onSegment &cb) {
+    _onSegment = cb;
+}
+
+int64_t TSSegment::onRecvHeader(const char *data, uint64_t len) {
+    _onSegment(data, len);
+    return 0;
+}
+
+const char *TSSegment::onSearchPacketTail(const char *data, int len) {
+    if (len < _size + 1) {
+        if (len == _size && ((uint8_t *) data)[0] == 0x47) {
+            return data + _size;
+        }
+        return nullptr;
+    }
+    //下一个包头
+    if (((uint8_t *) data)[_size] == 0x47) {
+        return data + _size;
+    }
+
+    auto pos = memchr(data + _size, 0x47, len - _size);
+    if (pos) {
+        return (char *) pos;
+    }
+    return nullptr;
+}
+
+////////////////////////////////////////////////////////////////
+
+TSDecoder::TSDecoder() : _ts_segment(TS_PACKET_SIZE) {
+    _ts_segment.setOnSegment([this](const char *data,uint64_t len){
+        if(((uint8_t*)data)[0] != 0x47 || len != TS_PACKET_SIZE ){
+            WarnL << "不是ts包:" << (int)(data[0]) << " " << len;
+            return;
+        }
+        ts_demuxer_input(_demuxer_ctx,(uint8_t*)data,len);
+    });
+    _demuxer_ctx = ts_demuxer_create([](void* param, int program, int stream, int codecid, int flags, int64_t pts, int64_t dts, const void* data, size_t bytes){
+        TSDecoder *thiz = (TSDecoder*)param;
+        if(thiz->_on_decode){
+            thiz->_on_decode(stream,codecid,flags,pts,dts,data,bytes);
+        }
+        return 0;
+    },this);
+}
+
+TSDecoder::~TSDecoder() {
+    ts_demuxer_destroy(_demuxer_ctx);
+}
+
+int TSDecoder::input(const uint8_t *data, int bytes) {
+    if(bytes == TS_PACKET_SIZE && ((uint8_t*)data)[0] == 0x47){
+        return ts_demuxer_input(_demuxer_ctx,(uint8_t*)data,bytes);
+    }
+    _ts_segment.input((char*)data,bytes);
+    return bytes;
+}
+
+void TSDecoder::setOnDecode(const Decoder::onDecode &decode) {
+    _on_decode = decode;
+}
+
+}//namespace mediakit
+#endif//defined(ENABLE_RTPPROXY)

src/Rtp/TSDecoder.h

@@ -0,0 +1,68 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 xiongziliang <771730766@qq.com>
+ *
+ * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+#ifndef ZLMEDIAKIT_TSDECODER_H
+#define ZLMEDIAKIT_TSDECODER_H
+
+#if defined(ENABLE_RTPPROXY)
+#include "Util/logger.h"
+#include "Http/HttpRequestSplitter.h"
+#include "Decoder.h"
+
+using namespace toolkit;
+namespace mediakit {
+
+//ts包拆分器
+class TSSegment : public HttpRequestSplitter {
+public:
+    typedef std::function<void(const char *data,uint64_t len)> onSegment;
+    TSSegment(int size = 188) : _size(size){}
+    ~TSSegment(){}
+    void setOnSegment(const onSegment &cb);
+protected:
+    int64_t onRecvHeader(const char *data, uint64_t len) override ;
+    const char *onSearchPacketTail(const char *data, int len) override ;
+private:
+    int _size;
+    onSegment _onSegment;
+};
+
+//ts解析器
+class TSDecoder : public Decoder {
+public:
+    TSDecoder();
+    ~TSDecoder();
+    int input(const uint8_t* data, int bytes) override ;
+    void setOnDecode(const onDecode &decode) override;
+private:
+    TSSegment _ts_segment;
+    struct ts_demuxer_t* _demuxer_ctx = nullptr;
+    onDecode _on_decode;
+};
+
+}//namespace mediakit
+#endif//defined(ENABLE_RTPPROXY)
+#endif //ZLMEDIAKIT_TSDECODER_H

tests/test_rtp.cpp

@@ -98,8 +98,6 @@ int main(int argc,char *argv[]) {
     rtspSrv->start<RtspSession>(554);//默认554
     rtmpSrv->start<RtmpSession>(1935);//默认1935
     httpSrv->start<HttpSession>(80);//默认80
-    //此处可以选择MP4V-ES或MP2P
-    mINI::Instance()[RtpProxy::kRtpType] = "MP4V-ES";
     //此处选择是否导出调试文件
 //    mINI::Instance()[RtpProxy::kDumpDir] = "/Users/xzl/Desktop/";