From 32222df46086ac7761026fbca6bf68c960de5000 Mon Sep 17 00:00:00 2001 From: sladro Date: Mon, 22 Dec 2025 14:05:51 +0800 Subject: [PATCH] update config --- configs/sample_cam1.json | 1 + plugins/publish/publish_node.cpp | 275 +++++++++++++++++++------------ 2 files changed, 173 insertions(+), 103 deletions(-) diff --git a/configs/sample_cam1.json b/configs/sample_cam1.json index 93aa91f..d277eec 100644 --- a/configs/sample_cam1.json +++ b/configs/sample_cam1.json @@ -29,6 +29,7 @@ "use_mpp": true, "use_ffmpeg_mux": true, "outputs": [ + { "proto": "rtsp", "port": 8554, "path": "/live/cam1" }, { "proto": "hls", "path": "/home/orangepi/Desktop/OrangePi3588Media/hls/cam1/index.m3u8", "segment_sec": 2 } ] } diff --git a/plugins/publish/publish_node.cpp b/plugins/publish/publish_node.cpp index 15316dd..d64bfac 100644 --- a/plugins/publish/publish_node.cpp +++ b/plugins/publish/publish_node.cpp @@ -10,6 +10,8 @@ #include #include +#include +#include #include "node.h" #if defined(RK3588_ENABLE_FFMPEG) @@ -49,83 +51,34 @@ struct EncodedPacket { #if defined(RK3588_ENABLE_FFMPEG) class AvMuxer { public: + AvMuxer() = default; + ~AvMuxer() { Close(); } + bool Init(const OutputConfig& cfg, AVCodecID codec_id, int width, int height, int fps, const std::vector& extradata) { - proto_ = cfg.proto.empty() ? "rtsp" : cfg.proto; + cfg_ = cfg; + codec_id_ = codec_id; + width_ = width; + height_ = height; fps_ = fps > 0 ? fps : 25; + extradata_ = extradata; // Copy extradata + + proto_ = cfg.proto.empty() ? "rtsp" : cfg.proto; url_ = BuildUrl(cfg); + + // Global init (safe to call multiple times) avformat_network_init(); - const char* fmt_name = proto_ == "hls" ? "hls" : "rtsp"; - if (avformat_alloc_output_context2(&fmt_, nullptr, fmt_name, url_.c_str()) < 0 || !fmt_) { - std::cerr << "[publish] avformat_alloc_output_context2 failed for " << url_ << "\n"; - return false; - } + running_ = true; + monitor_thread_ = std::thread(&AvMuxer::MonitorLoop, this); - stream_ = avformat_new_stream(fmt_, nullptr); - if (!stream_) { - std::cerr << "[publish] avformat_new_stream failed for " << url_ << "\n"; - Close(); - return false; - } - - stream_->time_base = AVRational{1, 1000}; - stream_->avg_frame_rate = AVRational{fps_, 1}; - stream_->codecpar->codec_type = AVMEDIA_TYPE_VIDEO; - stream_->codecpar->codec_id = codec_id; - stream_->codecpar->width = width; - stream_->codecpar->height = height; - stream_->codecpar->format = AV_PIX_FMT_YUV420P; - stream_->codecpar->codec_tag = 0; - - if (!extradata.empty()) { - stream_->codecpar->extradata_size = static_cast(extradata.size()); - stream_->codecpar->extradata = - static_cast(av_mallocz(extradata.size() + AV_INPUT_BUFFER_PADDING_SIZE)); - if (stream_->codecpar->extradata) { - std::memcpy(stream_->codecpar->extradata, extradata.data(), extradata.size()); - } - } - - AVDictionary* opts = nullptr; - if (proto_ == "rtsp") { - av_dict_set(&opts, "rtsp_flags", "listen", 0); - av_dict_set(&opts, "rtsp_transport", "tcp", 0); - } else if (proto_ == "hls") { - std::string seg = std::to_string(std::max(1, cfg.segment_sec)); - av_dict_set(&opts, "hls_time", seg.c_str(), 0); - av_dict_set(&opts, "hls_list_size", "0", 0); - av_dict_set(&opts, "hls_flags", "delete_segments+append_list", 0); - // Ensure HLS output directory exists (HLS muxer uses AVFMT_NOFILE) - std::filesystem::path p(url_); - if (!p.has_extension()) p /= "index.m3u8"; - std::error_code ec; - std::filesystem::create_directories(p.parent_path(), ec); - } - - // For non-NOFILE formats, open IO context manually. - if (!(fmt_->oformat->flags & AVFMT_NOFILE)) { - if (av_io_open(url_, opts) < 0) { - av_dict_free(&opts); - Close(); - return false; - } - } - - if (avformat_write_header(fmt_, &opts) < 0) { - std::cerr << "[publish] avformat_write_header failed for " << url_ << "\n"; - av_dict_free(&opts); - Close(); - return false; - } - av_dict_free(&opts); - ready_ = true; - std::cout << "[publish] mux start " << proto_ << " -> " << url_ << "\n"; + std::cout << "[publish] Muxer initialized async for " << url_ << "\n"; return true; } bool WriteFrame(const EncodedPacket& pkt) { - if (!ready_ || !stream_) return false; + std::lock_guard lock(mutex_); + if (!ready_ || !fmt_ || !stream_) return false; if (pkt.data.empty()) return false; AVPacket out; @@ -136,55 +89,161 @@ public: std::memcpy(out.data, pkt.data.data(), pkt.data.size()); out.stream_index = stream_->index; out.flags = pkt.key ? AV_PKT_FLAG_KEY : 0; + + // Simple PTS mapping int64_t pts = av_rescale_q(pkt.pts_ms, AVRational{1, 1000}, stream_->time_base); - // Ensure monotonically increasing PTS/DTS - if (pts <= last_pts_) { - pts = last_pts_ + 1; - } + if (pts <= last_pts_) pts = last_pts_ + 1; last_pts_ = pts; + out.pts = pts; out.dts = pts; out.duration = av_rescale_q(1000 / std::max(1, fps_), AVRational{1, 1000}, stream_->time_base); int ret = av_interleaved_write_frame(fmt_, &out); av_packet_unref(&out); - if (ret < 0 && !warned_) { - char errbuf[128]; - av_strerror(ret, errbuf, sizeof(errbuf)); - std::cerr << "[publish] av_interleaved_write_frame failed for " << url_ << " ret=" - << ret << " (" << errbuf << ")\n"; - warned_ = true; + + if (ret < 0) { + if (!warned_) { + char errbuf[128]; + av_strerror(ret, errbuf, sizeof(errbuf)); + std::cerr << "[publish] write failed for " << url_ << ": " << errbuf << ", resetting...\n"; + warned_ = true; + } + // Signal monitor thread to reset + ready_ = false; + cv_.notify_all(); } return ret >= 0; } void Close() { - if (fmt_) { - if (ready_) av_write_trailer(fmt_); - if (!(fmt_->oformat->flags & AVFMT_NOFILE) && fmt_->pb) { - avio_closep(&fmt_->pb); - } - avformat_free_context(fmt_); - } - fmt_ = nullptr; - stream_ = nullptr; - ready_ = false; - warned_ = false; + running_ = false; + // Break the av_io_open wait if possible (via CheckInterrupt) + cv_.notify_all(); + if (monitor_thread_.joinable()) monitor_thread_.join(); } private: - int av_io_open(const std::string& url, AVDictionary* opts) { - if (proto_ == "hls") { - std::filesystem::path p(url); - if (!p.has_extension()) p /= "index.m3u8"; - if (!p.is_absolute()) p = std::filesystem::current_path() / p; - std::error_code ec; - std::filesystem::create_directories(p.parent_path(), ec); - resolved_path_ = p.string(); - return avio_open2(&fmt_->pb, resolved_path_.c_str(), AVIO_FLAG_WRITE, nullptr, &opts); + void MonitorLoop() { + while (running_) { + if (TryOpen()) { + { + std::lock_guard lock(mutex_); + ready_ = true; + warned_ = false; + std::cout << "[publish] Server ready: " << url_ << "\n"; + } + + // Wait until error occurs or stop requested + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { return !running_ || !ready_; }); + } + + // Cleanup context + { + std::lock_guard lock(mutex_); + if (fmt_) { + // Try to write trailer if logical end (not socket error) + // But usually we are here because of error, so av_write_trailer might hang/fail. + // We skip trailer on error reset to avoid blocking. + if (fmt_->pb) { + avio_closep(&fmt_->pb); + } + avformat_free_context(fmt_); + fmt_ = nullptr; + stream_ = nullptr; + } + ready_ = false; + } + + if (running_) { + // Delay before retry + std::this_thread::sleep_for(std::chrono::seconds(1)); + } } - resolved_path_ = url; - return avio_open2(&fmt_->pb, url.c_str(), AVIO_FLAG_WRITE, nullptr, &opts); + } + + bool TryOpen() { + AVFormatContext* fmt = nullptr; + const char* fmt_name = proto_ == "hls" ? "hls" : "rtsp"; + if (avformat_alloc_output_context2(&fmt, nullptr, fmt_name, url_.c_str()) < 0 || !fmt) { + std::cerr << "[publish] alloc context failed " << url_ << "\n"; + return false; + } + + // Set interrupt callback to allow breaking blocking calls + fmt->interrupt_callback.callback = CheckInterrupt; + fmt->interrupt_callback.opaque = this; + + AVStream* stream = avformat_new_stream(fmt, nullptr); + if (!stream) { + avformat_free_context(fmt); + return false; + } + + stream->time_base = AVRational{1, 1000}; + stream->avg_frame_rate = AVRational{fps_, 1}; + stream->codecpar->codec_type = AVMEDIA_TYPE_VIDEO; + stream->codecpar->codec_id = codec_id_; + stream->codecpar->width = width_; + stream->codecpar->height = height_; + stream->codecpar->format = AV_PIX_FMT_YUV420P; + + if (!extradata_.empty()) { + stream->codecpar->extradata_size = static_cast(extradata_.size()); + stream->codecpar->extradata = static_cast(av_mallocz(extradata_.size() + AV_INPUT_BUFFER_PADDING_SIZE)); + std::memcpy(stream->codecpar->extradata, extradata_.data(), extradata_.size()); + } + + AVDictionary* opts = nullptr; + if (proto_ == "rtsp") { + av_dict_set(&opts, "rtsp_flags", "listen", 0); + av_dict_set(&opts, "rtsp_transport", "tcp", 0); + } else if (proto_ == "hls") { + std::string seg = std::to_string(std::max(1, cfg_.segment_sec)); + av_dict_set(&opts, "hls_time", seg.c_str(), 0); + av_dict_set(&opts, "hls_list_size", "0", 0); + av_dict_set(&opts, "hls_flags", "delete_segments+append_list", 0); + std::filesystem::create_directories(std::filesystem::path(url_).parent_path()); + } + + // For listener, av_io_open blocks until connection + if (!(fmt->oformat->flags & AVFMT_NOFILE)) { + // std::cout << "[publish] Waiting for connection on " << url_ << "...\n"; + if (av_io_open_helper(fmt, url_, opts) < 0) { + av_dict_free(&opts); + avformat_free_context(fmt); + return false; + } + } + + if (avformat_write_header(fmt, &opts) < 0) { + av_dict_free(&opts); + if (fmt->pb) avio_closep(&fmt->pb); + avformat_free_context(fmt); + return false; + } + av_dict_free(&opts); + + { + std::lock_guard lock(mutex_); + fmt_ = fmt; + stream_ = stream; + } + return true; + } + + int av_io_open_helper(AVFormatContext* fmt, const std::string& url, AVDictionary* opts) { + if (proto_ == "hls") { + // HLS open is slightly different handled in logic but avio_open2 can take callbacks + return avio_open2(&fmt->pb, url.c_str(), AVIO_FLAG_WRITE, &fmt->interrupt_callback, &opts); + } + return avio_open2(&fmt->pb, url.c_str(), AVIO_FLAG_WRITE, &fmt->interrupt_callback, &opts); + } + + static int CheckInterrupt(void* opaque) { + auto* self = static_cast(opaque); + return self->running_ ? 0 : 1; } std::string BuildUrl(const OutputConfig& cfg) const { @@ -194,14 +253,24 @@ private: return "rtsp://0.0.0.0:" + std::to_string(cfg.port) + path; } - AVFormatContext* fmt_ = nullptr; - AVStream* stream_ = nullptr; - std::string url_; - std::string resolved_path_; - std::string proto_; + OutputConfig cfg_; + AVCodecID codec_id_; + int width_ = 0; + int height_ = 0; int fps_ = 25; + std::vector extradata_; + std::string proto_; + std::string url_; + + std::atomic running_{false}; + std::thread monitor_thread_; + std::mutex mutex_; + std::condition_variable cv_; + bool ready_ = false; bool warned_ = false; + AVFormatContext* fmt_ = nullptr; + AVStream* stream_ = nullptr; int64_t last_pts_ = -1; };