Add how_much_to_skip option to the Source element

lib/membrane_http_adaptive_stream/source.ex

@@ -104,6 +104,17 @@ defmodule Membrane.HTTPAdaptiveStream.Source do
 
                 Defaults to `:highest_resolution`.
                 """
+              ],
+              start_at: [
+                spec: Membrane.Time.t(),
+                default: 0,
+                description: """
+                Specifies the decoding timestamp of 
+                the first sample that should be read from each of the tracks.
+
+                If there is no sample with exactly such a timestamp, that sample
+                will be the first sample with DTS greater than provided timestamp.
+                """
               ]
 
   @impl true
@@ -134,7 +145,11 @@ defmodule Membrane.HTTPAdaptiveStream.Source do
   @impl true
   def handle_setup(_ctx, state) do
     {:ok, client_genserver} =
-      ClientGenServer.start_link(state.url, state.variant_selection_policy)
+      ClientGenServer.start_link(
+        state.url,
+        state.variant_selection_policy,
+        Membrane.Time.as_milliseconds(state.start_at, :round)
+      )
 
     {[], %{state | client_genserver: client_genserver}}
   end

lib/membrane_http_adaptive_stream/source/client_genserver.ex

@@ -8,13 +8,15 @@ defmodule Membrane.HTTPAdaptiveStream.Source.ClientGenServer do
 
   @spec start_link(
           String.t(),
-          Membrane.HTTPAdaptiveStream.Source.variant_selection_policy()
+          Membrane.HTTPAdaptiveStream.Source.variant_selection_policy(),
+          Membrane.Time.t()
         ) ::
           GenServer.on_start()
-  def start_link(url, variant_selection_policy) do
+  def start_link(url, variant_selection_policy, start_at_ms) do
     GenServer.start_link(__MODULE__,
       url: url,
-      variant_selection_policy: variant_selection_policy
+      variant_selection_policy: variant_selection_policy,
+      start_at_ms: start_at_ms
     )
   end
 
@@ -37,10 +39,11 @@ defmodule Membrane.HTTPAdaptiveStream.Source.ClientGenServer do
   end
 
   @impl true
-  def init(url: url, variant_selection_policy: variant_selection_policy) do
+  def init(url: url, variant_selection_policy: variant_selection_policy, start_at_ms: start_at_ms) do
     state = %{
       url: url,
       variant_selection_policy: variant_selection_policy,
+      start_at_ms: start_at_ms,
       client: nil
     }
 
@@ -50,7 +53,7 @@ defmodule Membrane.HTTPAdaptiveStream.Source.ClientGenServer do
   @impl true
   def handle_continue(:setup, state) do
     state =
-      %{state | client: Client.new(state.url)}
+      %{state | client: Client.new(state.url, state.start_at_ms)}
       |> choose_variant()
 
     {:noreply, state}

mix.exs

@@ -69,7 +69,7 @@ defmodule Membrane.HTTPAdaptiveStream.MixProject do
       {:membrane_aac_plugin, "~> 0.19.0"},
       {:membrane_h26x_plugin, "~> 0.10.0"},
       {:ex_hls,
-       github: "membraneframework-labs/ex_hls", ref: "88847ba9a547366fd3f4868d49b72bd1e68ae412"},
+       github: "membraneframework-labs/ex_hls", ref: "5b0227b898559678df25962424d1469f49f48810"},
       {:bunch, "~> 1.6"},
       {:qex, "~> 0.5"},
       {:membrane_hackney_plugin, "~> 0.11.0", only: :test},

mix.lock

@@ -12,7 +12,7 @@
   "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"},
   "erlex": {:hex, :erlex, "0.2.7", "810e8725f96ab74d17aac676e748627a07bc87eb950d2b83acd29dc047a30595", [:mix], [], "hexpm", "3ed95f79d1a844c3f6bf0cea61e0d5612a42ce56da9c03f01df538685365efb0"},
   "ex_doc": {:hex, :ex_doc, "0.38.2", "504d25eef296b4dec3b8e33e810bc8b5344d565998cd83914ffe1b8503737c02", [:mix], [{:earmark_parser, "~> 1.4.44", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "732f2d972e42c116a70802f9898c51b54916e542cc50968ac6980512ec90f42b"},
-  "ex_hls": {:git, "https://github.com/membraneframework-labs/ex_hls.git", "88847ba9a547366fd3f4868d49b72bd1e68ae412", [ref: "88847ba9a547366fd3f4868d49b72bd1e68ae412"]},
+  "ex_hls": {:git, "https://github.com/membraneframework-labs/ex_hls.git", "5b0227b898559678df25962424d1469f49f48810", [ref: "5b0227b898559678df25962424d1469f49f48810"]},
   "ex_m3u8": {:hex, :ex_m3u8, "0.15.3", "c10427f450b2ed7bfd85808d8dce21214f1fe9fa18927591cbbf96fea0a6a8aa", [:mix], [{:nimble_parsec, "~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}, {:typed_struct, "~> 0.3.0", [hex: :typed_struct, repo: "hexpm", optional: false]}], "hexpm", "99f20c0b44bab130dc6aca71fefe0d1a174413ae9ac2763220994b29bd310939"},
   "file_system": {:hex, :file_system, "1.1.0", "08d232062284546c6c34426997dd7ef6ec9f8bbd090eb91780283c9016840e8f", [:mix], [], "hexpm", "bfcf81244f416871f2a2e15c1b515287faa5db9c6bcf290222206d120b3d43f6"},
   "finch": {:hex, :finch, "0.20.0", "5330aefb6b010f424dcbbc4615d914e9e3deae40095e73ab0c1bb0968933cadf", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.6.2 or ~> 1.7", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "2658131a74d051aabfcba936093c903b8e89da9a1b63e430bee62045fa9b2ee2"},

test/membrane_http_adaptive_stream/integration_test/source_test.exs

@@ -70,6 +70,38 @@ defmodule Membrane.HTTPAdaptiveStream.Source.Test do
       assert_track(audio_result_file, @mpeg_ts_audio_ref_file, 40_000)
       assert_track(video_result_file, @mpeg_ts_video_ref_file, 70_000)
     end
+
+    @tag :tmp_dir
+    @tag :sometag
+    test "(MPEG-TS) with start_at option", %{tmp_dir: tmp_dir} do
+      audio_result_file = Path.join(tmp_dir, "audio.aac")
+      video_result_file = Path.join(tmp_dir, "video.h264")
+      start_at = Membrane.Time.seconds(10)
+
+      spec =
+        hls_to_file_pipeline_spec(
+          @mpegts_url,
+          %Membrane.Transcoder{
+            assumed_input_stream_format: %Membrane.AAC{
+              encapsulation: :ADTS
+            },
+            output_stream_format: Membrane.AAC
+          },
+          audio_result_file,
+          video_result_file,
+          start_at
+        )
+
+      pipeline = Testing.Pipeline.start_link_supervised!(spec: spec)
+      Process.sleep(10_000)
+      Testing.Pipeline.terminate(pipeline)
+
+      # reference files created locally with a quite good internet connection have
+      #  - 78_732 bytes for audio
+      #  - 136_754 bytes for video
+      assert_track(audio_result_file, @mpeg_ts_audio_ref_file, 40_000)
+      assert_track(video_result_file, @mpeg_ts_video_ref_file, 70_000)
+    end
   end
 
   describe "Membrane.HTTPAdaptiveStream.Source sends :new_tracks notification" do
@@ -157,11 +189,18 @@ defmodule Membrane.HTTPAdaptiveStream.Source.Test do
     Testing.Pipeline.terminate(pipeline)
   end
 
-  defp hls_to_file_pipeline_spec(url, audio_transcoder, audio_result_file, video_result_file) do
+  defp hls_to_file_pipeline_spec(
+         url,
+         audio_transcoder,
+         audio_result_file,
+         video_result_file,
+         start_at \\ 0
+       ) do
     [
       child(:hls_source, %Membrane.HTTPAdaptiveStream.Source{
         url: url,
-        variant_selection_policy: :lowest_resolution
+        variant_selection_policy: :lowest_resolution,
+        start_at: start_at
       })
       |> via_out(:video_output)
       |> child(%Membrane.Transcoder{