update config

This commit is contained in:
sladro 2025-12-22 14:05:51 +08:00
parent 065c001a97
commit 32222df460
2 changed files with 173 additions and 103 deletions

View File

@ -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 }
]
}

View File

@ -10,6 +10,8 @@
#include <thread>
#include <vector>
#include <condition_variable>
#include <mutex>
#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<uint8_t>& 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<int>(extradata.size());
stream_->codecpar->extradata =
static_cast<uint8_t*>(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<std::mutex> 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<std::mutex> lock(mutex_);
ready_ = true;
warned_ = false;
std::cout << "[publish] Server ready: " << url_ << "\n";
}
// Wait until error occurs or stop requested
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, [this] { return !running_ || !ready_; });
}
// Cleanup context
{
std::lock_guard<std::mutex> 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<int>(extradata_.size());
stream->codecpar->extradata = static_cast<uint8_t*>(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<std::mutex> 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<AvMuxer*>(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<uint8_t> extradata_;
std::string proto_;
std::string url_;
std::atomic<bool> 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;
};