Negotiate many video codecs

README.md

@@ -15,7 +15,7 @@ The package can be installed by adding `membrane_webrtc_plugin` to your list of
 ```elixir
 def deps do
   [
-    {:membrane_webrtc_plugin, "~> 0.22.0"}
+    {:membrane_webrtc_plugin, "~> 0.22.1"}
   ]
 end
 ```

examples/file_to_browser.exs

@@ -41,7 +41,10 @@ defmodule Example.Pipeline do
 
     if state.audio_track && state.video_track do
       spec = [
-        child(:webrtc, %WebRTC.Sink{signaling: {:websocket, port: state.port}}),
+        child(:webrtc, %WebRTC.Sink{
+          signaling: {:websocket, port: state.port},
+          video_codec: [:vp8, :h264]
+        }),
         get_child(:demuxer)
         |> via_out(Pad.ref(:output, state.video_track))
         |> child({:realtimer, :video_track}, Membrane.Realtimer)

lib/membrane_webrtc/ex_webrtc/sink.ex

@@ -191,7 +191,12 @@ defmodule Membrane.WebRTC.ExWebRTCSink do
     to_notify =
       negotiating_tracks |> Enum.filter(& &1.notify) |> Enum.map(&Map.take(&1, [:id, :kind]))
 
-    actions = if to_notify == [], do: [], else: [notify_parent: {:new_tracks, to_notify}]
+    new_tracks_notification =
+      if to_notify == [], do: [], else: [notify_parent: {:new_tracks, to_notify}]
+
+    actions =
+      new_tracks_notification ++
+        [notify_parent: {:negotiated_video_codecs, get_negotiated_video_codecs(sdp)}]
 
     negotiated_tracks = negotiated_tracks ++ negotiating_tracks
 
@@ -265,4 +270,20 @@ defmodule Membrane.WebRTC.ExWebRTCSink do
     seq_num = rem(params.seq_num + 1, @max_rtp_seq_num + 1)
     put_in(state.input_tracks[pad], {id, %{params | seq_num: seq_num}})
   end
+
+  defp get_negotiated_video_codecs(sdp_answer) do
+    ex_sdp = ExSDP.parse!(sdp_answer.sdp)
+
+    ex_sdp.media
+    |> Enum.flat_map(fn
+      %{type: :video, attributes: attributes} -> attributes
+      _media -> []
+    end)
+    |> Enum.flat_map(fn
+      %ExSDP.Attribute.RTPMapping{encoding: "H264"} -> [:h264]
+      %ExSDP.Attribute.RTPMapping{encoding: "VP8"} -> [:vp8]
+      _attribute -> []
+    end)
+    |> Enum.uniq()
+  end
 end

lib/membrane_webrtc/ex_webrtc/utils.ex

@@ -3,7 +3,10 @@ defmodule Membrane.WebRTC.ExWebRTCUtils do
 
   alias ExWebRTC.RTPCodecParameters
 
-  @spec codec_params(:opus | :h264 | :vp8) :: [RTPCodecParameters.t()]
+  @type codec :: :opus | :h264 | :vp8
+  @type codec_or_codecs :: codec() | [codec()]
+
+  @spec codec_params(codec_or_codecs()) :: [RTPCodecParameters.t()]
   def codec_params(:opus),
     do: [
       %RTPCodecParameters{
@@ -40,8 +43,22 @@ defmodule Membrane.WebRTC.ExWebRTCUtils do
     ]
   end
 
-  @spec codec_clock_rate(:opus | :h264 | :vp8) :: pos_integer()
+  def codec_params(codecs) when is_list(codecs) do
+    codecs |> Enum.flat_map(&codec_params/1)
+  end
+
+  @spec codec_clock_rate(codec_or_codecs()) :: pos_integer()
   def codec_clock_rate(:opus), do: 48_000
   def codec_clock_rate(:vp8), do: 90_000
   def codec_clock_rate(:h264), do: 90_000
+
+  def codec_clock_rate(codecs) when is_list(codecs) do
+    cond do
+      codecs == [:opus] ->
+        48_000
+
+      codecs != [] and Enum.all?(codecs, &(&1 in [:vp8, :h264])) ->
+        90_000
+    end
+  end
 end

lib/membrane_webrtc/sink.ex

@@ -17,6 +17,7 @@ defmodule Membrane.WebRTC.Sink do
   """
   use Membrane.Bin
 
+  alias __MODULE__.VideoDispatcher
   alias Membrane.WebRTC.{ExWebRTCSink, SignalingChannel, SimpleWebSocketServer}
 
   @typedoc """
@@ -53,8 +54,18 @@ defmodule Membrane.WebRTC.Sink do
                 """
               ],
               video_codec: [
-                spec: :vp8 | :h264,
-                default: :vp8
+                spec: :vp8 | :h264 | [:vp8 | :h264],
+                default: [:vp8, :h264],
+                description: """
+                Video codecs, that #{inspect(__MODULE__)} will try to negotiatie in SDP
+                message exchange. Even if `[:vp8, :h264]` is passed to this option, there
+                is a chance, that one of these codecs won't be approved by the second
+                WebRTC peer.
+
+                After SDP messages exchange, #{inspect(__MODULE__)} will send a parent
+                notification `{:negotiated_video_codecs, codecs}`, where codecs is
+                the list of the video codecs, that might be received by this component.
+                """
               ],
               ice_servers: [
                 spec: [ExWebRTC.PeerConnection.Configuration.ice_server()],
@@ -104,14 +115,37 @@ defmodule Membrane.WebRTC.Sink do
   end
 
   @impl true
-  def handle_pad_added(Pad.ref(:input, _pid) = pad_ref, ctx, state) do
-    %{kind: kind} = ctx.pad_options
-
+  def handle_pad_added(Pad.ref(:input, pid) = pad_ref, %{pad_options: %{kind: kind}}, state) do
     spec =
-      bin_input(pad_ref)
-      |> then(if state.payload_rtp, do: &child(&1, get_payloader(kind, state)), else: & &1)
-      |> via_in(pad_ref, options: [kind: kind])
-      |> get_child(:webrtc)
+      cond do
+        not state.payload_rtp ->
+          bin_input(pad_ref)
+          |> via_in(pad_ref, options: [kind: kind])
+          |> get_child(:webrtc)
+
+        kind == :audio ->
+          bin_input(pad_ref)
+          |> child({:rtp_opus_payloader, pid}, Membrane.RTP.Opus.Payloader)
+          |> via_in(pad_ref, options: [kind: :audio])
+          |> get_child(:webrtc)
+
+        kind == :video ->
+          [
+            bin_input(pad_ref)
+            |> child({:video_dispatcher, pid}, VideoDispatcher)
+            |> via_out(:h264_output)
+            |> child({:rtp_h264_payloader, pid}, %Membrane.RTP.H264.Payloader{
+              max_payload_size: 1000
+            })
+            |> child({:funnel, pid}, %Membrane.Funnel{end_of_stream: :on_last_pad})
+            |> via_in(pad_ref, options: [kind: :video])
+            |> get_child(:webrtc),
+            get_child({:video_dispatcher, pid})
+            |> via_out(:vp8_output)
+            |> child({:rtp_vp8_payloader, pid}, Membrane.RTP.VP8.Payloader)
+            |> get_child({:funnel, pid})
+          ]
+      end
 
     {[spec: spec], state}
   end
@@ -126,6 +160,11 @@ defmodule Membrane.WebRTC.Sink do
     {[notify_parent: {:new_tracks, tracks}], state}
   end
 
+  @impl true
+  def handle_child_notification({:negotiated_video_codecs, codecs}, :webrtc, _ctx, state) do
+    {[notify_parent: {:negotiated_video_codecs, codecs}], state}
+  end
+
   @impl true
   def handle_parent_notification({:add_tracks, tracks}, _ctx, state) do
     {[notify_child: {:webrtc, {:add_tracks, tracks}}], state}
@@ -140,11 +179,4 @@ defmodule Membrane.WebRTC.Sink do
   def handle_element_end_of_stream(_name, _pad, _ctx, state) do
     {[], state}
   end
-
-  defp get_payloader(:audio, _state), do: Membrane.RTP.Opus.Payloader
-
-  defp get_payloader(:video, %{video_codec: :h264}),
-    do: %Membrane.RTP.H264.Payloader{max_payload_size: 1000}
-
-  defp get_payloader(:video, %{video_codec: :vp8}), do: Membrane.RTP.VP8.Payloader
 end

lib/membrane_webrtc/sink/video_dispatcher.ex

@@ -0,0 +1,64 @@
+defmodule Membrane.WebRTC.Sink.VideoDispatcher do
+  @moduledoc false
+  use Membrane.Filter
+
+  alias Membrane.{H264, RemoteStream, VP8}
+
+  def_input_pad :input, accepted_format: any_of(H264, VP8, %RemoteStream{content_format: VP8})
+  def_output_pad :h264_output, accepted_format: H264
+  def_output_pad :vp8_output, accepted_format: any_of(VP8, %RemoteStream{content_format: VP8})
+
+  @impl true
+  def handle_init(ctx, _opts) do
+    buffered_events = Map.keys(ctx.pads) |> Map.new(&{&1, []})
+    {[], %{selected_output_pad: nil, buffered_events: buffered_events}}
+  end
+
+  @impl true
+  def handle_stream_format(:input, stream_format, _ctx, state) do
+    selected_output_pad =
+      case stream_format do
+        %H264{} -> :h264_output
+        %VP8{} -> :vp8_output
+        %RemoteStream{content_format: VP8} -> :vp8_output
+      end
+
+    event_actions =
+      state.buffered_events
+      |> Enum.flat_map(fn
+        {_pad, []} -> []
+        {:input, events} -> [event: {selected_output_pad, Enum.reverse(events)}]
+        {^selected_output_pad, events} -> [event: {:input, Enum.reverse(events)}]
+        {_ignored_output_pad, _events} -> []
+      end)
+
+    state = %{state | selected_output_pad: selected_output_pad, buffered_events: %{}}
+    actions = event_actions ++ [stream_format: {selected_output_pad, stream_format}]
+    {actions, state}
+  end
+
+  @impl true
+  def handle_buffer(:input, buffer, _ctx, state) do
+    {[buffer: {state.selected_output_pad, buffer}], state}
+  end
+
+  @impl true
+  def handle_end_of_stream(:input, _ctx, state) do
+    {[end_of_stream: :h264_output, end_of_stream: :vp8_output], state}
+  end
+
+  @impl true
+  def handle_event(pad, event, _ctx, %{selected_output_pad: nil} = state) do
+    state = update_in(state, [:buffered_events, pad], &[event | &1])
+    {[], state}
+  end
+
+  @impl true
+  def handle_event(pad, event, _ctx, %{selected_output_pad: selected_output_pad} = state) do
+    case pad do
+      :input -> {[event: {selected_output_pad, event}], state}
+      ^selected_output_pad -> {[event: {:input, event}], state}
+      _ignored_output_pad -> {[], state}
+    end
+  end
+end

mix.exs

@@ -1,7 +1,7 @@
 defmodule Membrane.WebRTC.Plugin.Mixfile do
   use Mix.Project
 
-  @version "0.22.0"
+  @version "0.22.1"
   @github_url "https://github.com/membraneframework/membrane_webrtc_plugin"
 
   def project do

test/membrane_webrtc/integration_test.exs

@@ -188,11 +188,16 @@ defmodule Membrane.WebRTC.IntegrationTest do
 
     import Utils
 
+    @tag :dupa
     @tag :tmp_dir
     test "send and receive a file", %{tmp_dir: tmp_dir} do
       signaling = SignalingChannel.new()
       send_pipeline = Testing.Pipeline.start_link_supervised!()
-      prepare_input(send_pipeline, webrtc: %WebRTC.Sink{signaling: signaling})
+
+      prepare_input(send_pipeline,
+        webrtc: %WebRTC.Sink{signaling: signaling, video_codec: [:vp8, :h264]}
+      )
+
       receive_pipeline = Testing.Pipeline.start_link_supervised!()
 
       prepare_output(receive_pipeline, tmp_dir, webrtc: %WebRTC.Source{signaling: signaling})

test/test_helper.exs

@@ -1 +1,2 @@
 ExUnit.start(capture_log: true)
+# ExUnit.start()