同步代码

解决媒体注销但还保持无限重试推流的bug

3rdpart/ZLToolKit

@@ -1 +1 @@
-Subproject commit 493d14e1682cc0c79e72b3893dd98f865d45b8e9
+Subproject commit 7302286cf4be39d416b023fec3fd4ca9c54af762

CMakeLists.txt

@@ -44,6 +44,7 @@ option(ENABLE_FFMPEG "Enable FFmpeg" OFF)
 option(ENABLE_HLS "Enable HLS" ON)
 option(ENABLE_JEMALLOC_STATIC "Enable static linking to the jemalloc library" OFF)
 option(ENABLE_JEMALLOC_DUMP "Enable jemalloc to dump malloc statistics" OFF)
+option(ENABLE_TCMALLOC "Enable linking to the tcmalloc library" OFF)
 option(ENABLE_MEM_DEBUG "Enable Memory Debug" OFF)
 option(ENABLE_MP4 "Enable MP4" ON)
 option(ENABLE_MSVC_MT "Enable MSVC Mt/Mtd lib" ON)
@@ -257,8 +258,8 @@ endif()
 # Multiple modules depend on ffmpeg related libraries, unified search
 if(ENABLE_FFMPEG)
   find_package(PkgConfig QUIET)
-  # 查找 ffmpeg/libutil 是否安装
-  # find ffmpeg/libutil installed
+  # 查找 ffmpeg/libavutil 是否安装
+  # find ffmpeg/libavutil installed
   if(PKG_CONFIG_FOUND)
     pkg_check_modules(AVUTIL QUIET IMPORTED_TARGET libavutil)
     if(AVUTIL_FOUND)
@@ -297,8 +298,19 @@ if(ENABLE_FFMPEG)
     endif()
   endif()
 
-  # 查找 ffmpeg/libutil 是否安装
-  # find ffmpeg/libutil installed
+  # 查找 ffmpeg/libavfilter 是否安装
+  # find ffmpeg/libavfilter installed
+  if(PKG_CONFIG_FOUND)
+    pkg_check_modules(AVFILTER QUIET IMPORTED_TARGET libavfilter)
+    if(AVFILTER_FOUND)
+      update_cached_list(MK_LINK_LIBRARIES PkgConfig::AVFILTER)
+      message(STATUS "found library: ${AVFILTER_LIBRARIES}")
+    endif()
+  endif()
+
+
+  # 查找 ffmpeg/libavutil 是否安装
+  # find ffmpeg/libavutil installed
   if(NOT AVUTIL_FOUND)
     find_package(AVUTIL QUIET)
     if(AVUTIL_FOUND)
@@ -341,7 +353,16 @@ if(ENABLE_FFMPEG)
     endif()
   endif()
 
-  if(AVUTIL_FOUND AND AVCODEC_FOUND AND SWSCALE_FOUND AND SWRESAMPLE_FOUND)
+  if(NOT AVFILTER_FOUND)
+    find_package(AVFILTER QUIET)
+    if(AVFILTER_FOUND)
+      include_directories(SYSTEM ${AVFILTER_INCLUDE_DIR})
+      update_cached_list(MK_LINK_LIBRARIES ${AVFILTER_LIBRARIES})
+      message(STATUS "found library: ${AVFILTER_LIBRARIES}")
+    endif()
+  endif()
+
+  if(AVUTIL_FOUND AND AVCODEC_FOUND AND SWSCALE_FOUND AND SWRESAMPLE_FOUND AND AVFILTER_FOUND)
     update_cached_list(MK_COMPILE_DEFINITIONS ENABLE_FFMPEG)
     update_cached_list(MK_LINK_LIBRARIES ${CMAKE_DL_LIBS})
   else()
@@ -402,6 +423,19 @@ if(JEMALLOC_FOUND)
   endif ()
 endif()
 
+# 查找 tcmalloc 是否安装
+# find tcmalloc installed
+if(ENABLE_TCMALLOC)
+  find_package(TCMALLOC QUIET)
+  if(TCMALLOC_FOUND)
+    message(STATUS "Link with tcmalloc library: ${TCMALLOC_LIBRARIES}")
+    update_cached_list(MK_LINK_LIBRARIES ${TCMALLOC_LIBRARIES})
+  else()
+    set(ENABLE_TCMALLOC OFF)
+    message(WARNING "tcmalloc 相关功能未找到")
+  endif()
+endif()
+
 # 查找 openssl 是否安装
 # find openssl installed
 find_package(OpenSSL QUIET)
@@ -580,6 +614,9 @@ if (ENABLE_PYTHON)
   )
 endif ()
 
+if (ENABLE_FFMPEG)
+  file(COPY "${CMAKE_CURRENT_SOURCE_DIR}/DejaVuSans.ttf" DESTINATION ${EXECUTABLE_OUTPUT_PATH})
+endif ()
 # 拷贝VideoStack 无视频流时默认填充的背景图片
 # Copy the default background image used by VideoStack when there is no video stream
 if (ENABLE_VIDEOSTACK AND ENABLE_FFMPEG AND ENABLE_X264)

api/source/mk_common.cpp

@@ -327,6 +327,7 @@ API_EXPORT uint16_t API_CALL mk_ice_server_start(uint16_t port){
         iceServer_udp = std::make_shared<UdpServer>();
         iceServer_udp->start<IceSession>(port);
         iceServer_tcp->start<IceSession>(port);
+        return 0;
     } catch (std::exception &ex) {
         iceServer_udp = nullptr;
         iceServer_tcp = nullptr;

cmake/FindAVFILTER.cmake

@@ -0,0 +1,16 @@
+find_path(AVFILTER_INCLUDE_DIR
+        NAMES libavfilter/avfilter.h
+        HINTS ${FFMPEG_PATH_ROOT}
+        PATH_SUFFIXES include)
+
+find_library(AVFILTER_LIBRARY
+        NAMES avfilter
+        HINTS ${FFMPEG_PATH_ROOT}
+        PATH_SUFFIXES bin lib)
+
+set(AVFILTER_LIBRARIES ${AVFILTER_LIBRARY})
+set(AVFILTER_INCLUDE_DIRS ${AVFILTER_INCLUDE_DIR})
+
+include(FindPackageHandleStandardArgs)
+
+find_package_handle_standard_args(AVFILTER DEFAULT_MSG AVFILTER_LIBRARY AVFILTER_INCLUDE_DIR)

cmake/FindTCMALLOC.cmake

@@ -0,0 +1,16 @@
+find_path(Tcmalloc_INCLUDE_DIR
+  NAMES google/tcmalloc.h
+)
+
+find_library(Tcmalloc_LIBRARY
+  NAMES tcmalloc_minimal tcmalloc
+)
+
+set(TCMALLOC_LIBRARIES ${Tcmalloc_LIBRARY})
+set(TCMALLOC_INCLUDE_DIRS ${Tcmalloc_INCLUDE_DIR})
+
+INCLUDE(FindPackageHandleStandardArgs)
+FIND_PACKAGE_HANDLE_STANDARD_ARGS(TCMALLOC
+  DEFAULT_MSG
+  TCMALLOC_LIBRARIES TCMALLOC_INCLUDE_DIRS
+)

conf/config.ini

@@ -412,10 +412,14 @@ nackMaxMS=3000
 nackMaxCount=15
 #nack重传频率，rtt的倍数
 nackIntervalRatio=1.0
-#nack包中rtp个数，减小此值可以让nack包响应更灵敏
+#视频nack包中rtp个数，减小此值可以让nack包响应更灵敏
 nackRtpSize=8
+#音频nack包中rtp个数，减小此值可以让nack包响应更灵敏
+nackAudioRtpSize=4
 #是否尝试过滤 b帧
 bfilter=0
+# 是否优先采用webrtc over tcp模式
+preferred_tcp=0
 
 [srt]
 #srt播放推流、播放超时时间,单位秒

ext-codec/AV1.h

@@ -57,7 +57,7 @@ public:
     toolkit::Buffer::Ptr getExtraData() const override;
     void setExtraData(const uint8_t *data, size_t size) override;
 protected:
-    aom_av1_t _context = {0};
+    aom_av1_t _context {};
 };
 
 } // namespace mediakit

ext-codec/G711.cpp

@@ -20,7 +20,7 @@ using namespace toolkit;
 namespace mediakit {
 
 Buffer::Ptr G711Track::getExtraData() const {
-    struct wave_format_t wav = {0};
+    struct wave_format_t wav {};
     wav.wFormatTag = getCodecId() == CodecG711A ? WAVE_FORMAT_ALAW : WAVE_FORMAT_MULAW;
     wav.nChannels = getAudioChannel();
     wav.nSamplesPerSec = getAudioSampleRate();

ext-codec/Opus.cpp

@@ -28,7 +28,7 @@ void OpusTrack::setExtraData(const uint8_t *data, size_t size) {
 }
 
 Buffer::Ptr OpusTrack::getExtraData() const {
-    struct opus_head_t opus = { 0 };
+    struct opus_head_t opus {};
     opus.version = 1;
     opus.channels = getAudioChannel();
     opus.input_sample_rate = getAudioSampleRate();

ext-codec/VP8.h

@@ -41,7 +41,7 @@ public:
     toolkit::Buffer::Ptr getExtraData() const override;
     void setExtraData(const uint8_t *data, size_t size) override;
 private:
-    webm_vpx_t _vpx = {0};
+    webm_vpx_t _vpx {};
 };
 
 } // namespace mediakit

ext-codec/VP8Rtp.cpp

@@ -334,7 +334,7 @@ bool VP8RtpEncoder::inputFrame(const Frame::Ptr &frame) {
     bool key = frame->keyFrame();
     bool mark = false;
     for (size_t pos = 0; pos < len; pos += pdu_size) {
-        if (len - pos <= pdu_size) {
+        if (static_cast<int>(len - pos) <= pdu_size) {
             pdu_size = len - pos;
             mark = true;
         }

ext-codec/VP8Rtp.h

@@ -56,9 +56,6 @@ public:
     using Ptr = std::shared_ptr<VP8RtpEncoder>;
 
     bool inputFrame(const Frame::Ptr &frame) override;
-
-private:
-    uint16_t _pic_id = 0;
 };
 
 }//namespace mediakit

ext-codec/VP9.h

@@ -41,7 +41,7 @@ public:
     toolkit::Buffer::Ptr getExtraData() const override;
     void setExtraData(const uint8_t *data, size_t size) override;
 private:
-    webm_vpx_t _vpx = {0};
+    webm_vpx_t _vpx {};
 };
 
 } // namespace mediakit

ext-codec/VP9Rtp.cpp

@@ -297,7 +297,7 @@ bool VP9RtpEncoder::inputFrame(const Frame::Ptr &frame) {
     int pdu_size = getRtpInfo().getMaxSize() - nheader;
 
     bool mark = false;
-    for (size_t pos = 0; pos < len; pos += pdu_size) {
+    for (int pos = 0; pos < len; pos += pdu_size) {
         if (len - pos <= pdu_size) {
             pdu_size = len - pos;
             header[0] |= kEBit;

server/CMakeLists.txt

@@ -50,6 +50,10 @@ target_compile_definitions(MediaServer
 target_compile_options(MediaServer
   PRIVATE ${COMPILE_OPTIONS_DEFAULT})
 
+if(MINGW)
+  update_cached_list(MK_LINK_LIBRARIES dbghelp)
+endif()
+
 if(CMAKE_SYSTEM_NAME MATCHES "Linux")
   target_link_libraries(MediaServer -Wl,--start-group ${MK_LINK_LIBRARIES} -Wl,--end-group)
 else()

server/WebApi.cpp

@@ -115,8 +115,6 @@ static void responseApi(int code, const string &msg, const HttpSession::HttpResp
     responseApi(res, invoker);
 }
 
-static ApiArgsType getAllArgs(const Parser &parser);
-
 static HttpApi toApi(const function<void(API_ARGS_MAP_ASYNC)> &cb) {
     return [cb](const Parser &parser, const HttpSession::HttpResponseInvoker &invoker, SockInfo &sender) {
         GET_CONFIG(string, charSet, Http::kCharSet);
@@ -215,7 +213,7 @@ void api_regist(const string &api_path, const function<void(API_ARGS_STRING_ASYN
 
 // 获取HTTP请求中url参数、content参数  [AUTO-TRANSLATED:d161a1e1]
 // Get URL parameters and content parameters from the HTTP request
-static ApiArgsType getAllArgs(const Parser &parser) {
+ApiArgsType getAllArgs(const Parser &parser) {
     ApiArgsType allArgs;
     if (parser["Content-Type"].find("application/x-www-form-urlencoded") == 0) {
         auto contentArgs = parser.parseArgs(parser.content());
@@ -1079,6 +1077,7 @@ void installWebApi() {
             }
             fillSockInfo(jsession, session.get());
             jsession["id"] = id;
+            jsession["type"] = session->getSock()->sockType() == SockNum::Sock_TCP ? "tcp" : "udp";
             jsession["typeid"] = toolkit::demangle(typeid(*session).name());
             val["data"].append(jsession);
         });

server/WebApi.h

@@ -238,6 +238,7 @@ uint16_t openRtpServer(uint16_t local_port, const mediakit::MediaTuple &tuple, i
 #endif
 
 Json::Value makeMediaSourceJson(mediakit::MediaSource &media);
+ApiArgsType getAllArgs(const mediakit::Parser &parser);
 void getStatisticJson(const std::function<void(Json::Value &val)> &cb);
 void addStreamProxy(const mediakit::MediaTuple &tuple, const std::string &url, int retry_count,
                     const mediakit::ProtocolOption &option, int rtp_type, float timeout_sec, const toolkit::mINI &args,

server/WebHook.cpp

@@ -18,6 +18,7 @@
 #include "Http/HttpRequester.h"
 #include "Network/Session.h"
 #include "Rtsp/RtspSession.h"
+#include "Player/PlayerProxy.h"
 #include "WebHook.h"
 #include "WebApi.h"
 
@@ -500,10 +501,6 @@ void installWebHook() {
     // 监听rtsp、rtmp源注册或注销事件  [AUTO-TRANSLATED:6396afa8]
     // Listen to rtsp, rtmp source registration or deregistration events
     NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastMediaChanged, [](BroadcastMediaChangedArgs) {
-        GET_CONFIG(string, hook_stream_changed, Hook::kOnStreamChanged);
-        if (!hook_enable || hook_stream_changed.empty()) {
-            return;
-        }
         GET_CONFIG_FUNC(std::set<std::string>, stream_changed_set, Hook::kStreamChangedSchemas, [](const std::string &str) {
             std::set<std::string> ret;
             auto vec = split(str, "/");
@@ -520,6 +517,15 @@ void installWebHook() {
             // This protocol registration deregistration event is ignored
             return;
         }
+#if defined(ENABLE_PYTHON)
+        if (PythonInvoker::Instance().on_media_changed(bRegist, sender)) {
+            return;
+        }
+#endif
+        GET_CONFIG(string, hook_stream_changed, Hook::kOnStreamChanged);
+        if (!hook_enable || hook_stream_changed.empty()) {
+            return;
+        }
 
         ArgsType body;
         if (bRegist) {
@@ -799,6 +805,14 @@ void installWebHook() {
         do_http_hook(rtp_server_timeout, body);
     });
 
+    NoticeCenter::Instance().addListener(&web_hook_tag, Broadcast::kBroadcastPlayerProxyFailed, [](BroadcastPlayerProxyFailedArgs) {
+#if defined(ENABLE_PYTHON)
+        if (PythonInvoker::Instance().on_player_proxy_failed(sender, ex)) {
+            return;
+        }
+#endif
+    });
+
     // 汇报服务器重新启动  [AUTO-TRANSLATED:bd7d83df]
     // Report server restart
     reportServerStarted();

server/main.cpp

@@ -273,6 +273,16 @@ int start_main(int argc,char *argv[]) {
         }
 #endif //! defined(_WIN32)
 
+        // 设置poller线程数和cpu亲和性,该函数必须在使用ZLToolKit网络相关对象之前调用才能生效  [AUTO-TRANSLATED:7f03a1e5]
+        // Set the number of poller threads and CPU affinity. This function must be called before using ZLToolKit network related objects to take effect.
+        // 如果需要调用getSnap和addFFmpegSource接口，可以关闭cpu亲和性  [AUTO-TRANSLATED:7629f7bc]
+        // If you need to call the getSnap and addFFmpegSource interfaces, you can turn off CPU affinity
+
+        EventPollerPool::setPoolSize(threads);
+        WorkThreadPool::setPoolSize(threads);
+        EventPollerPool::enableCpuAffinity(affinity);
+        WorkThreadPool::enableCpuAffinity(affinity);
+
         // 开启崩溃捕获等  [AUTO-TRANSLATED:9c7c759c]
         // Enable crash capture, etc.
         System::systemSetup();
@@ -329,15 +339,6 @@ int start_main(int argc,char *argv[]) {
         uint16_t httpsPort = mINI::Instance()[Http::kSSLPort];
         uint16_t rtpPort = mINI::Instance()[RtpProxy::kPort];
 
-        // 设置poller线程数和cpu亲和性,该函数必须在使用ZLToolKit网络相关对象之前调用才能生效  [AUTO-TRANSLATED:7f03a1e5]
-        // Set the number of poller threads and CPU affinity. This function must be called before using ZLToolKit network related objects to take effect.
-        // 如果需要调用getSnap和addFFmpegSource接口，可以关闭cpu亲和性  [AUTO-TRANSLATED:7629f7bc]
-        // If you need to call the getSnap and addFFmpegSource interfaces, you can turn off CPU affinity
-
-        EventPollerPool::setPoolSize(threads);
-        WorkThreadPool::setPoolSize(threads);
-        EventPollerPool::enableCpuAffinity(affinity);
-
         // 简单的telnet服务器，可用于服务器调试，但是不能使用23端口，否则telnet上了莫名其妙的现象  [AUTO-TRANSLATED:f9324c6e]
         // Simple telnet server, can be used for server debugging, but cannot use port 23, otherwise telnet will have inexplicable phenomena
         // 测试方法:telnet 127.0.0.1 9000  [AUTO-TRANSLATED:de0ac883]
@@ -508,9 +509,11 @@ int start_main(int argc,char *argv[]) {
 #endif
 
 #if defined(ENABLE_PYTHON)
+        // 初始化python解释器
+        auto &ref = PythonInvoker::Instance();
         auto py_plugin = mINI::Instance()[Python::kPlugin];
         if (!py_plugin.empty()) {
-            PythonInvoker::Instance().load(py_plugin);
+            ref.load(py_plugin);
         }
 #endif
         sem.wait();
@@ -519,6 +522,10 @@ int start_main(int argc,char *argv[]) {
     unInstallWebHook();
     onProcessExited();
 
+#if defined(ENABLE_PYTHON)
+    PythonInvoker::release();
+#endif
+
     // 休眠1秒再退出，防止资源释放顺序错误  [AUTO-TRANSLATED:1b11a74f]
     // sleep for 1 second before exiting, to prevent resource release order errors
     InfoL << "程序退出中,请等待...";

server/pyinvoker.cpp

@@ -7,7 +7,10 @@
 #include <iostream>
 #include <string>
 #include <type_traits>
+#include "WebApi.h"
 #include "WebHook.h"
+#include "Util/util.h"
+#include "Util/File.h"
 #include "Common/Parser.h"
 #include "Http/HttpSession.h"
 
@@ -70,6 +73,11 @@ py::dict to_python(const SockInfo &info) {
     return jsonToPython(json);
 }
 
+template <typename T>
+std::shared_ptr<T> to_python2(const T &t) {
+    return std::shared_ptr<T>(const_cast<T *>(&t), py::nodelete());
+}
+
 template <typename T>
 T &to_native(const py::capsule &cap) {
     static auto name_str = toolkit::demangle(typeid(T).name());
@@ -110,6 +118,22 @@ void handle_http_request(const py::object &check_route, const py::object &submit
     }
     consumed = true;
 
+    // http api被python拦截了，再api统一鉴权
+    try {
+        auto args = getAllArgs(parser);
+        auto allArgs = ArgsMap(parser, args);
+        GET_CONFIG(std::string, api_secret, API::kSecret);
+        CHECK_SECRET(); // 检测secret
+    } catch (std::exception &ex) {
+        Json::Value val;
+        val["code"] = API::Exception;
+        val["msg"] = ex.what();
+        HttpSession::KeyValue headerOut;
+        headerOut["Content-Type"] = "application/json";
+        invoker(200, headerOut, val.toStyledString());
+        return;
+    }
+
     StrCaseMap resp_headers;
     std::string resp_body;
     int status = 500;
@@ -152,11 +176,18 @@ PYBIND11_EMBEDDED_MODULE(mk_loader, m) {
         }
         return "";
     });
+
+    m.def("get_full_path", [](const std::string &path, const std::string &current_path) -> std::string {
+        py::gil_scoped_release release;
+        return File::absolutePath(path, current_path);
+    });
+
     m.def("set_config", [](const std::string &key, const std::string &value) -> bool {
         py::gil_scoped_release release;
         mINI::Instance()[key]= value;
         return true;
     });
+
     m.def("update_config", []() {
         NOTICE_EMIT(BroadcastReloadConfigArgs, Broadcast::kBroadcastReloadConfig);
         mINI::Instance().dumpFile(g_ini_file);
@@ -171,6 +202,7 @@ PYBIND11_EMBEDDED_MODULE(mk_loader, m) {
         auto &invoker = to_native<Broadcast::PublishAuthInvoker>(cap);
         invoker(err, option);
     });
+
     m.def("auth_invoker_do", [](const py::capsule &cap, const std::string &err) {
         // 执行c++代码时释放gil锁
         py::gil_scoped_release release;
@@ -186,6 +218,46 @@ PYBIND11_EMBEDDED_MODULE(mk_loader, m) {
         });
     });
 
+    py::enum_<TrackType>(m, "TrackType")
+        .value("Invalid", TrackInvalid)
+        .value("Video", TrackVideo)
+        .value("Audio", TrackAudio)
+        .value("Title", TrackTitle)
+        .value("Application", TrackApplication)
+        .export_values();
+
+    py::class_<MediaSource, MediaSource::Ptr>(m, "MediaSource")
+        .def("getSchema", &MediaSource::getSchema)
+        .def("getUrl", &MediaSource::getUrl)
+        .def("getMediaTuple", &MediaSource::getMediaTuple)
+        .def("getTimeStamp", &MediaSource::getTimeStamp)
+        .def("setTimeStamp", &MediaSource::setTimeStamp)
+        .def("getBytesSpeed", &MediaSource::getBytesSpeed)
+        .def("getTotalBytes", &MediaSource::getTotalBytes)
+        .def("getCreateStamp", &MediaSource::getCreateStamp)
+        .def("getAliveSecond", &MediaSource::getAliveSecond)
+        .def("readerCount", &MediaSource::readerCount)
+        .def("totalReaderCount", &MediaSource::totalReaderCount)
+        .def("getOriginType", &MediaSource::getOriginType)
+        .def("getOriginUrl", &MediaSource::getOriginUrl)
+        .def("getOriginSock", &MediaSource::getOriginSock)
+        .def("seekTo", &MediaSource::seekTo)
+        .def("pause", &MediaSource::pause)
+        .def("speed", &MediaSource::speed)
+        .def("close", &MediaSource::close)
+        .def("setupRecord", &MediaSource::setupRecord)
+        .def("isRecording", &MediaSource::isRecording)
+        .def("stopSendRtp", &MediaSource::stopSendRtp)
+        .def("getLossRate", &MediaSource::getLossRate);
+
+    py::class_<MediaTuple, std::shared_ptr<MediaTuple>>(m, "MediaTuple")
+        .def_readwrite("vhost", &MediaTuple::vhost)
+        .def_readwrite("app", &MediaTuple::app)
+        .def_readwrite("stream", &MediaTuple::stream)
+        .def_readwrite("params", &MediaTuple::params)
+        .def("shortUrl", &MediaTuple::shortUrl);
+
+    py::class_<SockException, std::shared_ptr<SockException>>(m, "SockException").def("what", &SockException::what).def("code", &SockException::getErrCode);
 }
 
 namespace mediakit {
@@ -215,9 +287,18 @@ bool set_python_path() {
     return true;
 }
 
+static std::shared_ptr<PythonInvoker> g_instance;
+
 PythonInvoker &PythonInvoker::Instance() {
-    static std::shared_ptr<PythonInvoker> instance(new PythonInvoker);
-    return *instance;
+    static toolkit::onceToken s_token([]() {
+        g_instance.reset(new PythonInvoker);
+    });
+
+    return *g_instance;
+}
+
+void PythonInvoker::release() {
+    g_instance = nullptr;
 }
 
 PythonInvoker::PythonInvoker() {
@@ -244,32 +325,34 @@ PythonInvoker::~PythonInvoker() {
         }
         _on_exit = py::object();
         _on_publish = py::object();
+        _on_play = py::object();
+        _on_flow_report = py::object();
+        _on_reload_config = py::object();
+        _on_media_changed = py::object();
+        _on_player_proxy_failed = py::object();
         _module = py::module();
     }
-
     delete _rel;
     delete _interpreter;
 }
 
+#define GET_FUNC(instance, name) \
+    if (hasattr(instance, #name)) { \
+        _##name = instance.attr(#name); \
+    }
+
 void PythonInvoker::load(const std::string &module_name) {
     try {
         py::gil_scoped_acquire gil; // 加锁
         _module = py::module::import(module_name.c_str());
-        if (hasattr(_module, "on_exit")) {
-            _on_exit = _module.attr("on_exit");
-        }
-        if (hasattr(_module, "on_publish")) {
-            _on_publish = _module.attr("on_publish");
-        }
-        if (hasattr(_module, "on_play")) {
-            _on_play = _module.attr("on_play");
-        }
-        if (hasattr(_module, "on_flow_report")) {
-            _on_flow_report = _module.attr("on_flow_report");
-        }
-        if (hasattr(_module, "on_reload_config")) {
-            _on_reload_config = _module.attr("on_reload_config");
-        }
+        GET_FUNC(_module, on_exit);
+        GET_FUNC(_module, on_publish);
+        GET_FUNC(_module, on_play);
+        GET_FUNC(_module, on_flow_report);
+        GET_FUNC(_module, on_reload_config);
+        GET_FUNC(_module, on_media_changed);
+        GET_FUNC(_module, on_player_proxy_failed);
+
         if (hasattr(_module, "on_start")) {
             py::object on_start = _module.attr("on_start");
             if (on_start) {
@@ -305,6 +388,22 @@ bool PythonInvoker::on_flow_report(BroadcastFlowReportArgs) const {
     return _on_flow_report(to_python(args), totalBytes, totalDuration, isPlayer, to_python(sender)).cast<bool>();
 }
 
+bool PythonInvoker::on_media_changed(BroadcastMediaChangedArgs) const {
+    py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
+    if (!_on_media_changed) {
+        return false;
+    }
+    return _on_media_changed(bRegist, to_python2(sender)).cast<bool>();
+}
+
+bool PythonInvoker::on_player_proxy_failed(BroadcastPlayerProxyFailedArgs) const {
+    py::gil_scoped_acquire gil; // 确保在 Python 调用期间持有 GIL
+    if (!_on_player_proxy_failed) {
+        return false;
+    }
+    return _on_player_proxy_failed(sender.getUrl(), to_python2(sender.getMediaTuple()), to_python2(ex)).cast<bool>();
+}
+
 } // namespace mediakit
 
 #endif

server/pyinvoker.h

@@ -11,6 +11,7 @@
 #include "Util/logger.h"
 #include "Common/config.h"
 #include "Common/MediaSource.h"
+#include "Player/PlayerProxy.h"
 
 namespace py = pybind11;
 
@@ -21,11 +22,14 @@ public:
     ~PythonInvoker();
 
     static PythonInvoker& Instance();
+    static void release();
 
     void load(const std::string &module_name);
     bool on_publish(BroadcastMediaPublishArgs) const;
     bool on_play(BroadcastMediaPlayedArgs) const;
     bool on_flow_report(BroadcastFlowReportArgs) const;
+    bool on_media_changed(BroadcastMediaChangedArgs) const;
+    bool on_player_proxy_failed(BroadcastPlayerProxyFailedArgs) const;
 
 private:
     PythonInvoker();
@@ -46,6 +50,10 @@ private:
     py::object _on_flow_report;
     // 配置文件热更新回调
     py::object _on_reload_config;
+    // 媒体注册注销
+    py::object _on_media_changed;
+    // 拉流代理失败
+    py::object _on_player_proxy_failed;
 };
 
 } // namespace mediakit

src/Codec/Transcode.cpp

@@ -231,10 +231,6 @@ FFmpegFrame::FFmpegFrame(std::shared_ptr<AVFrame> frame) {
 }
 
 FFmpegFrame::~FFmpegFrame() {
-    if (_data) {
-        delete[] _data;
-        _data = nullptr;
-    }
 }
 
 AVFrame *FFmpegFrame::get() const {
@@ -242,9 +238,9 @@ AVFrame *FFmpegFrame::get() const {
 }
 
 void FFmpegFrame::fillPicture(AVPixelFormat target_format, int target_width, int target_height) {
-    assert(_data == nullptr);
-    _data = new char[av_image_get_buffer_size(target_format, target_width, target_height, 32)];
-    av_image_fill_arrays(_frame->data, _frame->linesize, (uint8_t *) _data,  target_format, target_width, target_height, 32);
+    auto buffer_size = av_image_get_buffer_size(target_format, target_width, target_height, 32);
+    _data = std::unique_ptr<char[]>(new char[buffer_size]);
+    av_image_fill_arrays(_frame->data, _frame->linesize, (uint8_t *)_data.get(), target_format, target_width, target_height, 32);
 }
 
 int FFmpegFrame::getChannels() const {
@@ -256,6 +252,14 @@ int FFmpegFrame::getChannels() const {
 #endif
 }
 
+// 资源池复用前调用
+void FFmpegFrame::reset() {
+    _data.reset();
+    if (_frame) {
+        av_frame_unref(_frame.get()); // 清理AVFrame数据引用
+    }
+}
+
 ///////////////////////////////////////////////////////////////////////////
 
 template<bool decoder = true>
@@ -734,6 +738,7 @@ FFmpegFrame::Ptr FFmpegSws::inputFrame(const FFmpegFrame::Ptr &frame, int &ret,
     }
     if (_ctx) {
         auto out = _sws_frame_pool.obtain2();
+        out->reset(); // 清理旧数据和帧引用
         if (!out->get()->data[0]) {
             if (data) {
                 av_image_fill_arrays(out->get()->data, out->get()->linesize, data, _target_format, target_width, target_height, 32);
@@ -756,45 +761,16 @@ FFmpegFrame::Ptr FFmpegSws::inputFrame(const FFmpegFrame::Ptr &frame, int &ret,
     return nullptr;
 }
 
-std::tuple<bool, std::string> FFmpegUtils::saveFrame(const FFmpegFrame::Ptr &frame, const char *filename, AVPixelFormat fmt) {
+std::tuple<bool, std::string> FFmpegUtils::saveFrame(const FFmpegFrame::Ptr &frame, const char *filename, AVPixelFormat fmt, int w, int h, const char *font_path) {
+    std::shared_ptr<AVFilterGraph> _filter_graph;
+    AVFilterContext *buffersrc_ctx = nullptr;
+    AVFilterContext *buffersink_ctx = nullptr;
+    const AVFilter *buffersrc = nullptr;
+    const AVFilter *buffersink = nullptr;
+    // kServerName
+    const string mark = "ZLMediaKit"; 
+    char drawtext_args1[512];
     _StrPrinter ss;
-    const AVCodec *jpeg_codec = avcodec_find_encoder(fmt == AV_PIX_FMT_YUVJ420P ? AV_CODEC_ID_MJPEG : AV_CODEC_ID_PNG);
-    std::unique_ptr<AVCodecContext, void (*)(AVCodecContext *)> jpeg_codec_ctx(
-        jpeg_codec ? avcodec_alloc_context3(jpeg_codec) : nullptr, [](AVCodecContext *ctx) { avcodec_free_context(&ctx); });
-
-    if (!jpeg_codec_ctx) {
-        ss << "Could not allocate JPEG/PNG codec context";
-        DebugL << ss;
-        return make_tuple<bool, std::string>(false, ss.data());
-    }
-
-    jpeg_codec_ctx->width = frame->get()->width;
-    jpeg_codec_ctx->height = frame->get()->height;
-    jpeg_codec_ctx->pix_fmt = fmt;
-    jpeg_codec_ctx->time_base = { 1, 1 };
-
-    auto ret = avcodec_open2(jpeg_codec_ctx.get(), jpeg_codec, NULL);
-    if (ret < 0) {
-        ss << "Could not open JPEG/PNG codec, " << ffmpeg_err(ret);
-        DebugL << ss;
-        return make_tuple<bool, std::string>(false, ss.data());
-    }
-
-    FFmpegSws sws(fmt, 0, 0);
-    auto new_frame = sws.inputFrame(frame);
-    if (!new_frame) {
-        ss << "Could not scale the frame: " << ffmpeg_err(ret);
-        DebugL << ss;
-        return make_tuple<bool, std::string>(false, ss.data());
-    }
-
-    auto pkt = alloc_av_packet();
-    ret = avcodec_send_frame(jpeg_codec_ctx.get(), new_frame->get());
-    if (ret < 0) {
-        ss << "Error sending a frame for encoding, " << ffmpeg_err(ret);
-        DebugL << ss;
-        return make_tuple<bool, std::string>(false, ss.data());
-    }
 
     std::unique_ptr<FILE, void (*)(FILE *)> tmp_save_file_jpg(File::create_file(filename, "wb"), [](FILE *fp) {
         if (fp) {
@@ -808,10 +784,104 @@ std::tuple<bool, std::string> FFmpegUtils::saveFrame(const FFmpegFrame::Ptr &fra
         return make_tuple<bool, std::string>(false, ss.data());
     }
 
-    while (avcodec_receive_packet(jpeg_codec_ctx.get(), pkt.get()) == 0) {
-        fwrite(pkt.get()->data, pkt.get()->size, 1, tmp_save_file_jpg.get());
+    std::string fontfile("");
+    if (font_path && File::fileExist(font_path)) {
+        fontfile = font_path;
+    } else {
+        // Fallback to common default
+        fontfile = exeDir() + "/DejaVuSans.ttf";
+    }
+
+    snprintf(drawtext_args1, sizeof(drawtext_args1), "text='%s':fontfile='%s':fontcolor=white@0.1:fontsize=h/50:x=w*0.02:y=h-th-h*0.02", mark.data(), fontfile.c_str());
+
+    const AVCodec *jpeg_codec = avcodec_find_encoder(fmt == AV_PIX_FMT_YUVJ420P ? AV_CODEC_ID_MJPEG : AV_CODEC_ID_PNG);
+    std::unique_ptr<AVCodecContext, void (*)(AVCodecContext *)> jpeg_codec_ctx(
+        jpeg_codec ? avcodec_alloc_context3(jpeg_codec) : nullptr, [](AVCodecContext *ctx) { avcodec_free_context(&ctx); });
+
+    if (!jpeg_codec_ctx) {
+        ss << "Could not allocate JPEG/PNG codec context";
+        DebugL << ss;
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    jpeg_codec_ctx->width = (w > 0 && w < 8192) ? w : frame->get()->width;
+    jpeg_codec_ctx->height = (h > 0 && h < 4320) ? h : frame->get()->height;
+    jpeg_codec_ctx->pix_fmt = fmt;
+    jpeg_codec_ctx->time_base = { 1, 1 };
+
+    auto ret = avcodec_open2(jpeg_codec_ctx.get(), jpeg_codec, NULL);
+    if (ret < 0) {
+        ss << "Could not open JPEG/PNG codec, " << ffmpeg_err(ret);
+        DebugL << ss;
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    FFmpegSws sws(fmt, jpeg_codec_ctx->width, jpeg_codec_ctx->height);
+    auto new_frame = sws.inputFrame(frame);
+    if (!new_frame) {
+        ss << "Could not scale the frame";
+        DebugL << ss;
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    _filter_graph.reset(avfilter_graph_alloc(), [](AVFilterGraph *ctx) { avfilter_graph_free(&ctx); });
+    if (!_filter_graph) {
+        ss << "avfilter_graph_alloc failed";
+        DebugL << ss;
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    char args[512];
+    snprintf(
+        args, sizeof(args), "video_size=%dx%d:pix_fmt=%d:time_base=%d/%d:pixel_aspect=%d/%d", jpeg_codec_ctx->width, jpeg_codec_ctx->height,
+        jpeg_codec_ctx->pix_fmt, jpeg_codec_ctx->time_base.num, jpeg_codec_ctx->time_base.den, jpeg_codec_ctx->sample_aspect_ratio.num,
+        jpeg_codec_ctx->sample_aspect_ratio.den);
+
+    buffersrc = avfilter_get_by_name("buffer");
+
+    if ((ret = avfilter_graph_create_filter(&buffersrc_ctx, buffersrc, "in", args, NULL, _filter_graph.get())) < 0) {
+        ss << "avfilter_graph_create_filter buffersrc failed: " << ret << " " << ffmpeg_err(ret);
+        DebugL << ss;
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    buffersink = avfilter_get_by_name("buffersink");
+    if ((ret = avfilter_graph_create_filter(&buffersink_ctx, buffersink, "out", NULL, NULL, _filter_graph.get())) < 0) {
+        ss << "avfilter_graph_create_filter buffersink failed: " << ret << " " << ffmpeg_err(ret);
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    AVFilterContext *drawtext_ctx1 = nullptr;
+
+    const AVFilter *drawtext_filter = avfilter_get_by_name("drawtext");
+    if ((ret = avfilter_graph_create_filter(&drawtext_ctx1, drawtext_filter, "drawtext", drawtext_args1, NULL, _filter_graph.get())) < 0) {
+        ss << "avfilter_graph_create_filter drawtext_filter failed: " << ret << " " << ffmpeg_err(ret);
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    if ((ret = avfilter_link(buffersrc_ctx, 0, drawtext_ctx1, 0) < 0 || avfilter_link(drawtext_ctx1, 0, buffersink_ctx, 0))< 0) {
+        ss << "avfilter_link: " << ret << " " << ffmpeg_err(ret);
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    if ((ret = avfilter_graph_config(_filter_graph.get(), NULL)) < 0) {
+        ss << "avfilter_graph_config failed: " << ret << " " << ffmpeg_err(ret);
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    if ((ret = av_buffersrc_add_frame_flags(buffersrc_ctx, new_frame->get(), 0)) < 0) {
+        ss << "av_buffersink_get_frame failed: " << ret << " " << ffmpeg_err(ret);
+        return make_tuple<bool, std::string>(false, ss.data());
+    }
+
+    auto pkt = alloc_av_packet();
+    while (av_buffersink_get_frame(buffersink_ctx, new_frame->get()) >= 0) {
+        if (avcodec_send_frame(jpeg_codec_ctx.get(), new_frame->get()) == 0) {
+            while (avcodec_receive_packet(jpeg_codec_ctx.get(), pkt.get()) == 0) {
+                fwrite(pkt.get()->data, pkt.get()->size, 1, tmp_save_file_jpg.get());
+            }
+        }
     }
-    DebugL << "Screenshot successful: " << filename;
     return make_tuple<bool, std::string>(true, "");
 }
 

src/Codec/Transcode.h

@@ -27,6 +27,9 @@ extern "C" {
 #include "libavutil/audio_fifo.h"
 #include "libavutil/imgutils.h"
 #include "libavutil/frame.h"
+#include "libavfilter/avfilter.h"
+#include "libavfilter/buffersink.h"
+#include "libavfilter/buffersrc.h"
 #ifdef __cplusplus
 }
 #endif
@@ -45,9 +48,10 @@ public:
     AVFrame *get() const;
     void fillPicture(AVPixelFormat target_format, int target_width, int target_height);
     int getChannels() const;
+    void reset();
 
 private:
-    char *_data = nullptr;
+    std::unique_ptr<char[]> _data;
     std::shared_ptr<AVFrame> _frame;
 };
 
@@ -170,9 +174,11 @@ public:
      * @param frame 解码后的帧
      * @param filename 保存文件路径
      * @param fmt jpg:AV_PIX_FMT_YUVJ420P，PNG:AV_PIX_FMT_RGB24
+     * @param w h (可选)裁剪的图片大小，默认和输入源一致
+     * @param font_path (可选), default DejaVuSans.ttf
      * @return
      */
-    static std::tuple<bool, std::string> saveFrame(const FFmpegFrame::Ptr &frame, const char *filename, AVPixelFormat fmt = AV_PIX_FMT_YUVJ420P);
+    static std::tuple<bool, std::string> saveFrame(const FFmpegFrame::Ptr &frame, const char *filename, AVPixelFormat fmt = AV_PIX_FMT_YUVJ420P, int w = 0, int h = 0, const char *font_path = nullptr);
 };
 
 }//namespace mediakit

src/Common/MediaSink.cpp

@@ -176,7 +176,9 @@ void MediaSink::checkTrackIfReady() {
 }
 
 void MediaSink::addTrackCompleted() {
-    setMaxTrackCount(_track_map.size());
+    if (!_track_map.empty()) {
+        setMaxTrackCount(_track_map.size());
+    }
 }
 
 void MediaSink::setMaxTrackCount(size_t i) {

src/Common/MediaSource.h

@@ -287,6 +287,36 @@ public:
     // Maximum number of tracks
     size_t max_track = 2;
 
+#define OPT_VALUE(XX)           \
+        XX(modify_stamp)        \
+        XX(enable_audio)        \
+        XX(add_mute_audio)      \
+        XX(auto_close)          \
+        XX(continue_push_ms)    \
+        XX(paced_sender_ms)     \
+                                \
+        XX(enable_hls)          \
+        XX(enable_hls_fmp4)     \
+        XX(enable_mp4)          \
+        XX(enable_rtsp)         \
+        XX(enable_rtmp)         \
+        XX(enable_ts)           \
+        XX(enable_fmp4)         \
+                                \
+        XX(hls_demand)          \
+        XX(rtsp_demand)         \
+        XX(rtmp_demand)         \
+        XX(ts_demand)           \
+        XX(fmp4_demand)         \
+                                \
+        XX(mp4_max_second)      \
+        XX(mp4_as_player)       \
+        XX(mp4_save_path)       \
+                                \
+        XX(hls_save_path)       \
+        XX(stream_replace)      \
+        XX(max_track)
+
     template <typename MAP>
     ProtocolOption(const MAP &allArgs) : ProtocolOption() {
         load(allArgs);
@@ -294,35 +324,18 @@ public:
 
     template <typename MAP>
     void load(const MAP &allArgs) {
-#define GET_OPT_VALUE(key) getArgsValue(allArgs, #key, key)
-        GET_OPT_VALUE(modify_stamp);
-        GET_OPT_VALUE(enable_audio);
-        GET_OPT_VALUE(add_mute_audio);
-        GET_OPT_VALUE(auto_close);
-        GET_OPT_VALUE(continue_push_ms);
-        GET_OPT_VALUE(paced_sender_ms);
+#define GET(key) getArgsValue(allArgs, #key, key);
+        OPT_VALUE(GET)
+#undef GET
+    }
 
-        GET_OPT_VALUE(enable_hls);
-        GET_OPT_VALUE(enable_hls_fmp4);
-        GET_OPT_VALUE(enable_mp4);
-        GET_OPT_VALUE(enable_rtsp);
-        GET_OPT_VALUE(enable_rtmp);
-        GET_OPT_VALUE(enable_ts);
-        GET_OPT_VALUE(enable_fmp4);
-
-        GET_OPT_VALUE(hls_demand);
-        GET_OPT_VALUE(rtsp_demand);
-        GET_OPT_VALUE(rtmp_demand);
-        GET_OPT_VALUE(ts_demand);
-        GET_OPT_VALUE(fmp4_demand);
-
-        GET_OPT_VALUE(mp4_max_second);
-        GET_OPT_VALUE(mp4_as_player);
-        GET_OPT_VALUE(mp4_save_path);
-
-        GET_OPT_VALUE(hls_save_path);
-        GET_OPT_VALUE(stream_replace);
-        GET_OPT_VALUE(max_track);
+    template <typename MAP>
+    MAP as() {
+        MAP ret;
+#define SET(key) ret[#key] = key;
+        OPT_VALUE(SET)
+#undef SET
+        return ret;
     }
 };
 

src/Common/MultiMediaSourceMuxer.cpp

@@ -401,15 +401,20 @@ bool MultiMediaSourceMuxer::setupRecord(Recorder::type type, bool start, const s
     }
 }
 
-std::string MultiMediaSourceMuxer::startRecord(const std::string &file_path, uint32_t back_time_ms, uint32_t forward_time_ms) {
+std::string MultiMediaSourceMuxer::startRecord(const std::string &file_path, int back_time_ms, int forward_time_ms) {
 #if !defined(ENABLE_MP4)
     throw std::invalid_argument("mp4相关功能未打开，请开启ENABLE_MP4宏后编译再测试");
 #else
     if (!_ring) {
         throw std::runtime_error("frame gop cache disabled, start record event video failed");
     }
-    auto path = Recorder::getRecordPath(Recorder::type_mp4, _tuple, _option.mp4_save_path);
-    path += file_path;
+    std::string path;
+    if (!start_with(file_path, "/")) {
+        path = Recorder::getRecordPath(Recorder::type_mp4, _tuple, _option.mp4_save_path);
+        path += file_path;
+    } else {
+        path = file_path;
+    }
     TraceL << "mp4 save path: " << path;
 
     auto muxer = std::make_shared<MP4Muxer>();
@@ -419,68 +424,124 @@ std::string MultiMediaSourceMuxer::startRecord(const std::string &file_path, uin
     }
     muxer->addTrackCompleted();
 
-    std::list<Frame::Ptr> history;
-    _ring->flushGop([&](const Frame::Ptr &frame) { history.emplace_back(frame); });
-    if (!history.empty()) {
-        auto now_dts = history.back()->dts();
-
-        decltype(history)::iterator pos = history.end();
-        for (auto it = history.rbegin(); it != history.rend(); ++it) {
-            auto &frame = *it;
-            if (frame->getTrackType() != TrackVideo || (!frame->configFrame() && !frame->keyFrame())) {
-                continue;
-            }
-            // 如果视频关键帧到末尾的时长超过一定的时间，那前面的数据应该全部删除
-            if (frame->dts() + back_time_ms < now_dts) {
-                pos = it.base();
-                --pos;
-                break;
-            }
-        }
-        if (pos != history.end()) {
-            // 移除前面过多的数据
-            TraceL << "clear history video: " << history.front()->dts() << " -> " << (*pos)->dts();
-            history.erase(history.begin(), pos);
-        }
+    bool have_history = false;
+    if (back_time_ms > 0) {
+        // 回溯录制
+        std::list<Frame::Ptr> history;
+        _ring->flushGop([&](const Frame::Ptr &frame) { history.emplace_back(frame); });
         if (!history.empty()) {
-            auto &front = history.front();
-            InfoL << "start record: " << path
-                  << ", start_dts: " << front->dts() << ", key_frame: " << front->keyFrame() << ", config_frame: " << front->configFrame()
-                  << ", now_dts: " << now_dts;
-        }
+            auto now_dts = history.back()->dts();
 
-        for (auto &frame : history) {
-            muxer->inputFrame(frame);
+            decltype(history)::iterator pos = history.end();
+            for (auto it = history.rbegin(); it != history.rend(); ++it) {
+                auto &frame = *it;
+                if (frame->getTrackType() != TrackVideo || (!frame->configFrame() && !frame->keyFrame())) {
+                    continue;
+                }
+                // 如果视频关键帧到末尾的时长超过一定的时间，那前面的数据应该全部删除
+                if (frame->dts() + back_time_ms < now_dts) {
+                    pos = it.base();
+                    --pos;
+                    break;
+                }
+            }
+            if (pos != history.end()) {
+                // 移除历史视频前面过多的数据
+                DebugL << "clear history front video: " << history.front()->dts() << " -> " << (*pos)->dts();
+                history.erase(history.begin(), pos);
+            }
+
+            if (forward_time_ms < 0) {
+                // 如果后向录制时长为负，说明回溯录制要截取一段尾部
+                pos = history.end();
+                for (auto it = history.rbegin(); it != history.rend(); ++it) {
+                    auto &frame = *it;
+                    if (frame->getTrackType() != TrackVideo) {
+                        continue;
+                    }
+                    if (frame->dts() < now_dts + forward_time_ms) {
+                        pos = it.base();
+                        ++pos;
+                        break;
+                    }
+                }
+
+                if (pos != history.end()) {
+                    // 移除历史视频后面过多的数据
+                    DebugL << "clear history tail video: " << (*pos)->dts() << " -> " << now_dts;
+                    history.erase(pos, history.end());
+                }
+            }
+
+            if (!history.empty()) {
+                auto &front = history.front();
+                InfoL << "start record: " << path
+                      << ", start_dts: " << front->dts() << ", key_frame: " << front->keyFrame() << ", config_frame: " << front->configFrame()
+                      << ", now_dts: " << now_dts;
+                have_history = true;
+            }
+
+            for (auto &frame : history) {
+                muxer->inputFrame(frame);
+            }
         }
     }
 
-    auto reader = _ring->attach(MultiMediaSourceMuxer::getOwnerPoller(MediaSource::NullMediaSource()), false);
-    uint64_t now_dts = 0;
-    int selected_index = -1;
-    Ticker ticker;
-    bool is_live_stream = _dur_sec < 0.01;
-    reader->setReadCB([muxer, now_dts, selected_index, forward_time_ms, reader, path, ticker, is_live_stream](const Frame::Ptr &frame) mutable {
-        // 循环引用自身
-        if (!now_dts) {
-            now_dts = frame->dts();
-            selected_index = frame->getIndex();
+    if (forward_time_ms > 0) {
+        if (!have_history) {
+            InfoL << "start record: " << path << ", back_time_ms: " << back_time_ms << ", forward_time_ms: " << forward_time_ms;
         }
-        // 新增兜底机制，如果直播录制任务时长超过预期时间3秒，不管数据时间戳是否增长是否达到预期，都强制停止录制
-        if ((frame->getIndex() == selected_index && now_dts + forward_time_ms < frame->dts()) || (is_live_stream && ticker.createdTime() > forward_time_ms + 3000)) {
-            InfoL << "stop record: " << path << ", end dts: " << frame->dts();
-            WorkThreadPool::Instance().getPoller()->async([muxer]() { muxer->closeMP4(); });
-            reader = nullptr;
-            return;
+
+        weak_ptr<MultiMediaSourceMuxer> weak_self = shared_from_this();
+        auto lam = [weak_self, muxer, forward_time_ms, have_history, path]() {
+            auto strong_self = weak_self.lock();
+            if (!strong_self) {
+                return;
+            }
+            uint64_t now_dts = 0;
+            int selected_index = -1;
+            Ticker ticker;
+            bool is_live_stream = strong_self->_dur_sec < 0.01;
+            auto reader = strong_self->_ring->attach(strong_self->MultiMediaSourceMuxer::getOwnerPoller(MediaSource::NullMediaSource()), !have_history, 1);
+            reader->setReadCB([muxer, now_dts, selected_index, forward_time_ms, reader, path, ticker, is_live_stream](const Frame::Ptr &frame) mutable {
+                if (!reader) {
+                    // 已经关闭录制
+                    return;
+                }
+                // 循环引用自身
+                if (!now_dts) {
+                    now_dts = frame->dts();
+                    selected_index = frame->getIndex();
+                }
+                // 新增兜底机制，如果直播录制任务时长超过预期时间3秒，不管数据时间戳是否增长是否达到预期，都强制停止录制
+                if ((frame->getIndex() == selected_index && now_dts + forward_time_ms < frame->dts())
+                    || (is_live_stream && ticker.createdTime() > forward_time_ms + 3000ULL)) {
+                    InfoL << "stop record: " << path << ", end dts: " << frame->dts();
+                    WorkThreadPool::Instance().getPoller()->async([muxer]() { muxer->closeMP4(); });
+                    reader = nullptr;
+                    return;
+                }
+                muxer->inputFrame(frame);
+            });
+            std::weak_ptr<RingType::RingReader> weak_reader = reader;
+            reader->setDetachCB([weak_reader]() {
+                if (auto strong_reader = weak_reader.lock()) {
+                    // 防止循环引用
+                    strong_reader->setReadCB(nullptr);
+                }
+            });
+        };
+        if (back_time_ms >= 0) {
+            // 立即前向录制
+            lam();
+        } else {
+            // 延时启动录制
+            MultiMediaSourceMuxer::getOwnerPoller(MediaSource::NullMediaSource())->doDelayTask(-back_time_ms, [lam]() {
+                lam();
+                return 0;
+            });
         }
-        muxer->inputFrame(frame);
-    });
-    std::weak_ptr<RingType::RingReader> weak_reader = reader;
-    reader->setDetachCB([weak_reader]() {
-        if (auto strong_reader = weak_reader.lock()) {
-            // 防止循环引用
-            strong_reader->setReadCB(nullptr);
-        }
-    });
+    }
 
     return path;
 #endif

src/Common/MultiMediaSourceMuxer.h

@@ -129,7 +129,7 @@ public:
      * @param forward_time_ms 后续录制时长
      * @return 录制文件绝对路径
      */
-    std::string startRecord(const std::string &file_path, uint32_t back_time_ms, uint32_t forward_time_ms);
+    std::string startRecord(const std::string &file_path, int back_time_ms, int forward_time_ms);
 
     /**
      * 获取录制状态

src/Common/config.cpp

@@ -81,6 +81,7 @@ const string kBroadcastRtcSctpClosed = "kBroadcastRtcSctpClosed";
 const string kBroadcastRtcSctpSend = "kBroadcastRtcSctpSend";
 const string kBroadcastRtcSctpReceived = "kBroadcastRtcSctpReceived";
 const string kBroadcastPlayerCountChanged = "kBroadcastPlayerCountChanged";
+const string kBroadcastPlayerProxyFailed = "kBroadcastPlayerProxyFailed";
 
 } // namespace Broadcast
 

src/Common/config.h

@@ -161,6 +161,9 @@ extern const std::string kBroadcastRtcSctpReceived;
 extern const std::string kBroadcastPlayerCountChanged;
 #define BroadcastPlayerCountChangedArgs const MediaTuple& args, const int& count
 
+extern const std::string kBroadcastPlayerProxyFailed;
+#define BroadcastPlayerProxyFailedArgs const PlayerProxy& sender, const toolkit::SockException &ex
+
 #define ReloadConfigTag ((void *)(0xFF))
 #define RELOAD_KEY(arg, key)                                                                                           \
     do {                                                                                                               \

src/Player/PlayerProxy.cpp

@@ -110,7 +110,9 @@ void PlayerProxy::play(const string &strUrlTmp) {
         if (!strongSelf) {
             return;
         }
-
+        if (err) {
+            NOTICE_EMIT(BroadcastPlayerProxyFailedArgs, Broadcast::kBroadcastPlayerProxyFailed, *strongSelf, err);
+        }
         if (strongSelf->_on_play) {
             strongSelf->_on_play(err);
             strongSelf->_on_play = nullptr;
@@ -146,6 +148,9 @@ void PlayerProxy::play(const string &strUrlTmp) {
         if (!strongSelf) {
             return;
         }
+        if (err) {
+            NOTICE_EMIT(BroadcastPlayerProxyFailedArgs, Broadcast::kBroadcastPlayerProxyFailed, *strongSelf, err);
+        }
 
         // 注销直接拉流代理产生的流：#532  [AUTO-TRANSLATED:c6343a3b]
         // Unregister the stream generated by the direct stream proxy: #532

src/Player/PlayerProxy.h

@@ -18,8 +18,7 @@
 
 namespace mediakit {
 
-struct StreamInfo
-{
+struct StreamInfo {
     TrackType codec_type;
     std::string codec_name;
     int bitrate;
@@ -30,8 +29,7 @@ struct StreamInfo
     int video_height;
     float video_fps;
 
-    StreamInfo()
-    {
+    StreamInfo() {
         codec_type = TrackInvalid;
         codec_name = "none";
         bitrate = -1;
@@ -44,14 +42,12 @@ struct StreamInfo
     }
 };
 
-struct TranslationInfo
-{
+struct TranslationInfo {
     std::vector<StreamInfo> stream_info;
     int byte_speed;
     uint64_t start_time_stamp;
 
-    TranslationInfo()
-    {
+    TranslationInfo() {
         byte_speed = -1;
         start_time_stamp = 0;
     }

src/Pusher/PusherProxy.cpp

@@ -115,7 +115,13 @@ void PusherProxy::rePublish(const string &dst_url, int failed_cnt) {
                 return false;
             }
             WarnL << "推流重试[" << failed_cnt << "]:" << dst_url;
-            strong_self->MediaPusher::publish(dst_url);
+            try {
+                strong_self->MediaPusher::publish(dst_url);
+            } catch (std::exception &e) {
+                WarnL << e.what();
+                // 回调推流失败，一般是媒体注销了
+                strong_self->_on_close(SockException(Err_other, e.what()));
+            }
             return false;
         },
         getPoller());

src/Record/MP4Muxer.cpp

@@ -19,7 +19,11 @@ using namespace toolkit;
 namespace mediakit {
 
 MP4Muxer::~MP4Muxer() {
-    closeMP4();
+    try {
+        closeMP4();
+    } catch (std::exception &e) {
+        WarnL << e.what();
+    }
 }
 
 void MP4Muxer::openMP4(const string &file) {

src/Rtmp/Rtmp.cpp

@@ -56,30 +56,33 @@ AudioMeta::AudioMeta(const AudioTrack::Ptr &audio) {
 }
 
 uint8_t getCodecFlags(CodecId cid) {
-    switch(cid) {
-#define XX(a, b, c) case a: return static_cast<uint8_t>(b);
-    RTMP_CODEC_MAP(XX)
+    switch (cid) {
+#define XX(a, b, c)                                                                                                                                            \
+    case a: return static_cast<uint8_t>(b);
+        RTMP_CODEC_MAP(XX)
 #undef XX
+        default: return 0;
     }
-    return 0;
 }
 
 uint32_t getCodecFourCC(CodecId cid) {
-    switch(cid) {
-#define XX(a, b, c) case a: return static_cast<uint32_t>(c);
-    RTMP_CODEC_MAP(XX)
+    switch (cid) {
+#define XX(a, b, c)                                                                                                                                            \
+    case a: return static_cast<uint32_t>(c);
+        RTMP_CODEC_MAP(XX)
 #undef XX
+        default: return 0;
     }
-    return 0;
 }
 
 CodecId getFourccCodec(uint32_t id) {
-    switch(id) {
-#define XX(a, b, c) case (uint32_t)c: return a;
-    RTMP_CODEC_MAP(XX)
+    switch (id) {
+#define XX(a, b, c)                                                                                                                                            \
+    case (uint32_t)c: return a;
+        RTMP_CODEC_MAP(XX)
 #undef XX
+        default: return CodecInvalid;
     }
-    return CodecInvalid;
 }
 
 uint8_t getAudioRtmpFlags(const Track::Ptr &track) {
@@ -195,12 +198,10 @@ bool RtmpPacket::isConfigFrame() const {
     switch (type_id) {
         case MSG_AUDIO: {
             switch ((RtmpAudioCodec)getRtmpCodecId()) {
-            case RtmpAudioCodec::aac:
-                return (RtmpAACPacketType)buffer[1] == RtmpAACPacketType::aac_config_header;
-            case RtmpAudioCodec::ex_header:
-                return (RtmpPacketType)(buffer[0] & 0x0f) == RtmpPacketType::PacketTypeSequenceStart;
+                case RtmpAudioCodec::aac: return (RtmpAACPacketType)buffer[1] == RtmpAACPacketType::aac_config_header;
+                case RtmpAudioCodec::ex_header: return (RtmpPacketType)(buffer[0] & 0x0f) == RtmpPacketType::PacketTypeSequenceStart;
+                default: return false;
             }
-            return false;
         }
         case MSG_VIDEO: {
             if (!isVideoKeyFrame()) {

src/Rtmp/RtmpProtocol.cpp

@@ -26,11 +26,35 @@ using namespace toolkit;
 #define S2_FMS_KEY_SIZE 68
 #define C1_OFFSET_SIZE 4
 
+
 #ifdef ENABLE_OPENSSL
 #include "Util/SSLBox.h"
 #include <openssl/hmac.h>
 #include <openssl/opensslv.h>
 
+static uint8_t FMSKey[] = {
+    0x47, 0x65, 0x6e, 0x75, 0x69, 0x6e, 0x65, 0x20,
+    0x41, 0x64, 0x6f, 0x62, 0x65, 0x20, 0x46, 0x6c,
+    0x61, 0x73, 0x68, 0x20, 0x4d, 0x65, 0x64, 0x69,
+    0x61, 0x20, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72,
+    0x20, 0x30, 0x30, 0x31, // Genuine Adobe Flash Media Server 001
+    0xf0, 0xee, 0xc2, 0x4a, 0x80, 0x68, 0xbe, 0xe8,
+    0x2e, 0x00, 0xd0, 0xd1, 0x02, 0x9e, 0x7e, 0x57,
+    0x6e, 0xec, 0x5d, 0x2d, 0x29, 0x80, 0x6f, 0xab,
+    0x93, 0xb8, 0xe6, 0x36, 0xcf, 0xeb, 0x31, 0xae
+}; // 68
+
+static uint8_t FPKey[] = {
+    0x47, 0x65, 0x6E, 0x75, 0x69, 0x6E, 0x65, 0x20,
+    0x41, 0x64, 0x6F, 0x62, 0x65, 0x20, 0x46, 0x6C,
+    0x61, 0x73, 0x68, 0x20, 0x50, 0x6C, 0x61, 0x79,
+    0x65, 0x72, 0x20, 0x30, 0x30, 0x31, // Genuine Adobe Flash Player 001
+    0xF0, 0xEE, 0xC2, 0x4A, 0x80, 0x68, 0xBE, 0xE8,
+    0x2E, 0x00, 0xD0, 0xD1, 0x02, 0x9E, 0x7E, 0x57,
+    0x6E, 0xEC, 0x5D, 0x2D, 0x29, 0x80, 0x6F, 0xAB,
+    0x93, 0xB8, 0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE
+}; // 62
+
 static string openssl_HMACsha256(const void *key, size_t key_len, const void *data, size_t data_len){
     std::shared_ptr<char> out(new char[32], [](char *ptr) { delete[] ptr; });
     unsigned int out_len;
@@ -329,8 +353,16 @@ const char* RtmpProtocol::handle_S0S1S2(const char *data, size_t len, const func
     }
     // 发送 C2  [AUTO-TRANSLATED:e51c339e]
     // Send C2
-    const char *pcC2 = data + 1;
-    onSendRawData(obtainBuffer(pcC2, C1_HANDSHARK_SIZE));
+    uint8_t *pS1 = (uint8_t*)data + 1;    
+    RtmpHandshake c2(0);
+    memcpy(&c2, pS1, sizeof(c2));
+#ifdef ENABLE_OPENSSL
+    if(pS1[4] >=3){ // 复杂握手计算c2
+        handle_S1_complex((char*)pS1, c2);
+    }
+#endif
+
+    onSendRawData(obtainBuffer(&c2, C1_HANDSHARK_SIZE));
     // 握手结束  [AUTO-TRANSLATED:9df763ff]
     // Handshake finished
     _next_step_func = [this](const char *data, size_t len) {
@@ -408,7 +440,7 @@ void RtmpProtocol::handle_C1_complex(const char *data){
         check_C1_Digest(digest, c1_joined);
 
         send_complex_S0S1S2(0, digest);
-//		InfoL << "schema0";
+//      InfoL << "schema0";
     } catch (std::exception &) {
         // 貌似flash从来都不用schema1  [AUTO-TRANSLATED:2c6d140f]
         // It seems that flash never uses schema1
@@ -426,40 +458,70 @@ void RtmpProtocol::handle_C1_complex(const char *data){
             check_C1_Digest(digest, c1_joined);
 
             send_complex_S0S1S2(1, digest);
-//			InfoL << "schema1";
+//          InfoL << "schema1";
         } catch (std::exception &) {
-//			WarnL << "try rtmp complex schema1 failed:" <<  ex.what();
+			//WarnL << "try rtmp complex schema1 failed:" <<  ex.what();
             handle_C1_simple(data);
         }
     }
 }
 
-#if !defined(u_int8_t)
-#define u_int8_t unsigned char
-#endif // !defined(u_int8_t)
+void RtmpProtocol::check_S1_Digest(const std::string &digest,const std::string &data){
+    auto sha256 = openssl_HMACsha256(FMSKey, S1_FMS_KEY_SIZE, data.data(), data.size());
+    if (sha256 != digest) {
+        throw std::runtime_error("digest mismatched");
+    } else {
+        InfoL << "check rtmp complex handshark success!";
+    }
+}
 
-static u_int8_t FMSKey[] = {
-    0x47, 0x65, 0x6e, 0x75, 0x69, 0x6e, 0x65, 0x20,
-    0x41, 0x64, 0x6f, 0x62, 0x65, 0x20, 0x46, 0x6c,
-    0x61, 0x73, 0x68, 0x20, 0x4d, 0x65, 0x64, 0x69,
-    0x61, 0x20, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72,
-    0x20, 0x30, 0x30, 0x31, // Genuine Adobe Flash Media Server 001
-    0xf0, 0xee, 0xc2, 0x4a, 0x80, 0x68, 0xbe, 0xe8,
-    0x2e, 0x00, 0xd0, 0xd1, 0x02, 0x9e, 0x7e, 0x57,
-    0x6e, 0xec, 0x5d, 0x2d, 0x29, 0x80, 0x6f, 0xab,
-    0x93, 0xb8, 0xe6, 0x36, 0xcf, 0xeb, 0x31, 0xae
-}; // 68
+void RtmpProtocol::handle_S1_complex(const char *data,RtmpHandshake &c2){
 
-static u_int8_t FPKey[] = {
-    0x47, 0x65, 0x6E, 0x75, 0x69, 0x6E, 0x65, 0x20,
-    0x41, 0x64, 0x6F, 0x62, 0x65, 0x20, 0x46, 0x6C,
-    0x61, 0x73, 0x68, 0x20, 0x50, 0x6C, 0x61, 0x79,
-    0x65, 0x72, 0x20, 0x30, 0x30, 0x31, // Genuine Adobe Flash Player 001
-    0xF0, 0xEE, 0xC2, 0x4A, 0x80, 0x68, 0xBE, 0xE8,
-    0x2E, 0x00, 0xD0, 0xD1, 0x02, 0x9E, 0x7E, 0x57,
-    0x6E, 0xEC, 0x5D, 0x2D, 0x29, 0x80, 0x6F, 0xAB,
-    0x93, 0xB8, 0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE
-}; // 62
+    const char *s1_start = data;
+    const char *schema_start = s1_start + 8;
+    char *digest_start;
+    std::string digest;
+    try {
+        /* c1s1 schema0
+        time: 4bytes
+        version: 4bytes
+        key: 764bytes
+        digest: 764bytes
+         */
+        digest = get_C1_digest((uint8_t *) schema_start + C1_SCHEMA_SIZE, &digest_start);
+        string s1_joined(s1_start, C1_HANDSHARK_SIZE);
+        s1_joined.erase(digest_start - s1_start, C1_DIGEST_SIZE);
+        check_S1_Digest(digest, s1_joined);
+		//InfoL << "schema0";
+    } catch (std::exception &ex) {
+        // 貌似flash从来都不用schema1  [AUTO-TRANSLATED:2c6d140f]
+        // It seems that flash never uses schema1
+		//WarnL << "try rtmp complex schema0 failed:" << ex.what();
+        try {
+            /* c1s1 schema1
+            time: 4bytes
+            version: 4bytes
+            digest: 764bytes
+            key: 764bytes
+             */
+            digest = get_C1_digest((uint8_t *) schema_start, &digest_start);
+            string s1_joined(s1_start, C1_HANDSHARK_SIZE);
+            s1_joined.erase(digest_start - s1_start, C1_DIGEST_SIZE);
+            check_S1_Digest(digest, s1_joined);
+            //send_complex_S0S1S2(1, digest);
+			//InfoL << "schema1";
+        } catch (std::exception &ex) {
+			WarnL << "try rtmp complex schema1 failed:" <<  ex.what();
+            return;
+        }
+    }
+
+    //InfoL << "send complex C2";
+    auto c2_key = openssl_HMACsha256(FPKey, sizeof(FPKey), digest.data(), digest.size());
+    std::string c2_str((char*)(&c2), sizeof(c2)- C1_DIGEST_SIZE);
+    auto c2_digest = openssl_HMACsha256(c2_key.data(), c2_key.size(), c2_str.data(), c2_str.size());
+    memcpy(c2.random + RANDOM_LEN - C1_DIGEST_SIZE, c2_digest.data(), C1_DIGEST_SIZE);
+}
 
 void RtmpProtocol::check_C1_Digest(const string &digest,const string &data){
     auto sha256 = openssl_HMACsha256(FPKey, C1_FPKEY_SIZE, data.data(), data.size());

src/Rtmp/RtmpProtocol.h

@@ -63,14 +63,16 @@ protected:
     void sendRtmp(uint8_t type, uint32_t stream_index, const std::string &buffer, uint32_t stamp, int chunk_id);
     void sendRtmp(uint8_t type, uint32_t stream_index, const toolkit::Buffer::Ptr &buffer, uint32_t stamp, int chunk_id);
     toolkit::BufferRaw::Ptr obtainBuffer(const void *data = nullptr, size_t len = 0);
-
+    
 private:
     void handle_C1_simple(const char *data);
 #ifdef ENABLE_OPENSSL
+    void handle_S1_complex(const char *data, RtmpHandshake &c2);
     void handle_C1_complex(const char *data);
     std::string get_C1_digest(const uint8_t *ptr,char **digestPos);
     std::string get_C1_key(const uint8_t *ptr);
     void check_C1_Digest(const std::string &digest,const std::string &data);
+    void check_S1_Digest(const std::string &digest,const std::string &data);
     void send_complex_S0S1S2(int schemeType,const std::string &digest);
 #endif //ENABLE_OPENSSL
 

src/Rtsp/RtspPlayer.cpp

@@ -446,7 +446,11 @@ void RtspPlayer::sendOptions() {
 }
 
 void RtspPlayer::sendKeepAlive() {
-    _on_response = [](const Parser &parser) {};
+    if (_play_check_timer)
+    {
+        WarnL << "receive RTP packet before handleResPAUSE";
+    }
+    _on_keepalive_reponse = [](const Parser &parser) {};
     if (_supported_cmd.find("GET_PARAMETER") != _supported_cmd.end()) {
         // 支持GET_PARAMETER，用此命令保活  [AUTO-TRANSLATED:b45cd737]
         // Support GET_PARAMETER, use this command to keep alive
@@ -532,6 +536,10 @@ void RtspPlayer::onWholeRtspPacket(Parser &parser) {
     try {
         decltype(_on_response) func;
         _on_response.swap(func);
+        if (!func)
+        {
+            _on_keepalive_reponse.swap(func);
+        }
         if (func) {
             func(parser);
         }

src/Rtsp/RtspPlayer.h

@@ -162,6 +162,7 @@ private:
     float _speed = 0.0f;
     std::vector<SdpTrack::Ptr> _sdp_track;
     std::function<void(const Parser&)> _on_response;
+    std::function<void(const Parser&)> _on_keepalive_reponse;
  protected:   
     // RTP端口,trackid idx 为数组下标  [AUTO-TRANSLATED:77c186bb]
     // RTP port, trackid idx is the array subscript

tests/test_ps.cpp

@@ -64,7 +64,6 @@ public:
 
 private:
     MultiMediaSourceMuxer::Ptr _muxer;
-    uint64_t timeStamp = 0;
     uint64_t timeStamp_last = 0;
 };
 

tests/test_rtp_pcap.cpp

@@ -83,6 +83,10 @@ struct sniff_tcp {
 #define TH_URG 0x20
 #define TH_ECE 0x40
 #define TH_CWR 0x80
+
+#if defined(TH_FLAGS)
+#undef TH_FLAGS
+#endif
 #define TH_FLAGS (TH_FINTH_SYNTH_RSTTH_ACKTH_URGTH_ECETH_CWR)
     u_short th_win; /* TCP滑动窗口 */
     u_short th_sum; /* 头部校验和 */
@@ -154,7 +158,7 @@ static bool loadFile(const char *path, const EventPoller::Ptr &poller) {
         return false;
     }
     auto total_size = std::make_shared<size_t>(0);
-    struct pcap_pkthdr header = {0};
+    struct pcap_pkthdr header {};
     while (true) {
         const u_char *pkt_buff = pcap_next(handle.get(), &header);
         if (!pkt_buff) {

webrtc/IceTransport.cpp

@@ -1175,7 +1175,7 @@ void IceAgent::connectivityCheck(const Pair::Ptr &pair, CandidateTuple& candidat
 }
 
 void IceAgent::tryTriggerredCheck(const Pair::Ptr& pair) {
-    DebugL;
+    // DebugL;
     //FIXME 暂不实现,因为当前实现基本收到candidate就会发起check
 }
 

webrtc/Nack.cpp

@@ -46,6 +46,7 @@ const string kNackIntervalRatio = RTC_FIELD "nackIntervalRatio";
 // nack包中rtp个数，减小此值可以让nack包响应更灵敏  [AUTO-TRANSLATED:12393868]
 // Number of rtp in nack packet, reducing this value can make nack packet response more sensitive
 const string kNackRtpSize = RTC_FIELD "nackRtpSize";
+const string kNackAudioRtpSize = RTC_FIELD "nackAudioRtpSize";
 
 static onceToken token([]() {
     mINI::Instance()[kMaxRtpCacheMS] = 5 * 1000;
@@ -55,6 +56,7 @@ static onceToken token([]() {
     mINI::Instance()[kNackMaxCount] = 15;
     mINI::Instance()[kNackIntervalRatio] = 1.0f;
     mINI::Instance()[kNackRtpSize] = 8;
+    mINI::Instance()[kNackAudioRtpSize] = 4;
 });
 
 } // namespace Rtc
@@ -155,7 +157,8 @@ int64_t NackList::getNtpStamp(uint16_t seq) {
 
 ////////////////////////////////////////////////////////////////////////////////////////////////
 
-NackContext::NackContext() {
+NackContext::NackContext(TrackType type) {
+    _type = type;
     setOnNack(nullptr);
 }
 
@@ -217,7 +220,9 @@ void NackContext::makeNack(uint16_t max_seq, bool flush) {
     // 最多生成5个nack包，防止seq大幅跳跃导致一直循环  [AUTO-TRANSLATED:9cc5da25]
     // Generate at most 5 nack packets to prevent seq from jumping significantly and causing continuous loops
     auto max_nack = 5u;
-    GET_CONFIG(uint32_t, nack_rtpsize, Rtc::kNackRtpSize);
+    GET_CONFIG(uint32_t, nack_video_rtpsize, Rtc::kNackRtpSize);
+    GET_CONFIG(uint32_t, nack_audio_rtpsize, Rtc::kNackAudioRtpSize);
+    auto nack_rtpsize = _type == TrackVideo ? nack_video_rtpsize : nack_audio_rtpsize;
     // kNackRtpSize must between 0 and 16
     nack_rtpsize = std::min<uint32_t>(nack_rtpsize, FCI_NACK::kBitSize);
     while (_nack_seq != max_seq && max_nack--) {

webrtc/Nack.h

@@ -55,7 +55,7 @@ public:
     using Ptr = std::shared_ptr<NackContext>;
     using onNack = std::function<void(const FCI_NACK &nack)>;
 
-    NackContext();
+    NackContext(TrackType type = TrackVideo);
 
     void received(uint16_t seq, bool is_rtx = false);
     void setOnNack(onNack cb);
@@ -71,6 +71,7 @@ private:
 private:
     bool _started = false;
     int _rtt = 50;
+    TrackType _type;
     onNack _cb;
     std::set<uint16_t> _seq;
     // 最新nack包中的rtp seq值  [AUTO-TRANSLATED:6984d95a]

webrtc/SrtpSession.cpp

@@ -204,11 +204,16 @@ SrtpSession::SrtpSession(Type type, CryptoSuite cryptoSuite, uint8_t *key, size_
     policy.key = key;
     // Required for sending RTP retransmission without RTX.
     policy.allow_repeat_tx = 1;
+#if 0
     if (type == Type::OUTBOUND) {
         policy.window_size = 0x8000 - 1;
     } else {
         policy.window_size = 1024;
     }
+#else
+    // TODO 关闭防重放攻击
+    policy.window_size = 0x8000 - 1;
+#endif
     policy.next = nullptr;
 
     // Set the SRTP session.

webrtc/WebRtcSession.cpp

@@ -69,10 +69,11 @@ void WebRtcSession::onRecv_l(const char *data, size_t len) {
             auto sock = Socket::createSocket(transport->getPoller(), false);
             // 1、克隆socket(fd不变)，切换poller线程到WebRtcTransport所在线程  [AUTO-TRANSLATED:f930bfab]
             // 1. Clone socket (fd remains unchanged), switch poller thread to the thread where WebRtcTransport is located
-            sock->cloneSocket(*(getSock()));
+            auto on_complete = sock->cloneSocket(*(getSock()));
             auto server = _server;
             std::string str(data, len);
-            sock->getPoller()->async([sock, server, str](){
+            // on_complete在创建WebRtcSession后才析构(才开始网络事件监听)
+            sock->getPoller()->async([sock, server, str, on_complete](){
                 auto strong_server = server.lock();
                 if (strong_server) {
                     auto session = static_pointer_cast<WebRtcSession>(strong_server->createSession(sock));

webrtc/WebRtcTransport.cpp

@@ -82,6 +82,7 @@ const string kMinBitrate = RTC_FIELD "min_bitrate";
 // 数据通道设置  [AUTO-TRANSLATED:2dc48bc3]
 // Data channel setting
 const string kDataChannelEcho = RTC_FIELD "datachannel_echo";
+const string kPreferredTcp = RTC_FIELD "preferred_tcp";
 
 static onceToken token([]() {
     mINI::Instance()[kTimeOutSec] = 15;
@@ -105,6 +106,7 @@ static onceToken token([]() {
     mINI::Instance()[kIceTransportPolicy] = 0;  // 默认值：不限制(kAll)
     mINI::Instance()[kIceUfrag] = "ZLMediaKit";
     mINI::Instance()[kIcePwd] = "ZLMediaKit";
+    mINI::Instance()[kPreferredTcp] = 0;
 });
 
 } // namespace Rtc
@@ -1102,7 +1104,7 @@ void WebRtcTransportImp::setIceCandidate(vector<SdpAttrCandidate> cands) {
 
 class RtpChannel : public RtpTrackImp, public std::enable_shared_from_this<RtpChannel> {
 public:
-    RtpChannel(EventPoller::Ptr poller, RtpTrackImp::OnSorted cb, function<void(const FCI_NACK &nack)> on_nack) {
+    RtpChannel(TrackType type, EventPoller::Ptr poller, RtpTrackImp::OnSorted cb, function<void(const FCI_NACK &nack)> on_nack) : _nack_ctx(type){
         _poller = std::move(poller);
         _on_nack = std::move(on_nack);
         setOnSorted(std::move(cb));
@@ -1314,7 +1316,7 @@ void WebRtcTransportImp::createRtpChannel(const string &rid, uint32_t ssrc, Medi
     // rid --> RtpReceiverImp
     auto &ref = track.rtp_channel[rid];
     weak_ptr<WebRtcTransportImp> weak_self = static_pointer_cast<WebRtcTransportImp>(shared_from_this());
-    ref = std::make_shared<RtpChannel>(
+    ref = std::make_shared<RtpChannel>(track.media->type,
         getPoller(), [&track, this, rid](RtpPacket::Ptr rtp) mutable { onSortedRtp(track, rid, std::move(rtp)); },
         [&track, weak_self, ssrc](const FCI_NACK &nack) mutable {
             // nack发送可能由定时器异步触发  [AUTO-TRANSLATED:186d6723]
@@ -1631,6 +1633,7 @@ WebRtcPluginManager &WebRtcPluginManager::Instance() {
 }
 
 void WebRtcPluginManager::registerPlugin(const string &type, Plugin cb) {
+    InfoL << "Load webrtc plugin:" << type;
     lock_guard<mutex> lck(_mtx_creator);
     _map_creator[type] = std::move(cb);
 }
@@ -1667,6 +1670,7 @@ void echo_plugin(SocketHelper& sender, const WebRtcArgs &args, const onCreateWeb
     cb(*WebRtcEchoTest::create(EventPollerPool::Instance().getPoller()));
 }
 
+template<typename Type>
 void push_plugin(SocketHelper& sender, const WebRtcArgs &args, const onCreateWebRtc &cb) {
     MediaInfo info(args["url"]);
     Broadcast::PublishAuthInvoker invoker = [cb, info](const string &err, const ProtocolOption &option) mutable {
@@ -1711,7 +1715,7 @@ void push_plugin(SocketHelper& sender, const WebRtcArgs &args, const onCreateWeb
             push_src_ownership = push_src->getOwnership();
             push_src->setProtocolOption(option);
         }
-        auto rtc = WebRtcPusher::create(EventPollerPool::Instance().getPoller(), push_src, push_src_ownership, info, option, 
+        auto rtc = Type::create(EventPollerPool::Instance().getPoller(), push_src, push_src_ownership, info, option,
             WebRtcTransport::Role::PEER, WebRtcTransport::SignalingProtocols::WHEP_WHIP);
         push_src->setListener(rtc);
         cb(*rtc);
@@ -1782,9 +1786,12 @@ static void setWebRtcArgs(const WebRtcArgs &args, WebRtcInterface &rtc) {
         }
     }
 
-    bool preferred_tcp = args["preferred_tcp"];
-    {
+    auto preferred_tcp = args["preferred_tcp"];
+    if (!preferred_tcp.empty()) {
         rtc.setPreferredTcp(preferred_tcp);
+    } else {
+        GET_CONFIG(bool, s_preferred_tcp, Rtc::kPreferredTcp);
+        rtc.setPreferredTcp(s_preferred_tcp);
     }
 
     {
@@ -1832,7 +1839,7 @@ static onceToken s_rtc_auto_register([]() {
     // Enable echo plugin only in debug mode
     WebRtcPluginManager::Instance().registerPlugin("echo", echo_plugin);
 #endif
-    WebRtcPluginManager::Instance().registerPlugin("push", push_plugin);
+    WebRtcPluginManager::Instance().registerPlugin("push", push_plugin<WebRtcPusher>);
     WebRtcPluginManager::Instance().registerPlugin("play", play_plugin<WebRtcPlayer>);
     WebRtcPluginManager::Instance().registerPlugin("talk", play_plugin<WebRtcTalk>);
 

webrtc/WebRtcTransport.h

@@ -286,7 +286,7 @@ public:
 
 struct WrappedMediaTrack {
     MediaTrack::Ptr track;
-    explicit WrappedMediaTrack(MediaTrack::Ptr ptr): track(ptr) {}
+    explicit WrappedMediaTrack(MediaTrack::Ptr ptr): track(std::move(ptr)) {}
     virtual ~WrappedMediaTrack() {}
     virtual void inputRtp(const char *buf, size_t len, uint64_t stamp_ms, RtpHeader *rtp) = 0;
 };