WIP

Author: sgfn

assets/publisher.js

@@ -18,6 +18,8 @@ export function createPublisherHook(iceServers = []) {
       view.fps = document.getElementById("lex-fps");
       view.bitrate = document.getElementById("lex-bitrate");
 
+      view.recordStream = document.getElementById("lex-record-stream");
+
       view.previewPlayer = document.getElementById("lex-preview-player");
 
       view.audioBitrate = document.getElementById("lex-audio-bitrate");
@@ -93,6 +95,7 @@ export function createPublisherHook(iceServers = []) {
       view.audioApplyButton.disabled = true;
       view.videoApplyButton.disabled = true;
       view.bitrate.disabled = true;
+      if (view.recordStream) view.recordStream.disabled = true;
     },
 
     enableControls(view) {
@@ -107,6 +110,7 @@ export function createPublisherHook(iceServers = []) {
       view.audioApplyButton.disabled = false;
       view.videoApplyButton.disabled = false;
       view.bitrate.disabled = false;
+      if (view.recordStream) view.recordStream.disabled = false;
     },
 
     async findDevices(view) {
@@ -173,6 +177,11 @@ export function createPublisherHook(iceServers = []) {
       console.log(`Obtained stream with id: ${view.localStream.id}`);
 
       view.previewPlayer.srcObject = view.localStream;
+
+      const config = {
+        recordStream: view.recordStream ? view.recordStream.checked : false
+      };
+      view.pushEventTo(view.el, "config-update", config);
     },
 
     async startStreaming(view) {

lib/live_ex_webrtc/publisher.ex

@@ -68,10 +68,11 @@ defmodule LiveExWebRTC.Publisher do
   use Phoenix.LiveView
 
   alias LiveExWebRTC.Publisher
-  alias ExWebRTC.{ICECandidate, PeerConnection, SessionDescription}
+  alias ExWebRTC.{ICECandidate, PeerConnection, Recorder, SessionDescription}
   alias Phoenix.PubSub
 
   @type on_connected() :: (publisher_id :: String.t() -> any())
+  @type on_disconnected() :: (publisher_id :: String.t() -> any())
 
   @type on_packet() ::
           (publisher_id :: String.t(),
@@ -85,11 +86,14 @@ defmodule LiveExWebRTC.Publisher do
   defstruct id: nil,
             pc: nil,
             streaming?: false,
+            recording?: false,
             audio_track_id: nil,
             video_track_id: nil,
             on_packet: nil,
             on_connected: nil,
+            on_disconnected: nil,
             pubsub: nil,
+            recorder: nil,
             ice_servers: nil,
             ice_ip_filter: nil,
             ice_port_range: nil,
@@ -126,7 +130,9 @@ defmodule LiveExWebRTC.Publisher do
   * `id` - publisher id. This is typically your user id (if there is users database).
   It is used to identify live view and generated HTML elements.
   * `pubsub` - a pubsub that publisher live view will use for broadcasting audio and video packets received from a browser. See module doc for more.
+  * `recorder` - WRITEME `t:GenServer.server/0`,
   * `on_connected` - callback called when the underlying peer connection changes its state to the `:connected`. See `t:on_connected/0`.
+  * `on_disconnected` - callback called when the underlying peer connection process terminates. See `t:on_disconnected/0`.
   * `on_packet` - callback called for each audio and video RTP packet. Can be used to modify the packet before publishing it on a pubsub. See `t:on_packet/0`.
   * `ice_servers` - a list of `t:ExWebRTC.PeerConnection.Configuration.ice_server/0`,
   * `ice_ip_filter` - `t:ExICE.ICEAgent.ip_filter/0`,
@@ -142,8 +148,10 @@ defmodule LiveExWebRTC.Publisher do
         :id,
         :name,
         :pubsub,
+        :recorder,
         :on_packet,
         :on_connected,
+        :on_disconnected,
         :ice_servers,
         :ice_ip_filter,
         :ice_port_range,
@@ -155,8 +163,10 @@ defmodule LiveExWebRTC.Publisher do
     publisher = %Publisher{
       id: Keyword.fetch!(opts, :id),
       pubsub: Keyword.fetch!(opts, :pubsub),
+      recorder: Keyword.get(opts, :recorder),
       on_packet: Keyword.get(opts, :on_packet),
       on_connected: Keyword.get(opts, :on_connected),
+      on_disconnected: Keyword.get(opts, :on_disconnected),
       ice_servers: Keyword.get(opts, :ice_servers, [%{urls: "stun:stun.l.google.com:19302"}]),
       ice_ip_filter: Keyword.get(opts, :ice_ip_filter),
       ice_port_range: Keyword.get(opts, :ice_port_range),
@@ -255,7 +265,7 @@ defmodule LiveExWebRTC.Publisher do
               <input
                 type="text"
                 id="lex-fps"
-                value="24"
+                value="30"
                 class="rounded-lg disabled:text-gray-400 disabled:border-gray-400 focus:border-brand focus:outline-none focus:ring-0"
               />
             </div>
@@ -268,6 +278,10 @@ defmodule LiveExWebRTC.Publisher do
                 class="rounded-lg disabled:text-gray-400 disabled:border-gray-400 focus:border-brand focus:outline-none focus:ring-0"
               />
             </div>
+            <div :if={@publisher.recorder} class="flex gap-2.5 items-center">
+              <label for="lex-record-stream">Record stream:</label>
+              <input type="checkbox" id="lex-record-stream" class="rounded-full" checked />
+            </div>
           </div>
           <button id="lex-video-apply-button" class="rounded-lg px-10 py-2.5 bg-brand disabled:bg-brand/50 hover:bg-brand/90 text-white font-bold" disabled>Apply</button>
         </details>
@@ -358,37 +372,43 @@ defmodule LiveExWebRTC.Publisher do
   def handle_info({:ex_webrtc, _pc, {:rtp, track_id, nil, packet}}, socket) do
     %{publisher: publisher} = socket.assigns
 
-    case publisher do
-      %Publisher{video_track_id: ^track_id} ->
-        packet =
-          if publisher.on_packet,
-            do: publisher.on_packet.(publisher.id, :video, packet, socket),
-            else: packet
-
-        PubSub.broadcast(
-          publisher.pubsub,
-          "streams:video:#{publisher.id}",
-          {:live_ex_webrtc, :video, packet}
-        )
+    kind =
+      case publisher do
+        %Publisher{video_track_id: ^track_id} -> :video
+        %Publisher{audio_track_id: ^track_id} -> :audio
+      end
 
-        {:noreply, socket}
+    packet =
+      if publisher.on_packet,
+        do: publisher.on_packet.(publisher.id, kind, packet, socket),
+        else: packet
 
-      %Publisher{audio_track_id: ^track_id} ->
-        PubSub.broadcast(
-          publisher.pubsub,
-          "streams:audio:#{publisher.id}",
-          {:live_ex_webrtc, :audio, packet}
-        )
+    if publisher.recording?, do: Recorder.record(publisher.recorder, track_id, nil, packet)
 
-        if publisher.on_packet, do: publisher.on_packet.(publisher.id, :audio, packet, socket)
-        {:noreply, socket}
-    end
+    PubSub.broadcast(
+      publisher.pubsub,
+      "streams:#{kind}:#{publisher.id}",
+      {:live_ex_webrtc, kind, packet}
+    )
+
+    {:noreply, socket}
   end
 
   @impl true
   def handle_info({:ex_webrtc, _pid, {:connection_state_change, :connected}}, socket) do
     %{publisher: pub} = socket.assigns
+
+    if pub.recording? do
+      [
+        %{kind: :audio, receiver: %{track: audio_track}},
+        %{kind: :video, receiver: %{track: video_track}}
+      ] = PeerConnection.get_transceivers(pub.pc)
+
+      Recorder.add_tracks(pub.recorder, [audio_track, video_track])
+    end
+
     if pub.on_connected, do: pub.on_connected.(pub.id)
+
     {:noreply, socket}
   end
 
@@ -397,6 +417,40 @@ defmodule LiveExWebRTC.Publisher do
     {:noreply, socket}
   end
 
+  @impl true
+  def handle_info({:DOWN, _ref, :process, pid, _reason}, socket) do
+    %{publisher: pub} = socket.assigns
+
+    if pid == pub.pc do
+      recorder_result =
+        if pub.recording? do
+          {:ok, manifest, maybe_upload_ref} =
+            Recorder.end_tracks(pub.recorder, [pub.audio_track_id, pub.video_track_id])
+
+          {manifest, maybe_upload_ref}
+        end
+
+      if pub.on_disconnected, do: pub.on_disconnected.(pub.id, recorder_result)
+    end
+
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_event("config-update", config, socket) do
+    %{publisher: publisher} = socket.assigns
+
+    new_publisher = %Publisher{
+      publisher
+      | recording?: config["recordStream"]
+    }
+
+    # XXX I'm 95% sure there's a race condition here somewhere
+    {:noreply,
+     socket
+     |> assign(publisher: new_publisher)}
+  end
+
   @impl true
   def handle_event("start-streaming", _, socket) do
     {:noreply,
@@ -418,6 +472,7 @@ defmodule LiveExWebRTC.Publisher do
     %{publisher: publisher} = socket.assigns
     offer = SessionDescription.from_json(unsigned_params)
     {:ok, pc} = spawn_peer_connection(socket)
+    Process.monitor(pc)
 
     :ok = PeerConnection.set_remote_description(pc, offer)
 

mix.exs

@@ -42,7 +42,8 @@ defmodule LiveExWebrtc.MixProject do
     [
       {:phoenix_live_view, "~> 1.0"},
       {:jason, "~> 1.0"},
-      {:ex_webrtc, "~> 0.7.0"},
+      # {:ex_webrtc, "~> 0.7.0"},
+      {:ex_webrtc, path: "~/git/w/w/main/", override: true},
       {:ex_doc, "~> 0.31.0", only: :dev, runtime: false}
     ]
   end