#include “roll_polling.h”
#include <thread>
#include <vector>

namespace mediakit {

#define ErrorL std::cerr
#define InfoL std::cerr
#define WarnL std::cerr

int RollPollingWorker::Start(const std::string input, const std::string output) {
av_log_set_level(AV_LOG_INFO);
std::lock_guard<std::mutex> lock(mutex_);
source_ctx_ = OpenInputFile(input);
if (!source_ctx_) {
ErrorL << “open input file [” << input << “] failed”;
return -1;
}

if (OpenOutputFile(output) < 0) {
ErrorL << “open output file [” << output << “] failed”;
return -1;
}

filter_ctx_ = InitFilter(source_ctx_->codec_ctx_, ocodec_ctx_);
if (!filter_ctx_) {
ErrorL << “output file " << output << " init filter failed”;
return -1;
}

output_ = output;
state_ = kRunning;
std::cerr << “start roll polling task [” << output_ << “] successed \n”;
return 0;
}

int RollPollingWorker::Stop() {
state_ = kStoped;
return 0;
}

int RollPollingWorker::UpdateInput(const std::string input) {
std::string prefix = Prefix();
if (kRunning != state_) {
ErrorL << prefix << " is not running";
return -1;
}

if (input == source_ctx_->filename_) {
InfoL << prefix << " input file [" << input << " is same with preview input";
return 0;
}

std::shared_ptr<SourceContext> source_ctx = OpenInputFile(input);
if (!source_ctx) {
ErrorL << prefix << " open input file [" << input << “] failed”;
return -1;
}


std::shared_ptr<FilterContext> filter_ctx = InitFilter(source_ctx->codec_ctx_, ocodec_ctx_);
if (!filter_ctx) {
ErrorL << prefix << " init filter failed";
return -1;
}

std::lock_guard<std::mutex> lock(mutex_);
source_ctx_ = source_ctx;
filter_ctx_ = filter_ctx;
need_idr_ = true;
InfoL << “update input file " << input << " successed \n”;
return 0;
}

void RollPollingWorker::Runnable() {
std::string prefix = Prefix();
AVPacket pkt = av_packet_alloc();
while (kRunning == state_) {
std::lock_guard<std::mutex> lock(mutex_);
SourceContext source = source_ctx_.get();
FilterContext filter = filter_ctx_.get();

int ret = av_read_frame(source->fmt_ctx_, pkt);
if (ret < 0) {
if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF)
ret = 0;
break;
}

// only process video stream
if (pkt->stream_index != source->vid_stream_idx_)
continue;


if (filter_ctx_) {
ret = avcodec_send_packet(source->codec_ctx_, pkt);
if (ret < 0) {
if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF)
ret = 0;
break;
}

while (ret >= 0) {
ret = avcodec_receive_frame(source->codec_ctx_, source->decoded_frame_);
if (ret < 0) {
if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF)
ret = 0;
break;
}

FilterFrame(filter, source->decoded_frame_);
av_frame_unref(source->decoded_frame_);
}
} else {
//av_packet_rescale_ts(pkt,
// source->fmt_ctx_->streams[pkt->stream_index]->time_base,
// ofmt_ctx_->streams[0]->time_base); // only video stream

pkt->stream_index = 0;
pkt->pts = pkt->dts = (frame_cnt_++) * 40;
pkt->duration = 0;
pkt->pos = -1;
ret = av_interleaved_write_frame(ofmt_ctx_, pkt);
}

av_packet_unref(pkt);
//std::cerr << “write video packet size = " << pkt->size << " result = " << ret << " \n”;
if (ret < 0) {
ErrorL << prefix << " write packet to output file failed \n";
//break;
}
}

av_write_trailer(ofmt_ctx_);
av_packet_free(&pkt);

avcodec_free_context(&ocodec_ctx_);
if (ofmt_ctx_ && !(ofmt_ctx_->oformat->flags & AVFMT_NOFILE))
avio_closep(&ofmt_ctx_->pb);
avformat_free_context(ofmt_ctx_);
state_ = kStoped;
InfoL << “roll polling task finished”;
}

std::shared_ptr<RollPollingWorker::SourceContext> RollPollingWorker::OpenInputFile(const std::string filename) {
std::shared_ptr<SourceContext> source_ctx = std::make_shared<SourceContext>();
int ret = avformat_open_input(&source_ctx->fmt_ctx_, filename.c_str(), 0, nullptr);
if (ret < 0) {
ErrorL << “failed to open input file [” << filename << “]”;
return nullptr;
}

ret = avformat_find_stream_info(source_ctx->fmt_ctx_, 0);
if (ret < 0) {
ErrorL << “Fail to find input stream information in [” << filename << “]”;
return nullptr;
}

AVCodec dec = nullptr;
ret = av_find_best_stream(source_ctx->fmt_ctx_, AVMEDIA_TYPE_VIDEO, -1, -1, &dec, 0);
if (ret < 0) {
ErrorL << “Fail to av_find_best_stream in [” << filename << “]”;
return nullptr;
}

source_ctx->vid_stream_idx_ = ret;
source_ctx->codec_ctx_ = avcodec_alloc_context3(dec);
if (!source_ctx->codec_ctx_) {
ErrorL << “Fail to avcodec_alloc_context3 in [” << filename << “]”;
return nullptr;
}

int vid_stream_idx = source_ctx->vid_stream_idx_;
avcodec_parameters_to_context(source_ctx->codec_ctx_, source_ctx->fmt_ctx_->streams[vid_stream_idx]->codecpar);

// very important
source_ctx->codec_ctx_->framerate = av_guess_frame_rate(source_ctx->fmt_ctx_, source_ctx->fmt_ctx_->streams[vid_stream_idx], nullptr);
ret = avcodec_open2(source_ctx->codec_ctx_, dec, NULL);
if (ret < 0) {
ErrorL << “Fail to avcodec_open2 in [” << filename << “]”;
return nullptr;
}

source_ctx->decoded_frame_ = av_frame_alloc();
return source_ctx;
}

int RollPollingWorker::OpenOutputFile(const std::string filename) {
avformat_alloc_output_context2(&ofmt_ctx_, nullptr, “flv”, filename.c_str());
if (!ofmt_ctx_) {
ErrorL << “failed to open output file [” << filename << “]”;
return -1;
}

const AVCodec enc = avcodec_find_encoder(AV_CODEC_ID_H264);
if (!enc) {
ErrorL << “cannot find H.264 encoder”;
return -1;
}

ocodec_ctx_ = avcodec_alloc_context3(enc);
if (!ocodec_ctx_) {
ErrorL << “failed to allocate H.264 encoder context”;
return -1;
}

ocodec_ctx_->height = source_ctx_->codec_ctx_->height;
ocodec_ctx_->width = source_ctx_->codec_ctx_->width;
ocodec_ctx_->sample_aspect_ratio = source_ctx_->codec_ctx_->sample_aspect_ratio;
ocodec_ctx_->pix_fmt = source_ctx_->codec_ctx_->pix_fmt;
ocodec_ctx_->time_base = av_inv_q(source_ctx_->codec_ctx_->framerate);

AVDictionary enc_opts = nullptr;
av_dict_set(&enc_opts, “x264-params”, “crf=23:bframes=0”, 0);
if (avcodec_open2(ocodec_ctx_, enc, &enc_opts) < 0) {
ErrorL << “Cannot open video encoder”;
return -1;
}

AVStream* out_stream = avformat_new_stream(ofmt_ctx_, nullptr);
if (!out_stream) {
ErrorL << “Fail to avformat_new_stream, filename " << filename << " \n”;
return -1;
}

if (avcodec_parameters_from_context(out_stream->codecpar, ocodec_ctx_) < 0) {
ErrorL << “Failed to copy encoder parameters to output”;
return -1;
}

out_stream->time_base = ocodec_ctx_->time_base;
if (avio_open(&ofmt_ctx_->pb, filename.c_str(), AVIO_FLAG_WRITE) < 0) {
ErrorL << "Could not open output file " << filename << “\n”;
return -1;
}

AVDictionary* opts = nullptr;
av_dict_set(&opts, “flvflags”, “no_duration_filesize+no_metadata”, 0);
return avformat_write_header(ofmt_ctx_, &opts);
}

std::shared_ptr<RollPollingWorker::FilterContext> RollPollingWorker::InitFilter(
AVCodecContext* icodec_ctx, AVCodecContext* ocodec_ctx) {
std::shared_ptr<FilterContext> filter_ctx = std::make_shared<FilterContext>();
filter_ctx->filter_graph_ = avfilter_graph_alloc();
filter_ctx->filter_frame_ = av_frame_alloc();

char args[512];
snprintf(args, sizeof(args),
“video_size=%dx%d:pix_fmt=%d:time_base=%d/%d:pixel_aspect=%d/%d”,
icodec_ctx->width, icodec_ctx->height, icodec_ctx->pix_fmt,
icodec_ctx->time_base.num, icodec_ctx->time_base.den,
icodec_ctx->sample_aspect_ratio.num,
icodec_ctx->sample_aspect_ratio.den);
const AVFilter* buffersrc = avfilter_get_by_name(“buffer”);
int ret = avfilter_graph_create_filter(&filter_ctx->buffersrc_ctx_, buffersrc, “in”,
args, nullptr, filter_ctx->filter_graph_);
if (ret < 0) {
ErrorL << “Cannot create buffer source”;
return nullptr;
}

const AVFilter* buffersink = avfilter_get_by_name(“buffersink”);
ret = avfilter_graph_create_filter(&filter_ctx->buffersink_ctx_, buffersink, “out”,
nullptr, nullptr, filter_ctx->filter_graph_);
if (ret < 0) {
ErrorL << “Cannot create buffer sink”;
return nullptr;
}

ret = av_opt_set_bin(filter_ctx->buffersink_ctx_, “pix_fmts”,
(uint8_t*)&ocodec_ctx_->pix_fmt, sizeof(ocodec_ctx_->pix_fmt), AV_OPT_SEARCH_CHILDREN);
if (ret < 0) {
ErrorL << “Cannot set output pixel format”;
return nullptr;
}

AVFilterInOut* outputs = avfilter_inout_alloc();;
outputs->name = av_strdup(“in”);
outputs->filter_ctx = filter_ctx->buffersrc_ctx_;
outputs->pad_idx = 0;
outputs->next = nullptr;

AVFilterInOut inputs = avfilter_inout_alloc();
inputs->name = av_strdup(“out”);
inputs->filter_ctx = filter_ctx->buffersink_ctx_;
inputs->pad_idx = 0;
inputs->next = NULL;

std::string filter_spec = “[in]scale=” + std::to_string(kWidth) + “:” + std::to_string(kHeight) + “[out]”;
do {
ret = avfilter_graph_parse_ptr(filter_ctx->filter_graph_, filter_spec.c_str(), &inputs, &outputs, nullptr);
if (ret < 0) {
ErrorL << “avfilter graph parse failed”;
break;
}

ret = avfilter_graph_config(filter_ctx->filter_graph_, nullptr);
if (ret < 0) {
ErrorL << “avfilter graph config failed”;
break;
}
} while (0);

avfilter_inout_free(&inputs);
avfilter_inout_free(&outputs);
return ret < 0 ? nullptr : filter_ctx;
}


int RollPollingWorker::FilterFrame(FilterContext filter, AVFrame* frame) {
int ret = av_buffersrc_add_frame_flags(filter->buffersrc_ctx_, frame, 0);
if (ret < 0) {
ErrorL << “Error while feeding the filtergraph \n”;
return ret;
}

while (1) {
ret = av_buffersink_get_frame(filter->buffersink_ctx_, filter->filter_frame_);
if (ret < 0) {
if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF)
ret = 0;
break;
}

filter->filter_frame_->pict_type = AV_PICTURE_TYPE_NONE;
if (need_idr_) {
filter->filter_frame_->pict_type = AV_PICTURE_TYPE_I;
need_idr_ = false;
}
ret = EncodeFrame(filter->filter_frame_);
av_frame_unref(filter->filter_frame_);
if (ret < 0)
break;
}

return ret;
}

int RollPollingWorker::EncodeFrame(AVFrame* frame) {
if (avcodec_send_frame(ocodec_ctx_, frame) < 0)
return -1;

AVPacket* enc_pkt = av_packet_alloc();
int ret = 0;
while (ret >= 0) {
ret = avcodec_receive_packet(ocodec_ctx_, enc_pkt);
if (ret < 0) {
if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF)
ret = 0;
break;
}

enc_pkt->stream_index = 0;
enc_pkt->pts = enc_pkt->dts = (frame_cnt_++) * 40;
enc_pkt->duration = 0;
enc_pkt->pos = -1;

//std::cerr << “muxing frame: pts " << enc_pkt->pts << " duration " << enc_pkt->duration << “\n”;
ret = av_interleaved_write_frame(ofmt_ctx_, enc_pkt);
av_packet_unref(enc_pkt);
if (ret < 0) {
ErrorL << Prefix() << " write packet to output file failed \n”;
break;
}
}

av_packet_free(&enc_pkt);
return ret;
}


int RollPollingWorkerPool::StartWorker(const std::string input, const std::string output) {
std::shared_ptr<RollPollingWorker> worker = std::make_shared<RollPollingWorker>();
if (!worker->Start(input, output)) {
return -1;
}

std::weak_ptr<RollPollingWorker> weak_worker = worker;
auto task = [weak_worker](int64_t unused){
auto worker = weak_worker.lock();
if (worker)
worker->Runnable();
};

std::lock_guard<std::mutex> lock(mutex_);
worker_list_[output] = worker;
return 0;
}

int RollPollingWorkerPool::StopWorker(const std::string url) {
std::lock_guard<std::mutex> lock(mutex_);
auto it = worker_list_.find(url);
return it != worker_list_.end() ? it->second->Stop() : -1;
}

}


int PollingStart(int argc, char* argv[]) {
std::vector<std::string> inputs;
for (int i = 1; i < (argc - 1); i++)
inputs.push_back(argv[i]);
std::string output = argv[argc - 1];

mediakit::RollPollingWorker worker;
if (worker.Start(inputs[0], output) < 0)
return -1;


auto f = &{
worker.Runnable();
};

std::thread task(f);
task.detach();

size_t idx = 1;
while (worker.IsRunning()) {
std::this_thread::sleep_for(std::chrono::milliseconds(10000));
if (idx >= 4)
break;

if (inputs.size() > 1) {
std::string tmp = inputs[idx % inputs.size()];
std::cerr << “change input file to [” << tmp << “\n”;
worker.UpdateInput(tmp);
idx++;
}
}

worker.Stop();
//task.join();
return 0;
}
解释一下这段代码的含义