Set payload type of egress RTP packets WiP

Author: FelonEkonom

lib/membrane_webrtc/ex_webrtc/sink.ex

@@ -21,7 +21,7 @@ defmodule Membrane.WebRTC.ExWebRTCSink do
   def_input_pad :input,
     accepted_format: Membrane.RTP,
     availability: :on_request,
-    options: [kind: []]
+    options: [kind: [], payload_type: [default: nil]]
 
   @max_rtp_timestamp 2 ** 32 - 1
   @max_rtp_seq_num 2 ** 16 - 1
@@ -115,8 +115,8 @@ defmodule Membrane.WebRTC.ExWebRTCSink do
   end
 
   @impl true
-  def handle_buffer(pad, buffer, _ctx, state) do
-    state = send_buffer(pad, buffer, state)
+  def handle_buffer(pad, buffer, ctx, state) do
+    state = send_buffer(pad, buffer, ctx, state)
     {[], state}
   end
 
@@ -206,11 +206,12 @@ defmodule Membrane.WebRTC.ExWebRTCSink do
         state
       ) do
     Membrane.Logger.debug("Received SDP answer")
+    IO.puts("SDP ANSWER:  \n" <> sdp.sdp <> "\n")
     :ok = PeerConnection.set_remote_description(state.pc, sdp)
 
     %{negotiating_tracks: negotiating_tracks, negotiated_tracks: negotiated_tracks} = state
 
-    video_codecs = ExWebRTCUtils.get_video_codecs_from_sdp(sdp)
+    video_codecs = ExWebRTCUtils.get_video_codecs_from_sdp(sdp) |> IO.inspect(label: "VIDEO CODECS FROM SDP")
 
     to_notify =
       negotiating_tracks |> Enum.filter(& &1.notify) |> Enum.map(&Map.take(&1, [:id, :kind]))
@@ -266,12 +267,13 @@ defmodule Membrane.WebRTC.ExWebRTCSink do
     |> Enum.each(&PeerConnection.set_transceiver_direction(pc, &1.id, :sendonly))
 
     {:ok, offer} = PeerConnection.create_offer(pc)
+    # IO.puts("SDP OFFER:  \n" <> offer.sdp <> "\n")
     :ok = PeerConnection.set_local_description(pc, offer)
     SignalingChannel.signal(state.signaling, offer)
     %{state | negotiating_tracks: negotiating_tracks, queued_tracks: []}
   end
 
-  defp send_buffer(pad, buffer, state) do
+  defp send_buffer(pad, buffer, ctx, state) do
     {id, params} = state.input_tracks[pad]
 
     timestamp =
@@ -288,6 +290,12 @@ defmodule Membrane.WebRTC.ExWebRTCSink do
         sequence_number: params.seq_num
       )
 
+    packet =
+      case ctx.pads[pad].options.payload_type do
+        int when is_integer(int) -> %{packet | payload_type: int}
+        nil -> packet
+      end
+
     PeerConnection.send_rtp(state.pc, id, packet)
     seq_num = rem(params.seq_num + 1, @max_rtp_seq_num + 1)
     put_in(state.input_tracks[pad], {id, %{params | seq_num: seq_num}})

lib/membrane_webrtc/ex_webrtc/utils.ex

@@ -6,25 +6,32 @@ defmodule Membrane.WebRTC.ExWebRTCUtils do
   @type codec :: :opus | :h264 | :vp8
   @type codec_or_codecs :: codec() | [codec()]
 
+  @spec codec_payload_type(codec()) :: pos_integer()
+  def codec_payload_type(:opus), do: 111
+  def codec_payload_type(:h264), do: 96
+  def codec_payload_type(:vp8), do: 102
+
   @spec codec_params(codec_or_codecs()) :: [RTPCodecParameters.t()]
   def codec_params(:opus),
     do: [
       %RTPCodecParameters{
-        payload_type: 111,
+        payload_type: codec_payload_type(:opus),
         mime_type: "audio/opus",
         clock_rate: codec_clock_rate(:opus),
         channels: 2
       }
     ]
 
   def codec_params(:h264) do
+    payload_type = codec_payload_type(:h264)
+
     [
       %RTPCodecParameters{
-        payload_type: 96,
+        payload_type: payload_type,
         mime_type: "video/H264",
         clock_rate: codec_clock_rate(:h264),
         sdp_fmtp_line: %ExSDP.Attribute.FMTP{
-          pt: 96,
+          pt: payload_type,
           level_asymmetry_allowed: 1,
           packetization_mode: 1,
           profile_level_id: 0x42E01F
@@ -36,7 +43,7 @@ defmodule Membrane.WebRTC.ExWebRTCUtils do
   def codec_params(:vp8) do
     [
       %RTPCodecParameters{
-        payload_type: 102,
+        payload_type: codec_payload_type(:vp8),
         mime_type: "video/VP8",
         clock_rate: codec_clock_rate(:vp8)
       }

lib/membrane_webrtc/sink.ex

@@ -16,11 +16,14 @@ defmodule Membrane.WebRTC.Sink do
   the track received in `t:new_tracks/0` notification.
   """
   use Membrane.Bin
+  require Membrane.Logger
 
   alias Membrane.H264
   alias Membrane.RemoteStream
   alias Membrane.VP8
-  alias Membrane.WebRTC.{ExWebRTCSink, ForwardingFilter, SignalingChannel, SimpleWebSocketServer}
+  alias Membrane.WebRTC.{ExWebRTCSink, ExWebRTCUtils, ForwardingFilter, SignalingChannel, SimpleWebSocketServer}
+
+  # alias __MODULE__.PayloadTypeSetter
 
   @typedoc """
   Notification that should be sent to the bin to negotiate new tracks.
@@ -132,7 +135,7 @@ defmodule Membrane.WebRTC.Sink do
   end
 
   @impl true
-  def handle_pad_added(Pad.ref(:input, pid) = pad_ref, %{pad_options: %{kind: kind}}, state) do
+  def handle_pad_added(Pad.ref(:input, id) = pad_ref, %{pad_options: %{kind: kind}}, state) do
     spec =
       cond do
         not state.payload_rtp ->
@@ -141,9 +144,11 @@ defmodule Membrane.WebRTC.Sink do
           |> get_child(:webrtc)
 
         kind == :audio ->
+          payload_type = ExWebRTCUtils.codec_payload_type(:opus)
+
           bin_input(pad_ref)
-          |> child({:rtp_opus_payloader, pid}, Membrane.RTP.Opus.Payloader)
-          |> via_in(pad_ref, options: [kind: :audio])
+          |> child({:rtp_opus_payloader, id}, Membrane.RTP.Opus.Payloader)
+          |> via_in(pad_ref, options: [kind: :audio, payload_type: payload_type])
           |> get_child(:webrtc)
 
         kind == :video ->
@@ -161,17 +166,19 @@ defmodule Membrane.WebRTC.Sink do
         _ctx,
         state
       ) do
-    payoader =
+    {payloader, codec} =
       case stream_format do
-        %H264{} -> %Membrane.RTP.H264.Payloader{max_payload_size: 1000}
-        %VP8{} -> Membrane.RTP.VP8.Payloader
-        %RemoteStream{content_format: VP8} -> Membrane.RTP.VP8.Payloader
+        %H264{} ->{%Membrane.RTP.H264.Payloader{max_payload_size: 1000}, :h264}
+        %VP8{} -> {Membrane.RTP.VP8.Payloader, :vp8}
+        %RemoteStream{content_format: VP8} -> {Membrane.RTP.VP8.Payloader, :vp8}
       end
 
+    payload_type = ExWebRTCUtils.codec_payload_type(codec)
+
     spec =
       get_child({:forwarding_filter, pad_ref})
-      |> child({:rtp_payloader, pad_ref}, payoader)
-      |> via_in(pad_ref, options: [kind: :video])
+      |> child({:rtp_payloader, pad_ref}, payloader)
+      |> via_in(pad_ref, options: [kind: :video, payload_type: payload_type])
       |> get_child(:webrtc)
 
     {[spec: spec], state}
@@ -183,9 +190,24 @@ defmodule Membrane.WebRTC.Sink do
   end
 
   @impl true
-  def handle_child_notification({type, _content} = notification, :webrtc, _ctx, state)
-      when type in [:new_tracks, :negotiated_video_codecs] do
-    {[notify_parent: notification], state}
+  def handle_child_notification({:negotiated_video_codecs, codecs}, :webrtc, _ctx, state) do
+    if length(codecs) > 2 and state.payload_rtp do
+      Membrane.Logger.warning("""
+      Negotiated more then one video codec (#{inspect(codecs)}), while this bin will receive `Membrane.RTP` \
+      packets on the input, with no information about codecs used to encode video streams. \
+      Lack of this information might lead to setting inapropriate payload type of RTP packets and further \
+      errors.
+
+      You can fix this problem by setting `:video_codec` option to `:vp8` or to `:h264`.
+      """)
+    end
+
+    {[notify_parent: {:negotiated_video_codecs, codecs}], state}
+  end
+
+  @impl true
+  def handle_child_notification({:new_tracks, new_tracks}, :webrtc, _ctx, state) do
+    {[notify_parent: {:new_tracks, new_tracks}], state}
   end
 
   @impl true