Refactor hls tests (remove sleep())

lib/membrane_rtc_engine/endpoints/hls/stream_format_updater.ex

@@ -28,7 +28,7 @@ defmodule Membrane.RTC.Engine.Endpoint.HLS.StreamFormatUpdater do
   end
 
   @impl true
-  def handle_end_of_stream(_pad, _ctx, %{update__queue: 0} = state),
+  def handle_end_of_stream(_pad, _ctx, %{update_queue: 0} = state),
     do: {[end_of_stream: :output], state}
 
   @impl true
@@ -52,7 +52,7 @@ defmodule Membrane.RTC.Engine.Endpoint.HLS.StreamFormatUpdater do
     {actions, %{state | update_queue: state.update_queue - 1}}
   end
 
-  defp maybe_notify_end_of_stream(%{end_of_stream: true, update_queue: 0}),
+  defp maybe_notify_end_of_stream(%{end_of_stream: true, update_queue: 1}),
     do: [end_of_stream: :output]
 
   defp maybe_notify_end_of_stream(_state), do: []

lib/membrane_rtc_engine/endpoints/hls_endpoint.ex

@@ -95,7 +95,7 @@ if Enum.all?(
       :compositor,
       :encoder,
       :video_parser_out,
-      :hls_sink_bin
+      {:hls_sink_bin, :muxed}
     ]
 
     def_input_pad :input,
@@ -162,7 +162,7 @@ if Enum.all?(
       spec =
         generate_audio_mixer(state, context) ++
           generate_compositor(state, context) ++
-          get_hls_sink_spec(state, %{stream_id: nil}, state.output_directory)
+          get_hls_sink_spec(state)
 
       {[spec: spec], state}
     end
@@ -214,22 +214,31 @@ if Enum.all?(
     @impl true
     def handle_pad_added(Pad.ref(:input, track_id) = pad, ctx, state) do
       {offset, state} = get_track_offset(state)
-      track = Map.get(state.tracks, track_id)
-      directory = get_hls_stream_directory(state, track)
 
+      track = Map.get(state.tracks, track_id)
       track_spec = get_track_spec(offset, bin_input(pad), track, state, ctx)
 
       {spec, state} =
         if hls_sink_bin_exists?(track, ctx, state) do
           {track_spec, state}
         else
-          hls_sink_spec = get_hls_sink_spec(state, track, directory)
+          hls_sink_spec = get_hls_sink_spec(state, track.stream_id)
           {track_spec ++ hls_sink_spec, state}
         end
 
       {[spec: spec], state}
     end
 
+    @impl true
+    def handle_child_notification(
+          :end_of_stream,
+          {:hls_sink_bin, stream},
+          _ctx,
+          state
+        ) do
+      {[notify_parent: {:forward_to_parent, {:end_of_stream, stream}}], state}
+    end
+
     @impl true
     def handle_child_notification(
           {:update_layout, stream_format},
@@ -251,22 +260,19 @@ if Enum.all?(
     end
 
     def handle_child_notification(
-          {:track_playable, content_type},
-          :hls_sink_bin,
-          _ctx,
-          state
-        ) do
-      send(state.owner, {:playlist_playable, content_type, ""})
-      {[], state}
-    end
-
-    def handle_child_notification(
-          {:track_playable, {content_type, _track_id}},
+          {:track_playable, data},
           {:hls_sink_bin, stream_id},
           _ctx,
           state
         ) do
-      send(state.owner, {:playlist_playable, content_type, stream_id})
+      content_type =
+        case data do
+          {content_type, _track_id} -> content_type
+          content_type -> content_type
+        end
+
+      output_dir = get_hls_stream_directory(state, stream_id)
+      send(state.owner, {:playlist_playable, content_type, output_dir})
       {[], state}
     end
 
@@ -308,7 +314,9 @@ if Enum.all?(
       {[], state}
     end
 
-    defp get_hls_sink_spec(state, track, directory) do
+    defp get_hls_sink_spec(state, stream_id \\ nil) do
+      directory = get_hls_stream_directory(state, stream_id)
+
       File.rm_rf(directory)
       File.mkdir_p!(directory)
 
@@ -320,7 +328,9 @@ if Enum.all?(
       hls_sink = struct(Membrane.HTTPAdaptiveStream.SinkBin, Map.from_struct(config))
 
       child_name =
-        if is_nil(state.mixer_config), do: {:hls_sink_bin, track.stream_id}, else: :hls_sink_bin
+        if is_nil(state.mixer_config),
+          do: {:hls_sink_bin, stream_id},
+          else: {:hls_sink_bin, :muxed}
 
       [child(child_name, hls_sink)]
     end
@@ -489,8 +499,7 @@ if Enum.all?(
 
       defp generate_compositor(state, _ctx) do
         compositor = %Membrane.VideoCompositor{
-          stream_format: state.mixer_config.video.stream_format,
-          real_time: false
+          stream_format: state.mixer_config.video.stream_format
         }
 
         video_parser_out = %Membrane.H264.FFmpeg.Parser{
@@ -515,7 +524,7 @@ if Enum.all?(
               partial_segment_duration: state.hls_config.partial_segment_duration
             ]
           )
-          |> get_child(:hls_sink_bin)
+          |> get_child({:hls_sink_bin, :muxed})
         ]
       end
     else
@@ -544,7 +553,7 @@ if Enum.all?(
               partial_segment_duration: state.hls_config.partial_segment_duration
             ]
           )
-          |> get_child(:hls_sink_bin)
+          |> get_child({:hls_sink_bin, :muxed})
         ]
       end
     else
@@ -559,18 +568,19 @@ if Enum.all?(
     defp hls_sink_bin_exists?(track, ctx, %{mixer_config: nil}),
       do: Map.has_key?(ctx.children, {:hls_sink_bin, track.stream_id})
 
-    defp hls_sink_bin_exists?(_track, ctx, _state), do: Map.has_key?(ctx.children, :hls_sink_bin)
+    defp hls_sink_bin_exists?(_track, ctx, _state),
+      do: Map.has_key?(ctx.children, {:hls_sink_bin, :muxed})
 
     defp get_depayloader(track) do
       track
       |> Track.get_depayloader()
       |> tap(&unless &1, do: raise("Couldn't find depayloader for track #{inspect(track)}"))
     end
 
-    defp get_hls_stream_directory(%{mixer_config: nil} = state, track),
-      do: Path.join(state.output_directory, track.stream_id)
+    defp get_hls_stream_directory(%{mixer_config: nil} = state, stream_id),
+      do: Path.join(state.output_directory, stream_id)
 
-    defp get_hls_stream_directory(state, _track), do: state.output_directory
+    defp get_hls_stream_directory(state, _stream_id), do: state.output_directory
 
     defp compositor_update_layout(_action, _track, _state, _stream_format \\ nil)
 

lib/membrane_rtc_engine/engine.ex

@@ -224,7 +224,7 @@ defmodule Membrane.RTC.Engine do
   @typedoc """
   Membrane action that will cause RTC Engine to forward supplied message to the business logic.
   """
-  @type forward_to_parent_action_t() :: {:notify, {:forward_to_parent, message :: any()}}
+  @type forward_to_parent_action_t() :: {:notify_parent, {:forward_to_parent, message :: any()}}
 
   @typedoc """
   Membrane action that will inform RTC Engine about track readiness.

mix.exs

@@ -87,7 +87,7 @@ defmodule Membrane.RTC.Engine.MixProject do
       {:membrane_raw_audio_format, "~> 0.10.0", optional: true},
       {:membrane_h264_ffmpeg_plugin, "~> 0.25.2", optional: true},
       {:membrane_audio_filler_plugin, "~> 0.1.0", optional: true},
-      {:membrane_video_compositor_plugin, "~> 0.1.0", optional: true},
+      {:membrane_video_compositor_plugin, "~> 0.2.0", optional: true},
       {:membrane_http_adaptive_stream_plugin, "~> 0.10.0", optional: true},
 
       # Test deps

mix.lock

@@ -10,7 +10,7 @@
   "credo": {:hex, :credo, "1.6.7", "323f5734350fd23a456f2688b9430e7d517afb313fbd38671b8a4449798a7854", [:mix], [{:bunt, "~> 0.2.1", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2.8", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "41e110bfb007f7eda7f897c10bf019ceab9a0b269ce79f015d54b0dcf4fc7dd3"},
   "dialyxir": {:hex, :dialyxir, "1.1.0", "c5aab0d6e71e5522e77beff7ba9e08f8e02bad90dfbeffae60eaf0cb47e29488", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "07ea8e49c45f15264ebe6d5b93799d4dd56a44036cf42d0ad9c960bc266c0b9a"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.30", "0b938aa5b9bafd455056440cdaa2a79197ca5e693830b4a982beada840513c5f", [:mix], [], "hexpm", "3b5385c2d36b0473d0b206927b841343d25adb14f95f0110062506b300cd5a1b"},
-  "elixir_make": {:hex, :elixir_make, "0.7.3", "c37fdae1b52d2cc51069713a58c2314877c1ad40800a57efb213f77b078a460d", [:mix], [{:castore, "~> 0.1", [hex: :castore, repo: "hexpm", optional: true]}], "hexpm", "24ada3e3996adbed1fa024ca14995ef2ba3d0d17b678b0f3f2b1f66e6ce2b274"},
+  "elixir_make": {:hex, :elixir_make, "0.7.4", "5439110c964ffdd8212ca919b5b8beac423085a77ad33d5e394abe812c2d2d75", [:mix], [{:castore, "~> 0.1", [hex: :castore, repo: "hexpm", optional: true]}], "hexpm", "70c33052f7b00c813fd66d15a3cf1f7d1e122860c572ec81b8181b1276074157"},
   "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},
   "ex_doc": {:hex, :ex_doc, "0.28.3", "6eea2f69995f5fba94cd6dd398df369fe4e777a47cd887714a0976930615c9e6", [:mix], [{:earmark_parser, "~> 1.4.19", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "05387a6a2655b5f9820f3f627450ed20b4325c25977b2ee69bed90af6688e718"},
   "ex_dtls": {:hex, :ex_dtls, "0.11.1", "1c4d9e235cd17f688e99b8359da45d132ad52ec91d18d0bac4d3d741cc35f7b8", [:mix], [{:unifex, "~> 1.0", [hex: :unifex, repo: "hexpm", optional: false]}], "hexpm", "249443c51aea5e9560f791cf023d0454533a7ea3efe70b3cb3956d7871147295"},
@@ -58,7 +58,7 @@
   "membrane_rtp_vp8_plugin": {:hex, :membrane_rtp_vp8_plugin, "0.7.1", "8a9d41eb53e2af8595fa87994a88030d0eb299cd7976f7b3273a6118553c58d2", [:mix], [{:membrane_core, "~> 0.11.2", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_rtp_format, "~> 0.6.0", [hex: :membrane_rtp_format, repo: "hexpm", optional: false]}, {:membrane_vp8_format, "~> 0.4.0", [hex: :membrane_vp8_format, repo: "hexpm", optional: false]}], "hexpm", "e5fdca517aac0520636c6ce3c709d1eb190d1d266a58c1768cc0c8e1e806a197"},
   "membrane_tee_plugin": {:hex, :membrane_tee_plugin, "0.10.1", "a78aef5f04894495f845f4fdeac855320ae0fe21466ff7cb3f00446bcd2a9d0d", [:mix], [{:bunch, "~> 1.0", [hex: :bunch, repo: "hexpm", optional: false]}, {:membrane_core, "~> 0.11.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "6b97a7c3f67c467f0f6c02a048947aa7f579387f438a50fd06a6e1cbf5a5371d"},
   "membrane_telemetry_metrics": {:hex, :membrane_telemetry_metrics, "0.1.0", "cb93d28356b436b0597736c3e4153738d82d2a14ff547f831df7e9051e54fc06", [:mix], [{:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6.1", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "aba28dc8311f70ced95d984509be930fac55857d2d18bffcf768815e627be3f0"},
-  "membrane_video_compositor_plugin": {:hex, :membrane_video_compositor_plugin, "0.1.0", "f8b192bd3ed7e22653a578c054897125f6c6180653ea60222c3eaac30ad4978c", [:mix], [{:membrane_core, "~> 0.11.2", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_framerate_converter_plugin, "~> 0.6.0", [hex: :membrane_framerate_converter_plugin, repo: "hexpm", optional: false]}, {:membrane_raw_video_format, "~> 0.2.0", [hex: :membrane_raw_video_format, repo: "hexpm", optional: false]}, {:qex, "~> 0.5.1", [hex: :qex, repo: "hexpm", optional: false]}, {:rustler, "~> 0.26.0", [hex: :rustler, repo: "hexpm", optional: false]}, {:unifex, "~> 1.0", [hex: :unifex, repo: "hexpm", optional: false]}], "hexpm", "853a30170356097a1c25c046dff7c3a3b5d1f09fc29c364b20c56f6fb603409b"},
+  "membrane_video_compositor_plugin": {:hex, :membrane_video_compositor_plugin, "0.2.0", "b85a06fc856fb814b5d0b8ae7314a5fcd3035a79e47d98eab69f726b9bf6b09c", [:mix], [{:membrane_core, "~> 0.11.2", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_framerate_converter_plugin, "~> 0.6.0", [hex: :membrane_framerate_converter_plugin, repo: "hexpm", optional: false]}, {:membrane_raw_video_format, "~> 0.2.0", [hex: :membrane_raw_video_format, repo: "hexpm", optional: false]}, {:qex, "~> 0.5.1", [hex: :qex, repo: "hexpm", optional: false]}, {:rustler, "~> 0.26.0", [hex: :rustler, repo: "hexpm", optional: false]}, {:unifex, "~> 1.0", [hex: :unifex, repo: "hexpm", optional: false]}], "hexpm", "f34727b2b0f4d6bd5f4ccaa4991fdc82d77bdf56cee53f4e9ad1ba6dccff6b11"},
   "membrane_vp8_format": {:hex, :membrane_vp8_format, "0.4.0", "6c29ec67479edfbab27b11266dc92f18f3baf4421262c5c31af348c33e5b92c7", [:mix], [], "hexpm", "8bb005ede61db8fcb3535a883f32168b251c2dfd1109197c8c3b39ce28ed08e2"},
   "membrane_webrtc_plugin": {:git, "https://github.com/membraneframework/membrane_webrtc_plugin.git", "4a3ecde78a1454cde69e036aca85b2444de2d247", []},
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"},

test/hls_reference/muxed-segments/g2QABXZpZGVv.m3u8

@@ -1,11 +1,15 @@
 #EXTM3U
 #EXT-X-VERSION:7
-#EXT-X-TARGETDURATION:5
+#EXT-X-TARGETDURATION:4
 #EXT-X-MEDIA-SEQUENCE:0
 #EXT-X-DISCONTINUITY-SEQUENCE:0
 #EXT-X-MAP:URI="muxed_header_g2QABXZpZGVv_part_0.mp4"
-#EXTINF:4.985333333,
+#EXTINF:2.982666666,
 muxed_segment_0_g2QABXZpZGVv.m4s
-#EXTINF:4.996,
+#EXTINF:3.541494444,
 muxed_segment_1_g2QABXZpZGVv.m4s
-
+#EXTINF:3.004,
+muxed_segment_2_g2QABXZpZGVv.m4s
+#EXTINF:0.591833333,
+muxed_segment_3_g2QABXZpZGVv.m4s
+#EXT-X-ENDLIST

test/hls_reference/muxed-segments/index.m3u8

@@ -1,5 +1,5 @@
 #EXTM3U
 #EXT-X-VERSION:7
 #EXT-X-INDEPENDENT-SEGMENTS
-#EXT-X-STREAM-INF:BANDWIDTH=731429,CODECS="avc1.42e00a"
+#EXT-X-STREAM-INF:BANDWIDTH=3116969,CODECS="avc1.42e00a"
 g2QABXZpZGVv.m3u8

test/hls_reference/video_mixer/g2QABXZpZGVv.m3u8

@@ -8,4 +8,8 @@
 video_segment_0_g2QABXZpZGVv.m4s
 #EXTINF:3.541655555,
 video_segment_1_g2QABXZpZGVv.m4s
+#EXTINF:3.0,
+video_segment_2_g2QABXZpZGVv.m4s
+#EXTINF:0.581255556,
+video_segment_3_g2QABXZpZGVv.m4s
 #EXT-X-ENDLIST

test/hls_reference/video_mixer/index.m3u8

@@ -2,5 +2,5 @@
 #EXT-X-VERSION:7
 #EXT-X-INDEPENDENT-SEGMENTS
 #EXT-X-MEDIA:TYPE=AUDIO,NAME="audio_default_name",GROUP-ID="audio_default_id",AUTOSELECT=YES,DEFAULT=YES,URI="g2QABWF1ZGlv.m3u8"
-#EXT-X-STREAM-INF:BANDWIDTH=1433546,CODECS="avc1.42e00a",AUDIO="audio_default_id"
+#EXT-X-STREAM-INF:BANDWIDTH=3074515,CODECS="avc1.42e00a",AUDIO="audio_default_id"
 g2QABXZpZGVv.m3u8