muxer_file_muxer.cpp 24 KB

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