Handle TDEN frame in timed ID3v2 stream

lib/ex_hls/client.ex

@@ -24,7 +24,8 @@ defmodule ExHLS.Client do
     :live_reader,
     :live_forwarder,
     :how_much_to_skip_ms,
-    :segment_format
+    :segment_format,
+    :ultra_low_latency?
   ]
 
   defstruct @enforce_keys
@@ -57,6 +58,12 @@ defmodule ExHLS.Client do
   the client will treat HLS segments based on the extension in their name,
   falling back `MPEG-TS` if the cannot recognize the extension.
 
+  Passing `ultra_low_latency?: true` option turns on ultra low latency mode of the client.
+  In this mode the client starts playing the playlist as fast as possible, and skips to the most
+  recent segment
+  Please note that this is not compliant with the HLS specification and might cause playback stalls.
+  The ultra low latency mode is turned off by default.
+
   Note that there is no guarantee that exactly the specified amount of time will be skipped.
   The actual skipped duration may be slightly shorter, depending on the HLS segments durations.
   To get the actual skipped duration, you can use `get_skipped_segments_cumulative_duration_ms/1`
@@ -71,10 +78,16 @@ defmodule ExHLS.Client do
     %{
       parent_process: parent_process,
       how_much_to_skip_ms: how_much_to_skip_ms,
-      segment_format: segment_format
+      segment_format: segment_format,
+      ultra_low_latency?: ultra_low_latency?
     } =
       opts
-      |> Keyword.validate!(parent_process: self(), how_much_to_skip_ms: 0, segment_format: nil)
+      |> Keyword.validate!(
+        parent_process: self(),
+        how_much_to_skip_ms: 0,
+        segment_format: nil,
+        ultra_low_latency?: false
+      )
       |> Map.new()
 
     root_playlist_raw_content = Utils.download_or_read_file!(url)
@@ -100,7 +113,8 @@ defmodule ExHLS.Client do
       live_reader: nil,
       live_forwarder: nil,
       how_much_to_skip_ms: how_much_to_skip_ms,
-      segment_format: segment_format
+      segment_format: segment_format,
+      ultra_low_latency?: ultra_low_latency?
     }
     |> maybe_resolve_media_playlist()
   end
@@ -155,7 +169,8 @@ defmodule ExHLS.Client do
         ExHLS.Client.Live.Reader.start_link(
           client.media_playlist_url,
           forwarder,
-          client.segment_format
+          client.segment_format,
+          client.ultra_low_latency?
         )
 
       %{client | live_reader: reader, live_forwarder: forwarder, hls_mode: :live}

lib/ex_hls/client/live/reader.ex

@@ -9,20 +9,23 @@ defmodule ExHLS.Client.Live.Reader do
   alias ExHLS.Client.Live.Forwarder
   alias ExM3U8.Tags.{MediaInit, Segment}
 
-  @spec start_link(String.t(), Forwarder.t(), :ts | :cmaf | nil) :: {:ok, pid()} | {:error, any()}
-  def start_link(media_playlist_url, forwarder, segment_format) do
+  @spec start_link(String.t(), Forwarder.t(), :ts | :cmaf | nil, boolean()) ::
+          {:ok, pid()} | {:error, any()}
+  def start_link(media_playlist_url, forwarder, segment_format, ultra_low_latency?) do
     GenServer.start_link(__MODULE__, %{
       media_playlist_url: media_playlist_url,
       forwarder: forwarder,
-      segment_format: segment_format
+      segment_format: segment_format,
+      ultra_low_latency?: ultra_low_latency?
     })
   end
 
   @impl true
   def init(%{
         media_playlist_url: media_playlist_url,
         forwarder: forwarder,
-        segment_format: segment_format
+        segment_format: segment_format,
+        ultra_low_latency?: ultra_low_latency?
       }) do
     state = %{
       forwarder: forwarder,
@@ -39,7 +42,8 @@ defmodule ExHLS.Client.Live.Reader do
       playlist_check_scheduled?: false,
       timestamp_offset: nil,
       playing_started?: false,
-      segment_format: segment_format
+      segment_format: segment_format,
+      ultra_low_latency?: ultra_low_latency?
     }
 
     {:ok, state, {:continue, :setup}}
@@ -105,6 +109,10 @@ defmodule ExHLS.Client.Live.Reader do
     end
   end
 
+  defp should_start_playing?(%{ultra_low_latency?: true}) do
+    true
+  end
+
   defp should_start_playing?(state) do
     %ExM3U8.MediaPlaylist.Info{
       media_sequence: media_sequence,
@@ -165,7 +173,20 @@ defmodule ExHLS.Client.Live.Reader do
     {media_inits, %{state | media_init_downloaded?: true}}
   end
 
-  defp next_segment_to_download_seq_num(%{max_downloaded_seq_num: nil} = state) do
+  # in the ultra low latency mode it skips to the most recent segment
+  defp next_segment_to_download_seq_num(
+         %{max_downloaded_seq_num: nil, ultra_low_latency?: true} = state
+       ) do
+    how_many_segments =
+      state.media_playlist.timeline
+      |> Enum.count(&match?(%Segment{}, &1))
+
+    state.media_playlist.info.media_sequence + how_many_segments - 1
+  end
+
+  defp next_segment_to_download_seq_num(
+         %{max_downloaded_seq_num: nil, ultra_low_latency?: false} = state
+       ) do
     {segments_with_end_times, duration_sum} =
       state.media_playlist.timeline
       |> Enum.flat_map_reduce(0, fn

lib/ex_hls/demuxing_engine/mpeg_ts.ex

@@ -3,12 +3,13 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
   @behaviour ExHLS.DemuxingEngine
 
   use Bunch.Access
+  use Bunch
 
   require Logger
   alias Membrane.{AAC, H264, RemoteStream}
   alias MPEG.TS.Demuxer
 
-  @enforce_keys [:demuxer]
+  @enforce_keys [:demuxer, :last_tden_tag]
   defstruct @enforce_keys ++ [track_timestamps_data: %{}]
 
   # using it a boundary expressed in nanoseconds, instead of the usual 90kHz clock ticks,
@@ -17,7 +18,8 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
   @timestamp_range_size_ns div(2 ** 33 * 1_000_000_000, 90_000)
 
   @type t :: %__MODULE__{
-          demuxer: Demuxer.t()
+          demuxer: Demuxer.t(),
+          last_tden_tag: String.t() | nil
         }
 
   @impl true
@@ -32,7 +34,7 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
     # TODO - figure out how to do it properly
     demuxer = %{demuxer | waiting_random_access_indicator: false}
 
-    %__MODULE__{demuxer: demuxer}
+    %__MODULE__{demuxer: demuxer, last_tden_tag: nil}
   end
 
   @impl true
@@ -79,8 +81,11 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
   @impl true
   def pop_chunk(%__MODULE__{} = demuxing_engine, track_id) do
     with {[packet], demuxer} <- Demuxer.take(demuxing_engine.demuxer, track_id) do
+      {maybe_tden_tag, demuxer} = maybe_read_tden_tag(demuxer, packet.pts)
+      tden_tag = maybe_tden_tag || demuxing_engine.last_tden_tag
+
       {demuxing_engine, packet} =
-        %{demuxing_engine | demuxer: demuxer}
+        %{demuxing_engine | demuxer: demuxer, last_tden_tag: tden_tag}
         |> handle_possible_timestamps_rollover(track_id, packet)
 
       chunk = %ExHLS.Chunk{
@@ -90,7 +95,8 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
         track_id: track_id,
         metadata: %{
           discontinuity: packet.discontinuity,
-          is_aligned: packet.is_aligned
+          is_aligned: packet.is_aligned,
+          tden_tag: tden_tag
         }
       }
 
@@ -101,6 +107,34 @@ defmodule ExHLS.DemuxingEngine.MPEGTS do
     end
   end
 
+  defp maybe_read_tden_tag(demuxer, packet_pts) do
+    withl no_id3_stream:
+            {id3_track_id, _stream_description} <-
+              demuxer.pmt.streams
+              |> Enum.find(fn {_pid, stream_description} ->
+                stream_description.stream_type == :METADATA_IN_PES
+              end),
+          no_id3_data: {[id3], demuxer} <- Demuxer.take(demuxer, id3_track_id),
+          id3_not_in_timerange: true <- id3.pts <= packet_pts do
+      {parse_tden_tag(id3.data), demuxer}
+    else
+      no_id3_stream: nil -> {nil, demuxer}
+      no_id3_data: {[], updated_demuxer} -> {nil, updated_demuxer}
+      id3_not_in_timerange: false -> {nil, demuxer}
+    end
+  end
+
+  defp parse_tden_tag(payload) do
+    with {pos, _len} <- :binary.match(payload, "TDEN"),
+         <<_skip::binary-size(pos), "TDEN", tden::binary>> <- payload,
+         <<size::integer-size(4)-unit(8), _flags::16, _3, text::binary-size(size - 2), 0,
+           _rest::binary>> <- tden do
+      text
+    else
+      _error -> nil
+    end
+  end
+
   # value returned by Demuxer is represented in nanoseconds
   defp packet_ts_to_millis(ts), do: div(ts, 1_000_000)
 

mix.exs

@@ -40,9 +40,12 @@ defmodule ExHLS.Mixfile do
       {:ex_m3u8, "~> 0.15.4"},
       {:req, "~> 0.5.10"},
       {:qex, "~> 0.5.1"},
+      {:bunch, "~> 1.6"},
       {:membrane_mp4_plugin, "~> 0.36.0"},
       {:membrane_h26x_plugin, "~> 0.10.2"},
-      {:mpeg_ts, "~> 2.0.0"},
+      {:mpeg_ts,
+       github: "membraneframework-labs/kim_mpeg_ts",
+       branch: "varsill/fix_pes_optional_header_resolving"},
       {:ex_doc, ">= 0.0.0", only: :dev, runtime: false},
       {:dialyxir, ">= 0.0.0", only: :dev, runtime: false},
       {:credo, ">= 0.0.0", only: :dev, runtime: false},

mix.lock

@@ -32,8 +32,8 @@
   "membrane_timestamp_queue": {:hex, :membrane_timestamp_queue, "0.2.2", "1c831b2273d018a6548654aa9f7fa7c4b683f71d96ffe164934ef55f9d11f693", [:mix], [{:heap, "~> 2.0", [hex: :heap, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "7c830e760baaced0988421671cd2c83c7cda8d1bd2b61fd05332711675d1204f"},
   "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"},
   "mint": {:hex, :mint, "1.7.1", "113fdb2b2f3b59e47c7955971854641c61f378549d73e829e1768de90fc1abf1", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "fceba0a4d0f24301ddee3024ae116df1c3f4bb7a563a731f45fdfeb9d39a231b"},
+"mpeg_ts": {:git, "https://github.com/membraneframework-labs/kim_mpeg_ts.git", "c8c770e0e7714c72b3faa7f20088fdbd76f5bade", [branch: "varsill/fix_pes_optional_header_resolving"]},
   "mock": {:hex, :mock, "0.3.9", "10e44ad1f5962480c5c9b9fa779c6c63de9bd31997c8e04a853ec990a9d841af", [:mix], [{:meck, "~> 0.9.2", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm", "9e1b244c4ca2551bb17bb8415eed89e40ee1308e0fbaed0a4fdfe3ec8a4adbd3"},
-  "mpeg_ts": {:hex, :mpeg_ts, "2.0.2", "87f7d3b38c962fc367edbb4b1419f5f314be41a0d512b95437f11c4f60c931f4", [:mix], [], "hexpm", "5b7f1245a945de647c29abc9453e3d9d7eca1b0001d3d582f4feb11fc09b2792"},
   "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"},
   "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},

test/client_test.exs

@@ -4,11 +4,13 @@ defmodule ExHLS.Client.Test do
   alias ExHLS.Client
   alias Membrane.{AAC, H264, RemoteStream}
 
-  @fixtures "https://raw.githubusercontent.com/membraneframework-labs/ex_hls/refs/heads/master/test/fixtures/"
+  @fixtures "https://raw.githubusercontent.com/membraneframework/ex_hls/refs/heads/master/test/fixtures/"
   @fmp4_url @fixtures <> "fmp4/output.m3u8"
   @fmp4_only_video_url @fixtures <> "fmp4_only_video/output.m3u8"
   @mpegts_only_video_url @fixtures <> "mpeg_ts_only_video/output_playlist.m3u8"
+  @mpegts_with_tden_url "test/fixtures/mpeg_ts_with_tden/output_playlist.m3u8"
   @mpegts_url "https://test-streams.mux.dev/x36xhzz/x36xhzz.m3u8"
+  @mpegts_live_url "./test/fixtures/mpeg_ts_live/output_playlist.m3u8"
 
   describe "if client reads video and audio chunks of the HLS" do
     test "(MPEGTS) stream" do
@@ -128,7 +130,36 @@ defmodule ExHLS.Client.Test do
              183, 150, 44, 216, 32, 217, 35, 238, 239, 120, 50, 54, 52, 32, 45, 32, 99, 111, 114,
              101, 32, 49, 54, 52, 32, 114>> <> _rest = video_chunk.payload
 
-    assert video_chunk.metadata == %{discontinuity: false, is_aligned: false}
+    assert video_chunk.metadata == %{discontinuity: false, is_aligned: false, tden_tag: nil}
+  end
+
+  @tag :sometag
+  test "(MPEGTS) stream with ID3v2.4 TDEN tag" do
+    client = Client.new(@mpegts_with_tden_url)
+
+    assert Client.get_variants(client) == %{}
+
+    chunks = Client.generate_stream(client) |> Enum.take(381)
+
+    first_audio_chunk_after_tden =
+      Enum.find(
+        chunks,
+        &(&1.metadata.tden_tag != nil and &1.media_type == :audio)
+      )
+
+    first_video_chunk_after_tden =
+      Enum.find(
+        chunks,
+        &(&1.metadata.tden_tag != nil and &1.media_type == :video)
+      )
+
+    assert first_audio_chunk_after_tden.pts_ms == 3328
+    assert first_audio_chunk_after_tden.dts_ms == 3328
+    assert first_audio_chunk_after_tden.metadata.tden_tag == "2025-10-21T08:07:50"
+
+    assert first_video_chunk_after_tden.pts_ms == 3233
+    assert first_video_chunk_after_tden.dts_ms == 3233
+    assert first_video_chunk_after_tden.metadata.tden_tag == "2025-10-21T08:07:50"
   end
 
   test "(fMP4) stream with only video" do
@@ -203,6 +234,28 @@ defmodule ExHLS.Client.Test do
              215, 198, 77, 184, 229, 170, 157, 115, 169, 223>> <> _rest = audio_chunk.payload
   end
 
+  test "(MPEGTS) stream with ultra low latency mode" do
+    client = Client.new(@mpegts_live_url, ultra_low_latency?: true)
+
+    assert Client.get_variants(client) == %{}
+    assert {:ok, tracks_info, client} = Client.get_tracks_info(client)
+
+    assert [%Membrane.RemoteStream{content_format: Membrane.H264, type: :bytestream}] =
+             tracks_info |> Map.values()
+
+    chunks = Client.generate_stream(client) |> Enum.take(1)
+    [video_chunk | _rest_video_chunks] = chunks
+
+    assert %{pts_ms: 11_081, dts_ms: 11_001} = video_chunk
+    assert byte_size(video_chunk.payload) == 28_699
+
+    assert <<0, 0, 0, 1, 9, 240, 0, 0, 0, 1, 103, 100, 0, 21, 172, 217, 65, 224, 143, 235, 1, 106,
+             12, 2, 13, 110, 0, 0, 9, 154, 0, 1, 224, 0, 30, 44, 91, 44, 0, 0, 0, 1, 104, 234,
+             225, 178, 200, 176, 0, 0>> <> _rest = video_chunk.payload
+
+    assert video_chunk.metadata == %{discontinuity: false, is_aligned: false, tden_tag: nil}
+  end
+
   defp assert_chunks_are_in_proper_order(chunks) do
     iteration_state = %{
       last_dts: %{audio: nil, video: nil},

test/demuxing_engine_mpegts_test.exs

@@ -26,7 +26,8 @@ defmodule ExHLS.DemuxingEngine.MPEGTS.Test do
 
     demuxer = %{
       waiting_random_access_indicator: nil,
-      packet_buffers: %{1 => packets, 2 => packets}
+      packet_buffers: %{1 => packets, 2 => packets},
+      pmt: %{streams: %{}}
     }
 
     new = fn -> demuxer end

test/fixtures/mpeg_ts_live/output_playlist.m3u8

@@ -0,0 +1,10 @@
+#EXTM3U
+#EXT-X-VERSION:3
+#EXT-X-TARGETDURATION:6
+#EXT-X-MEDIA-SEQUENCE:0
+#EXTINF:5.760933,
+video_segment_000.ts
+#EXTINF:3.840633,
+video_segment_001.ts
+#EXTINF:0.400111,
+video_segment_002.ts

test/fixtures/mpeg_ts_with_tden/output_playlist.m3u8

@@ -0,0 +1,18 @@
+#EXTM3U
+#EXT-X-VERSION:3
+#EXT-X-TARGETDURATION:2
+#EXT-X-MEDIA-SEQUENCE:0
+#EXT-X-PLAYLIST-TYPE:VOD
+#EXTINF:2.000000,
+output_playlist0.ts
+#EXTINF:2.000000,
+output_playlist1.ts
+#EXTINF:2.000000,
+output_playlist2.ts
+#EXTINF:2.000000,
+output_playlist3.ts
+#EXTINF:2.000000,
+output_playlist4.ts
+#EXTINF:0.033333,
+output_playlist5.ts
+#EXT-X-ENDLIST