Add delete session endpoint

Author: gBillal

lib/plug/router/webrtc.ex

@@ -53,5 +53,10 @@ if Code.ensure_loaded?(Plug) do
           send_resp(conn, 415, "Unsupported Media Type")
       end
     end
+
+    delete "/:source_id/whep/:session_id" do
+      Shinkai.Sources.remove_webrtc_peer(source_id, session_id)
+      send_resp(conn, 204, "")
+    end
   end
 end

lib/shinkai/pipeline.ex

@@ -17,6 +17,10 @@ defmodule Shinkai.Pipeline do
     Sink.WebRTC.handle_peer_answer(:"webrtc_sink_#{source_id}", session_id, sdp_answer)
   end
 
+  def remove_webrtc_peer(source_id, session_id) do
+    Sink.WebRTC.remove_peer(:"webrtc_sink_#{source_id}", session_id)
+  end
+
   def stop(source_id) do
     Supervisor.stop(:"#{source_id}")
   end

lib/shinkai/sink/webrtc.ex

@@ -33,22 +33,24 @@ defmodule Shinkai.Sink.WebRTC do
     GenServer.call(server, {:handle_peer_answer, session_id, sdp})
   end
 
+  @spec remove_peer(server :: pid() | atom(), session_id :: String.t()) :: :ok
+  def remove_peer(server, session_id) do
+    GenServer.cast(server, {:remove_peer, session_id})
+  end
+
   @impl true
   def init(opts) do
     source_id = opts[:id]
     {:ok, peer_manager} = PeerManager.start_link(source_id: source_id)
 
     PubSub.subscribe(Shinkai.PubSub, tracks_topic(source_id))
 
-    {:ok, sock} = :gen_udp.open(0, [:binary, active: true])
-
     {:ok,
      %{
        peer_manager: peer_manager,
        source_id: source_id,
        packets_topic: packets_topic(source_id),
-       tracks: %{},
-       sock: sock
+       tracks: %{}
      }}
   end
 
@@ -67,6 +69,12 @@ defmodule Shinkai.Sink.WebRTC do
     {:noreply, state}
   end
 
+  @impl true
+  def handle_cast({:remove_peer, session_id}, state) do
+    :ok = PeerManager.remove_peer(state.peer_manager, session_id)
+    {:noreply, state}
+  end
+
   @impl true
   def handle_info({:tracks, tracks}, state) do
     {tracks, unsupported_tracks} = Enum.split_with(tracks, &(&1.codec in @supported_codecs))
@@ -87,7 +95,7 @@ defmodule Shinkai.Sink.WebRTC do
       |> Enum.reject(&is_nil/1)
       |> Enum.reduce(state, fn track, state ->
         media_stream = ExWebRTC.MediaStreamTrack.new(track.type, [stream_id])
-        webrtc_track = webrtc_track(track, stream_id)
+        webrtc_track = webrtc_track(track)
 
         payloader_mod = payloader_mod(track.codec)
 
@@ -168,13 +176,15 @@ defmodule Shinkai.Sink.WebRTC do
     %{track_ctx | payloader_state: payloader_state}
   end
 
-  defp webrtc_track(track, stream_id) do
+  defp webrtc_track(track) do
+    pt = payload_type(track.codec)
+
     %RTPCodecParameters{
-      payload_type: payload_type(track.codec),
+      payload_type: pt,
       mime_type: mime_type(track.codec),
       clock_rate: clock_rate(track),
       channels: if(track.type == :audio, do: 1, else: nil),
-      sdp_fmtp_line: sdp_fmtp_line(track.codec, payload_type)
+      sdp_fmtp_line: sdp_fmtp_line(track.codec, pt)
     }
   end
 

lib/shinkai/sink/webrtc/peer_manager.ex

@@ -36,6 +36,11 @@ defmodule Shinkai.Sink.WebRTC.PeerManager do
     GenServer.cast(manager, {:handle_peer_answer, from, session_id, sdp_answer})
   end
 
+  @spec remove_peer(server :: pid() | atom(), session_id :: String.t()) :: :ok
+  def remove_peer(manager, session_id) do
+    GenServer.call(manager, {:remove_peer, session_id})
+  end
+
   @impl true
   def init(opts) do
     {:ok,
@@ -57,6 +62,20 @@ defmodule Shinkai.Sink.WebRTC.PeerManager do
     {:reply, :ok, %{state | audio_track: track}}
   end
 
+  def handle_call({:remove_peer, session_id}, _from, state) do
+    Logger.info("Removing WebRTC peer with session ID: #{session_id}")
+
+    case Registry.match(Shinkai.Registry, {:webrtc, state.source_id}, {:_, session_id}) do
+      [{_pid, {pc, _session_id}}] -> unregister(state.source_id, pc)
+      [] -> :ok
+    end
+
+    {pc, sessions} = Map.pop(state.sessions, session_id)
+    if pc, do: PeerConnection.stop(pc)
+
+    {:reply, :ok, %{state | sessions: sessions}}
+  end
+
   @impl true
   def handle_cast({:add_peer, from}, state) do
     video_tracks = if state.video_track, do: [elem(state.video_track, 1)], else: []
@@ -125,10 +144,12 @@ defmodule Shinkai.Sink.WebRTC.PeerManager do
   end
 
   def handle_info({:ex_webrtc, pid, {:connection_state_change, connection_state}}, state)
-      when connection_state in [:failed, :closed, :disconnected] do
-    Logger.info("WebRTC PeerConnection #{inspect(pid)} connection state: #{connection_state}")
-    PeerConnection.stop(pid)
-    Registry.unregister_match(Shinkai.Registry, {:webrtc, state.source_id}, {pid, :_})
+      when connection_state in [:failed, :closed] do
+    Logger.debug(
+      "WebRTC PeerConnection #{inspect(pid)} connection state changed to: #{connection_state}"
+    )
+
+    unregister(state.source_id, pid)
     {:noreply, state}
   end
 
@@ -137,7 +158,7 @@ defmodule Shinkai.Sink.WebRTC.PeerManager do
   end
 
   def handle_info({:ex_webrtc, _pid, msg}, state) do
-    Logger.info("Unhandled ExWebRTC message: #{inspect(msg)}")
+    # Logger.info("Unhandled ExWebRTC message: #{inspect(msg)}")
     {:noreply, state}
   end
 
@@ -153,4 +174,9 @@ defmodule Shinkai.Sink.WebRTC.PeerManager do
       end
     end)
   end
+
+  defp unregister(source_id, pc) do
+    PeerConnection.stop(pc)
+    Registry.unregister_match(Shinkai.Registry, {:webrtc, source_id}, {pc, :_})
+  end
 end

lib/shinkai/sources.ex

@@ -54,6 +54,7 @@ defmodule Shinkai.Sources do
 
   defdelegate add_webrtc_peer(source_id), to: Shinkai.Pipeline
   defdelegate handle_webrtc_peer_answer(source_id, session_id, sdp_answer), to: Shinkai.Pipeline
+  defdelegate remove_webrtc_peer(source_id, session_id), to: Shinkai.Pipeline
 
   defp storage_impl do
     Application.get_env(:shinkai, :storage_impl, Shinkai.Sources.Storage.File)