fftools/ffmpeg: add a live mux modelive

Track the wallclock time at which each input packet is demuxed and
propagate it through decoding and encoding.
When the live mux option is used, drop all packets demuxed before the
muxer is opened. This is intended to avoid latency when opening the
muxer takes a long time.

1 files changed, 37 insertions, 0 deletions

diff --git a/fftools/ffmpeg_mux.c b/fftools/ffmpeg_mux.c
index 08a76f0066..24f638d0e5 100644
--- a/fftools/ffmpeg_mux.c
+++ b/fftools/ffmpeg_mux.c
@@ -29,6 +29,7 @@
 #include "libavutil/intreadwrite.h"
 #include "libavutil/log.h"
 #include "libavutil/mem.h"
+#include "libavutil/time.h"
 #include "libavutil/timestamp.h"
 #include "libavutil/thread.h"
 
@@ -69,6 +70,10 @@ struct Muxer {
     atomic_int_least64_t last_filesize;
     int header_written;
 
+    int live_started;
+    int64_t header_write_wallclock;
+    int64_t live_ts_offset;
+
     AVPacket *sq_pkt;
 };
 
@@ -100,6 +105,37 @@ static int write_packet(OutputFile *of, OutputStream *ost, AVPacket *pkt)
     if (fs >= of->mux->limit_filesize)
         return AVERROR_EOF;
 
+    if (of->live) {
+        if (!of->mux->live_started) {
+            int64_t ts = av_rescale_q(pkt->dts, ost->mux_timebase, AV_TIME_BASE_Q);
+
+            of->mux->live_ts_offset = FFMAX(of->mux->live_ts_offset, ts);
+
+            // drop all packets that were demuxed before avformat_write_header()
+            if ((int64_t)pkt->opaque < of->mux->header_write_wallclock) {
+                fprintf(stderr, "live drop %s pts %g\n",
+                        av_get_media_type_string(st->codecpar->codec_type),
+                        pkt->pts * av_q2d(ost->mux_timebase));
+                av_packet_unref(pkt);
+                return 0;
+            }
+
+            of->mux->live_started = 1;
+            fprintf(stderr, "live started, ts offset %g\n", of->mux->live_ts_offset / (double)AV_TIME_BASE);
+        }
+
+        pkt->pts -= av_rescale_q(of->mux->live_ts_offset, AV_TIME_BASE_Q, ost->mux_timebase);
+        pkt->dts -= av_rescale_q(of->mux->live_ts_offset, AV_TIME_BASE_Q, ost->mux_timebase);
+
+        if (pkt->pts < 0) {
+            fprintf(stderr, "live drop %s pts %g\n",
+                    av_get_media_type_string(st->codecpar->codec_type),
+                    pkt->pts * av_q2d(ost->mux_timebase));
+            av_packet_unref(pkt);
+            return 0;
+        }
+    }
+
     if ((st->codecpar->codec_type == AVMEDIA_TYPE_VIDEO && ost->vsync_method == VSYNC_DROP) ||
         (st->codecpar->codec_type == AVMEDIA_TYPE_AUDIO && audio_sync_method < 0))
         pkt->pts = pkt->dts = AV_NOPTS_VALUE;
@@ -478,6 +514,7 @@ int of_check_init(OutputFile *of)
     }
     //assert_avoptions(of->opts);
     of->mux->header_written = 1;
+    of->mux->header_write_wallclock = av_gettime_relative();
 
     av_dump_format(fc, of->index, fc->url, 1);
     nb_output_dumped++;