From 2b1183e1454486025a634229911d2791cc73554d Mon Sep 17 00:00:00 2001 From: Romain Beauxis Date: Sat, 14 Sep 2024 20:25:11 -0500 Subject: [PATCH 1/4] Expand HLS segment name metadata --- CHANGES.md | 1 + doc/content/liq/output.file.hls.liq | 61 +++++------ doc/content/migrating.md | 13 +++ src/core/outputs/hls_output.ml | 156 +++++++++++++++++----------- src/libs/hls.liq | 8 +- 5 files changed, 139 insertions(+), 100 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 7bf23f1af2..5b345bc6a4 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -38,6 +38,7 @@ New: one stream is encoded. - Allow trailing commas in record definition (#3300). - Add `metadata.getter.source.float` (#3356). +- BREAKING: Added `duration` and `ticks` to metadata available when computing HLS segment names (#4135) - Added optional `main_playlist_writer` to `output.file.hls` and derivated operator (#3484) - Added `is_nan`, `is_infinite`, `ceil`, `floor`, `sign` and `round` (#3407) diff --git a/doc/content/liq/output.file.hls.liq b/doc/content/liq/output.file.hls.liq index e09e5d4cf0..368e33f712 100644 --- a/doc/content/liq/output.file.hls.liq +++ b/doc/content/liq/output.file.hls.liq @@ -1,44 +1,37 @@ s = mksafe(playlist("playlist")) -aac_lofi = %ffmpeg(format="mpegts", - %audio( - codec="aac", - channels=2, - ar=44100 - )) +aac_lofi = + %ffmpeg(format = "mpegts", %audio(codec = "aac", channels = 2, ar = 44100)) -aac_midfi = %ffmpeg(format="mpegts", - %audio( - codec="aac", - channels=2, - ar=44100, - b="96k" - )) +aac_midfi = + %ffmpeg( + format = "mpegts", + %audio(codec = "aac", channels = 2, ar = 44100, b = "96k") + ) -aac_hifi = %ffmpeg(format="mpegts", - %audio( - codec="aac", - channels=2, - ar=44100, - b="192k" - )) +aac_hifi = + %ffmpeg( + format = "mpegts", + %audio(codec = "aac", channels = 2, ar = 44100, b = "192k") + ) -streams = [("aac_lofi",aac_lofi), - ("aac_midfi", aac_midfi), - ("aac_hifi", aac_hifi)] +streams = + [("aac_lofi", aac_lofi), ("aac_midfi", aac_midfi), ("aac_hifi", aac_hifi)] -def segment_name(~position,~extname,stream_name) = +def segment_name(metadata) = timestamp = int_of_float(time()) - duration = 2 + let {stream_name, duration, timestamp, position, extname} = metadata "#{stream_name}_#{duration}_#{timestamp}_#{position}.#{extname}" end -output.file.hls(playlist="live.m3u8", - segment_duration=2.0, - segments=5, - segments_overhead=5, - segment_name=segment_name, - persist_at="/tmp/path/to/state.config", - "/tmp/path/to/hls/directory", - streams, - s) +output.file.hls( + playlist="live.m3u8", + segment_duration=2.0, + segments=5, + segments_overhead=5, + segment_name=segment_name, + persist_at="/tmp/path/to/state.config", + "/tmp/path/to/hls/directory", + streams, + s +) diff --git a/doc/content/migrating.md b/doc/content/migrating.md index e884f591ec..5ec412fbb0 100644 --- a/doc/content/migrating.md +++ b/doc/content/migrating.md @@ -115,6 +115,19 @@ Known incompatibilities include: - `(?Ppattern)` for named captures is not supported. `(?pattern)` should be used instead. +### `segment_name` in HLS outputs + +To make segment name more flexible, `duration` (segment duration in seconds) and `ticks` (segment exact duration in liquidsoap's main ticks) have been added +to the data available when calling `segment_name`. + +To prevent any further breakage of this function, its arguments have been changed to a single record containing all the available attributes: + +```liquidsoap +def segment_name(metadata) = + "#{metadata.stream_name}_#{metadata.position}.#{metadata.extname}" +end +``` + ### `on_air` metadata Request `on_air` and `on_air_timestamp` metadata are deprecated. These values were never reliable. They are set at the request level when `request.dynamic` diff --git a/src/core/outputs/hls_output.ml b/src/core/outputs/hls_output.ml index 2679e206f8..024d68b1ba 100644 --- a/src/core/outputs/hls_output.ml +++ b/src/core/outputs/hls_output.ml @@ -29,7 +29,7 @@ let log = Log.make ["hls"; "output"] let default_name = Lang.eval ~cache:false ~typecheck:false ~stdlib:`Disabled - {|fun (~position, ~extname, base) -> "#{base}_#{position}.#{extname}"|} + {|fun (metadata) -> "#{metadata.stream_name}_#{metadata.position}.#{metadata.extname}"|} let hls_proto frame_t = let main_playlist_writer_t = @@ -58,9 +58,16 @@ let hls_proto frame_t = let segment_name_t = Lang.fun_t [ - (false, "position", Lang.int_t); - (false, "extname", Lang.string_t); - (false, "", Lang.string_t); + ( false, + "", + Lang.record_t + [ + ("position", Lang.int_t); + ("extname", Lang.string_t); + ("duration", Lang.float_t); + ("ticks", Lang.int_t); + ("stream_name", Lang.string_t); + ] ); ] Lang.string_t in @@ -117,8 +124,9 @@ let hls_proto frame_t = segment_name_t, Some default_name, Some - "Segment name. Default: `fun (~position,~extname,stream_name) -> \ - \"#{stream_name}_#{position}.#{extname}\"`" ); + "Segment name. Default: `fun (metadata) -> \ + \"#{metadata.stream_name}_#{metadata.position}.#{metadata.extname}\"`" + ); ( "segments_overhead", Lang.nullable_t Lang.int_t, Some (Lang.int 5), @@ -180,6 +188,7 @@ type atomic_out_channel = ; output_substring : string -> int -> int -> unit ; position : int ; truncate : int -> unit + ; saved_filename : string option ; read : int -> int -> string ; close : unit > @@ -187,10 +196,11 @@ type segment = { id : int; discontinuous : bool; current_discontinuity : int; - filename : string; segment_extra_tags : string list; mutable init_filename : string option; + mutable filename : string option; mutable out_channel : atomic_out_channel option; + (* Segment length in main ticks. *) mutable len : int; mutable last_segmentable_position : (int * int) option; } @@ -222,8 +232,8 @@ let json_of_segment id; discontinuous; current_discontinuity; - filename; init_filename; + filename; segment_extra_tags; len; last_segmentable_position; @@ -233,9 +243,9 @@ let json_of_segment ("id", `Int id); ("discontinuous", `Bool discontinuous); ("current_discontinuity", `Int current_discontinuity); - ("filename", `String filename); ] @ json_optional "init_filename" (fun s -> `String s) init_filename + @ json_optional "filename" (fun s -> `String s) filename @ [ ("extra_tags", `Tuple (List.map (fun s -> `String s) segment_extra_tags)); ("len", `Int len); @@ -249,7 +259,6 @@ let segment_of_json = function let id = parse_json_int "id" l in let discontinuous = parse_json_bool "discontinuous" l in let current_discontinuity = parse_json_int "current_discontinuity" l in - let filename = parse_json_string "filename" l in let segment_extra_tags = parse_json "extra_tags" (function @@ -266,6 +275,11 @@ let segment_of_json = function (function `String s -> s | _ -> raise Invalid_state) l in + let filename = + parse_json_optional "filename" + (function `String s -> s | _ -> raise Invalid_state) + l + in let last_segmentable_position = parse_json_optional "last_segmentable_position" (function @@ -277,10 +291,10 @@ let segment_of_json = function id; discontinuous; current_discontinuity; - filename; init_filename; len; segment_extra_tags; + filename; out_channel = None; last_segmentable_position; } @@ -394,16 +408,20 @@ class hls_output p = (Lang.to_option (List.assoc "main_playlist_writer" p)) in let directory_val = Lang.assoc "" 1 p in - let directory = Lang_string.home_unrelate (Lang.to_string directory_val) in + let hls_directory = + Lang_string.home_unrelate (Lang.to_string directory_val) + in let perms = Lang.to_int (List.assoc "perm" p) in let dir_perm = Lang.to_int (List.assoc "dir_perm" p) in let temp_dir = Lang.to_valued_option Lang.to_string (List.assoc "temp_dir" p) in let () = - if (not (Sys.file_exists directory)) || not (Sys.is_directory directory) + if + (not (Sys.file_exists hls_directory)) + || not (Sys.is_directory hls_directory) then ( - try Utils.mkdir ~perm:dir_perm directory + try Utils.mkdir ~perm:dir_perm hls_directory with _ -> raise (Error.Invalid_value @@ -415,7 +433,7 @@ class hls_output p = let filename = Lang.to_string filename in let filename = if Filename.is_relative filename then - Filename.concat directory filename + Filename.concat hls_directory filename else filename in let dir = Filename.dirname filename in @@ -440,15 +458,20 @@ class hls_output p = let segment_main_duration = segment_ticks * Lazy.force Frame.size in let segment_duration = Frame.seconds_of_main segment_main_duration in let segment_name = Lang.to_fun (List.assoc "segment_name" p) in - let segment_name ~position ~extname sname = - directory - ^^ Lang.to_string - (segment_name - [ - ("position", Lang.int position); - ("extname", Lang.string extname); - ("", Lang.string sname); - ]) + let segment_name ~position ~extname ~duration ~ticks sname = + Lang.to_string + (segment_name + [ + ( "", + Lang.record + [ + ("position", Lang.int position); + ("extname", Lang.string extname); + ("duration", Lang.float duration); + ("ticks", Lang.int ticks); + ("stream_name", Lang.string sname); + ] ); + ]) in let streams = let streams = Lang.assoc "" 2 p in @@ -598,7 +621,6 @@ class hls_output p = in let source = Lang.assoc "" 3 p in let main_playlist_filename = Lang.to_string (List.assoc "playlist" p) in - let main_playlist_filename = directory ^^ main_playlist_filename in let main_playlist_extra_tags = List.map (fun s -> String.trim (Lang.to_string s)) @@ -636,10 +658,7 @@ class hls_output p = | `Streaming, _ -> state <- `Streaming method private open_out filename = - let state = if Sys.file_exists filename then `Updated else `Created in - let temp_dir = - Option.value ~default:(Filename.dirname filename) temp_dir - in + let temp_dir = Option.value ~default:hls_directory temp_dir in let tmp_file = Filename.temp_file ~temp_dir "liq" "tmp" in Unix.chmod tmp_file perms; let fd = @@ -668,12 +687,20 @@ class hls_output p = in Bytes.sub_string b 0 (f 0) + val mutable saved_filename = None + method saved_filename = saved_filename + method close = (try Unix.close fd with _ -> ()); Fun.protect ~finally:(fun () -> try Sys.remove tmp_file with _ -> ()) (fun () -> - (try Unix.rename tmp_file filename + let fname = Filename.concat hls_directory (filename ()) in + saved_filename <- Some fname; + let state = + if Sys.file_exists fname then `Updated else `Created + in + (try Unix.rename tmp_file fname with Unix.Unix_error (Unix.EXDEV, _, _) -> self#log#important "Rename failed! Directory for temporary files appears to be \ @@ -682,9 +709,9 @@ class hls_output p = operations!"; Utils.copy ~mode:[Open_creat; Open_trunc; Open_binary] - ~perms tmp_file filename; + ~perms tmp_file fname; Sys.remove tmp_file); - on_file_change ~state filename) + on_file_change ~state fname) end method private unlink filename = @@ -695,11 +722,15 @@ class hls_output p = self#log#important "Could not remove file %s: %s" filename (Unix.error_message e) + method private unlink_segment = + function { filename = Some filename } -> self#unlink filename | _ -> () + method private close_segment = function | { current_segment = Some ({ out_channel = Some oc } as segment) } as s -> oc#close; + segment.filename <- oc#saved_filename; segment.out_channel <- None; let segments = List.assoc s.name segments in push_segment segment segments; @@ -709,7 +740,7 @@ class hls_output p = | Some max_segments -> List.length !segments >= max_segments then ( let segment = remove_segment segments in - self#unlink segment.filename; + self#unlink_segment segment; match segment.init_filename with | None -> () | Some filename -> @@ -728,20 +759,6 @@ class hls_output p = method private open_segment s = self#log#debug "Opening segment %d for stream %s." s.position s.name; - let filename = - segment_name ~position:s.position ~extname:s.extname s.name - in - let directory = Filename.dirname filename in - let () = - if (not (Sys.file_exists directory)) || not (Sys.is_directory directory) - then ( - try Utils.mkdir ~perm:dir_perm directory - with exn -> - let bt = Printexc.get_raw_backtrace () in - Lang.raise_as_runtime ~bt ~kind:"file" exn) - in - let out_channel = self#open_out filename in - Strings.iter out_channel#output_substring (s.encoder.Encoder.header ()); let discontinuous, current_discontinuity = if state = `Restarted then (true, s.discontinuity_count + 1) else (false, s.discontinuity_count) @@ -754,14 +771,23 @@ class hls_output p = discontinuous; current_discontinuity; len = 0; - filename; segment_extra_tags; init_filename = (match s.init_state with `Has_init f -> Some f | _ -> None); - out_channel = Some out_channel; + filename = None; + out_channel = None; last_segmentable_position = None; } in + let filename () = + let ticks = segment.len in + let duration = Frame.seconds_of_main ticks in + segment_name ~position:s.position ~extname:s.extname ~duration ~ticks + s.name + in + let out_channel = self#open_out filename in + Strings.iter out_channel#output_substring (s.encoder.Encoder.header ()); + segment.out_channel <- Some out_channel; s.current_segment <- Some segment; s.discontinuity_count <- current_discontinuity; s.position <- s.position + 1; @@ -799,7 +825,7 @@ class hls_output p = method private cleanup_streams = List.iter - (fun (_, s) -> List.iter (fun s -> self#unlink s.filename) !s) + (fun (_, s) -> List.iter (fun s -> self#unlink_segment s) !s) segments; List.iter (fun s -> @@ -812,29 +838,35 @@ class hls_output p = (fun filename -> if Sys.file_exists filename then self#unlink filename) segment.init_filename); - self#unlink segment.filename) + self#unlink_segment segment) s.current_segment); s.current_segment <- None) streams - method private playlist_name s = directory ^^ s.name ^ ".m3u8" + method private playlist_name s = s.name ^ ".m3u8" method private write_playlist s = + let segments = + List.filter_map + (function + | { filename = Some fname } as s -> Some (fname, s) | _ -> None) + (List.rev !(List.assoc s.name segments)) + in let segments = List.fold_left (fun cur el -> if List.length cur < segments_per_playlist then el :: cur else cur) - [] - (List.rev !(List.assoc s.name segments)) + [] segments in let discontinuity_sequence, media_sequence = match segments with - | { current_discontinuity; id } :: _ -> (current_discontinuity, id - 1) + | (_, { current_discontinuity; id }) :: _ -> + (current_discontinuity, id - 1) | [] -> (0, 0) in let filename = self#playlist_name s in self#log#debug "Writing playlist %s.." s.name; - let oc = self#open_out filename in + let oc = self#open_out (fun () -> filename) in Fun.protect ~finally:(fun () -> oc#close) (fun () -> @@ -855,7 +887,7 @@ class hls_output p = oc#output_string "\r\n") s.stream_extra_tags; List.iteri - (fun pos segment -> + (fun pos (filename, segment) -> if 0 < pos && segment.discontinuous then oc#output_string "#EXT-X-DISCONTINUITY\r\n"; if pos = 0 || segment.discontinuous then ( @@ -878,8 +910,7 @@ class hls_output p = oc#output_string "\r\n") segment.segment_extra_tags; oc#output_string - (Printf.sprintf "%s%s\r\n" prefix - (Filename.basename segment.filename))) + (Printf.sprintf "%s%s\r\n" prefix (Filename.basename filename))) segments) val mutable main_playlist_written = false @@ -903,7 +934,7 @@ class hls_output p = | Some playlist -> self#log#debug "Writing main playlist %s.." main_playlist_filename; - let oc = self#open_out main_playlist_filename in + let oc = self#open_out (fun () -> main_playlist_filename) in oc#output_string playlist; oc#close)); main_playlist_written <- true @@ -1033,9 +1064,10 @@ class hls_output p = | None -> s.init_state <- `No_init | Some data when not (Strings.is_empty data) -> let init_filename = - segment_name ~position:init_position ~extname name + segment_name ~position:init_position ~extname ~duration:0. + ~ticks:0 name in - let oc = self#open_out init_filename in + let oc = self#open_out (fun () -> init_filename) in Fun.protect ~finally:(fun () -> oc#close) (fun () -> Strings.iter oc#output_substring data); diff --git a/src/libs/hls.liq b/src/libs/hls.liq index 4d666f80e9..0df4eda4b0 100644 --- a/src/libs/hls.liq +++ b/src/libs/hls.liq @@ -254,8 +254,8 @@ end def replaces output.harbor.hls( %argsof(output.file.hls[!segment_name]), ~segment_name=( - fun (~position, ~extname, stream_name) -> - "#{stream_name}_#{position}.#{extname}" + fun (metadata) -> + "#{metadata.stream_name}_#{metadata.position}.#{metadata.extname}" ), ~headers=[("Access-Control-Allow-Origin", "*")], ~port=8000, @@ -298,8 +298,8 @@ end def output.harbor.hls.https( %argsof(output.file.hls[!segment_name]), ~segment_name=( - fun (~position, ~extname, stream_name) -> - "#{stream_name}_#{position}.#{extname}" + fun (metadata) -> + "#{metadata.stream_name}_#{metadata.position}.#{metadata.extname}" ), ~headers=[("Access-Control-Allow-Origin", "*")], ~port=8000, From 955ea87b409f20817a9f98ec5b67d7b9fd8135e7 Mon Sep 17 00:00:00 2001 From: Romain Beauxis Date: Sun, 15 Sep 2024 10:04:17 -0500 Subject: [PATCH 2/4] Adapt tests. --- tests/media/ffmpeg_distributed_hls.liq | 3 ++- tests/media/ffmpeg_drop_tracks.liq | 3 ++- tests/media/ffmpeg_raw_hls.liq | 3 ++- tests/streams/hls_id3v2.liq | 5 +++-- 4 files changed, 9 insertions(+), 5 deletions(-) diff --git a/tests/media/ffmpeg_distributed_hls.liq b/tests/media/ffmpeg_distributed_hls.liq index 0beff54e84..1c7bc98442 100644 --- a/tests/media/ffmpeg_distributed_hls.liq +++ b/tests/media/ffmpeg_distributed_hls.liq @@ -103,7 +103,8 @@ def check_stream() = end end -def segment_name(~position, ~extname, stream_name) = +def segment_name(metadata) = + let {position, extname, stream_name} = metadata if position > 10 then check_stream() end timestamp = int_of_float(time()) "#{stream_name}_#{timestamp}_#{position}.#{extname}" diff --git a/tests/media/ffmpeg_drop_tracks.liq b/tests/media/ffmpeg_drop_tracks.liq index 0738ead3d7..8f6f687bb5 100644 --- a/tests/media/ffmpeg_drop_tracks.liq +++ b/tests/media/ffmpeg_drop_tracks.liq @@ -100,7 +100,8 @@ def check_stream() = end end -def segment_name(~position, ~extname, stream_name) = +def segment_name(metadata) = + let {position, extname, stream_name} = metadata if position > 2 then check_stream() end timestamp = int_of_float(time()) "#{stream_name}_#{timestamp}_#{position}.#{extname}" diff --git a/tests/media/ffmpeg_raw_hls.liq b/tests/media/ffmpeg_raw_hls.liq index 3bf8e7543b..b999fbaf0b 100644 --- a/tests/media/ffmpeg_raw_hls.liq +++ b/tests/media/ffmpeg_raw_hls.liq @@ -98,7 +98,8 @@ def check_stream() = end end -def segment_name(~position, ~extname, stream_name) = +def segment_name(metadata) = + let {position, extname, stream_name} = metadata if position > 2 then check_stream() end timestamp = int_of_float(time()) "#{stream_name}_#{timestamp}_#{position}.#{extname}" diff --git a/tests/streams/hls_id3v2.liq b/tests/streams/hls_id3v2.liq index 0bd7f91a36..a653f14b14 100644 --- a/tests/streams/hls_id3v2.liq +++ b/tests/streams/hls_id3v2.liq @@ -180,8 +180,9 @@ s = mksafe(s) check_running = ref(false) segments_created = ref(0) -def segment_name(~position, ~extname, stream) = - "segment-#{stream}_#{position}.#{extname}" +def segment_name(metadata) = + let {position, extname, stream_name} = metadata + "segment-#{stream_name}_#{position}.#{extname}" end def on_file_change(~state, fname) = From df1c53438ff030538315333f447e521e36a6f2fa Mon Sep 17 00:00:00 2001 From: Romain Beauxis Date: Sun, 15 Sep 2024 10:21:20 -0500 Subject: [PATCH 3/4] Fix --- doc/content/liq/output.file.hls.liq | 2 +- src/core/outputs/hls_output.ml | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/doc/content/liq/output.file.hls.liq b/doc/content/liq/output.file.hls.liq index 368e33f712..59f239a282 100644 --- a/doc/content/liq/output.file.hls.liq +++ b/doc/content/liq/output.file.hls.liq @@ -20,7 +20,7 @@ streams = def segment_name(metadata) = timestamp = int_of_float(time()) - let {stream_name, duration, timestamp, position, extname} = metadata + let {stream_name, duration, position, extname} = metadata "#{stream_name}_#{duration}_#{timestamp}_#{position}.#{extname}" end diff --git a/src/core/outputs/hls_output.ml b/src/core/outputs/hls_output.ml index 024d68b1ba..f3569baf1f 100644 --- a/src/core/outputs/hls_output.ml +++ b/src/core/outputs/hls_output.ml @@ -779,11 +779,11 @@ class hls_output p = last_segmentable_position = None; } in + let { position; extname } = s in let filename () = let ticks = segment.len in let duration = Frame.seconds_of_main ticks in - segment_name ~position:s.position ~extname:s.extname ~duration ~ticks - s.name + segment_name ~position ~extname ~duration ~ticks s.name in let out_channel = self#open_out filename in Strings.iter out_channel#output_substring (s.encoder.Encoder.header ()); From ae06ca87d2880a45679f1b4fde7eb289b3cf33f7 Mon Sep 17 00:00:00 2001 From: Romain Beauxis Date: Sun, 15 Sep 2024 11:36:46 -0500 Subject: [PATCH 4/4] Fix. --- tests/media/ffmpeg_raw_hls.liq | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/media/ffmpeg_raw_hls.liq b/tests/media/ffmpeg_raw_hls.liq index b999fbaf0b..f97a90c1d8 100644 --- a/tests/media/ffmpeg_raw_hls.liq +++ b/tests/media/ffmpeg_raw_hls.liq @@ -100,7 +100,7 @@ end def segment_name(metadata) = let {position, extname, stream_name} = metadata - if position > 2 then check_stream() end + if position > 1 then check_stream() end timestamp = int_of_float(time()) "#{stream_name}_#{timestamp}_#{position}.#{extname}" end