Add stop_transceiver/2 (#55)

lib/ex_webrtc/rtp_transceiver.ex

@@ -20,24 +20,30 @@ defmodule ExWebRTC.RTPTransceiver do
   @type t() :: %__MODULE__{
           id: id(),
           mid: String.t() | nil,
+          mline_idx: non_neg_integer() | nil,
           direction: direction(),
           current_direction: direction() | nil,
           fired_direction: direction() | nil,
           kind: kind(),
           rtp_hdr_exts: [ExSDP.Attribute.Extmap.t()],
           codecs: [RTPCodecParameters.t()],
           receiver: RTPReceiver.t(),
-          sender: RTPSender.t()
+          sender: RTPSender.t(),
+          stopping: boolean(),
+          stopped: boolean()
         }
 
   @enforce_keys [:id, :direction, :kind, :sender, :receiver]
   defstruct @enforce_keys ++
               [
                 :mid,
+                :mline_idx,
                 :current_direction,
                 :fired_direction,
                 codecs: [],
-                rtp_hdr_exts: []
+                rtp_hdr_exts: [],
+                stopping: false,
+                stopped: false
               ]
 
   @doc false
@@ -77,8 +83,8 @@ defmodule ExWebRTC.RTPTransceiver do
   end
 
   @doc false
-  @spec from_mline(ExSDP.Media.t(), Configuration.t()) :: t()
-  def from_mline(mline, config) do
+  @spec from_mline(ExSDP.Media.t(), non_neg_integer(), Configuration.t()) :: t()
+  def from_mline(mline, mline_idx, config) do
     codecs = get_codecs(mline, config)
     rtp_hdr_exts = get_rtp_hdr_extensions(mline, config)
     {:mid, mid} = ExSDP.get_attribute(mline, :mid)
@@ -88,6 +94,7 @@ defmodule ExWebRTC.RTPTransceiver do
     %__MODULE__{
       id: Utils.generate_id(),
       mid: mid,
+      mline_idx: mline_idx,
       direction: :recvonly,
       kind: mline.type,
       codecs: codecs,
@@ -115,30 +122,63 @@ defmodule ExWebRTC.RTPTransceiver do
   @doc false
   @spec to_answer_mline(t(), ExSDP.Media.t(), Keyword.t()) :: ExSDP.Media.t()
   def to_answer_mline(transceiver, mline, opts) do
-    if transceiver.codecs == [] do
-      # reject mline and skip further processing
-      # see RFC 8299 sec. 5.3.1 and RFC 3264 sec. 6
-      %ExSDP.Media{mline | port: 0}
-    else
-      offered_direction = SDPUtils.get_media_direction(mline)
-      direction = get_direction(offered_direction, transceiver.direction)
-      opts = Keyword.put(opts, :direction, direction)
-      to_mline(transceiver, opts)
+    offered_direction = SDPUtils.get_media_direction(mline)
+    direction = get_direction(offered_direction, transceiver.direction)
+    opts = Keyword.put(opts, :direction, direction)
+
+    # Reject mline. See RFC 8829 sec. 5.3.1 and RFC 3264 sec. 6.
+    # We could reject earlier (as RFC suggests) but we generate
+    # answer mline at first to have consistent fingerprint, ice_ufrag and
+    # ice_pwd values across mlines.
+    cond do
+      transceiver.codecs == [] ->
+        # there has to be at least one format so take it from the offer
+        codecs = SDPUtils.get_rtp_codec_parameters(mline)
+        transceiver = %__MODULE__{transceiver | codecs: codecs}
+        mline = to_mline(transceiver, opts)
+        %ExSDP.Media{mline | port: 0}
+
+      transceiver.stopping == true or transceiver.stopped == true ->
+        mline = to_mline(transceiver, opts)
+        %ExSDP.Media{mline | port: 0}
+
+      true ->
+        to_mline(transceiver, opts)
     end
   end
 
   @doc false
   @spec to_offer_mline(t(), Keyword.t()) :: ExSDP.Media.t()
   def to_offer_mline(transceiver, opts) do
-    to_mline(transceiver, opts)
+    mline = to_mline(transceiver, opts)
+    if transceiver.stopping, do: %ExSDP.Media{mline | port: 0}, else: mline
   end
 
   @doc false
   # asssings mid to the transceiver and its sender
   @spec assign_mid(t(), String.t()) :: t()
   def assign_mid(transceiver, mid) do
     sender = %RTPSender{transceiver.sender | mid: mid}
-    %{transceiver | mid: mid, sender: sender}
+    %__MODULE__{transceiver | mid: mid, sender: sender}
+  end
+
+  @spec stop(t(), (-> term())) :: t()
+  def stop(transceiver, on_track_ended) do
+    tr =
+      if transceiver.stopping,
+        do: transceiver,
+        else: stop_sending_and_receiving(transceiver, on_track_ended)
+
+    # should we reset stopping or leave it as true?
+    %__MODULE__{tr | stopped: true, stopping: false, current_direction: nil}
+  end
+
+  @spec stop_sending_and_receiving(t(), (-> term())) :: t()
+  def stop_sending_and_receiving(transceiver, on_track_ended) do
+    # TODO send RTCP BYE for each RTP stream
+    # TODO stop receiving media
+    on_track_ended.()
+    %__MODULE__{transceiver | direction: :inactive, stopping: true}
   end
 
   defp to_mline(transceiver, opts) do

lib/ex_webrtc/sdp_utils.ex

@@ -253,6 +253,25 @@ defmodule ExWebRTC.SDPUtils do
     |> Map.new()
   end
 
+  @spec find_mline_by_mid(ExSDP.t(), binary()) :: ExSDP.Media.t() | nil
+  def find_mline_by_mid(sdp, mid) do
+    Enum.find(sdp.media, fn mline ->
+      {:mid, mline_mid} = ExSDP.get_attribute(mline, :mid)
+      mline_mid == mid
+    end)
+  end
+
+  @spec find_free_mline_idx(ExSDP.t(), [non_neg_integer()]) :: non_neg_integer() | nil
+  def find_free_mline_idx(sdp, indices) do
+    sdp.media
+    |> Stream.with_index()
+    |> Enum.find_index(fn {mline, idx} -> mline.port == 0 and idx not in indices end)
+  end
+
+  @spec is_rejected(ExSDP.Media.t()) :: boolean()
+  def is_rejected(%ExSDP.Media{port: 0}), do: true
+  def is_rejected(%ExSDP.Media{}), do: false
+
   defp do_get_ice_credentials(sdp_or_mline) do
     ice_ufrag =
       case ExSDP.get_attribute(sdp_or_mline, :ice_ufrag) do

mix.exs

@@ -9,6 +9,7 @@ defmodule ExWebRTC.MixProject do
       app: :ex_webrtc,
       version: @version,
       elixir: "~> 1.15",
+      elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,
       description: "Implementation of WebRTC",
       package: package(),
@@ -36,6 +37,9 @@ defmodule ExWebRTC.MixProject do
     ]
   end
 
+  defp elixirc_paths(:test), do: ["lib", "test/support"]
+  defp elixirc_paths(_env), do: ["lib"]
+
   def package do
     [
       licenses: ["Apache-2.0"],

test/ex_webrtc/peer_connection_test.exs

@@ -1,6 +1,8 @@
 defmodule ExWebRTC.PeerConnectionTest do
   use ExUnit.Case, async: true
 
+  import ExWebRTC.Support.TestUtils
+
   alias ExWebRTC.{
     RTPTransceiver,
     RTPSender,
@@ -441,6 +443,119 @@ defmodule ExWebRTC.PeerConnectionTest do
     end
   end
 
+  describe "stop_transceiver/2" do
+    test "before the first offer" do
+      {:ok, pc1} = PeerConnection.start_link()
+      {:ok, tr} = PeerConnection.add_transceiver(pc1, :audio)
+      :ok = PeerConnection.stop_transceiver(pc1, tr.id)
+      assert_receive {:ex_webrtc, ^pc1, {:track_ended, track_id}}
+      assert tr.receiver.track.id == track_id
+      {:ok, offer} = PeerConnection.create_offer(pc1)
+      sdp = ExSDP.parse!(offer.sdp)
+      assert sdp.media == []
+    end
+
+    test "with renegotiation" do
+      {:ok, pc1} = PeerConnection.start_link()
+      {:ok, pc2} = PeerConnection.start_link()
+      {:ok, tr} = PeerConnection.add_transceiver(pc1, :audio)
+
+      assert_receive {:ex_webrtc, ^pc1, :negotiation_needed}
+
+      :ok = negotiate(pc1, pc2)
+
+      :ok = PeerConnection.stop_transceiver(pc1, tr.id)
+
+      assert_receive {:ex_webrtc, ^pc1, :negotiation_needed}
+      assert_receive {:ex_webrtc, ^pc1, {:track_ended, track_id}}
+      assert tr.receiver.track.id == track_id
+
+      assert [
+               %RTPTransceiver{
+                 current_direction: :sendonly,
+                 direction: :inactive,
+                 stopping: true,
+                 stopped: false
+               }
+             ] = PeerConnection.get_transceivers(pc1)
+
+      {:ok, offer} = PeerConnection.create_offer(pc1)
+      :ok = PeerConnection.set_local_description(pc1, offer)
+
+      # nothing should change
+      assert [
+               %RTPTransceiver{
+                 current_direction: :sendonly,
+                 direction: :inactive,
+                 stopping: true,
+                 stopped: false
+               }
+             ] = PeerConnection.get_transceivers(pc1)
+
+      # on the remote side, transceiver should be stopped
+      # immediately after setting remote description
+      :ok = PeerConnection.set_remote_description(pc2, offer)
+
+      assert_receive {:ex_webrtc, ^pc2, {:track_ended, _track_id}}
+
+      assert [
+               %RTPTransceiver{
+                 current_direction: nil,
+                 direction: :inactive,
+                 stopping: false,
+                 stopped: true
+               }
+             ] = PeerConnection.get_transceivers(pc2)
+
+      {:ok, answer} = PeerConnection.create_answer(pc2)
+      :ok = PeerConnection.set_local_description(pc2, answer)
+
+      assert [] == PeerConnection.get_transceivers(pc2)
+
+      :ok = PeerConnection.set_remote_description(pc1, answer)
+
+      assert [] == PeerConnection.get_transceivers(pc1)
+
+      # renegotiate without changes
+      {:ok, offer} = PeerConnection.create_offer(pc1)
+      sdp = ExSDP.parse!(offer.sdp)
+      assert Enum.count(sdp.media) == 1
+
+      :ok = PeerConnection.set_local_description(pc1, offer)
+      assert [] == PeerConnection.get_transceivers(pc1)
+
+      # on setting remote description, a stopped transceiver
+      # should be created and on setting local description
+      # it should be removed
+      :ok = PeerConnection.set_remote_description(pc2, offer)
+
+      [
+        %RTPTransceiver{
+          current_direction: nil,
+          direction: :inactive,
+          stopped: true,
+          stopping: false
+        }
+      ] = PeerConnection.get_transceivers(pc2)
+
+      {:ok, answer} = PeerConnection.create_answer(pc2)
+      sdp = ExSDP.parse!(answer.sdp)
+      assert Enum.count(sdp.media) == 1
+
+      :ok = PeerConnection.set_local_description(pc2, answer)
+      assert [] == PeerConnection.get_transceivers(pc2)
+
+      :ok = PeerConnection.set_remote_description(pc1, answer)
+      assert [] == PeerConnection.get_transceivers(pc1)
+    end
+
+    test "with invalid transceiver id" do
+      {:ok, pc} = PeerConnection.start_link()
+      {:ok, tr} = PeerConnection.add_transceiver(pc, :audio)
+      assert {:error, :invalid_transceiver_id} == PeerConnection.stop_transceiver(pc, tr.id + 1)
+    end
+  end
+
   describe "add_track/2" do
     test "with no available transceivers" do
       {:ok, pc} = PeerConnection.start_link()
@@ -785,14 +900,4 @@ defmodule ExWebRTC.PeerConnectionTest do
     assert [%RTPTransceiver{direction: :inactive, current_direction: :inactive}] =
              PeerConnection.get_transceivers(pc2)
   end
-
-  defp negotiate(pc1, pc2) do
-    {:ok, offer} = PeerConnection.create_offer(pc1)
-    :ok = PeerConnection.set_local_description(pc1, offer)
-    :ok = PeerConnection.set_remote_description(pc2, offer)
-    {:ok, answer} = PeerConnection.create_answer(pc2)
-    :ok = PeerConnection.set_local_description(pc2, answer)
-    :ok = PeerConnection.set_remote_description(pc1, answer)
-    :ok
-  end
 end