Fix credo

Author: FelonEkonom

lib/membrane_http_adaptive_stream/hls_source.ex

@@ -12,15 +12,11 @@ defmodule Membrane.HLS.Source do
 
   def_output_pad :video_output,
     accepted_format: any_of(H264, %RemoteStream{content_format: H264}),
-    availability: :on_request,
-    max_instances: 1,
     flow_control: :manual,
     demand_unit: :buffers
 
   def_output_pad :audio_output,
     accepted_format: any_of(AAC, %RemoteStream{content_format: AAC}),
-    availability: :on_request,
-    max_instances: 1,
     flow_control: :manual,
     demand_unit: :buffers
 
@@ -34,7 +30,6 @@ defmodule Membrane.HLS.Source do
 
   def_options url: [spec: String.t()],
               buffer_size: [spec: pos_integer(), default: 0],
-              container: [spec: :mpeg_ts | :fmp4, default: :mpeg_ts],
               variant_selection_policy: [
                 spec: variant_selection_policy(),
                 default: :highest_resolution
@@ -45,8 +40,8 @@ defmodule Membrane.HLS.Source do
     state =
       Map.from_struct(opts)
       |> Map.merge(%{
-        audio_output: %{pad_ref: nil, requested: 0, qex: Qex.new(), qex_size: 0},
-        video_output: %{pad_ref: nil, requested: 0, qex: Qex.new(), qex_size: 0},
+        audio_output: %{requested: 0, qex: Qex.new(), qex_size: 0},
+        video_output: %{requested: 0, qex: Qex.new(), qex_size: 0},
         client_genserver: nil
       })
 
@@ -55,30 +50,14 @@ defmodule Membrane.HLS.Source do
 
   @impl true
   def handle_setup(_ctx, state) do
-    demuxing_engine =
-      case state.container do
-        :mpeg_ts -> ExHLS.DemuxingEngine.MPEGTS
-        :fmp4 -> ExHLS.DemuxingEngine.CMAF
-      end
-
-    {:ok, clinet_genserver} =
-      ClientGenServer.start_link(state.url, demuxing_engine, state.variant_selection_policy)
-
-    {[], %{state | client_genserver: clinet_genserver}}
-  end
+    {:ok, client_genserver} =
+      ClientGenServer.start_link(state.url, state.variant_selection_policy)
 
-  @impl true
-  def handle_pad_added(Pad.ref(pad_name, _id) = pad_ref, _ctx, state) do
-    state = state |> put_in([pad_name, :pad_ref], pad_ref)
-    {[], state}
+    {[], %{state | client_genserver: client_genserver}}
   end
 
   @impl true
   def handle_playing(ctx, state) do
-    if ctx.pads |> map_size() < 2 do
-      raise "HLS Source requires both audio and video output pads to be present"
-    end
-
     {[audio_stream_format], [video_stream_format]} =
       ClientGenServer.get_tracks_info(state.client_genserver)
       |> Map.values()
@@ -90,8 +69,8 @@ defmodule Membrane.HLS.Source do
       end)
 
     actions = [
-      stream_format: {state.audio_output.pad_ref, audio_stream_format},
-      stream_format: {state.video_output.pad_ref, video_stream_format}
+      stream_format: {:audio_output, audio_stream_format},
+      stream_format: {:video_output, video_stream_format}
     ]
 
     {actions, state}
@@ -105,42 +84,42 @@ defmodule Membrane.HLS.Source do
   end
 
   @impl true
-  def handle_info({stream_type, frame}, _ctx, state)
-      when stream_type in [:audio_stream, :video_stream] do
-    pad_name =
-      case stream_type do
-        :audio_stream -> :audio_output
-        :video_stream -> :video_output
+  def handle_info({data_type, sample}, _ctx, state)
+      when data_type in [:audio_sample_, :video_sample] do
+    pad_ref =
+      case data_type do
+        :audio_sample -> :audio_output
+        :video_sample -> :video_output
       end
 
     state =
       state
-      |> update_in([pad_name, :qex], &Qex.push(&1, frame))
-      |> update_in([pad_name, :qex_size], &(&1 + 1))
-      |> update_in([pad_name, :requested], &(&1 - 1))
+      |> update_in([pad_ref, :qex], &Qex.push(&1, sample))
+      |> update_in([pad_ref, :qex_size], &(&1 + 1))
+      |> update_in([pad_ref, :requested], &(&1 - 1))
 
-    {[redemand: state[pad_name].pad_ref], state}
+    {[redemand: pad_ref], state}
   end
 
-  defp pop_buffers(Pad.ref(pad_name, _id), demand, state) do
-    range_upperbound = min(state[pad_name].qex_size, demand)
+  defp pop_buffers(pad_ref, demand, state) do
+    range_upperbound = min(state[pad_ref].qex_size, demand)
 
     if range_upperbound > 0 do
       1..range_upperbound
       |> Enum.map_reduce(state, fn _i, state ->
-        {frame, qex} = state[pad_name].qex |> Qex.pop!()
+        {%ExHLS.Sample{} = sample, qex} = state[pad_ref].qex |> Qex.pop!()
 
         buffer = %Membrane.Buffer{
-          payload: frame.payload,
-          pts: frame.pts |> Membrane.Time.milliseconds(),
-          dts: frame.dts |> Membrane.Time.milliseconds(),
-          metadata: frame.metadata
+          payload: sample.payload,
+          pts: sample.pts_ms |> Membrane.Time.milliseconds(),
+          dts: sample.dts_ms |> Membrane.Time.milliseconds(),
+          metadata: sample.metadata
         }
 
         state =
           state
-          |> put_in([pad_name, :qex], qex)
-          |> update_in([pad_name, :qex_size], &(&1 - 1))
+          |> put_in([pad_ref, :qex], qex)
+          |> update_in([pad_ref, :qex_size], &(&1 - 1))
 
         {buffer, state}
       end)
@@ -149,30 +128,24 @@ defmodule Membrane.HLS.Source do
     end
   end
 
-  defp request_frames(state) do
+  defp request_samples(state) do
     [:audio_output, :video_output]
-    |> Enum.reduce(state, fn pad_name, state ->
-      request_size = state.buffer_size - state[pad_name].qex_size - state[pad_name].requested
-      :ok = do_request(state, pad_name, request_size)
+    |> Enum.reduce(state, fn pad_ref, state ->
+      request_size = state.buffer_size - state[pad_ref].qex_size - state[pad_ref].requested
+
+      if request_size > 0 do
+        1..request_size
+        |> Enum.each(fn _i -> reuqest_single_sample(pad_ref, state) end)
+      end
 
       state
-      |> update_in([pad_name, :requested], &(&1 + request_size))
+      |> update_in([pad_ref, :requested], &(&1 + request_size))
     end)
   end
 
-  defp do_request(_state, _pad_name, request_size) when request_size < 1, do: :ok
+  defp request_single_sample(:audio_output, state),
+    do: ClientGenServer.request_audio_sample(state.client_genserver)
 
-  defp do_request(state, :audio_output, request_size) do
-    1..request_size
-    |> Enum.each(fn _i ->
-      ClientGenServer.request_audio(state.client_genserver)
-    end)
-  end
-
-  defp do_request(state, :video_output, request_size) do
-    1..request_size
-    |> Enum.each(fn _i ->
-      ClientGenServer.request_video(state.client_genserver)
-    end)
-  end
+  defp request_single_sample(:video_output, state),
+    do: ClientGenServer.request_video_sample(state.client_genserver)
 end

lib/membrane_http_adaptive_stream/hls_source/client_genserver.ex

@@ -8,26 +8,24 @@ defmodule Membrane.HLS.Source.ClientGenServer do
 
   @spec start_link(
           String.t(),
-          ExHLS.DemuxingEngine.CMAF | ExHLS.DemuxingEngine.MPEGTS,
           Membrane.HLS.Source.variant_selection_policy()
         ) ::
           GenServer.on_start()
-  def start_link(url, demuxing_engine, variant_selection_policy) do
+  def start_link(url, variant_selection_policy) do
     GenServer.start_link(__MODULE__,
       url: url,
-      demuxing_engine: demuxing_engine,
       variant_selection_policy: variant_selection_policy
     )
   end
 
-  @spec request_audio(pid()) :: :ok
-  def request_audio(client_genserver) do
-    GenServer.cast(client_genserver, {:request_audio, self()})
+  @spec request_audio_sample(pid()) :: :ok
+  def request_audio_sample(client_genserver) do
+    GenServer.cast(client_genserver, {:request_audio_sample, self()})
   end
 
-  @spec request_video(pid()) :: :ok
-  def request_video(client_genserver) do
-    GenServer.cast(client_genserver, {:request_video, self()})
+  @spec request_video_sample(pid()) :: :ok
+  def request_video_sample(client_genserver) do
+    GenServer.cast(client_genserver, {:request_video_sample, self()})
   end
 
   @spec get_tracks_info(pid()) :: map()
@@ -36,37 +34,48 @@ defmodule Membrane.HLS.Source.ClientGenServer do
   end
 
   @impl true
-  def init(
-        url: url,
-        demuxing_engine: demuxing_engine,
-        variant_selection_policy: variant_selection_policy
-      ) do
+  def init(url: url, variant_selection_policy: variant_selection_policy) do
     state = %{
-      client: Client.new(url, demuxing_engine),
-      variant_selection_policy: variant_selection_policy
+      url: url,
+      variant_selection_policy: variant_selection_policy,
+      client: nil
     }
 
-    {:ok, state |> choose_variant()}
+    # let's create Client and choose variant asnychronously
+    # beyond init/1, because it requires doing some HTTP requests
+    # so it can take some time
+    self() |> send(:setup)
+
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_info(:setup, state) do
+    state =
+      %{state | client: Client.new(state.url)}
+      |> choose_variant()
+
+    {:noreply, state}
   end
 
   defp choose_variant(state) do
     variants = Client.get_variants(state.client)
-    get_resolution = fn {_id, %{resolution: {width, height}}} -> width * height end
-    get_bandwidth = fn {_id, %{bandwidth: bandwidth}} -> bandwidth end
+    get_resolution_fn = fn {_id, %{resolution: {width, height}}} -> width * height end
+    get_bandwidth_fn = fn {_id, %{bandwidth: bandwidth}} -> bandwidth end
 
     chosen_variant_id =
       case state.variant_selection_policy do
         :lowest_resolution ->
-          variants |> Enum.min_by(get_resolution) |> elem(0)
+          variants |> Enum.min_by(get_resolution_fn) |> elem(0)
 
         :highest_resolution ->
-          variants |> Enum.max_by(get_resolution) |> elem(0)
+          variants |> Enum.max_by(get_resolution_fn) |> elem(0)
 
         :lowest_bandwidth ->
-          variants |> Enum.min_by(get_bandwidth) |> elem(0)
+          variants |> Enum.min_by(get_bandwidth_fn) |> elem(0)
 
         :highest_bandwidth ->
-          variants |> Enum.max_by(get_bandwidth) |> elem(0)
+          variants |> Enum.max_by(get_bandwidth_fn) |> elem(0)
 
         custom_policy when is_function(custom_policy, 1) ->
           variants |> custom_policy.()
@@ -77,16 +86,16 @@ defmodule Membrane.HLS.Source.ClientGenServer do
   end
 
   @impl true
-  def handle_cast({:request_audio, pid}, state) do
-    {frame, client} = Client.read_audio_frame(state.client)
-    send(pid, {:audio_stream, frame})
+  def handle_cast({:request_audio_sample, pid}, state) do
+    {sample, client} = Client.read_audio_sample(state.client)
+    send(pid, {:audio_sample, sample})
     {:noreply, %{state | client: client}}
   end
 
   @impl true
-  def handle_cast({:request_video, pid}, state) do
-    {frame, client} = Client.read_video_frame(state.client)
-    send(pid, {:video_stream, frame})
+  def handle_cast({:request_video_sample, pid}, state) do
+    {sample, client} = Client.read_video_sample(state.client)
+    send(pid, {:video_sample, sample})
     {:noreply, %{state | client: client}}
   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: "d72569595f3a4eeb9d1521dfbc8b2445b6e50943"},
+       github: "membraneframework-labs/ex_hls", ref: "8de4292b21ebca8080496aade56233fb69e1ba29"},
       {:bunch, "~> 1.6"},
       {:qex, "~> 0.5"},
       {:membrane_hackney_plugin, "~> 0.11.0", only: :test},

mix.lock

@@ -9,8 +9,8 @@
   "earmark_parser": {:hex, :earmark_parser, "1.4.44", "f20830dd6b5c77afe2b063777ddbbff09f9759396500cdbe7523efd58d7a339c", [:mix], [], "hexpm", "4778ac752b4701a5599215f7030989c989ffdc4f6df457c5f36938cc2d2a2750"},
   "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", "d72569595f3a4eeb9d1521dfbc8b2445b6e50943", [ref: "d72569595f3a4eeb9d1521dfbc8b2445b6e50943"]},
-  "ex_m3u8": {:git, "https://github.com/membraneframework/ex_m3u8.git", "8a83e4f6dd14e749aadf551bf1adb7ad4d4eb495", [branch: "allow-empty-default"]},
+  "ex_hls": {:git, "https://github.com/membraneframework-labs/ex_hls.git", "8de4292b21ebca8080496aade56233fb69e1ba29", [ref: "8de4292b21ebca8080496aade56233fb69e1ba29"]},
+  "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.19.0", "c644641491ea854fc5c1bbaef36bfc764e3f08e7185e1f084e35e0672241b76d", [: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", "fc5324ce209125d1e2fa0fcd2634601c52a787aff1cd33ee833664a5af4ea2b6"},
   "hackney": {:hex, :hackney, "1.24.1", "f5205a125bba6ed4587f9db3cc7c729d11316fa8f215d3e57ed1c067a9703fa9", [:rebar3], [{:certifi, "~> 2.15.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~> 6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~> 1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.4", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.4.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "f4a7392a0b53d8bbc3eb855bdcc919cd677358e65b2afd3840b5b3690c4c8a39"},
@@ -41,7 +41,7 @@
   "mimerl": {:hex, :mimerl, "1.4.0", "3882a5ca67fbbe7117ba8947f27643557adec38fa2307490c4c4207624cb213b", [:rebar3], [], "hexpm", "13af15f9f68c65884ecca3a3891d50a7b57d82152792f3e19d88650aa126b144"},
   "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"},
   "mockery": {:hex, :mockery, "2.3.3", "3dba87bd0422a513e6af6e0d811383f38f82ac6be5d3d285a5fcca9c299bd0ac", [:mix], [], "hexpm", "17282be00613286254298117cd25e607a39f15ac03b41c631f60e52f5b5ec974"},
-  "mpeg_ts": {:git, "https://github.com/kim-company/kim_mpeg_ts.git", "f3e529b4945c185001884fbdc50d668f1af02c5b", []},
+  "mpeg_ts": {:git, "https://github.com/membraneframework-labs/kim_mpeg_ts.git", "8c036fca6558a4339033a5a8697ebf147728f36b", [branch: "backport-v1.0.3"]},
   "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"},