muxer_file_muxer.cpp 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855
  1. #include "muxer_file_muxer.h"
  2. #include "../base/logger.h"
  3. #include "../base/media_common.h"
  4. #include <filesystem>
  5. #include <sstream>
  6. #include <iomanip>
  7. #include <algorithm>
  8. extern "C" {
  9. #include <libavformat/avformat.h>
  10. #include <libavutil/opt.h>
  11. #include <libavutil/time.h>
  12. #include <libavutil/mathematics.h>
  13. }
  14. namespace av {
  15. namespace muxer {
  16. FileMuxer::FileMuxer() {
  17. AV_LOGGER_DEBUG("创建文件复用器");
  18. lastFlushTime_ = std::chrono::steady_clock::now();
  19. }
  20. FileMuxer::~FileMuxer() {
  21. close();
  22. AV_LOGGER_DEBUG("文件复用器已销毁");
  23. }
  24. ErrorCode FileMuxer::initialize(const MuxerParams& params) {
  25. if (params.type != MuxerType::FILE_MUXER) {
  26. AV_LOGGER_ERROR("参数类型不是文件复用器");
  27. return ErrorCode::INVALID_ARGUMENT;
  28. }
  29. fileMuxerParams_ = static_cast<const FileMuxerParams&>(params);
  30. if (!validateParams(fileMuxerParams_)) {
  31. return ErrorCode::INVALID_PARAMS;
  32. }
  33. // 设置输出文件
  34. if (!fileMuxerParams_.outputFile.empty()) {
  35. currentOutputFile_ = fileMuxerParams_.outputFile;
  36. } else {
  37. currentOutputFile_ = fileMuxerParams_.outputPath;
  38. }
  39. ErrorCode result = validateFilePath(currentOutputFile_);
  40. if (result != ErrorCode::OK) {
  41. return result;
  42. }
  43. // 设置分段
  44. if (fileMuxerParams_.enableSegmentation) {
  45. segmentationEnabled_ = true;
  46. if (fileMuxerParams_.segmentPattern.empty()) {
  47. // 生成默认分段模式
  48. std::filesystem::path path(currentOutputFile_);
  49. std::string stem = path.stem().string();
  50. std::string ext = path.extension().string();
  51. fileMuxerParams_.segmentPattern = stem + "_%03d" + ext;
  52. }
  53. }
  54. setState(MuxerState::INITIALIZED);
  55. AV_LOGGER_INFOF("文件复用器初始化成功: {}", currentOutputFile_);
  56. return ErrorCode::OK;
  57. }
  58. ErrorCode FileMuxer::start() {
  59. if (getState() != MuxerState::INITIALIZED) {
  60. AV_LOGGER_ERROR("复用器状态无效,无法启动");
  61. return ErrorCode::INVALID_STATE;
  62. }
  63. ErrorCode result = setupOutput();
  64. if (result != ErrorCode::OK) {
  65. return result;
  66. }
  67. result = writeHeader();
  68. if (result != ErrorCode::OK) {
  69. return result;
  70. }
  71. // 启动写入线程
  72. shouldStopWriting_ = false;
  73. try {
  74. writeThread_ = std::thread(&FileMuxer::writeThreadFunc, this);
  75. setState(MuxerState::STARTED);
  76. fileStartTime_ = std::chrono::steady_clock::now();
  77. AV_LOGGER_INFO("文件复用器已启动");
  78. return ErrorCode::OK;
  79. } catch (const std::exception& e) {
  80. AV_LOGGER_ERRORF("启动写入线程失败: {}", e.what());
  81. return ErrorCode::THREAD_ERROR;
  82. }
  83. }
  84. ErrorCode FileMuxer::stop() {
  85. if (getState() != MuxerState::STARTED && getState() != MuxerState::PAUSED) {
  86. return ErrorCode::OK;
  87. }
  88. // 停止写入线程
  89. shouldStopWriting_ = true;
  90. queueCondition_.notify_all();
  91. if (writeThread_.joinable()) {
  92. writeThread_.join();
  93. }
  94. // 刷新剩余数据
  95. flush();
  96. // 写入文件尾
  97. ErrorCode result = writeTrailer();
  98. if (result != ErrorCode::OK) {
  99. AV_LOGGER_ERRORF("写入文件尾失败: {}", static_cast<int>(result));
  100. }
  101. setState(MuxerState::STOPPED);
  102. AV_LOGGER_INFO("文件复用器已停止");
  103. return ErrorCode::OK;
  104. }
  105. ErrorCode FileMuxer::close() {
  106. stop();
  107. ErrorCode result = closeOutputFile();
  108. // 清空队列
  109. {
  110. std::lock_guard<std::mutex> lock(queueMutex_);
  111. while (!packetQueue_.empty()) {
  112. packetQueue_.pop();
  113. }
  114. }
  115. setState(MuxerState::IDLE);
  116. AV_LOGGER_INFO("文件复用器已关闭");
  117. return result;
  118. }
  119. ErrorCode FileMuxer::writePacket(AVPacket* packet) {
  120. if (!packet) {
  121. AV_LOGGER_ERROR("包指针为空");
  122. return ErrorCode::INVALID_ARGUMENT;
  123. }
  124. if (getState() != MuxerState::STARTED) {
  125. AV_LOGGER_ERROR("复用器未启动,无法写入包");
  126. return ErrorCode::INVALID_STATE;
  127. }
  128. // 检查队列大小
  129. {
  130. std::lock_guard<std::mutex> lock(queueMutex_);
  131. if (packetQueue_.size() >= MAX_QUEUE_SIZE) {
  132. AV_LOGGER_WARNING("包队列已满,丢弃包");
  133. {
  134. std::lock_guard<std::mutex> statsLock(statsMutex_);
  135. stats_.droppedPackets++;
  136. }
  137. return ErrorCode::QUEUE_FULL;
  138. }
  139. }
  140. // 复制包
  141. AVPacket* packetCopy = av_packet_alloc();
  142. if (!packetCopy) {
  143. AV_LOGGER_ERROR("分配包内存失败");
  144. return ErrorCode::OUT_OF_MEMORY;
  145. }
  146. int ret = av_packet_ref(packetCopy, packet);
  147. if (ret < 0) {
  148. av_packet_free(&packetCopy);
  149. AV_LOGGER_ERRORF("复制包失败: {}", ffmpeg_utils::errorToString(ret));
  150. return static_cast<ErrorCode>(ret);
  151. }
  152. // 添加到队列
  153. auto queueItem = std::make_unique<PacketQueueItem>(packetCopy, packet->stream_index);
  154. {
  155. std::lock_guard<std::mutex> lock(queueMutex_);
  156. packetQueue_.push(std::move(queueItem));
  157. queuedPackets_++;
  158. }
  159. queueCondition_.notify_one();
  160. return ErrorCode::OK;
  161. }
  162. ErrorCode FileMuxer::writeFrame(AVFrame* frame, int streamIndex) {
  163. if (!frame) {
  164. AV_LOGGER_ERROR("帧指针为空");
  165. return ErrorCode::INVALID_ARGUMENT;
  166. }
  167. // 这里需要编码器将帧编码为包,然后调用writePacket
  168. // 由于这是复用器,通常接收已编码的包
  169. AV_LOGGER_WARNING("文件复用器不直接支持写入原始帧,请先编码为包");
  170. return ErrorCode::NOT_SUPPORTED;
  171. }
  172. ErrorCode FileMuxer::flush() {
  173. if (!formatCtx_) {
  174. return ErrorCode::OK;
  175. }
  176. std::lock_guard<std::mutex> lock(fileMutex_);
  177. int ret = av_write_frame(formatCtx_, nullptr); // 刷新
  178. if (ret < 0) {
  179. AV_LOGGER_ERRORF("刷新复用器失败: {}", ffmpeg_utils::errorToString(ret));
  180. return static_cast<ErrorCode>(ret);
  181. }
  182. if (formatCtx_->pb) {
  183. avio_flush(formatCtx_->pb);
  184. }
  185. lastFlushTime_ = std::chrono::steady_clock::now();
  186. AV_LOGGER_DEBUG("复用器已刷新");
  187. return ErrorCode::OK;
  188. }
  189. ErrorCode FileMuxer::addStream(const StreamInfo& streamInfo) {
  190. if (getState() != MuxerState::INITIALIZED) {
  191. AV_LOGGER_ERROR("只能在初始化状态下添加流");
  192. return ErrorCode::INVALID_STATE;
  193. }
  194. std::lock_guard<std::mutex> lock(streamsMutex_);
  195. // 检查流索引是否已存在
  196. auto it = std::find_if(streams_.begin(), streams_.end(),
  197. [&streamInfo](const StreamInfo& info) {
  198. return info.index == streamInfo.index;
  199. });
  200. if (it != streams_.end()) {
  201. AV_LOGGER_ERRORF("流索引 {} 已存在", streamInfo.index);
  202. return ErrorCode::STREAM_EXISTS;
  203. }
  204. streams_.push_back(streamInfo);
  205. AV_LOGGER_INFOF("已添加流: 索引={}, 类型={}, 编解码器={}",
  206. streamInfo.index, static_cast<int>(streamInfo.type),
  207. streamInfo.codecName);
  208. return ErrorCode::OK;
  209. }
  210. ErrorCode FileMuxer::setOutputFile(const std::string& filename) {
  211. if (getState() != MuxerState::IDLE && getState() != MuxerState::INITIALIZED) {
  212. AV_LOGGER_ERROR("无法在运行时更改输出文件");
  213. return ErrorCode::INVALID_STATE;
  214. }
  215. ErrorCode result = validateFilePath(filename);
  216. if (result != ErrorCode::OK) {
  217. return result;
  218. }
  219. currentOutputFile_ = filename;
  220. fileMuxerParams_.outputFile = filename;
  221. AV_LOGGER_INFOF("输出文件已设置为: {}", filename);
  222. return ErrorCode::OK;
  223. }
  224. std::string FileMuxer::getOutputFile() const {
  225. return currentOutputFile_;
  226. }
  227. int64_t FileMuxer::getCurrentFileSize() const {
  228. return currentFileSize_;
  229. }
  230. double FileMuxer::getCurrentDuration() const {
  231. auto now = std::chrono::steady_clock::now();
  232. return std::chrono::duration<double>(now - fileStartTime_).count();
  233. }
  234. ErrorCode FileMuxer::enableSegmentation(bool enable, int duration) {
  235. if (getState() == MuxerState::STARTED) {
  236. AV_LOGGER_ERROR("无法在运行时更改分段设置");
  237. return ErrorCode::INVALID_STATE;
  238. }
  239. segmentationEnabled_ = enable;
  240. fileMuxerParams_.enableSegmentation = enable;
  241. fileMuxerParams_.segmentDuration = duration;
  242. AV_LOGGER_INFOF("分段已{}: 时长={}秒", enable ? "启用" : "禁用", duration);
  243. return ErrorCode::OK;
  244. }
  245. ErrorCode FileMuxer::forceNewSegment() {
  246. if (!segmentationEnabled_) {
  247. AV_LOGGER_ERROR("分段功能未启用");
  248. return ErrorCode::INVALID_STATE;
  249. }
  250. if (getState() != MuxerState::STARTED) {
  251. AV_LOGGER_ERROR("复用器未启动");
  252. return ErrorCode::INVALID_STATE;
  253. }
  254. return createNewSegment();
  255. }
  256. std::vector<std::string> FileMuxer::getSegmentFiles() const {
  257. return segmentFiles_;
  258. }
  259. ErrorCode FileMuxer::setFastStart(bool enable) {
  260. fileMuxerParams_.enableFastStart = enable;
  261. if (enable) {
  262. fileMuxerParams_.movFlags |= 0x01; // AVFMT_FLAG_FASTSTART
  263. } else {
  264. fileMuxerParams_.movFlags &= ~0x01;
  265. }
  266. AV_LOGGER_INFOF("快速开始已{}", enable ? "启用" : "禁用");
  267. return ErrorCode::OK;
  268. }
  269. ErrorCode FileMuxer::setMovFlags(int flags) {
  270. fileMuxerParams_.movFlags = flags;
  271. AV_LOGGER_INFOF("MOV标志已设置为: 0x{:X}", flags);
  272. return ErrorCode::OK;
  273. }
  274. ErrorCode FileMuxer::setupOutput() {
  275. // 分配格式上下文
  276. const char* formatName = fileMuxerParams_.format.empty() ? nullptr : fileMuxerParams_.format.c_str();
  277. const char* filename = currentOutputFile_.c_str();
  278. int ret = avformat_alloc_output_context2(&formatCtx_, nullptr, formatName, filename);
  279. if (ret < 0) {
  280. AV_LOGGER_ERRORF("分配输出格式上下文失败: {}", ffmpeg_utils::errorToString(ret));
  281. return static_cast<ErrorCode>(ret);
  282. }
  283. if (!formatCtx_->oformat) {
  284. AV_LOGGER_ERROR("无法确定输出格式");
  285. return ErrorCode::FORMAT_NOT_SUPPORTED;
  286. }
  287. AV_LOGGER_INFOF("输出格式: {} ({})", formatCtx_->oformat->name, formatCtx_->oformat->long_name);
  288. // 设置流
  289. ErrorCode result = setupStreams();
  290. if (result != ErrorCode::OK) {
  291. return result;
  292. }
  293. // 设置元数据
  294. for (const auto& meta : fileMuxerParams_.metadata) {
  295. av_dict_set(&formatCtx_->metadata, meta.first.c_str(), meta.second.c_str(), 0);
  296. }
  297. // 打开输出文件
  298. return openOutputFile();
  299. }
  300. ErrorCode FileMuxer::writeHeader() {
  301. if (!formatCtx_) {
  302. AV_LOGGER_ERROR("格式上下文未初始化");
  303. return ErrorCode::INVALID_STATE;
  304. }
  305. // 设置格式选项
  306. AVDictionary* options = nullptr;
  307. for (const auto& option : fileMuxerParams_.options) {
  308. av_dict_set(&options, option.first.c_str(), option.second.c_str(), 0);
  309. }
  310. // 设置MOV标志
  311. if (fileMuxerParams_.movFlags != 0) {
  312. av_dict_set_int(&options, "movflags", fileMuxerParams_.movFlags, 0);
  313. }
  314. // 写入文件头
  315. int ret = avformat_write_header(formatCtx_, &options);
  316. av_dict_free(&options);
  317. if (ret < 0) {
  318. AV_LOGGER_ERRORF("写入文件头失败: {}", ffmpeg_utils::errorToString(ret));
  319. return static_cast<ErrorCode>(ret);
  320. }
  321. AV_LOGGER_INFO("文件头已写入");
  322. return ErrorCode::OK;
  323. }
  324. ErrorCode FileMuxer::writeTrailer() {
  325. if (!formatCtx_) {
  326. return ErrorCode::OK;
  327. }
  328. std::lock_guard<std::mutex> lock(fileMutex_);
  329. int ret = av_write_trailer(formatCtx_);
  330. if (ret < 0) {
  331. AV_LOGGER_ERRORF("写入文件尾失败: {}", ffmpeg_utils::errorToString(ret));
  332. return static_cast<ErrorCode>(ret);
  333. }
  334. AV_LOGGER_INFO("文件尾已写入");
  335. return ErrorCode::OK;
  336. }
  337. ErrorCode FileMuxer::openOutputFile() {
  338. if (!formatCtx_) {
  339. return ErrorCode::INVALID_STATE;
  340. }
  341. // 检查是否需要覆盖文件
  342. if (!fileMuxerParams_.overwrite && std::filesystem::exists(currentOutputFile_)) {
  343. AV_LOGGER_ERRORF("文件已存在且不允许覆盖: {}", currentOutputFile_);
  344. return ErrorCode::FILE_EXISTS;
  345. }
  346. // 打开输出文件
  347. if (!(formatCtx_->oformat->flags & AVFMT_NOFILE)) {
  348. int ret = avio_open(&formatCtx_->pb, currentOutputFile_.c_str(), AVIO_FLAG_WRITE);
  349. if (ret < 0) {
  350. AV_LOGGER_ERRORF("打开输出文件失败: {} ({})",
  351. ffmpeg_utils::errorToString(ret), currentOutputFile_);
  352. return static_cast<ErrorCode>(ret);
  353. }
  354. }
  355. currentFileSize_ = 0;
  356. AV_LOGGER_INFOF("输出文件已打开: {}", currentOutputFile_);
  357. return ErrorCode::OK;
  358. }
  359. ErrorCode FileMuxer::closeOutputFile() {
  360. if (!formatCtx_) {
  361. return ErrorCode::OK;
  362. }
  363. std::lock_guard<std::mutex> lock(fileMutex_);
  364. // 关闭文件
  365. if (formatCtx_->pb && !(formatCtx_->oformat->flags & AVFMT_NOFILE)) {
  366. avio_closep(&formatCtx_->pb);
  367. }
  368. // 释放格式上下文
  369. avformat_free_context(formatCtx_);
  370. formatCtx_ = nullptr;
  371. // 清理流映射
  372. streamMap_.clear();
  373. AV_LOGGER_INFO("输出文件已关闭");
  374. return ErrorCode::OK;
  375. }
  376. ErrorCode FileMuxer::createNewSegment() {
  377. if (!segmentationEnabled_) {
  378. return ErrorCode::OK;
  379. }
  380. // 写入当前段的尾部
  381. ErrorCode result = writeTrailer();
  382. if (result != ErrorCode::OK) {
  383. return result;
  384. }
  385. // 关闭当前文件
  386. result = closeOutputFile();
  387. if (result != ErrorCode::OK) {
  388. return result;
  389. }
  390. // 添加到段文件列表
  391. segmentFiles_.push_back(currentOutputFile_);
  392. // 生成新的段文件名
  393. currentSegmentIndex_++;
  394. currentOutputFile_ = generateSegmentFilename(currentSegmentIndex_);
  395. // 重新设置输出
  396. result = setupOutput();
  397. if (result != ErrorCode::OK) {
  398. return result;
  399. }
  400. // 写入新段的头部
  401. result = writeHeader();
  402. if (result != ErrorCode::OK) {
  403. return result;
  404. }
  405. fileStartTime_ = std::chrono::steady_clock::now();
  406. AV_LOGGER_INFOF("创建新段: {}", currentOutputFile_);
  407. return ErrorCode::OK;
  408. }
  409. ErrorCode FileMuxer::setupStreams() {
  410. std::lock_guard<std::mutex> lock(streamsMutex_);
  411. for (const auto& streamInfo : streams_) {
  412. AVStream* stream = createAVStream(streamInfo);
  413. if (!stream) {
  414. AV_LOGGER_ERRORF("创建流失败: 索引={}", streamInfo.index);
  415. return ErrorCode::STREAM_CREATE_FAILED;
  416. }
  417. streamMap_[streamInfo.index] = stream;
  418. }
  419. AV_LOGGER_INFOF("已设置 {} 个流", streams_.size());
  420. return ErrorCode::OK;
  421. }
  422. AVStream* FileMuxer::createAVStream(const StreamInfo& streamInfo) {
  423. if (!formatCtx_) {
  424. return nullptr;
  425. }
  426. AVStream* stream = avformat_new_stream(formatCtx_, nullptr);
  427. if (!stream) {
  428. AV_LOGGER_ERROR("创建新流失败");
  429. return nullptr;
  430. }
  431. stream->id = streamInfo.index;
  432. stream->time_base = streamInfo.timeBase;
  433. // 设置编解码器参数
  434. AVCodecParameters* codecpar = stream->codecpar;
  435. codecpar->codec_id = streamInfo.codecId;
  436. codecpar->codec_type = (streamInfo.type == StreamType::VIDEO) ? AVMEDIA_TYPE_VIDEO : AVMEDIA_TYPE_AUDIO;
  437. if (streamInfo.type == StreamType::VIDEO) {
  438. codecpar->width = streamInfo.width;
  439. codecpar->height = streamInfo.height;
  440. codecpar->format = streamInfo.pixelFormat;
  441. codecpar->bit_rate = streamInfo.bitrate;
  442. // 设置帧率
  443. stream->avg_frame_rate = streamInfo.frameRate;
  444. stream->r_frame_rate = streamInfo.frameRate;
  445. } else if (streamInfo.type == StreamType::AUDIO) {
  446. codecpar->sample_rate = streamInfo.sampleRate;
  447. av_channel_layout_default(&codecpar->ch_layout, streamInfo.channels);
  448. codecpar->format = streamInfo.sampleFormat;
  449. codecpar->bit_rate = streamInfo.bitrate;
  450. }
  451. AV_LOGGER_INFOF("创建流成功: 索引={}, 类型={}", streamInfo.index, static_cast<int>(streamInfo.type));
  452. return stream;
  453. }
  454. ErrorCode FileMuxer::processPacket(AVPacket* packet) {
  455. if (!packet || !formatCtx_) {
  456. return ErrorCode::INVALID_ARGUMENT;
  457. }
  458. // 检查流索引
  459. auto it = streamMap_.find(packet->stream_index);
  460. if (it == streamMap_.end()) {
  461. AV_LOGGER_ERRORF("未找到流索引: {}", packet->stream_index);
  462. return ErrorCode::STREAM_NOT_FOUND;
  463. }
  464. // 检查是否需要创建新段
  465. if (shouldCreateNewSegment()) {
  466. ErrorCode result = createNewSegment();
  467. if (result != ErrorCode::OK) {
  468. return result;
  469. }
  470. }
  471. return writePacketInternal(packet);
  472. }
  473. ErrorCode FileMuxer::writePacketInternal(AVPacket* packet) {
  474. std::lock_guard<std::mutex> lock(fileMutex_);
  475. auto startTime = std::chrono::steady_clock::now();
  476. // 写入包
  477. int ret = av_interleaved_write_frame(formatCtx_, packet);
  478. if (ret < 0) {
  479. AV_LOGGER_ERRORF("写入包失败: {}", ffmpeg_utils::errorToString(ret));
  480. return static_cast<ErrorCode>(ret);
  481. }
  482. // 更新统计信息
  483. updateStats(packet);
  484. currentFileSize_ += packet->size;
  485. writtenPackets_++;
  486. // 更新平均写入时间
  487. auto endTime = std::chrono::steady_clock::now();
  488. double writeTime = std::chrono::duration<double, std::milli>(endTime - startTime).count();
  489. averageWriteTime_ = (averageWriteTime_ * 0.9) + (writeTime * 0.1);
  490. // 定期刷新
  491. if (!fileMuxerParams_.syncMode) {
  492. auto now = std::chrono::steady_clock::now();
  493. auto elapsed = std::chrono::duration<double>(now - lastFlushTime_).count();
  494. if (elapsed >= fileMuxerParams_.flushInterval) {
  495. flush();
  496. }
  497. }
  498. return ErrorCode::OK;
  499. }
  500. void FileMuxer::writeThreadFunc() {
  501. AV_LOGGER_INFO("写入线程已启动");
  502. while (!shouldStopWriting_) {
  503. std::unique_ptr<PacketQueueItem> item;
  504. // 从队列获取包
  505. {
  506. std::unique_lock<std::mutex> lock(queueMutex_);
  507. queueCondition_.wait(lock, [this] {
  508. return !packetQueue_.empty() || shouldStopWriting_;
  509. });
  510. if (shouldStopWriting_ && packetQueue_.empty()) {
  511. break;
  512. }
  513. if (!packetQueue_.empty()) {
  514. item = std::move(packetQueue_.front());
  515. packetQueue_.pop();
  516. }
  517. }
  518. if (item && item->packet) {
  519. ErrorCode result = processPacket(item->packet);
  520. if (result != ErrorCode::OK) {
  521. onError(result, "写入包失败");
  522. }
  523. }
  524. }
  525. // 处理剩余的包
  526. while (true) {
  527. std::unique_ptr<PacketQueueItem> item;
  528. {
  529. std::lock_guard<std::mutex> lock(queueMutex_);
  530. if (packetQueue_.empty()) {
  531. break;
  532. }
  533. item = std::move(packetQueue_.front());
  534. packetQueue_.pop();
  535. }
  536. if (item && item->packet) {
  537. processPacket(item->packet);
  538. }
  539. }
  540. AV_LOGGER_INFO("写入线程已退出");
  541. }
  542. std::string FileMuxer::generateSegmentFilename(int segmentIndex) {
  543. if (fileMuxerParams_.segmentPattern.empty()) {
  544. std::filesystem::path path(currentOutputFile_);
  545. std::string stem = path.stem().string();
  546. std::string ext = path.extension().string();
  547. std::ostringstream oss;
  548. oss << stem << "_" << std::setfill('0') << std::setw(3) << segmentIndex << ext;
  549. return oss.str();
  550. } else {
  551. char buffer[1024];
  552. snprintf(buffer, sizeof(buffer), fileMuxerParams_.segmentPattern.c_str(), segmentIndex);
  553. return std::string(buffer);
  554. }
  555. }
  556. bool FileMuxer::shouldCreateNewSegment() const {
  557. if (!segmentationEnabled_) {
  558. return false;
  559. }
  560. // 检查时长
  561. if (fileMuxerParams_.segmentDuration > 0) {
  562. double currentDuration = getCurrentDuration();
  563. if (currentDuration >= fileMuxerParams_.segmentDuration) {
  564. return true;
  565. }
  566. }
  567. // 检查文件大小
  568. if (fileMuxerParams_.maxFileSize > 0) {
  569. if (currentFileSize_ >= fileMuxerParams_.maxFileSize) {
  570. return true;
  571. }
  572. }
  573. return false;
  574. }
  575. ErrorCode FileMuxer::validateFilePath(const std::string& path) {
  576. if (path.empty()) {
  577. AV_LOGGER_ERROR("文件路径不能为空");
  578. return ErrorCode::INVALID_ARGUMENT;
  579. }
  580. try {
  581. std::filesystem::path filePath(path);
  582. // 检查父目录是否存在
  583. std::filesystem::path parentDir = filePath.parent_path();
  584. if (!parentDir.empty() && !std::filesystem::exists(parentDir)) {
  585. // 尝试创建目录
  586. std::filesystem::create_directories(parentDir);
  587. }
  588. // 检查文件扩展名
  589. std::string extension = filePath.extension().string();
  590. if (extension.empty()) {
  591. AV_LOGGER_WARNING("文件没有扩展名,可能导致格式检测问题");
  592. }
  593. } catch (const std::exception& e) {
  594. AV_LOGGER_ERRORF("文件路径验证失败: {}", e.what());
  595. return ErrorCode::INVALID_PATH;
  596. }
  597. return ErrorCode::OK;
  598. }
  599. bool FileMuxer::validateParams(const MuxerParams& params) {
  600. if (!AbstractMuxer::validateParams(params)) {
  601. return false;
  602. }
  603. const auto& fileParams = static_cast<const FileMuxerParams&>(params);
  604. if (fileParams.outputFile.empty() && fileParams.outputPath.empty()) {
  605. AV_LOGGER_ERROR("输出文件路径不能为空");
  606. return false;
  607. }
  608. if (fileParams.segmentDuration < 0) {
  609. AV_LOGGER_ERROR("分段时长不能为负数");
  610. return false;
  611. }
  612. if (fileParams.maxFileSize < 0) {
  613. AV_LOGGER_ERROR("最大文件大小不能为负数");
  614. return false;
  615. }
  616. if (fileParams.flushInterval <= 0) {
  617. AV_LOGGER_ERROR("刷新间隔必须大于0");
  618. return false;
  619. }
  620. return true;
  621. }
  622. // FileMuxerFactory 实现
  623. std::unique_ptr<AbstractMuxer> FileMuxer::FileMuxerFactory::createMuxer(MuxerType type) {
  624. if (type == MuxerType::FILE_MUXER) {
  625. return std::make_unique<FileMuxer>();
  626. }
  627. return nullptr;
  628. }
  629. bool FileMuxer::FileMuxerFactory::isTypeSupported(MuxerType type) const {
  630. return type == MuxerType::FILE_MUXER;
  631. }
  632. std::vector<MuxerType> FileMuxer::FileMuxerFactory::getSupportedTypes() const {
  633. return {MuxerType::FILE_MUXER};
  634. }
  635. std::unique_ptr<FileMuxer> FileMuxer::FileMuxerFactory::createFileMuxer(const std::string& filename) {
  636. auto muxer = std::make_unique<FileMuxer>();
  637. FileMuxerParams params;
  638. params.outputFile = filename;
  639. params.format = AbstractMuxer::getFormatFromExtension(filename);
  640. ErrorCode result = muxer->initialize(params);
  641. if (result != ErrorCode::OK) {
  642. AV_LOGGER_ERRORF("创建文件复用器失败: {}", static_cast<int>(result));
  643. return nullptr;
  644. }
  645. return muxer;
  646. }
  647. std::unique_ptr<FileMuxer> FileMuxer::FileMuxerFactory::createSegmentedMuxer(const std::string& pattern, int duration) {
  648. auto muxer = std::make_unique<FileMuxer>();
  649. FileMuxerParams params;
  650. params.outputFile = pattern;
  651. params.enableSegmentation = true;
  652. params.segmentDuration = duration;
  653. params.segmentPattern = pattern;
  654. // 从模式中推断格式
  655. std::string firstFile = pattern;
  656. size_t pos = firstFile.find("%");
  657. if (pos != std::string::npos) {
  658. firstFile.replace(pos, firstFile.find('d', pos) - pos + 1, "001");
  659. }
  660. params.format = AbstractMuxer::getFormatFromExtension(firstFile);
  661. ErrorCode result = muxer->initialize(params);
  662. if (result != ErrorCode::OK) {
  663. AV_LOGGER_ERRORF("创建分段复用器失败: {}", static_cast<int>(result));
  664. return nullptr;
  665. }
  666. return muxer;
  667. }
  668. } // namespace muxer
  669. } // namespace av