startSendRtp相关接口新增enable_origin_recv_limit参数控制是否开启源端限流

多倍速rtp转发时，在tcp模式下，如果转发速度跟不上接收速度将导致转发失败，通过开启该参数，可以确保历史视频多倍速下载成功

ext-codec/H264.cpp

@@ -157,7 +157,6 @@ bool H264Track::inputFrame(const Frame::Ptr &frame) {
         // AUD帧丢弃
         return false;
     }
-
     if ((type == H264Frame::NAL_B_P || type == H264Frame::NAL_IDR) && ready()) {
         return inputFrame_l(frame);
     }

postman/ZLMediaKit.postman_collection.json

@@ -310,6 +310,53 @@
 			},
 			"response": []
 		},
+		{
+			"name": "删除截图(deleteSnapDirectory)",
+			"request": {
+				"method": "GET",
+				"header": [],
+				"url": {
+					"raw": "{{ZLMediaKit_URL}}/index/api/deleteSnapDirectory?secret={{ZLMediaKit_secret}}&vhost={{defaultVhost}}&app=live&stream=test&file=71_1740828613.jpg",
+					"host": [
+						"{{ZLMediaKit_URL}}"
+					],
+					"path": [
+						"index",
+						"api",
+						"deleteSnapDirectory"
+					],
+					"query": [
+						{
+							"key": "secret",
+							"value": "{{ZLMediaKit_secret}}",
+							"description": "api操作密钥(配置文件配置)"
+						},
+						{
+							"key": "vhost",
+							"value": "{{defaultVhost}}",
+							"description": "筛选虚拟主机，例如__defaultVhost__"
+						},
+						{
+							"key": "app",
+							"value": "live",
+							"description": "筛选应用名，例如 live"
+						},
+						{
+							"key": "stream",
+							"value": "test",
+							"description": "筛选流id，例如 test"
+						},
+						{
+							"key": "file",
+							"value": "",
+							"disabled": true,
+							"description": "文件名，非必选"
+						}
+					]
+				}
+			},
+			"response": []
+		},
 		{
 			"name": "关断单个流(close_stream)",
 			"request": {
@@ -522,7 +569,7 @@
 			"response": []
 		},
 		{
-			"name": "添加rtsp/rtmp/hls/srt拉流代理(addStreamProxy)",
+			"name": "添加拉流代理(addStreamProxy)",
 			"request": {
 				"method": "GET",
 				"header": [],
@@ -560,7 +607,7 @@
 						{
 							"key": "url",
 							"value": "rtmp://live.hkstv.hk.lxdns.com/live/hks2",
-							"description": "拉流地址，例如rtmp://live.hkstv.hk.lxdns.com/live/hks2"
+							"description": "拉流地址，支持rtsp/rtmp/hls/srt/http-flv/http-ts协议"
 						},
 						{
 							"key": "rtp_type",
@@ -1333,6 +1380,62 @@
 			},
 			"response": []
 		},
+		{
+			"name": "开始事件视频录制(startRecordTask)",
+			"request": {
+				"method": "GET",
+				"header": [],
+				"url": {
+					"raw": "{{ZLMediaKit_URL}}/index/api/startRecordTask?secret={{ZLMediaKit_secret}}&vhost={{defaultVhost}}&app=live&stream=test&path=1.mp4&back_ms=10000&forward_ms=10000",
+					"host": [
+						"{{ZLMediaKit_URL}}"
+					],
+					"path": [
+						"index",
+						"api",
+						"startRecordTask"
+					],
+					"query": [
+						{
+							"key": "secret",
+							"value": "{{ZLMediaKit_secret}}",
+							"description": "api操作密钥(配置文件配置)"
+						},
+						{
+							"key": "vhost",
+							"value": "{{defaultVhost}}",
+							"description": "虚拟主机，例如__defaultVhost__"
+						},
+						{
+							"key": "app",
+							"value": "live",
+							"description": "应用名，例如 live"
+						},
+						{
+							"key": "stream",
+							"value": "test",
+							"description": "流id，例如 obs"
+						},
+						{
+							"key": "path",
+							"value": "1.mp4",
+							"description": "录像文件保存相对路径，包括名称"
+						},
+						{
+							"key": "back_ms",
+							"value": "10000",
+							"description": "回溯录制时长"
+						},
+						{
+							"key": "forward_ms",
+							"value": "10000",
+							"description": "后续录制时长"
+						}
+					]
+				}
+			},
+			"response": []
+		},
 		{
 			"name": "设置录像速度(setRecordSpeed)",
 			"request": {
@@ -2092,6 +2195,12 @@
 							"value": "",
 							"description": "发送rtp同时接收，一般用于双向语言对讲, 如果不为空，说明开启接收，值为接收流的id",
 							"disabled": true
+						},
+						{
+							"key": "enable_origin_recv_limit",
+							"value": "1",
+							"description": "转发rtp(tcp模式)时，如果发送不出去，是否限制源端收流速度，此参数在多倍速rtp转发时作用较大",
+							"disabled": true
 						}
 					]
 				}
@@ -2186,6 +2295,12 @@
 							"value": "5000",
 							"description": "等待tcp连接超时时间，单位毫秒，默认5000毫秒",
 							"disabled": true
+						},
+						{
+							"key": "enable_origin_recv_limit",
+							"value": "1",
+							"description": "转发rtp(tcp模式)时，如果发送不出去，是否限制源端收流速度，此参数在多倍速rtp转发时作用较大",
+							"disabled": true
 						}
 					]
 				}
@@ -2261,6 +2376,12 @@
 							"value": "1",
 							"description": "rtp es方式打包时，是否只打包音频；该参数非必选参数",
 							"disabled": true
+						},
+						{
+							"key": "enable_origin_recv_limit",
+							"value": "1",
+							"description": "转发rtp(tcp模式)时，如果发送不出去，是否限制源端收流速度，此参数在多倍速rtp转发时作用较大",
+							"disabled": true
 						}
 					]
 				}

server/WebApi.cpp

@@ -1603,6 +1603,7 @@ void installWebApi() {
         // Record the app and vhost of the sending stream
         args.recv_stream_app = allArgs["app"];
         args.recv_stream_vhost = allArgs["vhost"];
+        args.enable_origin_recv_limit = allArgs["enable_origin_recv_limit"];
         src->getOwnerPoller()->async([=]() mutable {
             try {
                 src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable {
@@ -1649,6 +1650,7 @@ void installWebApi() {
         args.recv_stream_id = allArgs["recv_stream_id"];
         args.recv_stream_app = allArgs["app"];
         args.recv_stream_vhost = allArgs["vhost"];
+        args.enable_origin_recv_limit = allArgs["enable_origin_recv_limit"];
 
         src->getOwnerPoller()->async([=]() mutable {
             try {
@@ -1776,6 +1778,30 @@ void installWebApi() {
         });
     });
 
+    api_regist("/index/api/startRecordTask",[](API_ARGS_MAP_ASYNC){
+        CHECK_SECRET();
+        CHECK_ARGS("vhost", "app", "stream", "path", "back_ms", "forward_ms");
+
+        auto src = MediaSource::find(allArgs["vhost"], allArgs["app"], allArgs["stream"]);
+        if (!src) {
+            throw ApiRetException("can not find the stream", API::NotFound);
+        }
+
+        src->getOwnerPoller()->async([=]() mutable {
+            std::string err;
+            std::string path;
+            try {
+                path = src->getMuxer()->startRecord(allArgs["path"], allArgs["back_ms"], allArgs["forward_ms"]);
+            } catch (std::exception &ex) {
+                err = ex.what();
+            }
+            val["code"] = err.empty() ? API::Success : API::OtherFailed;
+            val["data"]["path"] = path;
+            val["msg"] = err;
+            invoker(200, headerOut, val.toStyledString());
+        });
+    });
+
     // 设置录像流播放速度  [AUTO-TRANSLATED:a8d82298]
     // Set the playback speed of the recording stream
     api_regist("/index/api/setRecordSpeed", [](API_ARGS_MAP_ASYNC) {
@@ -1888,11 +1914,13 @@ void installWebApi() {
     // http://127.0.0.1/index/api/deleteRecordDirectroy?vhost=__defaultVhost__&app=live&stream=ss&period=2020-01-01
     api_regist("/index/api/deleteRecordDirectory", [](API_ARGS_MAP) {
         CHECK_SECRET();
-        CHECK_ARGS("vhost", "app", "stream", "period");
+        CHECK_ARGS("vhost", "app", "stream");
         auto tuple = MediaTuple{allArgs["vhost"], allArgs["app"], allArgs["stream"], ""};
         auto record_path = Recorder::getRecordPath(Recorder::type_mp4, tuple, allArgs["customized_path"]);
         auto period = allArgs["period"];
-        record_path = record_path + period + "/";
+        if (!period.empty()) {
+            record_path = record_path + period + "/";
+        }
 
         bool recording = false;
         auto name = allArgs["name"];
@@ -1927,6 +1955,15 @@ void installWebApi() {
         File::deleteEmptyDir(record_path);
     });
 
+    api_regist("/index/api/deleteSnapDirectory", [](API_ARGS_MAP) {
+        CHECK_SECRET();
+        CHECK_ARGS("vhost", "app", "stream");
+        GET_CONFIG(std::string, root, API::kSnapRoot);
+        auto path = File::absolutePath(allArgs["vhost"] + "/" + allArgs["app"] + "/" + allArgs["stream"] + "/" + allArgs["file"], root);
+        InfoL << "delete " << path;
+        File::delete_file(path, true);
+    });
+
     // 获取录像文件夹列表或mp4文件列表  [AUTO-TRANSLATED:f7e299bc]
     // Get the list of recording folders or mp4 files
     //http://127.0.0.1/index/api/getMP4RecordFile?vhost=__defaultVhost__&app=live&stream=ss&period=2020-01

src/Common/MultiMediaSourceMuxer.cpp

@@ -11,6 +11,7 @@
 #include <math.h>
 #include "Common/config.h"
 #include "MultiMediaSourceMuxer.h"
+#include "Thread/WorkThreadPool.h"
 
 using namespace std;
 using namespace toolkit;
@@ -400,6 +401,88 @@ bool MultiMediaSourceMuxer::setupRecord(MediaSource &sender, Recorder::type type
     }
 }
 
+std::string MultiMediaSourceMuxer::startRecord(const std::string &file_path, uint32_t back_time_ms, uint32_t 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;
+    TraceL << "mp4 save path: " << path;
+
+    auto muxer = std::make_shared<MP4Muxer>();
+    muxer->openMP4(path);
+    for (auto &track : MediaSink::getTracks()) {
+        muxer->addTrack(track);
+    }
+    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);
+        }
+        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;
+        }
+
+        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;
+    reader->setReadCB([muxer, now_dts, selected_index, forward_time_ms, reader, path](const Frame::Ptr &frame) mutable {
+        // 循环引用自身
+        if (!now_dts) {
+            now_dts = frame->dts();
+            selected_index = frame->getIndex();
+        }
+        if (frame->getIndex() == selected_index && now_dts + forward_time_ms < frame->dts()) {
+            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);
+        }
+    });
+
+    return path;
+#endif
+}
+
 // 此函数可能跨线程调用  [AUTO-TRANSLATED:e8c5f74d]
 // This function may be called across threads
 bool MultiMediaSourceMuxer::isRecording(MediaSource &sender, Recorder::type type) {

src/Common/MultiMediaSourceMuxer.h

@@ -122,6 +122,15 @@ public:
      */
     bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const std::string &custom_path, size_t max_second) override;
 
+    /**
+     * 开始录制mp4
+     * @param file_path mp4相对路径
+     * @param back_time_ms 回溯录制时长
+     * @param forward_time_ms 后续录制时长
+     * @return 录制文件绝对路径
+     */
+    std::string startRecord(const std::string &file_path, uint32_t back_time_ms, uint32_t forward_time_ms);
+
     /**
      * 获取录制状态
      * @param type 录制类型

src/Common/Stamp.cpp

@@ -148,10 +148,11 @@ void Stamp::revise_l(int64_t dts, int64_t pts, int64_t &dts_out, int64_t &pts_ou
         if (ABS(dts_diff) < 5000 || _need_sync > 3) {
             // 两种时间戳相差不得大于300ms
             dts_diff = _relative_stamp - _sync_master->_relative_stamp;
+            // 强制同步音视频
             if (dts_diff > 300) {
-                dts_diff = 300;
+                dts_diff = 0;
             } else if (dts_diff < -300) {
-                dts_diff = -300;
+                dts_diff = 0;
             }
             // 如果绝对时间戳小于5秒，那么说明他们的起始时间戳是一致的，那么强制同步  [AUTO-TRANSLATED:5d11ef6a]
             // If the absolute timestamp is less than 5 seconds, then it means that their starting timestamps are consistent, then force synchronization

src/Extension/Frame.cpp

@@ -205,11 +205,7 @@ bool FrameMerger::willFlush(const Frame::Ptr &frame) const{
 
         case mp4_nal_size:
         case h264_prefix: {
-            if (frame->dropAble() && !_have_config_frame) {
+            if (!_have_decode_able_frame) {
-                // 遇到SEI帧且未缓存配置帧，flush之前的帧
-                return true;
-            }
-            if (!_have_decode_able_frame && !_have_drop_able_frame) {
                 // 缓存中没有有效的能解码的帧，所以这次不flush  [AUTO-TRANSLATED:5d860722]
                 // There are no valid frames that can be decoded in the cache, so no flush this time.
                 return _frame_cache.size() > kMaxFrameCacheSize;
@@ -294,8 +290,6 @@ bool FrameMerger::inputFrame(const Frame::Ptr &frame, onOutput cb, BufferLikeStr
         cb(back->dts(), back->pts(), merged_frame, have_key_frame);
         _frame_cache.clear();
         _have_decode_able_frame = false;
-        _have_drop_able_frame = false;
-        _have_config_frame = false;
     }
 
     if (!frame) {
@@ -305,12 +299,6 @@ bool FrameMerger::inputFrame(const Frame::Ptr &frame, onOutput cb, BufferLikeStr
     if (frame->decodeAble()) {
         _have_decode_able_frame = true;
     }
-    if (frame->dropAble()) {
-        _have_drop_able_frame = true;
-    }
-    if (frame->configFrame()) {
-        _have_config_frame = true;
-    }
     _cb = std::move(cb);
     _frame_cache.emplace_back(Frame::getCacheAbleFrame(frame));
     return true;

src/Extension/Frame.h

@@ -660,8 +660,6 @@ private:
 private:
     int _type;
     bool _have_decode_able_frame = false;
-    bool _have_drop_able_frame = false;
-    bool _have_config_frame = false;
     onOutput _cb;
     toolkit::List<Frame::Ptr> _frame_cache;
 };

src/Rtp/RtpServer.cpp

@@ -207,8 +207,8 @@ void RtpServer::start(uint16_t local_port, const char *local_ip, const MediaTupl
     TcpServer::Ptr tcp_server;
     if (tcp_mode == PASSIVE || tcp_mode == ACTIVE) {
         auto processor = helper ? helper->getProcess() : nullptr;
-        // 如果共享同一个processor对象，那么tcp server深圳为单线程模式确保线程安全  [AUTO-TRANSLATED:68bdd877]
+        // 如果共享同一个processor对象，那么tcp server声明为单线程模式确保线程安全  [AUTO-TRANSLATED:68bdd877]
-        // If the same processor object is shared, then the TCP server Shenzhen is in single-threaded mode to ensure thread safety
+        // If the same processor object is shared, declare the TCP server in single-threaded mode to ensure thread safety.
         tcp_server = std::make_shared<TcpServer>(processor ? poller : nullptr);
         (*tcp_server)[RtpSession::kVhost] = tuple.vhost;
         (*tcp_server)[RtpSession::kApp] = tuple.app;