Allow to receive raw RTP in Sink / Allow to negotiate AV1 codecs

lib/membrane_webrtc/ex_webrtc/sink.ex

@@ -130,7 +130,8 @@ defmodule Membrane.WebRTC.ExWebRTCSink do
         :ok
 
       %{kind: :video, codec: nil} ->
-        supported_video_codecs = get_transceiver(state, pad)
+        transceiver = get_transceiver(state, pad)
+        supported_video_codecs = transceiver.codecs
 
         if length(supported_video_codecs) > 1 do
           raise """
@@ -151,8 +152,8 @@ defmodule Membrane.WebRTC.ExWebRTCSink do
     {[], state}
   end
 
-  defp set_pc_sender_video_codec(state, pad, codec) when codec in [:vp8, :h264] do
-    mime_type = if codec == :vp8, do: "video/VP8", else: "video/H264"
+  defp set_pc_sender_video_codec(state, pad, codec) when codec in [:vp8, :h264, :av1] do
+    mime_type = mime_type_from_codec(codec)
     transceiver = get_transceiver(state, pad)
 
     if transceiver == nil do
@@ -371,4 +372,8 @@ 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 mime_type_from_codec(:h264), do: "video/H264"
+  defp mime_type_from_codec(:vp8), do: "video/VP8"
+  defp mime_type_from_codec(:av1), do: "video/AV1"
 end

lib/membrane_webrtc/ex_webrtc/utils.ex

@@ -43,6 +43,22 @@ defmodule Membrane.WebRTC.ExWebRTCUtils do
     ]
   end
 
+  def codec_params(:av1) do
+    [
+      %RTPCodecParameters{
+        payload_type: 45,
+        mime_type: "video/AV1",
+        clock_rate: codec_clock_rate(:av1),
+        sdp_fmtp_line: %ExSDP.Attribute.FMTP{
+          pt: 45,
+          profile: 0,
+          level_idx: 5,
+          tier: 0
+        }
+      }
+    ]
+  end
+
   def codec_params(codecs) when is_list(codecs) do
     codecs |> Enum.flat_map(&codec_params/1)
   end
@@ -51,6 +67,7 @@ defmodule Membrane.WebRTC.ExWebRTCUtils do
   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(:av1), do: 90_000
 
   def codec_clock_rate(codecs) when is_list(codecs) do
     cond do
@@ -62,7 +79,7 @@ defmodule Membrane.WebRTC.ExWebRTCUtils do
     end
   end
 
-  @spec get_video_codecs_from_sdp(ExWebRTC.SessionDescription.t()) :: [:h264 | :vp8]
+  @spec get_video_codecs_from_sdp(ExWebRTC.SessionDescription.t()) :: [:h264 | :vp8 | :av1]
   def get_video_codecs_from_sdp(%ExWebRTC.SessionDescription{sdp: sdp}) do
     ex_sdp = ExSDP.parse!(sdp)
 
@@ -74,6 +91,7 @@ defmodule Membrane.WebRTC.ExWebRTCUtils do
     |> Enum.flat_map(fn
       %ExSDP.Attribute.RTPMapping{encoding: "H264"} -> [:h264]
       %ExSDP.Attribute.RTPMapping{encoding: "VP8"} -> [:vp8]
+      %ExSDP.Attribute.RTPMapping{encoding: "AV1"} -> [:av1]
       _attribute -> []
     end)
     |> Enum.uniq()

lib/membrane_webrtc/sink.ex

@@ -68,7 +68,7 @@ defmodule Membrane.WebRTC.Sink do
                 """
               ],
               video_codec: [
-                spec: :vp8 | :h264 | [:vp8 | :h264],
+                spec: :vp8 | :h264 | :av1 | [:vp8 | :h264 | :av1],
                 default: [:vp8, :h264],
                 description: """
                 Video codecs, that #{inspect(__MODULE__)} will try to negotiatie in SDP
@@ -103,6 +103,7 @@ defmodule Membrane.WebRTC.Sink do
       any_of(
         %Membrane.H264{alignment: :nalu},
         %Membrane.RemoteStream{content_format: Membrane.VP8},
+        %Membrane.RemoteStream{content_format: Membrane.RTP},
         Membrane.VP8,
         Membrane.Opus,
         Membrane.RTP
@@ -146,8 +147,10 @@ defmodule Membrane.WebRTC.Sink do
     spec =
       cond do
         not state.payload_rtp ->
+          codec = if kind == :video, do: ensure_single_video_codec(state.video_codec), else: :opus
+
           bin_input(pad_ref)
-          |> via_in(pad_ref, options: [kind: kind])
+          |> via_in(pad_ref, options: [kind: kind, codec: codec])
           |> get_child(:webrtc)
 
         kind == :audio ->
@@ -175,7 +178,6 @@ defmodule Membrane.WebRTC.Sink do
     {[], state}
   end
 
-  @impl true
   def handle_child_notification(
         {:stream_format, _connector_pad, stream_format},
         {:connector, pad_ref},
@@ -184,9 +186,21 @@ defmodule Membrane.WebRTC.Sink do
       ) do
     codec =
       case stream_format do
-        %H264{} -> :h264
-        %VP8{} -> :vp8
-        %RemoteStream{content_format: VP8} -> :vp8
+        %H264{} ->
+          :h264
+
+        %VP8{} ->
+          :vp8
+
+        %RemoteStream{content_format: VP8} ->
+          :vp8
+
+        other ->
+          raise """
+          Unsupported stream format for payloading: #{inspect(other)}
+          If you're sending raw RTP or using a codec without a built-in payloader (like AV1),
+          set `payload_rtp: false` in the Sink options.
+          """
       end
 
     payloader =
@@ -231,4 +245,8 @@ defmodule Membrane.WebRTC.Sink do
   end
 
   def default_ice_ip_filter(_ip), do: true
+
+  defp ensure_single_video_codec(codec) when is_atom(codec), do: codec
+  defp ensure_single_video_codec([codec]), do: codec
+  defp ensure_single_video_codec(_codec), do: nil
 end

lib/membrane_webrtc/source.ex

@@ -64,7 +64,7 @@ defmodule Membrane.WebRTC.Source do
                 """
               ],
               allowed_video_codecs: [
-                spec: :vp8 | :h264 | [:vp8 | :h264],
+                spec: :vp8 | :h264 | :av1 | [:vp8 | :h264 | :av1],
                 default: :vp8,
                 description: """
                 Specifies, which video codecs can be accepted by the source during the SDP
@@ -260,6 +260,14 @@ defmodule Membrane.WebRTC.Source do
 
       state.negotiated_video_codecs == nil ->
         raise "Cannot select depayloader before end of SDP messages exchange"
+
+      true ->
+        raise """
+        Unsupported video codec for depayloading: #{inspect(state.negotiated_video_codecs)}
+
+        If you're receiving raw RTP or using a codec without a built-in depayloader (like AV1),
+        set `depayload_rtp: false` in the Source options.
+        """
     end
   end
 

test/membrane_webrtc/integration_test.exs

@@ -69,7 +69,7 @@ defmodule Membrane.WebRTC.IntegrationTest do
   defmodule Utils do
     import ExUnit.Assertions
 
-    def fixture_processing_timeout, do: 30_000
+    def fixture_processing_timeout(), do: 30_000
 
     def prepare_input(pipeline, opts) do
       demuxer_name = {:demuxer, make_ref()}
@@ -504,4 +504,58 @@ defmodule Membrane.WebRTC.IntegrationTest do
       end)
     end
   end
+
+  defmodule AV1RawRTPPassthrough do
+    use ExUnit.Case, async: true
+
+    import Membrane.ChildrenSpec
+    import Membrane.Testing.Assertions
+
+    require Membrane.Pad, as: Pad
+
+    alias Membrane.Testing
+    alias Membrane.WebRTC
+    alias Membrane.WebRTC.Signaling
+
+    test "send and receive AV1 via raw RTP passthrough" do
+      signaling = Signaling.new()
+
+      send_pipeline = Testing.Pipeline.start_link_supervised!()
+      receive_pipeline = Testing.Pipeline.start_link_supervised!()
+
+      Testing.Pipeline.execute_actions(send_pipeline,
+        spec: [
+          child(:video_source, %KeyframeTestSource{
+            stream_format: %Membrane.RTP{}
+          })
+          |> via_in(:input, options: [kind: :video])
+          |> child(:webrtc, %WebRTC.Sink{
+            signaling: signaling,
+            payload_rtp: false,
+            video_codec: :av1,
+            tracks: [:video]
+          })
+        ]
+      )
+
+      Testing.Pipeline.execute_actions(receive_pipeline,
+        spec: [
+          child(:webrtc, %WebRTC.Source{
+            signaling: signaling,
+            depayload_rtp: false,
+            allowed_video_codecs: :av1
+          })
+          |> via_out(Pad.ref(:output, :video), options: [kind: :video])
+          |> child(:video_sink, KeyframeTestSink)
+        ]
+      )
+
+      assert_pipeline_notified(send_pipeline, :webrtc, {:negotiated_video_codecs, [:av1]})
+      assert_pipeline_notified(receive_pipeline, :webrtc, {:negotiated_video_codecs, [:av1]})
+      assert_pipeline_notified(receive_pipeline, :video_sink, {:buffer, _buffer}, 5_000)
+
+      Testing.Pipeline.terminate(send_pipeline)
+      Testing.Pipeline.terminate(receive_pipeline)
+    end
+  end
 end