Membrane Live dev

integration_test/test_videoroom/lib/test_videoroom/room.ex

@@ -57,7 +57,7 @@ defmodule TestVideoroom.Room do
       id: room_id
     ]
 
-    {:ok, _supervisor, pid} = Membrane.RTC.Engine.start(rtc_engine_options, [])
+    {:ok, pid} = Membrane.RTC.Engine.start(rtc_engine_options, [])
     Process.monitor(pid)
     Engine.register(pid, self())
 

lib/membrane_rtc_engine/endpoints/hls/config/audio_mixer_config.ex

@@ -0,0 +1,24 @@
+if Code.ensure_loaded?(Membrane.RawAudio) do
+  defmodule Membrane.RTC.Engine.Endpoint.HLS.AudioMixerConfig do
+    @moduledoc """
+    Module representing audio mixer configuration for the HLS endpoint.
+    """
+
+    alias Membrane.RawAudio
+
+    @typedoc """
+    * `stream_format` - defines audio mixer output stream_format.
+    * `background` - module generating background sound that will be played from the beginning of the stream.
+    """
+    @type t() :: %__MODULE__{
+            stream_format: RawAudio.t(),
+            background: struct() | nil
+          }
+    defstruct stream_format: %RawAudio{
+                channels: 1,
+                sample_rate: 48_000,
+                sample_format: :s16le
+              },
+              background: nil
+  end
+end

lib/membrane_rtc_engine/endpoints/hls/config/compositor_config.ex

@@ -0,0 +1,31 @@
+if Code.ensure_loaded?(Membrane.RawVideo) do
+  defmodule Membrane.RTC.Engine.Endpoint.HLS.CompositorConfig do
+    @moduledoc """
+    Module representing compositor configuration for the HLS endpoint.
+
+    Compositor is the element responsible for positioning input video streams
+    into one video output.
+    """
+
+    @typedoc """
+    * `stream_format` - output video specification in Membrane.RawVideo format
+    * `layout_module` - Module implementing `Membrane.RTC.Engine.Endpoint.HLS.VideoLayoutMaker`
+      behavior that should be used by the HLS endpoint.
+    * `background` - Struct representing module that generates video used as a background in composed video.
+    """
+    @type t() :: %__MODULE__{
+            stream_format: Membrane.RawVideo.t(),
+            layout_module: module(),
+            background: struct() | nil
+          }
+    defstruct stream_format: %Membrane.RawVideo{
+                width: 400,
+                height: 800,
+                pixel_format: :I420,
+                framerate: {24, 1},
+                aligned: true
+              },
+              layout_module: Membrane.RTC.Engine.Endpoint.HLS.MobileLayoutMaker,
+              background: nil
+  end
+end

lib/membrane_rtc_engine/endpoints/hls/config/mixer_config.ex

@@ -0,0 +1,24 @@
+if Enum.all?([Membrane.RawAudio, Membrane.RawVideo], &Code.ensure_loaded?/1) do
+  defmodule Membrane.RTC.Engine.Endpoint.HLS.MixerConfig do
+    @moduledoc """
+    Module representing mixer configuration for the HLS endpoint.
+    """
+
+    alias Membrane.RTC.Engine.Endpoint.HLS.{AudioMixerConfig, CompositorConfig}
+
+    @typedoc """
+    * `video` - video compositor configuration.
+    * `audio` - audio mixer configuration.
+    * `persist?` - if set to true, the stream will be continuous even if all tracks are removed, if then some new track is added it will belong to the same stream.
+    Otherwise if set to false, stream will automatically end when the last track is removed, adding then new track will start a new hls stream.
+    """
+    @type t() :: %__MODULE__{
+            video: CompositorConfig.t(),
+            audio: AudioMixerConfig.t(),
+            persist?: boolean()
+          }
+    defstruct video: %CompositorConfig{},
+              audio: %AudioMixerConfig{},
+              persist?: false
+  end
+end

lib/membrane_rtc_engine/endpoints/hls/config/sink_bin_config.ex

@@ -0,0 +1,46 @@
+if Code.ensure_loaded?(Membrane.HTTPAdaptiveStream.Manifest) do
+  defmodule Membrane.RTC.Engine.Endpoint.HLS.HLSConfig do
+    @moduledoc """
+    Module representing `Membrane.HTTPAdaptiveStream.SinkBin` configuration for the HLS endpoint.
+    """
+
+    alias Membrane.HTTPAdaptiveStream.{Manifest, Storage}
+    alias Membrane.HTTPAdaptiveStream.Sink.SegmentDuration
+    alias Membrane.Time
+
+    @typedoc """
+      To read more about config options go to module Membrane.HTTPAdaptiveStream.SinkBin and read options descriptions.
+      * `segment_duration` - The segment duration range  of the regular segments.
+      * `partial_segment_duration` - The segment duration range  of the partial segments. If not set then the bin won't produce any partial segments.
+    """
+    @type t() :: %__MODULE__{
+            manifest_name: String.t(),
+            manifest_module: Manifest.t(),
+            storage: (Path.t() -> Storage.config_t()),
+            target_window_duration: pos_integer | :infinity,
+            persist?: boolean,
+            mode: :live | :vod,
+            hls_mode: :muxed_av | :separate_av,
+            header_naming_fun: (Manifest.Track.t(), counter :: non_neg_integer() -> String.t()),
+            segment_naming_fun: (Manifest.Track.t() -> String.t()),
+            segment_duration: SegmentDuration.t(),
+            partial_segment_duration: SegmentDuration.t() | nil
+          }
+
+    defstruct manifest_name: "index",
+              manifest_module: Membrane.HTTPAdaptiveStream.HLS,
+              storage: &__MODULE__.default_storage/1,
+              target_window_duration: Membrane.Time.seconds(40),
+              persist?: false,
+              mode: :live,
+              hls_mode: :separate_av,
+              header_naming_fun: &Manifest.Track.default_header_naming_fun/2,
+              segment_naming_fun: &Manifest.Track.default_segment_naming_fun/1,
+              segment_duration: SegmentDuration.new(Time.seconds(4), Time.seconds(5)),
+              partial_segment_duration: nil
+
+    @spec default_storage(String.t()) :: any
+    def default_storage(directory),
+      do: %Membrane.HTTPAdaptiveStream.Storages.FileStorage{directory: directory}
+  end
+end

lib/membrane_rtc_engine/endpoints/hls/mobile_layout_maker.ex

@@ -0,0 +1,176 @@
+if Code.ensure_loaded?(Membrane.VideoCompositor) do
+  defmodule Membrane.RTC.Engine.Endpoint.HLS.MobileLayoutMaker do
+    @moduledoc """
+    Module representing function for updating video layout for the HLS stream.
+
+    1) Only main presenter
+         _ _ _ _
+        |       |
+        |       |
+        |       |
+        |       |
+        |       |
+         - - - -
+    2) Main presenter and one side presenter
+         _ _ _ _
+        |       |
+        |       |
+        |       |
+         - -    |
+        |   |   |
+         - - - -
+
+    3) Main presenter and two side presenters
+         _ _ _ _
+        |       |
+        |       |
+        |       |
+         - - - -
+        |   |   |
+         - - - -
+    """
+
+    @behaviour Membrane.RTC.Engine.Endpoint.HLS.VideoLayoutMaker
+
+    alias Membrane.VideoCompositor.RustStructs.BaseVideoPlacement
+    alias Membrane.VideoCompositor.VideoTransformations
+    alias Membrane.VideoCompositor.VideoTransformations.TextureTransformations.CornersRounding
+    alias Membrane.VideoCompositor.VideoTransformations.TextureTransformations.Cropping
+
+    @main_stream %{position: {0, 0}, z_value: 0.1, corner_radius: 0}
+    @side_stream %{z_value: 0.3, corner_radius: 20, padding: 5}
+
+    @impl true
+    def init(output_stream_format), do: %{tracks: %{}, output_stream_format: output_stream_format}
+
+    @impl true
+    def track_added(%{metadata: %{"mainPresenter" => true}} = track, stream_format, state) do
+      new_state = put_in(state, [:tracks, track.id], {track, stream_format})
+
+      {layout, transformations} =
+        get_track_layout(:main, nil, stream_format, state.output_stream_format)
+
+      updated_layout = [
+        {track.id, layout, transformations}
+      ]
+
+      {updated_layout, new_state}
+    end
+
+    @impl true
+    def track_added(track, stream_format, state) do
+      new_state = put_in(state, [:tracks, track.id], {track, stream_format})
+      {update_layout(new_state), new_state}
+    end
+
+    @impl true
+    def track_updated(track, stream_format, state), do: track_added(track, stream_format, state)
+
+    @impl true
+    def track_removed(track, state) do
+      {_, new_state} = pop_in(state, [:tracks, track.id])
+      {update_layout(new_state), new_state}
+    end
+
+    defp update_layout(%{tracks: tracks, output_stream_format: output_stream_format}) do
+      tracks
+      |> Enum.filter(fn {_id, {track, _stream_format}} ->
+        track.type == :video and not track.metadata["mainPresenter"]
+      end)
+      |> Enum.with_index()
+      |> Enum.flat_map(fn {{_id, {track, stream_format}}, index} ->
+        {layout, transcoding} =
+          get_track_layout(:side, index, stream_format, output_stream_format)
+
+        [
+          {track.id, layout, transcoding}
+        ]
+      end)
+    end
+
+    defp get_track_layout(:main, _index, stream_format, output_stream_format) do
+      placement =
+        get_placement(
+          output_stream_format,
+          stream_format,
+          @main_stream.position,
+          @main_stream.z_value
+        )
+
+      transformations =
+        get_transformations(output_stream_format, placement.size, @main_stream.corner_radius)
+
+      {placement, transformations}
+    end
+
+    defp get_track_layout(:side, index, stream_format, %{width: width, height: height}) do
+      # side track layout is a thumbnail that is placed on the bottom of the main stream.
+      output_stream_format = %{
+        width: round(1 / 2 * width) - @side_stream.padding * 2,
+        height: round(1 / 4 * height) - @side_stream.padding * 2
+      }
+
+      # x coordinate of a thumbnail can differ depends on which index they have, y coordinate is always the same.
+      position_x = round(index / 2 * width) + @side_stream.padding
+      position_y = height - round(1 / 4 * height) + @side_stream.padding
+
+      position = {position_x, position_y}
+
+      placement =
+        get_placement(output_stream_format, stream_format, position, @side_stream.z_value)
+
+      transformations =
+        get_transformations(output_stream_format, placement.size, @side_stream.corner_radius)
+
+      {placement, transformations}
+    end
+
+    defp video_proportion(%{width: width, height: height}), do: width / height
+
+    defp get_placement(output_stream_format, stream_format, position, z_value),
+      do: %BaseVideoPlacement{
+        position: position,
+        size: get_display_size(output_stream_format, stream_format),
+        z_value: z_value
+      }
+
+    defp get_display_size(output_stream_format, stream_format) do
+      if video_proportion(output_stream_format) > video_proportion(stream_format),
+        do:
+          {output_stream_format.width,
+           round(output_stream_format.width / stream_format.width * stream_format.height)},
+        else:
+          {round(output_stream_format.height / stream_format.height * stream_format.width),
+           output_stream_format.height}
+    end
+
+    defp get_transformations(output_stream_format, display_size, radius),
+      do: %VideoTransformations{
+        texture_transformations: [
+          get_cropping(output_stream_format, display_size),
+          get_corners_rounding(radius)
+        ]
+      }
+
+    defp get_corners_rounding(radius), do: %CornersRounding{border_radius: radius}
+
+    defp get_cropping(output_stream_format, display_size),
+      do: %Cropping{
+        crop_top_left_corner: get_cropping_position(output_stream_format, display_size),
+        crop_size: get_cropping_size(output_stream_format, display_size),
+        cropped_video_position: :input_position
+      }
+
+    defp get_cropping_position(output_stream_format, {width, height}) do
+      if output_stream_format.width == width,
+        do: {0.0, (height - output_stream_format.height) / (2 * height)},
+        else: {(width - output_stream_format.width) / (2 * width), 0.0}
+    end
+
+    defp get_cropping_size(output_stream_format, {width, height}) do
+      if output_stream_format.width == width,
+        do: {1.0, output_stream_format.height / height},
+        else: {output_stream_format.width / width, 1.0}
+    end
+  end
+end

lib/membrane_rtc_engine/endpoints/hls/stream_format_updater.ex

@@ -0,0 +1,59 @@
+defmodule Membrane.RTC.Engine.Endpoint.HLS.StreamFormatUpdater do
+  @moduledoc """
+    Element responsible for sending stream format for specific track to HLS Endpoint and waiting for Endpoint to send updated layout to compositor element.
+  """
+
+  use Membrane.Filter
+
+  def_input_pad :input,
+    accepted_format: _any,
+    demand_unit: :buffers,
+    demand_mode: :auto
+
+  def_output_pad :output,
+    accepted_format: _any,
+    demand_mode: :auto
+
+  @impl true
+  def handle_init(_ctx, _opts) do
+    {[], %{update_queue: 0, buffers: %{}, end_of_stream: false}}
+  end
+
+  @impl true
+  def handle_stream_format(_pad, stream_format, _ctx, state) do
+    state = put_in(state, [:buffers, state.update_queue + 1], [])
+
+    {[forward: stream_format, notify_parent: {:update_layout, stream_format}],
+     %{state | update_queue: state.update_queue + 1}}
+  end
+
+  @impl true
+  def handle_end_of_stream(_pad, _ctx, %{update__queue: 0} = state),
+    do: {[end_of_stream: :output], state}
+
+  @impl true
+  def handle_end_of_stream(_pad, _ctx, state), do: {[], %{state | end_of_stream: true}}
+
+  @impl true
+  def handle_process(_pad, buffer, _ctx, %{update_queue: 0} = state),
+    do: {[buffer: {:output, buffer}], state}
+
+  @impl true
+  def handle_process(_pad, buffer, _ctx, state) do
+    state = update_in(state, [:buffers, state.update_queue], &[buffer | &1])
+    {[], state}
+  end
+
+  @impl true
+  def handle_parent_notification(:layout_updated, _ctx, state) do
+    {buffers, state} = pop_in(state, [:buffers, state.update_queue])
+    buffers = Enum.reverse(buffers)
+    actions = [buffer: {:output, buffers}] ++ maybe_notify_end_of_stream(state)
+    {actions, %{state | update_queue: state.update_queue - 1}}
+  end
+
+  defp maybe_notify_end_of_stream(%{end_of_stream: true, update_queue: 0}),
+    do: [end_of_stream: :output]
+
+  defp maybe_notify_end_of_stream(_state), do: []
+end