Make encoder header more robust by using a callback. Fixes: #3265 (#3276

)

CHANGES.md

@@ -5,6 +5,8 @@ Fixed:
 - Fixed HTTP response status in `output.harbor` (#3255)
 - Make sure main HLS playlist is regenerated after being
   unlinked (#3275)
+- Fix `output.harbor` encoder header when encoding with
+  `%ogg`, `%vorbis` and etc. (#3276)
 
 ---
 

src/core/encoder/encoder.ml

@@ -314,10 +314,7 @@ let dummy_hls encode =
 
 type encoder = {
   insert_metadata : Export_metadata.metadata -> unit;
-  (* Encoder are all called from the main
-     thread so there's no need to protect this
-     value with a mutex so far.. *)
-  mutable header : Strings.t;
+  header : unit -> Strings.t;
   hls : hls;
   encode : Frame.t -> int -> int -> Strings.t;
   stop : unit -> Strings.t;
@@ -348,6 +345,7 @@ let get_factory fmt =
       (* Protect all functions with a mutex. *)
       let m = Mutex.create () in
       let insert_metadata = Tutils.mutexify m insert_metadata in
+      let header = Tutils.mutexify m header in
       let {
         init;
         init_encode;

src/core/encoder/encoder.mli

@@ -110,10 +110,7 @@ val dummy_hls : (Generator.t -> int -> int -> Strings.t) -> hls
 
 type encoder = {
   insert_metadata : Export_metadata.metadata -> unit;
-  (* Encoder are all called from the main
-   * thread so there's no need to protect this
-   * value with a mutex so far.. *)
-  mutable header : Strings.t;
+  header : unit -> Strings.t;
   hls : hls;
   encode : Frame.t -> int -> int -> Strings.t;
   stop : unit -> Strings.t;

src/core/encoder/encoders/avi_encoder.ml

@@ -116,7 +116,7 @@ let encoder ~pos:_ avi =
     Encoder.insert_metadata = (fun _ -> ());
     hls = Encoder.dummy_hls encode;
     encode;
-    header = Strings.of_string header;
+    header = (fun () -> Strings.of_string header);
     stop = (fun () -> Strings.empty);
   }
 

src/core/encoder/encoders/external_encoder.ml

@@ -164,10 +164,7 @@ let encoder ~pos:_ id ext =
   in
   {
     Encoder.insert_metadata;
-    (* External encoders do not support
-     * headers for now. They will probably
-     * never do.. *)
-    header = Strings.empty;
+    header = (fun () -> Strings.empty);
     hls = Encoder.dummy_hls encode;
     encode;
     stop;

src/core/encoder/encoders/fdkaac_encoder.ml

@@ -110,7 +110,7 @@ let encoder ~pos aac =
   in
   {
     Encoder.insert_metadata = (fun _ -> ());
-    header = Strings.empty;
+    header = (fun () -> Strings.empty);
     hls = Encoder.dummy_hls encode;
     encode;
     stop;

src/core/encoder/encoders/ffmpeg_encoder_common.ml

@@ -335,4 +335,10 @@ let encoder ~pos ~mk_streams ffmpeg meta =
       video_size;
     }
   in
-  { Encoder.insert_metadata; header = Strings.empty; hls; encode; stop }
+  {
+    Encoder.insert_metadata;
+    header = (fun () -> Strings.empty);
+    hls;
+    encode;
+    stop;
+  }

src/core/encoder/encoders/flac_encoder.ml

@@ -65,7 +65,7 @@ let encoder ~pos:_ flac meta =
     Encoder.insert_metadata = (fun _ -> ());
     (* Flac encoder do not support header
      * for now. It will probably never do.. *)
-    header = Strings.empty;
+    header = (fun () -> Strings.empty);
     hls = Encoder.dummy_hls encode;
     encode;
     stop;

src/core/encoder/encoders/gstreamer_encoder.ml

@@ -191,7 +191,7 @@ let encoder ext =
   in
   {
     Encoder.insert_metadata;
-    header = Strings.empty;
+    header = (fun () -> Strings.empty);
     hls = Encoder.dummy_hls encode;
     encode;
     stop;

src/core/encoder/encoders/lame_encoder.ml

@@ -119,7 +119,7 @@ let () =
       insert_metadata = (fun _ -> ());
       hls = Encoder.dummy_hls encode;
       encode;
-      header = Strings.empty;
+      header = (fun () -> Strings.empty);
       stop;
     }
   in

src/core/encoder/encoders/ogg_encoder.ml

@@ -105,7 +105,7 @@ let encoder ~pos { Ogg_format.audio; video } =
         Encoder.insert_metadata;
         hls = Encoder.dummy_hls (fun _ _ -> assert false);
         encode;
-        header = Strings.empty;
+        header = (fun () -> Ogg_muxer.get_header ogg_enc);
         stop;
       }
     and streams_start () =
@@ -115,14 +115,11 @@ let encoder ~pos { Ogg_format.audio; video } =
           | None -> track.id <- Some (track.reset ogg_enc meta)
       in
       List.iter f tracks;
-      Ogg_muxer.streams_start ogg_enc;
-      enc.Encoder.header <- Ogg_muxer.get_header ogg_enc
+      Ogg_muxer.streams_start ogg_enc
     and encode frame start len =
       (* We do a lazy start, to
        * avoid empty streams at beginning.. *)
-      if Ogg_muxer.state ogg_enc <> Ogg_muxer.Streaming then (
-        streams_start ();
-        enc.Encoder.header <- Ogg_muxer.get_header ogg_enc);
+      if Ogg_muxer.state ogg_enc <> Ogg_muxer.Streaming then streams_start ();
       let f track =
         track.encode ogg_enc (Option.get track.id) frame start len
       in
@@ -131,18 +128,15 @@ let encoder ~pos { Ogg_format.audio; video } =
     and ogg_stop () =
       let f track = track.id <- None in
       List.iter f tracks;
-      if Ogg_muxer.state ogg_enc = Ogg_muxer.Streaming then (
-        Ogg_muxer.end_of_stream ogg_enc;
-        enc.Encoder.header <- Strings.empty)
+      if Ogg_muxer.state ogg_enc = Ogg_muxer.Streaming then
+        Ogg_muxer.end_of_stream ogg_enc
     and stop () =
       ogg_stop ();
-      enc.Encoder.header <- Strings.empty;
       Ogg_muxer.get_data ogg_enc
     and insert_metadata m =
       ogg_stop ();
       let f track = track.id <- Some (track.reset ogg_enc m) in
-      List.iter f tracks;
-      enc.Encoder.header <- Ogg_muxer.get_header ogg_enc
+      List.iter f tracks
     in
     { enc with hls = Encoder.dummy_hls encode }
 

src/core/encoder/encoders/shine_encoder.ml

@@ -66,7 +66,7 @@ let encoder ~pos:_ shine =
   let stop () = Strings.of_string (Shine.flush enc) in
   {
     Encoder.insert_metadata = (fun _ -> ());
-    header = Strings.empty;
+    header = (fun () -> Strings.empty);
     hls = Encoder.dummy_hls encode;
     encode;
     stop;

src/core/encoder/encoders/wav_encoder.ml

@@ -73,7 +73,7 @@ let encoder ~pos wav =
     Encoder.insert_metadata = (fun _ -> ());
     hls = Encoder.dummy_hls encode;
     encode;
-    header = Strings.of_string header;
+    header = (fun () -> Strings.of_string header);
     stop = (fun () -> Strings.empty);
   }
 

src/core/outputs/harbor_output.ml

@@ -456,7 +456,7 @@ class output p =
           data.format icyheader extra_headers
       in
       let buffer =
-        Strings.Mutable.of_strings (Option.get encoder).Encoder.header
+        Strings.Mutable.of_strings ((Option.get encoder).Encoder.header ())
       in
       let close () = try Harbor.close s with _ -> () in
       let rec client =

src/core/outputs/hls_output.ml

@@ -591,7 +591,7 @@ class hls_output p =
             Lang.raise_as_runtime ~bt ~kind:"file" exn)
       in
       let out_channel = self#open_out filename in
-      Strings.iter (output_substring out_channel) s.encoder.Encoder.header;
+      Strings.iter (output_substring out_channel) (s.encoder.Encoder.header ());
       let discontinuous = state = `Restarted in
       let segment =
         {

tests/core/stream_decoder_test.ml

@@ -34,7 +34,7 @@ let () =
   let encoder =
     create_encoder ~pos:None "test stream" Export_metadata.empty_metadata
   in
-  write encoder.Encoder.header;
+  write (encoder.Encoder.header ());
   try
     while true do
       try

tests/regression/GH3276.liq

@@ -0,0 +1,24 @@
+log.level := 4
+
+port = random.int(min=8000, max=10000)
+
+thread.run(delay=7., fun () -> begin
+  s = input.http("http://localhost:#{port}/test")
+
+  track_count = ref(0)
+  s = source.on_track(s, fun (_) -> begin
+    if track_count() > 1 then
+      test.pass()
+    end
+    track_count := track_count() + 1
+  end)
+
+  output.dummy(fallible=true,s)
+end)
+
+output.harbor(
+  %vorbis,
+  chop(every=5., sine()),
+  port=port,
+  mount="/test"
+)