Add outbound RTX (#101)

Author: LVala

examples/echo/lib/echo/peer_handler.ex

@@ -19,13 +19,14 @@ defmodule Echo.PeerHandler do
     %RTPCodecParameters{
       payload_type: 96,
       mime_type: "video/VP8",
-      clock_rate: 90_000
+      clock_rate: 90_000,
+      rtcp_fbs: [%ExSDP.Attribute.RTCPFeedback{pt: 96, feedback_type: :nack}]
     },
     %RTPCodecParameters{
       payload_type: 97,
       mime_type: "video/rtx",
       clock_rate: 90_000,
-      sdp_fmtp_line: %{pt: 97, apt: 96}
+      sdp_fmtp_line: %ExSDP.Attribute.FMTP{pt: 97, apt: 96}
     }
   ]
 
@@ -35,12 +36,6 @@ defmodule Echo.PeerHandler do
       mime_type: "audio/opus",
       clock_rate: 48_000,
       channels: 2
-    },
-    %RTPCodecParameters{
-      payload_type: 112,
-      mime_type: "audio/rtx",
-      clock_rate: 48_000,
-      sdp_fmtp_line: %{pt: 112, apt: 111}
     }
   ]
 

examples/echo/mix.lock

@@ -2,18 +2,19 @@
   "bandit": {:hex, :bandit, "1.2.3", "a98d664a96fec23b68e776062296d76a94b4459795b38209f4ae89cb4225709c", [:mix], [{:hpax, "~> 0.1.1", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "3e29150245a9b5f56944434e5240966e75c917dad248f689ab589b32187a81af"},
   "bunch": {:hex, :bunch, "1.6.1", "5393d827a64d5f846092703441ea50e65bc09f37fd8e320878f13e63d410aec7", [:mix], [], "hexpm", "286cc3add551628b30605efbe2fca4e38cc1bea89bcd0a1a7226920b3364fe4a"},
   "bunch_native": {:hex, :bunch_native, "0.5.0", "8ac1536789a597599c10b652e0b526d8833348c19e4739a0759a2bedfd924e63", [:mix], [{:bundlex, "~> 1.0", [hex: :bundlex, repo: "hexpm", optional: false]}], "hexpm", "24190c760e32b23b36edeb2dc4852515c7c5b3b8675b1a864e0715bdd1c8f80d"},
-  "bundlex": {:hex, :bundlex, "1.5.0", "295472109d6c8a2fb59712523f20f9502105e3306edc3b490aa1cc4df3f9c5cf", [:mix], [{:bunch, "~> 1.0", [hex: :bunch, repo: "hexpm", optional: false]}, {:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}, {:qex, "~> 0.5", [hex: :qex, repo: "hexpm", optional: false]}, {:req, "~> 0.4.0", [hex: :req, repo: "hexpm", optional: false]}, {:zarex, "~> 1.0", [hex: :zarex, repo: "hexpm", optional: false]}], "hexpm", "e5841de5380b9e99eef347656a71f3d6fab7ff34dee41da3e70c256d39cfdf77"},
+  "bundlex": {:hex, :bundlex, "1.5.1", "a85890a9d0a70366afa538c8589a4ba75e1319d32a771e1f5f3b7566beea9c26", [:mix], [{:bunch, "~> 1.0", [hex: :bunch, repo: "hexpm", optional: false]}, {:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}, {:qex, "~> 0.5", [hex: :qex, repo: "hexpm", optional: false]}, {:req, "~> 0.4.0", [hex: :req, repo: "hexpm", optional: false]}, {:zarex, "~> 1.0", [hex: :zarex, repo: "hexpm", optional: false]}], "hexpm", "aae447d63230fe1f3b788c429ac02bc696f30163d0f23f52fcfe6ed38372c7ea"},
   "castore": {:hex, :castore, "1.0.6", "ffc42f110ebfdafab0ea159cd43d31365fa0af0ce4a02ecebf1707ae619ee727", [:mix], [], "hexpm", "374c6e7ca752296be3d6780a6d5b922854ffcc74123da90f2f328996b962d33a"},
   "crc": {:hex, :crc, "0.10.5", "ee12a7c056ac498ef2ea985ecdc9fa53c1bfb4e53a484d9f17ff94803707dfd8", [:mix, :rebar3], [{:elixir_make, "~> 0.6", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "3e673b6495a9525c5c641585af1accba59a1eb33de697bedf341e247012c2c7f"},
   "elixir_make": {:hex, :elixir_make, "0.8.3", "d38d7ee1578d722d89b4d452a3e36bcfdc644c618f0d063b874661876e708683", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:certifi, "~> 2.0", [hex: :certifi, repo: "hexpm", optional: true]}], "hexpm", "5c99a18571a756d4af7a4d89ca75c28ac899e6103af6f223982f09ce44942cc9"},
   "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"},
   "ex_dtls": {:hex, :ex_dtls, "0.15.1", "34b3600ff13eebf6c96be033005cc110ea5beef98394631365ec26b493df80c5", [:mix], [{:unifex, "~> 1.0", [hex: :unifex, repo: "hexpm", optional: false]}], "hexpm", "9cfebdfe9111c0f68c77667cb9366e4a6f17e8a240975ffd100148de57478a29"},
-  "ex_ice": {:git, "https://github.com/elixir-webrtc/ex_ice.git", "1b57c79effb762ea71df79390605a52065b60821", []},
+  "ex_ice": {:git, "https://github.com/elixir-webrtc/ex_ice.git", "2d762caec5a00bdc99fb8df5051c0ff42b93c33f", []},
   "ex_libsrtp": {:hex, :ex_libsrtp, "0.7.2", "211bd89c08026943ce71f3e2c0231795b99cee748808ed3ae7b97cd8d2450b6b", [:mix], [{:bunch, "~> 1.6", [hex: :bunch, repo: "hexpm", optional: false]}, {:bundlex, "~> 1.3", [hex: :bundlex, repo: "hexpm", optional: false]}, {:membrane_precompiled_dependency_provider, "~> 0.1.0", [hex: :membrane_precompiled_dependency_provider, repo: "hexpm", optional: false]}, {:unifex, "~> 1.1", [hex: :unifex, repo: "hexpm", optional: false]}], "hexpm", "2e20645d0d739a4ecdcf8d4810a0c198120c8a2f617f2b75b2e2e704d59f492a"},
-  "ex_rtcp": {:git, "https://github.com/elixir-webrtc/ex_rtcp.git", "67fe8320302fa240e190eb97b95bbabf6bebc2fd", []},
+  "ex_rtcp": {:git, "https://github.com/elixir-webrtc/ex_rtcp.git", "323873e4d9be018bd742c633d3feed0db40625e8", []},
   "ex_rtp": {:git, "https://github.com/elixir-webrtc/ex_rtp.git", "eb9b49d8e53ba92bcd31e7d015473ae10a26707d", []},
   "ex_sdp": {:hex, :ex_sdp, "0.15.0", "53815fb5b5e4fae0f3b26de90f372446bb8e0eed62a3cc20394d3c29519698be", [:mix], [{:bunch, "~> 1.3", [hex: :bunch, repo: "hexpm", optional: false]}, {:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}], "hexpm", "d3f23596b73e7057521ff0f0d55b1189c6320a2f04388aa3a80a0aa97ffb379f"},
-  "ex_stun": {:hex, :ex_stun, "0.1.0", "252474bf4c8519fbf4bc0fbfc6a1b846a634b1478c65dbbfb4b6ab4e33c2a95a", [:mix], [], "hexpm", "629fc8be45b624a92522f81d85ba001877b1f0745889a2419bdb678790d7480c"},
+  "ex_stun": {:hex, :ex_stun, "0.2.0", "feb1fc7db0356406655b2a617805e6c712b93308c8ea2bf0ba1197b1f0866deb", [:mix], [], "hexpm", "1e01ba8290082ccbf37acaa5190d1f69b51edd6de2026a8d6d51368b29d115d0"},
+  "ex_turn": {:hex, :ex_turn, "0.1.0", "177405aadf3d754567d0d37cf881a83f9cacf8f45314d188633b04c4a9e7c1ec", [:mix], [{:ex_stun, "~> 0.2.0", [hex: :ex_stun, repo: "hexpm", optional: false]}], "hexpm", "d677737fb7d45274d5dac19fe3c26b9038b6effbc0a6b3e7417bccc76b6d1cd3"},
   "finch": {:hex, :finch, "0.18.0", "944ac7d34d0bd2ac8998f79f7a811b21d87d911e77a786bc5810adb75632ada4", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.3", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 0.2.6 or ~> 1.0", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "69f5045b042e531e53edc2574f15e25e735b522c37e2ddb766e15b979e03aa65"},
   "hpax": {:hex, :hpax, "0.1.2", "09a75600d9d8bbd064cdd741f21fc06fc1f4cf3d0fcc335e5aa19be1a7235c84", [:mix], [], "hexpm", "2c87843d5a23f5f16748ebe77969880e29809580efdaccd615cd3bed628a8c13"},
   "jason": {:hex, :jason, "1.4.1", "af1504e35f629ddcdd6addb3513c3853991f694921b1b9368b0bd32beb9f1b63", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "fbb01ecdfd565b56261302f7e1fcc27c4fb8f32d56eab74db621fc154604a7a1"},

examples/save_to_file/lib/save_to_file/peer_handler.ex

@@ -25,7 +25,14 @@ defmodule SaveToFile.PeerHandler do
     %RTPCodecParameters{
       payload_type: 96,
       mime_type: "video/VP8",
-      clock_rate: 90_000
+      clock_rate: 90_000,
+      rtcp_fbs: [%ExSDP.Attribute.RTCPFeedback{pt: 96, feedback_type: :nack}]
+    },
+    %RTPCodecParameters{
+      payload_type: 97,
+      mime_type: "video/rtx",
+      clock_rate: 90_000,
+      sdp_fmtp_line: %ExSDP.Attribute.FMTP{pt: 97, apt: 96}
     }
   ]
 

examples/send_from_file/lib/send_from_file/peer_handler.ex

@@ -30,7 +30,14 @@ defmodule SendFromFile.PeerHandler do
     %RTPCodecParameters{
       payload_type: 96,
       mime_type: "video/VP8",
-      clock_rate: 90_000
+      clock_rate: 90_000,
+      rtcp_fbs: [%ExSDP.Attribute.RTCPFeedback{pt: 96, feedback_type: :nack}]
+    },
+    %RTPCodecParameters{
+      payload_type: 97,
+      mime_type: "video/rtx",
+      clock_rate: 90_000,
+      sdp_fmtp_line: %ExSDP.Attribute.FMTP{pt: 97, apt: 96}
     }
   ]
 

examples/send_from_file/mix.lock

@@ -10,7 +10,7 @@
   "ex_dtls": {:hex, :ex_dtls, "0.15.1", "34b3600ff13eebf6c96be033005cc110ea5beef98394631365ec26b493df80c5", [:mix], [{:unifex, "~> 1.0", [hex: :unifex, repo: "hexpm", optional: false]}], "hexpm", "9cfebdfe9111c0f68c77667cb9366e4a6f17e8a240975ffd100148de57478a29"},
   "ex_ice": {:git, "https://github.com/elixir-webrtc/ex_ice.git", "1b57c79effb762ea71df79390605a52065b60821", []},
   "ex_libsrtp": {:hex, :ex_libsrtp, "0.7.2", "211bd89c08026943ce71f3e2c0231795b99cee748808ed3ae7b97cd8d2450b6b", [:mix], [{:bunch, "~> 1.6", [hex: :bunch, repo: "hexpm", optional: false]}, {:bundlex, "~> 1.3", [hex: :bundlex, repo: "hexpm", optional: false]}, {:membrane_precompiled_dependency_provider, "~> 0.1.0", [hex: :membrane_precompiled_dependency_provider, repo: "hexpm", optional: false]}, {:unifex, "~> 1.1", [hex: :unifex, repo: "hexpm", optional: false]}], "hexpm", "2e20645d0d739a4ecdcf8d4810a0c198120c8a2f617f2b75b2e2e704d59f492a"},
-  "ex_rtcp": {:git, "https://github.com/elixir-webrtc/ex_rtcp.git", "67fe8320302fa240e190eb97b95bbabf6bebc2fd", []},
+  "ex_rtcp": {:git, "https://github.com/elixir-webrtc/ex_rtcp.git", "323873e4d9be018bd742c633d3feed0db40625e8", []},
   "ex_rtp": {:git, "https://github.com/elixir-webrtc/ex_rtp.git", "eb9b49d8e53ba92bcd31e7d015473ae10a26707d", []},
   "ex_sdp": {:hex, :ex_sdp, "0.15.0", "53815fb5b5e4fae0f3b26de90f372446bb8e0eed62a3cc20394d3c29519698be", [:mix], [{:bunch, "~> 1.3", [hex: :bunch, repo: "hexpm", optional: false]}, {:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}], "hexpm", "d3f23596b73e7057521ff0f0d55b1189c6320a2f04388aa3a80a0aa97ffb379f"},
   "ex_stun": {:hex, :ex_stun, "0.1.0", "252474bf4c8519fbf4bc0fbfc6a1b846a634b1478c65dbbfb4b6ab4e33c2a95a", [:mix], [], "hexpm", "629fc8be45b624a92522f81d85ba001877b1f0745889a2419bdb678790d7480c"},

lib/ex_webrtc/peer_connection.ex

@@ -198,17 +198,18 @@ defmodule ExWebRTC.PeerConnection do
 
   @doc """
   Send an RTP packet to the remote peer using specified track or its id.
+
+  Options:
+    * `rtx?` - send the packet as if it was retransmited (use SSRC and payload type specific to RTX)
   """
   @spec send_rtp(
           peer_connection(),
-          MediaStreamTrack.t() | MediaStreamTrack.id(),
-          ExRTP.Packet.t()
+          MediaStreamTrack.id(),
+          ExRTP.Packet.t(),
+          rtx?: boolean()
         ) :: :ok
-  def send_rtp(peer_connection, %MediaStreamTrack{id: track_id}, packet),
-    do: send_rtp(peer_connection, track_id, packet)
-
-  def send_rtp(peer_connection, track_id, packet) when is_integer(track_id) do
-    GenServer.cast(peer_connection, {:send_rtp, track_id, packet})
+  def send_rtp(peer_connection, track_id, packet, opts \\ []) do
+    GenServer.cast(peer_connection, {:send_rtp, track_id, packet, opts})
   end
 
   @doc """
@@ -245,9 +246,6 @@ defmodule ExWebRTC.PeerConnection do
     # route data to the DTLSTransport
     :ok = DefaultICETransport.on_data(ice_pid, dtls_transport)
 
-    # TODO: should keep track of previous ice candidates
-    # and add them to local description
-
     state = %{
       owner: owner,
       config: config,
@@ -479,7 +477,9 @@ defmodule ExWebRTC.PeerConnection do
   @impl true
   def handle_call({:add_transceiver, kind, options}, _from, state)
       when kind in [:audio, :video] do
-    options = Keyword.put(options, :ssrc, generate_ssrc(state))
+    {ssrc, rtx_ssrc} = generate_ssrcs(state)
+    options = [{:ssrc, ssrc}, {:rtx_ssrc, rtx_ssrc} | options]
+
     transceiver = RTPTransceiver.new(kind, nil, state.config, options)
     state = %{state | transceivers: state.transceivers ++ [transceiver]}
 
@@ -490,7 +490,9 @@ defmodule ExWebRTC.PeerConnection do
 
   @impl true
   def handle_call({:add_transceiver, %MediaStreamTrack{} = track, options}, _from, state) do
-    options = Keyword.put(options, :ssrc, generate_ssrc(state))
+    {ssrc, rtx_ssrc} = generate_ssrcs(state)
+    options = [{:ssrc, ssrc}, {:rtx_ssrc, rtx_ssrc} | options]
+
     transceiver = RTPTransceiver.new(track.kind, track, state.config, options)
     state = %{state | transceivers: state.transceivers ++ [transceiver]}
 
@@ -558,16 +560,24 @@ defmodule ExWebRTC.PeerConnection do
           false
       end)
 
+    {ssrc, rtx_ssrc} = generate_ssrcs(state)
+
     {transceivers, sender} =
       case free_transceiver_idx do
         nil ->
-          options = [direction: :sendrecv, ssrc: generate_ssrc(state), added_by_add_track: true]
+          options = [
+            direction: :sendrecv,
+            added_by_add_track: true,
+            ssrc: ssrc,
+            rtx_ssrc: rtx_ssrc
+          ]
+
           tr = RTPTransceiver.new(kind, track, state.config, options)
           {state.transceivers ++ [tr], tr.sender}
 
         idx ->
           tr = Enum.at(state.transceivers, idx)
-          tr = RTPTransceiver.add_track(tr, track, generate_ssrc(state))
+          tr = RTPTransceiver.add_track(tr, track, ssrc, rtx_ssrc)
           {List.replace_at(state.transceivers, idx, tr), tr.sender}
       end
 
@@ -594,7 +604,8 @@ defmodule ExWebRTC.PeerConnection do
             {:reply, {:error, :invalid_track_type}, state}
 
           tr.direction in [:sendrecv, :sendonly] ->
-            tr = RTPTransceiver.replace_track(tr, track, generate_ssrc(state))
+            {ssrc, rtx_ssrc} = generate_ssrcs(state)
+            tr = RTPTransceiver.replace_track(tr, track, ssrc, rtx_ssrc)
             transceivers = List.replace_at(state.transceivers, tr_idx, tr)
             state = %{state | transceivers: transceivers}
             {:reply, :ok, state}
@@ -760,7 +771,9 @@ defmodule ExWebRTC.PeerConnection do
   end
 
   @impl true
-  def handle_cast({:send_rtp, track_id, packet}, state) do
+  def handle_cast({:send_rtp, track_id, packet, opts}, state) do
+    rtx? = Keyword.get(opts, :rtx?, false)
+
     # TODO: iterating over transceivers is not optimal
     # but this is, most likely, going to be refactored anyways
     {transceiver, idx} =
@@ -792,7 +805,7 @@ defmodule ExWebRTC.PeerConnection do
               {packet, state}
           end
 
-        {packet, transceiver} = RTPTransceiver.send_packet(transceiver, packet)
+        {packet, transceiver} = RTPTransceiver.send_packet(transceiver, packet, rtx?)
         :ok = DTLSTransport.send_rtp(state.dtls_transport, packet)
 
         transceivers = List.replace_at(state.transceivers, idx, transceiver)
@@ -932,13 +945,13 @@ defmodule ExWebRTC.PeerConnection do
   def handle_info({:dtls_transport, _pid, {:rtcp, data}}, state) do
     case ExRTCP.CompoundPacket.decode(data) do
       {:ok, packets} ->
-        transceivers =
-          Enum.reduce(packets, state.transceivers, fn packet, transceivers ->
-            handle_report(packet, transceivers)
+        state =
+          Enum.reduce(packets, state, fn packet, state ->
+            handle_rtcp_packet(state, packet)
           end)
 
         notify(state.owner, {:rtcp, packets})
-        {:noreply, %{state | transceivers: transceivers}}
+        {:noreply, state}
 
       {:error, _res} ->
         case data do
@@ -1588,23 +1601,47 @@ defmodule ExWebRTC.PeerConnection do
     end
   end
 
-  defp handle_report(%ExRTCP.Packet.SenderReport{} = report, transceivers) do
+  defp handle_rtcp_packet(state, %ExRTCP.Packet.SenderReport{} = report) do
     transceiver =
-      transceivers
+      state.transceivers
       |> Enum.with_index()
       |> Enum.find(fn {tr, _idx} -> tr.receiver.ssrc == report.ssrc end)
 
     case transceiver do
       nil ->
-        transceivers
+        state
 
       {tr, idx} ->
         tr = RTPTransceiver.receive_report(tr, report)
-        List.replace_at(transceivers, idx, tr)
+        transceivers = List.replace_at(state.transceivers, idx, tr)
+        %{state | transceivers: transceivers}
+    end
+  end
+
+  defp handle_rtcp_packet(state, %ExRTCP.Packet.TransportFeedback.NACK{} = nack) do
+    transceiver =
+      state.transceivers
+      |> Enum.with_index()
+      |> Enum.find(fn {tr, _idx} -> tr.sender.ssrc == nack.media_ssrc end)
+
+    case transceiver do
+      nil ->
+        state
+
+      # in case NACK was received, but RTX was not negotiated
+      # as NACK and RTX are negotited independently
+      {%RTPTransceiver{sender: %RTPSender{rtx_pt: nil}}, _idx} ->
+        state
+
+      {tr, idx} ->
+        {packets, tr} = RTPTransceiver.receive_nack(tr, nack)
+        for packet <- packets, do: send_rtp(self(), tr.sender.track.id, packet, rtx?: true)
+        transceivers = List.replace_at(state.transceivers, idx, tr)
+        %{state | transceivers: transceivers}
     end
   end
 
-  defp handle_report(_report, transceivers), do: transceivers
+  defp handle_rtcp_packet(state, _packet), do: state
 
   defp do_get_description(nil, _candidates), do: nil
 
@@ -1613,10 +1650,12 @@ defmodule ExWebRTC.PeerConnection do
     %SessionDescription{type: type, sdp: to_string(sdp)}
   end
 
-  defp generate_ssrc(state) do
+  defp generate_ssrcs(state) do
     rtp_sender_ssrcs = Enum.map(state.transceivers, & &1.sender.ssrc)
     ssrcs = MapSet.new(Map.keys(state.demuxer.ssrc_to_mid) ++ rtp_sender_ssrcs)
-    do_generate_ssrc(ssrcs, 200)
+    ssrc = do_generate_ssrc(ssrcs, 200)
+    rtx_ssrc = do_generate_ssrc(MapSet.put(ssrcs, ssrc), 200)
+    {ssrc, rtx_ssrc}
   end
 
   # this is practically impossible so it's easier to raise