Replace RTP.SessionBins with RTP.Muxers and RTP.Demuxers (#213)

* Bump deps * Replace SessionBin with Muxer in rtp_senders * Replace SessionBin with Demuxer in rtp_receivers
membraneframework · Jan 16, 2025 · 7ff5032 · 7ff5032
1 parent ddf16cf
commit 7ff5032
Show file tree

Hide file tree

Showing 7 changed files with 75 additions and 108 deletions.
diff --git a/README.md b/README.md
@@ -15,7 +15,7 @@ The package can be installed by adding `membrane_live_compositor_plugin` to your
 ```elixir
 def deps do
   [
-    {:membrane_live_compositor_plugin, "~> 0.10.0"}
+    {:membrane_live_compositor_plugin, "~> 0.10.1"}
   ]
 end
 ```

diff --git a/examples/mix.exs b/examples/mix.exs
@@ -34,9 +34,9 @@ defmodule Membrane.LiveCompositor.Examples.Mixfile do
       # Examples
       {:membrane_file_plugin, "~> 0.17.0"},
       {:membrane_realtimer_plugin, "~> 0.9.0"},
-      {:membrane_sdl_plugin, "~> 0.18.2"},
+      {:membrane_sdl_plugin, "~> 0.18.5"},
       {:membrane_h26x_plugin, "~> 0.10.1"},
-      {:membrane_h264_ffmpeg_plugin, "~> 0.31.6"},
+      {:membrane_h264_ffmpeg_plugin, "~> 0.32.0"},
       {:membrane_hackney_plugin, "~> 0.11.0"},
       {:membrane_mp4_plugin, "~> 0.34.2"},
       {:membrane_portaudio_plugin, "~> 0.19.2"},

diff --git a/examples/mix.lock b/examples/mix.lock
diff --git a/lib/membrane/live_compositor/api_client.ex b/lib/membrane/live_compositor/api_client.ex
@@ -34,7 +34,7 @@ defmodule Membrane.LiveCompositor.ApiClient do
   @spec send_request(request(), :inet.port_number()) :: request_result()
   def send_request(request, lc_port) do
     {method, route, body} = request
-    {:ok, _} = Application.ensure_all_started(:req)
+    {:ok, _apps} = Application.ensure_all_started(:req)
 
     retry_delay_ms = fn retry_count -> retry_count * 100 end
     req = Req.new(base_url: "http://#{@local_host_url}:#{lc_port}", retry_delay: retry_delay_ms)

diff --git a/lib/membrane/live_compositor/live_compositor.ex b/lib/membrane/live_compositor/live_compositor.ex
@@ -430,7 +430,7 @@ defmodule Membrane.LiveCompositor do
 
     opt.init_requests
     |> Enum.each(fn request ->
-      {:ok, _} =
+      {:ok, _response} =
         IntoRequest.into_request(request)
         |> ApiClient.send_request(lc_port)
     end)
@@ -461,11 +461,11 @@ defmodule Membrane.LiveCompositor do
 
     links =
       bin_input(input_ref)
-      |> via_in(Pad.ref(:input, ssrc),
-        options: [payloader: RTP.H264.Payloader]
+      |> child({:rtp_h264_payloader, pad_id}, RTP.H264.Payloader)
+      |> via_in(:input,
+        options: [ssrc: ssrc, payload_type: 96, clock_rate: 90_000]
       )
-      |> child({:rtp_sender, pad_id}, RTP.SessionBin)
-      |> via_out(Pad.ref(:rtp_output, ssrc), options: [payload_type: 96])
+      |> child({:rtp_sender, pad_id}, RTP.Muxer)
       |> child({:bye_sender, pad_id}, %RtcpByeSender{ssrc: ssrc})
       |> child({:tcp_encapsulator, pad_id}, RTP.TCP.Encapsulator)
       |> child({:tcp_sink, input_ref}, %TCP.Sink{
@@ -490,11 +490,11 @@ defmodule Membrane.LiveCompositor do
 
     links =
       bin_input(input_ref)
-      |> via_in(Pad.ref(:input, ssrc),
-        options: [payloader: RTP.Opus.Payloader]
+      |> child({:rtp_opus_payloader, pad_id}, RTP.Opus.Payloader)
+      |> via_in(:input,
+        options: [ssrc: ssrc, payload_type: 97, clock_rate: 48_000]
       )
-      |> child({:rtp_sender, pad_id}, RTP.SessionBin)
-      |> via_out(Pad.ref(:rtp_output, ssrc), options: [payload_type: 97, clock_rate: 48_000])
+      |> child({:rtp_sender, pad_id}, RTP.Muxer)
       |> child({:bye_sender, pad_id}, %RtcpByeSender{ssrc: ssrc})
       |> child({:tcp_encapsulator, pad_id}, RTP.TCP.Encapsulator)
       |> child({:tcp_sink, input_ref}, %TCP.Sink{
@@ -527,9 +527,11 @@ defmodule Membrane.LiveCompositor do
           connection_side: {:client, @local_host, port}
         })
         |> child({:tcp_decapsulator, pad_id}, RTP.TCP.Decapsulator)
-        |> via_in(Pad.ref(:rtp_input, pad_id))
-        |> child({:rtp_receiver, output_ref}, RTP.SessionBin),
-        child({:output_processor, pad_id}, %Membrane.LiveCompositor.VideoOutputProcessor{
+        |> child({:rtp_receiver, output_ref}, RTP.Demuxer)
+        |> via_out(:output, options: [stream_id: {:payload_type, 96}])
+        |> child(%Membrane.RTP.JitterBuffer{latency: 0, clock_rate: 90_000})
+        |> child(RTP.H264.Depayloader)
+        |> child({:output_processor, pad_id}, %Membrane.LiveCompositor.VideoOutputProcessor{
           output_stream_format: output_stream_format
         })
         |> bin_output(Pad.ref(:video_output, pad_id))
@@ -550,9 +552,11 @@ defmodule Membrane.LiveCompositor do
         connection_side: {:client, @local_host, port}
       })
       |> child({:tcp_decapsulator, pad_id}, RTP.TCP.Decapsulator)
-      |> via_in(Pad.ref(:rtp_input, pad_id))
-      |> child({:rtp_receiver, output_ref}, RTP.SessionBin),
-      child({:output_processor, pad_id}, Membrane.LiveCompositor.AudioOutputProcessor)
+      |> child({:rtp_receiver, output_ref}, RTP.Demuxer)
+      |> via_out(:output, options: [stream_id: {:payload_type, 97}])
+      |> child(%Membrane.RTP.JitterBuffer{latency: 0, clock_rate: 48_000})
+      |> child(RTP.Opus.Depayloader)
+      |> child({:output_processor, pad_id}, Membrane.LiveCompositor.AudioOutputProcessor)
       |> bin_output(Pad.ref(:audio_output, pad_id))
     ]
 
@@ -614,46 +618,6 @@ defmodule Membrane.LiveCompositor do
     {[], state}
   end
 
-  @impl true
-  def handle_child_notification(
-        {:new_rtp_stream, ssrc, _payload_type, _extensions},
-        {:rtp_receiver, ref = Pad.ref(:video_output, pad_id)},
-        _ctx,
-        state = %State{}
-      ) do
-    links =
-      get_child({:rtp_receiver, ref})
-      |> via_out(Pad.ref(:output, ssrc),
-        options: [depayloader: nil, clock_rate: 90_000]
-      )
-      |> child(%Membrane.RTP.JitterBuffer{latency: 0, clock_rate: 90_000})
-      |> child(RTP.H264.Depayloader)
-      |> get_child({:output_processor, pad_id})
-
-    actions = [spec: {links, group: output_group_id(pad_id)}]
-
-    {actions, state}
-  end
-
-  @impl true
-  def handle_child_notification(
-        {:new_rtp_stream, ssrc, _payload_type, _extensions},
-        {:rtp_receiver, ref = Pad.ref(:audio_output, pad_id)},
-        _ctx,
-        state = %State{}
-      ) do
-    links =
-      get_child({:rtp_receiver, ref})
-      |> via_out(Pad.ref(:output, ssrc),
-        options: [depayloader: nil, clock_rate: 48_000]
-      )
-      |> child(%Membrane.RTP.JitterBuffer{latency: 0, clock_rate: 48_000})
-      |> child(RTP.Opus.Depayloader)
-      |> get_child({:output_processor, pad_id})
-
-    {[spec: {links, group: output_group_id(pad_id)}], state}
-  end
-
   @impl true
   def handle_child_notification(
         {:connection_info, _ip, _port},

diff --git a/mix.exs b/mix.exs
@@ -1,7 +1,7 @@
 defmodule Membrane.LiveCompositor.Mixfile do
   use Mix.Project
 
-  @version "0.10.0"
+  @version "0.10.1"
   @github_url "https://github.com/membraneframework/membrane_live_compositor_plugin"
 
   def project do
@@ -45,11 +45,11 @@ defmodule Membrane.LiveCompositor.Mixfile do
       {:membrane_core, "~> 1.0"},
       {:membrane_raw_video_format, "~> 0.4.0"},
       ## RTP
-      {:membrane_rtp_plugin, "~> 0.29.0"},
+      {:membrane_rtp_plugin, "~> 0.30.0"},
       {:membrane_rtp_h264_plugin, "~> 0.20.0"},
       {:membrane_tcp_plugin, "~> 0.6.0"},
       {:membrane_opus_plugin, "~> 0.20.1"},
-      {:membrane_rtp_opus_plugin, "~> 0.9.0"},
+      {:membrane_rtp_opus_plugin, "~> 0.10.0"},
       # VC server start
       {:muontrap, "~> 1.0"},
       # VC API