Add stop_transceiver/2

Author: mickel8

lib/ex_webrtc/peer_connection.ex

@@ -41,6 +41,7 @@ defmodule ExWebRTC.PeerConnection do
            | {:connection_state_change, connection_state()}
            | {:track, MediaStreamTrack.t()}
            | {:track_muted, MediaStreamTrack.id()}
+           | {:track_ended, MediaStreamTrack.id()}
            | {:rtp, MediaStreamTrack.id(), ExRTP.Packet.t()}}
 
   @typedoc """
@@ -124,6 +125,12 @@ defmodule ExWebRTC.PeerConnection do
     GenServer.call(peer_connection, {:set_transceiver_direction, transceiver_id, direction})
   end
 
+  @spec stop_transceiver(peer_connection(), RTPTransceiver.id()) ::
+          :ok | {:error, :invalid_transceiver_id}
+  def stop_transceiver(peer_connection, transceiver_id) do
+    GenServer.call(peer_connection, {:stop_transceiver, transceiver_id})
+  end
+
   @spec add_track(peer_connection(), MediaStreamTrack.t()) :: {:ok, RTPSender.t()}
   def add_track(peer_connection, track) do
     GenServer.call(peer_connection, {:add_track, track})
@@ -206,9 +213,6 @@ defmodule ExWebRTC.PeerConnection do
       :ok = state.ice_transport.restart(state.ice_pid)
     end
 
-    next_mid = find_next_mid(state)
-    transceivers = assign_mids(state.transceivers, next_mid)
-
     {:ok, ice_ufrag, ice_pwd} =
       state.ice_transport.get_local_credentials(state.ice_pid)
 
@@ -229,7 +233,74 @@ defmodule ExWebRTC.PeerConnection do
         rtcp: true
       ]
 
-    mlines = Enum.map(transceivers, &RTPTransceiver.to_offer_mline(&1, opts))
+    next_mid = find_next_mid(state)
+
+    {transceivers, mlines} =
+      if state.current_local_desc == nil do
+        # TODO stopped transceivers
+        {transceivers, _next_mid} =
+          Enum.map_reduce(state.transceivers, next_mid, fn %{mid: nil} = tr, nm ->
+            if tr.stopped do
+              {tr, nm}
+            else
+              tr = RTPTransceiver.assign_mid(tr, to_string(nm))
+              tr = %RTPTransceiver{tr | mline_idx: nm}
+              {tr, nm + 1}
+            end
+          end)
+
+        mlines =
+          transceivers
+          |> Enum.reject(fn tr -> tr.stopped == true end)
+          |> Enum.map(&RTPTransceiver.to_offer_mline(&1, opts))
+
+        {transceivers, mlines}
+      else
+        last_answer = get_last_answer(state)
+
+        {transceivers, _next_mid} =
+          Enum.map_reduce(state.transceivers, {next_mid, []}, fn
+            %{mid: nil} = tr, {nm, mline_indices} ->
+              if tr.stopped do
+                {tr, nm}
+              else
+                tr = RTPTransceiver.assign_mid(tr, to_string(nm))
+                # idx might be nil, this means that
+                # there is no mline to recycle
+                idx = SDPUtils.find_free_mline_idx(last_answer, mline_indices)
+                tr = %RTPTransceiver{tr | mline_idx: idx}
+                {tr, {nm + 1, [idx | mline_indices]}}
+              end
+
+            tr, acc ->
+              {tr, acc}
+          end)
+
+        transceivers =
+          transceivers
+          |> Enum.reject(fn tr -> tr.stopped == true end)
+          |> Enum.sort_by(fn tr -> tr.mline_idx end)
+          |> Enum.with_index()
+          |> Enum.map(fn {tr, idx} -> %RTPTransceiver{tr | mline_idx: idx} end)
+
+        mlines = Enum.map(transceivers, fn tr -> RTPTransceiver.to_offer_mline(tr, opts) end)
+
+        final_mlines =
+          last_answer.media
+          |> Stream.with_index()
+          |> Enum.map(fn {answer_mline, idx} ->
+            case Enum.at(mlines, idx) do
+              nil -> answer_mline
+              mline -> mline
+            end
+          end)
+
+        mlines = Enum.slice(mlines, 0..(Enum.count(final_mlines) - 1))
+
+        mlines = final_mlines ++ (mlines -- final_mlines)
+
+        {transceivers, mlines}
+      end
 
     mids =
       Enum.map(mlines, fn mline ->
@@ -361,6 +432,17 @@ defmodule ExWebRTC.PeerConnection do
     {:reply, {:ok, transceiver}, state}
   end
 
+  @impl true
+  def handle_call({:add_transceiver, %MediaStreamTrack{} = track, options}, _from, state) do
+    options = Keyword.put(options, :ssrc, generate_ssrc(state))
+    transceiver = RTPTransceiver.new(track.kind, track, state.config, options)
+    state = %{state | transceivers: state.transceivers ++ [transceiver]}
+
+    state = update_negotiation_needed(state)
+
+    {:reply, {:ok, transceiver}, state}
+  end
+
   @impl true
   def handle_call({:set_transceiver_direction, tr_id, direction}, _from, state) do
     idx = Enum.find_index(state.transceivers, fn tr -> tr.id == tr_id end)
@@ -380,14 +462,28 @@ defmodule ExWebRTC.PeerConnection do
   end
 
   @impl true
-  def handle_call({:add_transceiver, %MediaStreamTrack{} = track, options}, _from, state) do
-    options = Keyword.put(options, :ssrc, generate_ssrc(state))
-    transceiver = RTPTransceiver.new(track.kind, track, state.config, options)
-    state = %{state | transceivers: state.transceivers ++ [transceiver]}
+  def handle_call({:stop_transceiver, tr_id}, _from, state) do
+    idx = Enum.find_index(state.transceivers, fn tr -> tr.id == tr_id end)
 
-    state = update_negotiation_needed(state)
+    case idx do
+      nil ->
+        {:reply, {:error, :invalid_transceiver_id}, state}
 
-    {:reply, {:ok, transceiver}, state}
+      idx ->
+        tr = Enum.at(state.transceivers, idx)
+
+        if tr.stopping do
+          {:reply, :ok, state}
+        else
+          # TODO send RTCP BYE for each RTP stream
+          # TODO stop receiving media
+          tr = %RTPTransceiver{tr | direction: :inactive, stopping: true}
+          transceivers = List.replace_at(state.transceivers, idx, tr)
+          state = %{state | transceivers: transceivers}
+          state = update_negotiation_needed(state)
+          {:reply, :ok, state}
+        end
+    end
   end
 
   @impl true
@@ -638,12 +734,15 @@ defmodule ExWebRTC.PeerConnection do
 
       transceivers = update_transceiver_directions(state.transceivers, sdp, :local, type)
 
+      # TODO re-think order of those functions
+      # and demuxer update
       state =
         state
         |> set_description(:local, type, sdp)
+        |> Map.replace!(:transceivers, transceivers)
+        |> remove_stopped_transceivers(type, sdp)
         |> update_signaling_state(next_sig_state)
         |> Map.update!(:demuxer, &Demuxer.update(&1, sdp))
-        |> Map.replace!(:transceivers, transceivers)
 
       if state.signaling_state == :stable do
         state = %{state | negotiation_needed: false}
@@ -672,9 +771,7 @@ defmodule ExWebRTC.PeerConnection do
       state = %{state | config: config}
 
       transceivers =
-        state
-        |> update_transceivers(sdp)
-        |> update_transceiver_directions(sdp, :remote, type)
+        update_transceivers(sdp.media, state.transceivers, type, state.config, state.owner)
 
       # TODO: this can result in ICE restart (when it should, e.g. when this is answer)
       :ok = state.ice_transport.set_remote_credentials(state.ice_pid, ice_ufrag, ice_pwd)
@@ -690,9 +787,10 @@ defmodule ExWebRTC.PeerConnection do
       state =
         state
         |> set_description(:remote, type, sdp)
+        |> Map.replace!(:transceivers, transceivers)
+        |> remove_stopped_transceivers(type, sdp)
         |> update_signaling_state(next_sig_state)
         |> Map.update!(:demuxer, &Demuxer.update(&1, sdp))
-        |> Map.replace!(:transceivers, transceivers)
 
       if state.signaling_state == :stable do
         state = %{state | negotiation_needed: false}
@@ -710,6 +808,29 @@ defmodule ExWebRTC.PeerConnection do
     end
   end
 
+  defp remove_stopped_transceivers(state, :answer, sdp) do
+    # see W3C 4.4.1.5-4.7.12 xd
+    transceivers =
+      Enum.reject(state.transceivers, fn tr ->
+        if tr.mid != nil do
+          # TODO refactor
+          mline =
+            Enum.find(sdp.media, fn mline ->
+              {:mid, mid} = ExSDP.get_attribute(mline, :mid)
+              mid == tr.mid
+            end)
+
+          tr.stopped == true and mline.port == 0
+        else
+          false
+        end
+      end)
+
+    %{state | transceivers: transceivers}
+  end
+
+  defp remove_stopped_transceivers(state, :offer, _sdp), do: state
+
   defp next_signaling_state(current_signaling_state, source, type)
   defp next_signaling_state(:stable, :remote, :offer), do: {:ok, :have_remote_offer}
   defp next_signaling_state(:stable, :local, :offer), do: {:ok, :have_local_offer}
@@ -797,25 +918,39 @@ defmodule ExWebRTC.PeerConnection do
   end
 
   # this function is only called when applying remote description
-  defp update_transceivers(state, sdp) do
-    Enum.reduce(sdp.media, state.transceivers, fn mline, transceivers ->
-      {:mid, mid} = ExSDP.get_attribute(mline, :mid)
+  defp update_transceivers(mlines, transceivers, sdp_type, config, owner)
+  defp update_transceivers([], transceivers, _sdp_type, _config, _owner), do: transceivers
 
-      direction = SDPUtils.get_media_direction(mline) |> reverse_direction()
+  defp update_transceivers([mline | mlines], transceivers, sdp_type, config, owner) do
+    {:mid, mid} = ExSDP.get_attribute(mline, :mid)
 
-      # TODO: consider recycled transceivers
-      case find_transceiver(transceivers, mid) do
-        {idx, %RTPTransceiver{} = tr} ->
-          new_tr = RTPTransceiver.update(tr, mline, state.config)
-          new_tr = process_remote_track(new_tr, direction, state.owner)
-          List.replace_at(transceivers, idx, new_tr)
+    direction =
+      if SDPUtils.is_rejected(mline),
+        do: :inactive,
+        else: SDPUtils.get_media_direction(mline) |> reverse_direction()
 
-        nil ->
-          new_tr = RTPTransceiver.from_mline(mline, state.config)
-          new_tr = process_remote_track(new_tr, direction, state.owner)
-          transceivers ++ [new_tr]
+    # TODO: consider recycled transceivers
+    # Note: in theory we should update transceiver codecs
+    # after processing remote track but this shouldn't have any impact
+    {idx, tr} =
+      case find_transceiver(transceivers, mid) do
+        {idx, %RTPTransceiver{} = tr} -> {idx, RTPTransceiver.update(tr, mline, config)}
+        nil -> {nil, RTPTransceiver.from_mline(mline, config)}
       end
-    end)
+
+    tr = process_remote_track(tr, direction, owner)
+    tr = if sdp_type == :answer, do: %RTPTransceiver{tr | current_direction: direction}, else: tr
+    tr = if SDPUtils.is_rejected(mline), do: RTPTransceiver.stop(tr), else: tr
+
+    case idx do
+      nil ->
+        transceivers = transceivers ++ [tr]
+        update_transceivers(mlines, transceivers, sdp_type, config, owner)
+
+      idx ->
+        transceivers = List.replace_at(transceivers, idx, tr)
+        update_transceivers(mlines, transceivers, sdp_type, config, owner)
+    end
   end
 
   # see W3C WebRTC 5.1.1
@@ -867,19 +1002,6 @@ defmodule ExWebRTC.PeerConnection do
     |> Enum.find(fn {_idx, tr} -> tr.mid == mid end)
   end
 
-  defp assign_mids(transceivers, next_mid) do
-    {new_transceivers, _next_mid} =
-      Enum.map_reduce(transceivers, next_mid, fn
-        %{mid: nil} = t, nm ->
-          {RTPTransceiver.assign_mid(t, to_string(nm)), nm + 1}
-
-        other, nm ->
-          {other, nm}
-      end)
-
-    new_transceivers
-  end
-
   defp find_next_mid(state) do
     # next mid must be unique, it's acomplished by looking for values
     # greater than any mid in remote description or our own transceivers
@@ -913,6 +1035,9 @@ defmodule ExWebRTC.PeerConnection do
     end)
   end
 
+  defp get_last_answer(%{current_local_desc: {:answer, desc}}), do: desc
+  defp get_last_answer(%{current_remote_desc: {:answer, desc}}), do: desc
+
   # TODO support :disconnected state - our ICE doesn't provide disconnected state for now
   # TODO support :closed state
   # the order of these clauses is important

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
@@ -115,9 +121,9 @@ 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
+    if transceiver.codecs == [] or transceiver.stopping == true or transceiver.stopped == true do
       # reject mline and skip further processing
-      # see RFC 8299 sec. 5.3.1 and RFC 3264 sec. 6
+      # see RFC 8829 sec. 5.3.1 and RFC 3264 sec. 6
       %ExSDP.Media{mline | port: 0}
     else
       offered_direction = SDPUtils.get_media_direction(mline)
@@ -130,7 +136,9 @@ defmodule ExWebRTC.RTPTransceiver do
   @doc false
   @spec to_offer_mline(t(), Keyword.t()) :: ExSDP.Media.t()
   def to_offer_mline(transceiver, opts) do
-    to_mline(transceiver, opts)
+    # TODO refactor
+    mline = to_mline(transceiver, opts)
+    if transceiver.stopping, do: %ExSDP.Media{mline | port: 0}, else: mline
   end
 
   @doc false
@@ -141,6 +149,20 @@ defmodule ExWebRTC.RTPTransceiver do
     %{transceiver | mid: mid, sender: sender}
   end
 
+  @spec stop(t()) :: t()
+  def stop(transceiver) do
+    tr = if transceiver.stopping, do: transceiver, else: stop_sending_and_receiving(transceiver)
+    # 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()) :: t()
+  def stop_sending_and_receiving(transceiver) do
+    # TODO send RTCP BYE for each RTP stream
+    # TODO stop receiving media
+    %__MODULE__{transceiver | direction: :inactive, stopping: true}
+  end
+
   defp to_mline(transceiver, opts) do
     pt = Enum.map(transceiver.codecs, fn codec -> codec.payload_type end)