Handle single media streams

Author: FelonEkonom

lib/membrane_http_adaptive_stream/hls_source.ex

@@ -25,12 +25,16 @@ defmodule Membrane.HLS.Source do
   def_output_pad :video_output,
     accepted_format: any_of(H264, %RemoteStream{content_format: H264}),
     flow_control: :manual,
-    demand_unit: :buffers
+    demand_unit: :buffers,
+    availability: :on_request,
+    max_instances: 1
 
   def_output_pad :audio_output,
     accepted_format: any_of(AAC, %RemoteStream{content_format: AAC}),
     flow_control: :manual,
-    demand_unit: :buffers
+    demand_unit: :buffers,
+    availability: :on_request,
+    max_instances: 1
 
   # The boundary on how many chunks of one stream will be requested
   # from Membrane.HLS.Source.ClientGenServer at once.
@@ -88,6 +92,7 @@ defmodule Membrane.HLS.Source do
   @impl true
   def handle_init(_ctx, opts) do
     initial_pad_state = %{
+      ref: nil,
       requested: 0,
       qex: Qex.new(),
       qex_size: 0,
@@ -106,6 +111,12 @@ defmodule Membrane.HLS.Source do
     {[], state}
   end
 
+  @impl true
+  def handle_pad_added(Pad.ref(pad_name, _id) = pad_ref, _ctx, state) do
+    state = state |> put_in([pad_name, :ref], pad_ref)
+    {[], state}
+  end
+
   @impl true
   def handle_setup(_ctx, state) do
     {:ok, client_genserver} =
@@ -119,20 +130,53 @@ defmodule Membrane.HLS.Source do
 
   @impl true
   def handle_playing(_ctx, state) do
-    {[audio_stream_format], [video_stream_format]} =
+    stream_formats =
       ClientGenServer.get_tracks_info(state.client_genserver)
       |> Map.values()
-      |> Enum.split_with(&audio_stream_format?/1)
 
-    actions = [
-      stream_format: {:audio_output, audio_stream_format},
-      stream_format: {:video_output, video_stream_format}
-    ]
+    :ok = ensure_pads_match_stream_formats!(stream_formats, state)
+
+    actions =
+      stream_formats
+      |> Enum.map(fn stream_format ->
+        pad_ref =
+          if audio_stream_format?(stream_format),
+            do: state.audio_output.ref,
+            else: state.video_output.ref
+
+        {:stream_format, {pad_ref, stream_format}}
+      end)
 
     state = request_media_chunks(state)
     {actions, state}
   end
 
+  defp ensure_pads_match_stream_formats!(stream_formats, state) do
+    audio_format_occurs? =
+      Enum.any?(stream_formats, &audio_stream_format?/1)
+
+    video_format_occurs? =
+      Enum.any?(stream_formats, &(not audio_stream_format?(&1)))
+
+    if audio_format_occurs? and state.audio_output.ref == nil do
+      raise "Audio pad should be linked"
+    end
+
+    if video_format_occurs? and state.video_output.ref == nil do
+      raise "Video pad should be linked"
+    end
+
+    if not audio_format_occurs? and state.audio_output.ref != nil do
+      raise "Audio pad should not be linked"
+    end
+
+    if not video_format_occurs? and state.video_output.ref != nil do
+      raise "Video pad should not be linked"
+    end
+
+    :ok
+  end
+
   defp audio_stream_format?(stream_format) do
     case stream_format do
       %RemoteStream{content_format: AAC} -> true
@@ -151,7 +195,7 @@ defmodule Membrane.HLS.Source do
 
   @impl true
   def handle_info({data_type, %ExHLS.Chunk{} = chunk}, _ctx, state) do
-    pad_ref = data_type_to_pad_ref(data_type)
+    pad_name = data_type_to_pad_name(data_type)
 
     buffer = %Buffer{
       payload: chunk.payload,
@@ -162,55 +206,55 @@ defmodule Membrane.HLS.Source do
 
     state =
       state
-      |> update_in([pad_ref, :qex], &Qex.push(&1, buffer))
-      |> update_in([pad_ref, :qex_size], &(&1 + 1))
-      |> update_in([pad_ref, :requested], &(&1 - 1))
-      |> update_in([pad_ref, :oldest_buffer_dts], fn
+      |> update_in([pad_name, :qex], &Qex.push(&1, buffer))
+      |> update_in([pad_name, :qex_size], &(&1 + 1))
+      |> update_in([pad_name, :requested], &(&1 - 1))
+      |> update_in([pad_name, :oldest_buffer_dts], fn
         nil -> buffer.dts
         oldest_dts -> oldest_dts
       end)
       |> request_media_chunks()
 
-    {[redemand: pad_ref], state}
+    {[redemand: state[pad_name].ref], state}
   end
 
   @impl true
   def handle_info({data_type, :end_of_stream}, _ctx, state) do
-    pad_ref = data_type_to_pad_ref(data_type)
+    pad_name = data_type_to_pad_name(data_type)
 
     state =
-      if state[pad_ref].eos_received? do
+      if state[pad_name].eos_received? do
         state
       else
         state
-        |> put_in([pad_ref, :eos_received?], true)
-        |> update_in([pad_ref, :qex], &Qex.push(&1, :end_of_stream))
-        |> update_in([pad_ref, :qex_size], &(&1 + 1))
+        |> put_in([pad_name, :eos_received?], true)
+        |> update_in([pad_name, :qex], &Qex.push(&1, :end_of_stream))
+        |> update_in([pad_name, :qex_size], &(&1 + 1))
       end
 
-    state = state |> update_in([pad_ref, :requested], &(&1 - 1))
+    state = state |> update_in([pad_name, :requested], &(&1 - 1))
 
-    {[redemand: pad_ref], state}
+    {[redemand: state[pad_name].ref], state}
   end
 
-  defp data_type_to_pad_ref(:audio_chunk), do: :audio_output
-  defp data_type_to_pad_ref(:video_chunk), do: :video_output
+  defp data_type_to_pad_name(:audio_chunk), do: :audio_output
+  defp data_type_to_pad_name(:video_chunk), do: :video_output
 
-  defp pop_buffers(pad_ref, demand, state) do
-    how_many_pop = min(state[pad_ref].qex_size, demand)
+  defp pop_buffers(Pad.ref(pad_name, _id) = pad_ref, demand, state) do
+    how_many_pop = min(state[pad_name].qex_size, demand)
 
     1..how_many_pop//1
     |> Enum.flat_map_reduce(state, fn _i, state ->
-      {buffer_or_eos, qex} = state[pad_ref].qex |> Qex.pop!()
+      {buffer_or_eos, qex} = state[pad_name].qex |> Qex.pop!()
 
       state =
         state
-        |> put_in([pad_ref, :qex], qex)
-        |> update_in([pad_ref, :qex_size], &(&1 - 1))
+        |> put_in([pad_name, :qex], qex)
+        |> update_in([pad_name, :qex_size], &(&1 - 1))
 
       case buffer_or_eos do
         %Buffer{} = buffer ->
-          state = state |> put_in([pad_ref, :oldest_buffer_dts], buffer.dts)
+          state = state |> put_in([pad_name, :oldest_buffer_dts], buffer.dts)
           {[buffer: {pad_ref, buffer}], state}
 
         :end_of_stream ->
@@ -221,13 +265,13 @@ defmodule Membrane.HLS.Source do
 
   defp request_media_chunks(state) do
     [:audio_output, :video_output]
-    |> Enum.reduce(state, fn pad_ref, state ->
-      oldest_dts = state[pad_ref].oldest_buffer_dts
-      eos_received? = state[pad_ref].eos_received?
+    |> Enum.reduce(state, fn pad_name, state ->
+      %{eos_received?: eos_received?, oldest_buffer_dts: oldest_dts, ref: pad_ref} =
+        state[pad_name]
 
       request_size =
-        case state[pad_ref].qex |> Qex.first() do
-          _any when eos_received? ->
+        case state[pad_name].qex |> Qex.first() do
+          _any when pad_ref == nil or eos_received? ->
             0
 
           # todo: maybe we should handle rollovers
@@ -236,14 +280,14 @@ defmodule Membrane.HLS.Source do
             0
 
           _empty_or_not_new_enough ->
-            @requested_chunks_boundary - state[pad_ref].requested
+            @requested_chunks_boundary - state[pad_name].requested
         end
 
       1..request_size//1
-      |> Enum.each(fn _i -> request_single_chunk(pad_ref, state) end)
+      |> Enum.each(fn _i -> request_single_chunk(pad_name, state) end)
 
       state
-      |> update_in([pad_ref, :requested], &(&1 + request_size))
+      |> update_in([pad_name, :requested], &(&1 + request_size))
     end)
   end
 

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: "bf78d233b2f0a8409c8453df8e6dcb61a4e32c29"},
+       github: "membraneframework-labs/ex_hls", ref: "981f02f5093979c3c1ef71d596ac53e204e2ef7d"},
       {: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", "bf78d233b2f0a8409c8453df8e6dcb61a4e32c29", [ref: "bf78d233b2f0a8409c8453df8e6dcb61a4e32c29"]},
+  "ex_hls": {:git, "https://github.com/membraneframework-labs/ex_hls.git", "981f02f5093979c3c1ef71d596ac53e204e2ef7d", [ref: "981f02f5093979c3c1ef71d596ac53e204e2ef7d"]},
   "ex_m3u8": {:hex, :ex_m3u8, "0.15.2", "7b13d5c719fa6bd58e1fd54d3d17b27e4c2a21fa10ad74c39e7ec36be165d58a", [: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", "160d31f132f15856fcd4a4c40e448784f3105e5fcd05ac39de6a4ccc8aa4a697"},
   "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"},