muxer_file_muxer.cpp 26 KB

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