优化hls媒体流注册与无人观看清理机制 (#1421)

* 优化hls媒体流注册与无人观看清理机制

* 优化hls cookie管理

* bug fixed

src/Http/HttpFileManager.cpp

@@ -38,15 +38,8 @@ public:
     string _path;
     //上次鉴权失败信息,为空则上次鉴权成功
     string _err_msg;
-    //本cookie是否为hls直播的
-    bool _is_hls = false;
     //hls直播时的其他一些信息，主要用于播放器个数计数以及流量计数
     HlsCookieData::Ptr _hls_data;
-    //如果是hls直播，那么判断该cookie是否使用过MediaSource::findAsync查找过
-    //如果程序未正常退出，会残余上次的hls文件，所以判断hls直播是否存在的关键不是文件存在与否
-    //而是应该判断HlsMediaSource是否已注册，但是这样会每次获取m3u8文件时都会用MediaSource::findAsync判断一次
-    //会导致程序性能低下，所以我们应该在cookie声明周期的第一次判断HlsMediaSource是否已经注册，后续通过文件存在与否判断
-    bool _have_find_media_source = false;
 };
 
 const string &HttpFileManager::getContentType(const char *name) {
@@ -245,40 +238,39 @@ public:
  * 5、触发kBroadcastHttpAccess事件
  */
 static void canAccessPath(TcpSession &sender, const Parser &parser, const MediaInfo &media_info, bool is_dir,
-                          const function<void(const string &errMsg, const HttpServerCookie::Ptr &cookie)> &callback) {
+                          const function<void(const string &err_msg, const HttpServerCookie::Ptr &cookie)> &callback) {
     //获取用户唯一id
     auto uid = parser.Params();
     auto path = parser.Url();
 
     //先根据http头中的cookie字段获取cookie
     HttpServerCookie::Ptr cookie = HttpCookieManager::Instance().getCookie(kCookieName, parser.getHeader());
-    //如果不是从http头中找到的cookie,我们让http客户端设置下cookie
+    //是否需要更新cookie
-    bool cookie_from_header = true;
+    bool update_cookie = false;
     if (!cookie && !uid.empty()) {
         //客户端请求中无cookie,再根据该用户的用户id获取cookie
         cookie = HttpCookieManager::Instance().getCookieByUid(kCookieName, uid);
-        cookie_from_header = false;
+        update_cookie = true;
     }
 
     if (cookie) {
-        //找到了cookie，对cookie上锁先
         auto& attach = cookie->getAttach<HttpCookieAttachment>();
         if (path.find(attach._path) == 0) {
             //上次cookie是限定本目录
             if (attach._err_msg.empty()) {
                 //上次鉴权成功
-                if (attach._is_hls) {
+                if (attach._hls_data) {
                     //如果播放的是hls，那么刷新hls的cookie(获取ts文件也会刷新)
                     cookie->updateTime();
-                    cookie_from_header = false;
+                    update_cookie = true;
                 }
-                callback("", cookie_from_header ? nullptr : cookie);
+                callback("", update_cookie ? cookie : nullptr);
                 return;
             }
             //上次鉴权失败，但是如果url参数发生变更，那么也重新鉴权下
             if (parser.Params().empty() || parser.Params() == cookie->getUid()) {
                 //url参数未变，或者本来就没有url参数，那么判断本次请求为重复请求，无访问权限
-                callback(attach._err_msg, cookie_from_header ? nullptr : cookie);
+                callback(attach._err_msg, update_cookie ? cookie : nullptr);
                 return;
             }
         }
@@ -312,13 +304,9 @@ static void canAccessPath(TcpSession &sender, const Parser &parser, const MediaI
             attach->_path = cookie_path;
             //记录能否访问
             attach->_err_msg = err_msg;
-            //记录访问的是否为hls
-            attach->_is_hls = is_hls;
             if (is_hls) {
                 // hls相关信息
                 attach->_hls_data = std::make_shared<HlsCookieData>(media_info, info);
-                // hls未查找MediaSource
-                attach->_have_find_media_source = false;
             }
             callback(err_msg, HttpCookieManager::Instance().addCookie(kCookieName, uid, life_second, attach));
         } else {
@@ -383,19 +371,19 @@ static void accessFile(TcpSession &sender, const Parser &parser, const MediaInfo
 
     weak_ptr<TcpSession> weakSession = sender.shared_from_this();
     //判断是否有权限访问该文件
-    canAccessPath(sender, parser, media_info, false, [cb, file_path, parser, is_hls, media_info, weakSession , file_exist](const string &errMsg, const HttpServerCookie::Ptr &cookie) {
+    canAccessPath(sender, parser, media_info, false, [cb, file_path, parser, is_hls, media_info, weakSession , file_exist](const string &err_msg, const HttpServerCookie::Ptr &cookie) {
         auto strongSession = weakSession.lock();
         if (!strongSession) {
             //http客户端已经断开，不需要回复
             return;
         }
-        if (!errMsg.empty()) {
+        if (!err_msg.empty()) {
             //文件鉴权失败
             StrCaseMap headerOut;
             if (cookie) {
                 headerOut["Set-Cookie"] = cookie->getCookie(cookie->getAttach<HttpCookieAttachment>()._path);
             }
-            cb(401, "text/html", headerOut, std::make_shared<HttpStringBody>(errMsg));
+            cb(401, "text/html", headerOut, std::make_shared<HttpStringBody>(err_msg));
             return;
         }
 
@@ -408,7 +396,7 @@ static void accessFile(TcpSession &sender, const Parser &parser, const MediaInfo
             HttpSession::HttpResponseInvoker invoker = [&](int code, const StrCaseMap &headerOut, const HttpBody::Ptr &body) {
                 if (cookie && file_exist) {
                     auto& attach = cookie->getAttach<HttpCookieAttachment>();
-                    if (attach._is_hls) {
+                    if (attach._hls_data) {
                         attach._hls_data->addByteUsage(body->remainSize());
                     }
                 }
@@ -417,50 +405,37 @@ static void accessFile(TcpSession &sender, const Parser &parser, const MediaInfo
             invoker.responseFile(parser.getHeader(), httpHeader, file_content.empty() ? file_path : file_content, !is_hls, file_content.empty());
         };
 
-        if (!is_hls) {
+        if (!is_hls || !cookie) {
-            //不是hls,直接回复文件或404
+            //不是hls或访问m3u8文件不带cookie, 直接回复文件或404
             response_file(cookie, cb, file_path, parser);
+            if (is_hls) {
+                WarnL << "access m3u8 file without cookie:" << file_path;
+            }
             return;
         }
 
-        //是hls直播，判断HLS直播流是否已经注册
+        auto src = cookie->getAttach<HttpCookieAttachment>()._hls_data->getMediaSource();
-        bool have_find_media_src = false;
-        if (cookie) {
-            auto& attach = cookie->getAttach<HttpCookieAttachment>();
-            have_find_media_src = attach._have_find_media_source;
-            if (!have_find_media_src) {
-                const_cast<HttpCookieAttachment &>(attach)._have_find_media_source = true;
-            } else {
-                auto src = attach._hls_data->getMediaSource();
         if (src) {
             //直接从内存获取m3u8索引文件(而不是从文件系统)
             response_file(cookie, cb, file_path, parser, src->getIndexFile());
             return;
         }
-            }
-        }
-        if (have_find_media_src) {
-            //之前该cookie已经通过MediaSource::findAsync查找过了，所以现在只以文件系统查找结果为准
-            response_file(cookie, cb, file_path, parser);
-            return;
-        }
-        //hls文件不存在，我们等待其生成并延后回复
-        MediaSource::findAsync(media_info, strongSession, [response_file, cookie, cb, file_path, parser](const MediaSource::Ptr &src) {
-            if (cookie) {
-                //尝试添加HlsMediaSource的观看人数(HLS是按需生成的，这样可以触发HLS文件的生成)
-                auto &attach = cookie->getAttach<HttpCookieAttachment>();
-                attach._hls_data->addByteUsage(0);
-                attach._hls_data->setMediaSource(dynamic_pointer_cast<HlsMediaSource>(src));
-            }
 
+        //hls流可能未注册，MediaSource::findAsync可以触发not_found事件，然后再按需推拉流
+        MediaSource::findAsync(media_info, strongSession, [response_file, cookie, cb, file_path, parser](const MediaSource::Ptr &src) {
             auto hls = dynamic_pointer_cast<HlsMediaSource>(src);
             if (!hls) {
-                //流不存在，那么直接返回文件(相当于纯粹的HLS文件服务器,但是会挂起播放器15秒左右(用于等待HLS流的注册))
+                //流不在线
                 response_file(cookie, cb, file_path, parser);
                 return;
             }
 
-            //可能异步获取m3u8索引文件
+            auto &attach = cookie->getAttach<HttpCookieAttachment>();
+            attach._hls_data->setMediaSource(hls);
+            //添加HlsMediaSource的观看人数(HLS是按需生成的，这样可以触发HLS文件的生成)
+            attach._hls_data->addByteUsage(0);
+
+            // m3u8文件可能不存在, 等待m3u8索引文件按需生成
             hls->getIndexFile([response_file, file_path, cookie, cb, parser](const string &file) {
                 response_file(cookie, cb, file_path, parser, file);
             });
@@ -519,15 +494,15 @@ void HttpFileManager::onAccessPath(TcpSession &sender, Parser &parser, const Htt
             return;
         }
         //判断是否有权限访问该目录
-        canAccessPath(sender, parser, media_info, true, [strMenu, cb](const string &errMsg, const HttpServerCookie::Ptr &cookie) mutable{
+        canAccessPath(sender, parser, media_info, true, [strMenu, cb](const string &err_msg, const HttpServerCookie::Ptr &cookie) mutable{
-            if (!errMsg.empty()) {
+            if (!err_msg.empty()) {
-                strMenu = errMsg;
+                strMenu = err_msg;
             }
             StrCaseMap headerOut;
             if (cookie) {
                 headerOut["Set-Cookie"] = cookie->getCookie(cookie->getAttach<HttpCookieAttachment>()._path);
             }
-            cb(errMsg.empty() ? 200 : 401, "text/html", headerOut, std::make_shared<HttpStringBody>(strMenu));
+            cb(err_msg.empty() ? 200 : 401, "text/html", headerOut, std::make_shared<HttpStringBody>(strMenu));
         });
         return;
     }

src/Record/HlsMakerImp.cpp

@@ -117,7 +117,7 @@ void HlsMakerImp::onWriteHls(const std::string &data) {
         fwrite(data.data(), data.size(), 1, hls.get());
         hls.reset();
         if (_media_src) {
-            _media_src->registHls(data);
+            _media_src->setIndexFile(data);
         }
     } else {
         WarnL << "create hls file failed," << _path_hls << " " << get_uv_errmsg();

src/Record/HlsMediaSource.cpp

@@ -23,7 +23,7 @@ HlsCookieData::HlsCookieData(const MediaInfo &info, const std::shared_ptr<SockIn
 
 void HlsCookieData::addReaderCount() {
     if (!*_added) {
-        auto src = std::dynamic_pointer_cast<HlsMediaSource>(MediaSource::find(HLS_SCHEMA, _info._vhost, _info._app, _info._streamid));
+        auto src = getMediaSource();
         if (src) {
             *_added = true;
             _ring_reader = src->getRing()->attach(EventPollerPool::Instance().getPoller());

src/Record/HlsMediaSource.h

@@ -39,12 +39,10 @@ public:
     int readerCount() override { return _ring ? _ring->readerCount() : 0; }
 
     /**
-     * 生成m3u8文件时触发
+     * 设置或清空m3u8索引文件内容
-     * @param index_file m3u8文件内容
      */
-    void registHls(std::string index_file) {
+    void setIndexFile(std::string index_file) {
-        if (!_is_regist) {
+        if (!_ring) {
-            _is_regist = true;
             std::weak_ptr<HlsMediaSource> weakSelf = std::dynamic_pointer_cast<HlsMediaSource>(shared_from_this());
             auto lam = [weakSelf](int size) {
                 auto strongSelf = weakSelf.lock();
@@ -58,19 +56,19 @@ public:
             regist();
         }
 
-        if (index_file.empty()) {
-            //没产生索引文件, 只是为了触发媒体注册
-            return;
-        }
-
         //赋值m3u8索引文件内容
         std::lock_guard<std::mutex> lck(_mtx_index);
         _index_file = std::move(index_file);
 
+        if (!_index_file.empty()) {
             _list_cb.for_each([&](const std::function<void(const std::string &str)> &cb) { cb(_index_file); });
             _list_cb.clear();
         }
+    }
 
+    /**
+     * 异步获取m3u8文件
+     */
     void getIndexFile(std::function<void(const std::string &str)> cb) {
         std::lock_guard<std::mutex> lck(_mtx_index);
         if (!_index_file.empty()) {
@@ -81,6 +79,9 @@ public:
         _list_cb.emplace_back(std::move(cb));
     }
 
+    /**
+     * 同步获取m3u8文件
+     */
     std::string getIndexFile() const {
         std::lock_guard<std::mutex> lck(_mtx_index);
         return _index_file;
@@ -89,9 +90,7 @@ public:
     void onSegmentSize(size_t bytes) { _speed[TrackVideo] += bytes; }
 
 private:
-    bool _is_regist = false;
     RingType::Ptr _ring;
-
     std::string _index_file;
     mutable std::mutex _mtx_index;
     toolkit::List<std::function<void(const std::string &)>> _list_cb;

src/Record/HlsRecorder.h

@@ -38,13 +38,9 @@ public:
     void setListener(const std::weak_ptr<MediaSourceEvent> &listener) {
         setDelegate(listener);
         _hls->getMediaSource()->setListener(shared_from_this());
-        //先注册媒体流，后续可以按需生成
-        _hls->getMediaSource()->registHls("");
     }
 
-    int readerCount() {
+    int readerCount() { return _hls->getMediaSource()->readerCount(); }
-        return _hls->getMediaSource()->readerCount();
-    }
 
     void onReaderChanged(MediaSource &sender, int size) override {
         GET_CONFIG(bool, hls_demand, General::kHlsDemand);
@@ -57,17 +53,13 @@ public:
         MediaSourceEventInterceptor::onReaderChanged(sender, size);
     }
 
-    bool isEnabled() {
-        GET_CONFIG(bool, hls_demand, General::kHlsDemand);
-        //缓存尚未清空时，还允许触发inputFrame函数，以便及时清空缓存
-        return hls_demand ? (_clear_cache ? true : _enabled) : true;
-    }
-
     bool inputFrame(const Frame::Ptr &frame) override {
         GET_CONFIG(bool, hls_demand, General::kHlsDemand);
         if (_clear_cache && hls_demand) {
             _clear_cache = false;
+            //清空旧的m3u8索引文件于ts切片
             _hls->clearCache();
+            _hls->getMediaSource()->setIndexFile("");
         }
         if (_enabled || !hls_demand) {
             return MpegMuxer::inputFrame(frame);
@@ -75,6 +67,12 @@ public:
         return false;
     }
 
+    bool isEnabled() {
+        GET_CONFIG(bool, hls_demand, General::kHlsDemand);
+        //缓存尚未清空时，还允许触发inputFrame函数，以便及时清空缓存
+        return hls_demand ? (_clear_cache ? true : _enabled) : true;
+    }
+
 private:
     void onWrite(std::shared_ptr<toolkit::Buffer> buffer, uint32_t timestamp, bool key_pos) override {
         if (!buffer) {
@@ -85,8 +83,7 @@ private:
     }
 
 private:
-    //默认不生成hls文件，有播放器时再生成
+    bool _enabled = true;
-    bool _enabled = false;
     bool _clear_cache = false;
     std::shared_ptr<HlsMakerImp> _hls;
 };