diff --git a/server/FFmpegSource.cpp b/server/FFmpegSource.cpp index 4300ae3e..98b17f2f 100644 --- a/server/FFmpegSource.cpp +++ b/server/FFmpegSource.cpp @@ -363,42 +363,88 @@ void FFmpegSource::onGetMediaSource(const MediaSource::Ptr &src) { #include "Player/MediaPlayer.h" #include "Codec/Transcode.h" -static void makeSnapAsync(const string &play_url, const string &save_path, float timeout_sec, const FFmpegSnap::onSnap &cb) { - struct Holder { - MediaPlayer::Ptr player; - }; - auto holder = std::make_shared(); - auto player = std::make_shared(); - (*player)[mediakit::Client::kTimeoutMS] = timeout_sec * 1000; +static std::string getStreamName(const std::string &input) { + // 先找 '/' + size_t pos = input.find('/'); + if (pos != std::string::npos) { + return input.substr(0, pos); + } - player->setOnPlayResult([holder, save_path, cb, timeout_sec](const SockException &ex) mutable { - onceToken token(nullptr, [&]() { holder->player = nullptr; }); - auto video = ex ? nullptr : dynamic_pointer_cast(holder->player->getTrack(TrackVideo, false)); - if (!video) { - cb(false, ex ? ex.what() : "none video track"); - return; - } - auto decoder = std::make_shared(video); - auto new_holder = std::make_shared(*holder); - auto timer = EventPollerPool::Instance().getPoller()->doDelayTask(1000 * timeout_sec, [cb, new_holder]() { - // 防止解码失败导致播放器无法释放 - new_holder->player = nullptr; - cb(false, "decode frame timeout"); - return 0; + // 如果没有 '/', 再找 '.' + pos = input.find('.'); + if (pos != std::string::npos) { + return input.substr(0, pos); + } + + // 都没有，返回原字符串 + return input; +} + +static bool makeSnapAsync(const string &play_url, const string &save_path, float timeout_sec, const FFmpegSnap::onSnap &cb) { + MediaInfo info(play_url); + info.stream = getStreamName(info.stream); + MediaSource::Ptr src; + Track::Ptr track; + MultiMediaSourceMuxer::Ptr muxer; + if (info.host == "127.0.0.1" || info.host == "localhost" || info.params.find("self=1") != std::string::npos) { + // 截图zlm本身 + src = MediaSource::find(info.vhost, info.app, info.stream, false); + track = src ? src->getTrack(TrackVideo) : nullptr; + muxer = src ? src->getMuxer() : nullptr; + } + // 要求开启转码额度 + if (src && track && muxer) { + DebugL << "start snap in fast mode: " << play_url; + src->getOwnerPoller()->async([muxer, track, timeout_sec, cb, save_path]() mutable { + auto done = std::make_shared>(false); + auto reader = muxer->getFrameReader(); + auto timer = EventPollerPool::Instance().getPoller()->doDelayTask(1000 * timeout_sec, [reader, cb, done]() { + if (*done) { + return 0; + } + *done = true; + // 防止解码失败导致播放器无法释放 + cb(false, "decode frame timeout"); + return 0; + }); + + // 单线程解码，没必要太快 + auto decoder = std::make_shared(track, 1); + decoder->setOnDecode([save_path, cb, done, timer](const FFmpegFrame::Ptr &frame) mutable { + if (*done) { + return; + } + *done = true; + timer->cancel(); + + auto ret = FFmpegUtils::saveFrame(frame, save_path.data()); + cb(std::get<0>(ret), std::get<1>(ret)); + }); + + auto decode_poller = WorkThreadPool::Instance().getPoller(); + reader->setReadCB([decoder, decode_poller, done](const Frame::Ptr &frame) mutable { + if (*done) { + // 销毁解码器，防止无谓的解码 + decoder = nullptr; + return; + } + if (frame->getTrackType() == TrackVideo) { + // 异步解码，防止阻塞网络线程 + auto copy_frame = Frame::getCacheAbleFrame(frame); + decode_poller->async([decoder, copy_frame, done]() mutable { + if (*done) { + // 销毁解码器，防止无谓的解码 + decoder = nullptr; + return; + } + decoder->inputFrame(copy_frame, false, false, true); + }); + } + }); }); - auto done = false; - decoder->setOnDecode([save_path, new_holder, cb, done, timer](const FFmpegFrame::Ptr &frame) mutable { - if (done) { - return; - } - onceToken token(nullptr, [&]() { new_holder->player = nullptr; timer->cancel(); done = true; }); - auto ret = FFmpegUtils::saveFrame(frame, save_path.data()); - cb(std::get<0>(ret), std::get<1>(ret)); - }); - video->addDelegate([decoder](const Frame::Ptr &frame) { return decoder->inputFrame(frame, false, true); }); - }); - player->play(play_url); - holder->player = std::move(player); + return true; + } + return false; } #endif @@ -406,8 +452,9 @@ static void makeSnapAsync(const string &play_url, const string &save_path, float void FFmpegSnap::makeSnap(bool async, const string &play_url, const string &save_path, float timeout_sec, const onSnap &cb) { #if defined(ENABLE_FFMPEG) if (async) { - makeSnapAsync(play_url, save_path, timeout_sec, cb); - return; + if (makeSnapAsync(play_url, save_path, timeout_sec, cb)) { + return; + } } #endif diff --git a/src/Common/MultiMediaSourceMuxer.cpp b/src/Common/MultiMediaSourceMuxer.cpp index 10479645..f56e8334 100644 --- a/src/Common/MultiMediaSourceMuxer.cpp +++ b/src/Common/MultiMediaSourceMuxer.cpp @@ -564,7 +564,7 @@ bool MultiMediaSourceMuxer::isRecording(Recorder::type type) { void MultiMediaSourceMuxer::startSendRtp(const MediaSourceEvent::SendRtpArgs &args, const std::function cb) { #if defined(ENABLE_RTPPROXY) - createGopCacheIfNeed(1); + createGopCacheIfNeed(); auto ring = _ring; auto ssrc = args.ssrc; @@ -636,6 +636,13 @@ bool MultiMediaSourceMuxer::stopSendRtp(const string &ssrc) { #endif//ENABLE_RTPPROXY } +MultiMediaSourceMuxer::RingType::RingReader::Ptr MultiMediaSourceMuxer::getFrameReader() { + auto poller = getOwnerPoller(MediaSource::NullMediaSource()); + CHECK(poller->isCurrentThread()); + createGopCacheIfNeed(); + return _ring->attach(poller); +} + EventPoller::Ptr MultiMediaSourceMuxer::getOwnerPoller(MediaSource &sender) { auto listener = getDelegate(); if (!listener) { @@ -754,13 +761,7 @@ void MultiMediaSourceMuxer::onAllTrackReady() { listener->onAllTrackReady(); } -#if defined(ENABLE_RTPPROXY) - GET_CONFIG(size_t, gop_cache, RtpProxy::kGopCache); - if (gop_cache > 0) { - createGopCacheIfNeed(gop_cache); - } -#endif - + createGopCacheIfNeed(); Stamp *first = nullptr; for (auto &pr : _stamps) { if (!first) { @@ -775,10 +776,11 @@ void MultiMediaSourceMuxer::onAllTrackReady() { InfoL << "stream: " << shortUrl() << " , codec info: " << getTrackInfoStr(this); } -void MultiMediaSourceMuxer::createGopCacheIfNeed(size_t gop_count) { +void MultiMediaSourceMuxer::createGopCacheIfNeed() { if (_ring) { return; } + GET_CONFIG(size_t, gop_cache, RtpProxy::kGopCache); weak_ptr weak_self = shared_from_this(); auto src = std::make_shared(weak_self.lock()); _ring = std::make_shared(1024, [weak_self, src](int size) { @@ -789,7 +791,7 @@ void MultiMediaSourceMuxer::createGopCacheIfNeed(size_t gop_count) { strong_self->onReaderChanged(*src, strong_self->totalReaderCount()); }); } - }, gop_count); + }, std::max(gop_cache, 1)); } void MultiMediaSourceMuxer::resetTracks() { diff --git a/src/Common/MultiMediaSourceMuxer.h b/src/Common/MultiMediaSourceMuxer.h index ea656156..38a8c95c 100644 --- a/src/Common/MultiMediaSourceMuxer.h +++ b/src/Common/MultiMediaSourceMuxer.h @@ -194,6 +194,9 @@ public: */ std::shared_ptr getMuxer(MediaSource &sender) const override; + // 获取frame ring reader + RingType::RingReader::Ptr getFrameReader(); + const ProtocolOption &getOption() const; const MediaTuple &getMediaTuple() const; std::string shortUrl() const; @@ -238,7 +241,7 @@ protected: bool onTrackFrame_l(const Frame::Ptr &frame); private: - void createGopCacheIfNeed(size_t gop_count); + void createGopCacheIfNeed(); std::shared_ptr makeRecorder(Recorder::type type); private: