Fix RC between publishers with duplicated ids (#15)

Author: mickel8

lib/live_ex_webrtc/player.ex

@@ -80,6 +80,9 @@ defmodule LiveExWebRTC.Player do
 
   @type t() :: struct()
 
+  @check_lock_timeout_ms 3000
+  @max_lock_timeout_ms 3000
+
   defstruct id: nil,
             publisher_id: nil,
             publisher_audio_track: nil,
@@ -101,7 +104,10 @@ defmodule LiveExWebRTC.Player do
             target_layer: nil,
             video_layers: [],
             # codec that will be used for video sending
-            video_send_codec: nil
+            video_send_codec: nil,
+            last_seen: nil,
+            locked: false,
+            lock_timer: nil
 
   alias ExWebRTC.{ICECandidate, MediaStreamTrack, PeerConnection, RTP.Munger, SessionDescription}
   alias ExRTCP.Packet.PayloadFeedback.PLI
@@ -351,16 +357,34 @@ defmodule LiveExWebRTC.Player do
         publisher_audio_track: ^publisher_audio_track,
         publisher_video_track: ^publisher_video_track
       } ->
-        # tracks are the same, do nothing
+        # tracks are the same, update last_seen and do nothing
+        player = %Player{player | last_seen: System.monotonic_time(:millisecond)}
+        socket = assign(socket, player: player)
+        {:noreply, socket}
+
+      %Player{locked: true} ->
+        # Different tracks but we are still receiving updates from old publisher. Ignore.
         {:noreply, socket}
 
       %Player{
         publisher_audio_track: old_publisher_audio_track,
         publisher_video_track: old_publisher_video_track,
-        video_layers: old_layers
+        video_layers: old_layers,
+        locked: false
       } ->
         if player.pc, do: PeerConnection.close(player.pc)
 
+        if player.lock_timer do
+          Process.cancel_timer(player.lock_timer)
+
+          # flush mailbox
+          receive do
+            :check_lock -> :ok
+          after
+            0 -> :ok
+          end
+        end
+
         video_layers = (publisher_video_track && publisher_video_track.rids) || ["h"]
 
         video_layers =
@@ -378,7 +402,10 @@ defmodule LiveExWebRTC.Player do
             layer: "h",
             target_layer: "h",
             video_layers: video_layers,
-            munger: nil
+            munger: nil,
+            last_seen: System.monotonic_time(:millisecond),
+            locked: true,
+            lock_timer: Process.send_after(self(), :check_lock, @check_lock_timeout_ms)
         }
 
         socket = assign(socket, :player, player)
@@ -406,6 +433,25 @@ defmodule LiveExWebRTC.Player do
     end
   end
 
+  @impl true
+  def handle_info({:live_ex_webrtc, :bye, publisher_audio_track, publisher_video_track}, socket) do
+    %{player: player} = socket.assigns
+
+    case player do
+      %Player{
+        publisher_audio_track: ^publisher_audio_track,
+        publisher_video_track: ^publisher_video_track
+      } ->
+        player = %Player{player | locked: false}
+        socket = assign(socket, player: player)
+        {:noreply, socket}
+
+      _ ->
+        {:noreply, socket}
+    end
+  end
+
+  @impl true
   def handle_info({:live_ex_webrtc, :audio, packet}, socket) do
     %{player: player} = socket.assigns
 
@@ -418,6 +464,7 @@ defmodule LiveExWebRTC.Player do
     {:noreply, socket}
   end
 
+  @impl true
   def handle_info({:live_ex_webrtc, :video, rid, packet}, socket) do
     %{player: player} = socket.assigns
 
@@ -461,6 +508,30 @@ defmodule LiveExWebRTC.Player do
     end
   end
 
+  @impl true
+  def handle_info(:check_lock, %{assigns: %{player: %Player{locked: true} = player}} = socket) do
+    now = System.monotonic_time(:millisecond)
+
+    if now - socket.assigns.player.last_seen > @max_lock_timeout_ms do
+      # unlock i.e. allow for track update
+      player = %Player{player | lock_timer: nil, locked: false}
+      socket = assign(socket, :player, player)
+      {:noreply, socket}
+    else
+      timer = Process.send_after(self(), :check_lock, @check_lock_timeout_ms)
+      player = %Player{player | lock_timer: timer}
+      socket = assign(socket, :player, player)
+      {:noreply, socket}
+    end
+  end
+
+  @impl true
+  def handle_info(:check_lock, socket) do
+    player = %Player{socket.assigns.player | lock_timer: nil}
+    socket = assign(socket, :player, player)
+    {:noreply, socket}
+  end
+
   @impl true
   def handle_event("toggle-settings", _params, socket) do
     socket =

lib/live_ex_webrtc/publisher.ex

@@ -25,7 +25,7 @@ defmodule LiveExWebRTC.Publisher do
   `{:live_ex_webrtc, :audio, ExRTP.Packet.t()}`. Packets for non-simulcast video tracks are always
   sent with "h" identifier.
   * `streams:info:#{publisher.id}"` - for receiving information about publisher tracks and their layers.
-  The message is in form of: `{:live_ex_webrtc, :info, audio_track :: ExWebRTC.MediaStreamTrack.t(), video_track :: ExWebRTC.MediaStreamTrack.t()}`.
+  The message is in form of: `{:live_ex_webrtc, :info | :bye, audio_track :: ExWebRTC.MediaStreamTrack.t(), video_track :: ExWebRTC.MediaStreamTrack.t()}`.
   * `publishers:#{publisher_id}` for sending keyframe request.
   The message must be in form of `{:live_ex_webrtc, :keyframe_req, "l" | "m" | "h"}`
   E.g.
@@ -139,6 +139,7 @@ defmodule LiveExWebRTC.Publisher do
 
   defstruct id: nil,
             pc: nil,
+            pc_mref: nil,
             streaming?: false,
             simulcast_supported?: nil,
             # record checkbox status
@@ -160,7 +161,8 @@ defmodule LiveExWebRTC.Publisher do
             ice_port_range: nil,
             audio_codecs: nil,
             video_codecs: nil,
-            pc_genserver_opts: nil
+            pc_genserver_opts: nil,
+            info_timer: nil
 
   attr(:socket, Phoenix.LiveView.Socket, required: true, doc: "Parent live view socket")
 
@@ -507,7 +509,6 @@ defmodule LiveExWebRTC.Publisher do
       socket =
         receive do
           {^ref, %Publisher{id: ^pub_id} = publisher} ->
-            Process.send_after(self(), :streams_info, 1000)
             codecs = publisher.video_codecs || PeerConnection.Configuration.default_video_codecs()
             publisher = %Publisher{publisher | simulcast_supported?: simulcast_supported?(codecs)}
             assign(socket, publisher: publisher)
@@ -522,7 +523,11 @@ defmodule LiveExWebRTC.Publisher do
   end
 
   @impl true
-  def handle_info({:live_ex_webrtc, :keyframe_req, layer}, socket) do
+  def handle_info(
+        {:live_ex_webrtc, :keyframe_req, layer},
+        %{assigns: %{publisher: %{pc: pc}}} = socket
+      )
+      when pc != nil do
     %{publisher: publisher} = socket.assigns
 
     # Non-simulcast tracks are always sent with "h" identifier
@@ -535,9 +540,7 @@ defmodule LiveExWebRTC.Publisher do
         layer
       end
 
-    if pc = publisher.pc do
-      :ok = PeerConnection.send_pli(pc, publisher.video_track.id, layer)
-    end
+    :ok = PeerConnection.send_pli(pc, publisher.video_track.id, layer)
 
     {:noreply, socket}
   end
@@ -575,6 +578,10 @@ defmodule LiveExWebRTC.Publisher do
   def handle_info({:ex_webrtc, _pid, {:connection_state_change, :connected}}, socket) do
     %{publisher: pub} = socket.assigns
 
+    info_timer = Process.send_after(self(), :streams_info, 1000)
+    pub = %Publisher{pub | info_timer: info_timer}
+    socket = assign(socket, :publisher, pub)
+
     if pub.record? do
       [
         %{kind: :audio, receiver: %{track: audio_track}},
@@ -589,6 +596,11 @@ defmodule LiveExWebRTC.Publisher do
     {:noreply, socket}
   end
 
+  @impl true
+  def handle_info({:ex_webrtc, _pid, {:connection_state_change, :failed}}, socket) do
+    {:noreply, bye(socket)}
+  end
+
   @impl true
   def handle_info({:ex_webrtc, _, _}, socket) do
     {:noreply, socket}
@@ -598,31 +610,33 @@ defmodule LiveExWebRTC.Publisher do
   def handle_info(:streams_info, socket) do
     %{publisher: publisher} = socket.assigns
 
-    PubSub.broadcast(
-      publisher.pubsub,
-      "streams:info:#{publisher.id}",
-      {:live_ex_webrtc, :info, publisher.audio_track, publisher.video_track}
-    )
-
-    Process.send_after(self(), :streams_info, 1_000)
-
-    {:noreply, socket}
+    if publisher.audio_track != nil or publisher.video_track != nil do
+      PubSub.broadcast(
+        publisher.pubsub,
+        "streams:info:#{publisher.id}",
+        {:live_ex_webrtc, :info, publisher.audio_track, publisher.video_track}
+      )
+
+      info_timer = Process.send_after(self(), :streams_info, 1_000)
+      publisher = %Publisher{publisher | info_timer: info_timer}
+      socket = assign(socket, :publisher, publisher)
+      {:noreply, socket}
+    else
+      {:noreply, socket}
+    end
   end
 
+  @impl true
   def handle_info(
         {:DOWN, _ref, :process, pc, _reason},
-        %{assigns: %{publisher: %{pc: pc} = pub}} = socket
+        %{assigns: %{publisher: %{pc: pc}}} = socket
       ) do
-    if pub.record? do
-      recorder_result =
-        Recorder.end_tracks(pub.recorder, [pub.audio_track.id, pub.video_track.id])
-
-      if pub.on_recording_finished, do: pub.on_recording_finished.(pub.id, recorder_result)
-    end
-
-    if pub.on_disconnected, do: pub.on_disconnected.(pub.id)
+    {:noreply, bye(socket)}
+  end
 
-    {:noreply, assign(socket, publisher: %Publisher{pub | streaming?: false})}
+  @impl true
+  def handle_info(_msg, socket) do
+    {:noreply, socket}
   end
 
   @impl true
@@ -647,27 +661,21 @@ defmodule LiveExWebRTC.Publisher do
 
   @impl true
   def handle_event("stop-streaming", _, socket) do
-    {:noreply,
-     socket
-     |> assign(publisher: %Publisher{socket.assigns.publisher | streaming?: false})
-     |> push_event("stop-streaming", %{})}
+    {:noreply, bye(socket)}
   end
 
   @impl true
   def handle_event("record-stream-change", params, socket) do
     record? = params["value"] == "on"
-
-    {:noreply,
-     socket
-     |> assign(publisher: %Publisher{socket.assigns.publisher | record?: record?})}
+    {:noreply, assign(socket, publisher: %Publisher{socket.assigns.publisher | record?: record?})}
   end
 
   @impl true
   def handle_event("offer", unsigned_params, socket) do
     %{publisher: publisher} = socket.assigns
     offer = SessionDescription.from_json(unsigned_params)
     {:ok, pc} = spawn_peer_connection(socket)
-    Process.monitor(pc)
+    pc_mref = Process.monitor(pc)
 
     :ok = PeerConnection.set_remote_description(pc, offer)
 
@@ -687,6 +695,7 @@ defmodule LiveExWebRTC.Publisher do
     new_publisher = %Publisher{
       publisher
       | pc: pc,
+        pc_mref: pc_mref,
         audio_track: audio_track,
         video_track: video_track
     }
@@ -731,6 +740,9 @@ defmodule LiveExWebRTC.Publisher do
     end
   end
 
+  @impl true
+  def terminate(_reason, socket), do: bye(socket)
+
   defp spawn_peer_connection(socket) do
     %{publisher: publisher} = socket.assigns
 
@@ -770,4 +782,58 @@ defmodule LiveExWebRTC.Publisher do
         false
     end)
   end
+
+  defp bye(socket) do
+    %{publisher: publisher} = socket.assigns
+
+    if publisher.audio_track != nil or publisher.video_track != nil do
+      PubSub.broadcast(
+        publisher.pubsub,
+        "streams:info:#{publisher.id}",
+        {:live_ex_webrtc, :bye, publisher.audio_track, publisher.video_track}
+      )
+    end
+
+    if publisher.info_timer != nil, do: Process.cancel_timer(publisher.info_timer)
+
+    receive do
+      :streams_info -> :ok
+    after
+      0 -> :ok
+    end
+
+    try do
+      Process.demonitor(publisher.pc_mref)
+      PeerConnection.close(publisher.pc)
+    catch
+      _, _ -> :ok
+    end
+
+    if publisher.record? and publisher.audio_track != nil and publisher.video_track != nil do
+      recorder_result =
+        Recorder.end_tracks(publisher.recorder, [
+          publisher.audio_track.id,
+          publisher.video_track.id
+        ])
+
+      if publisher.on_recording_finished,
+        do: publisher.on_recording_finished.(publisher.id, recorder_result)
+    end
+
+    if publisher.on_disconnected, do: publisher.on_disconnected.(publisher.id)
+
+    publisher = %Publisher{
+      publisher
+      | info_timer: nil,
+        audio_track: nil,
+        video_track: nil,
+        streaming?: false,
+        pc: nil,
+        pc_mref: nil
+    }
+
+    socket
+    |> assign(:publisher, publisher)
+    |> push_event("stop-streaming", %{})
+  end
 end