frontends (tools/webrtc): Add recording to file descriptor:

A new `record_fd` argument can now be used to specify a file descriptor. If set, the
incoming stream will be written there.

For now `record_fd` must be specified on `WebRTC` creation. In the future, it should
be possible to add or remove it it during a call.

libervia/frontends/tools/webrtc.py

@@ -17,6 +17,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from collections.abc import Awaitable
+import time
 import gi
 
 gi.require_versions({"Gst": "1.0", "GstWebRTC": "1.0"})
@@ -104,6 +105,7 @@ class WebRTC:
         target_size: tuple[int, int] | None = None,
         call_start_cb: Callable[[str, dict, str], Awaitable[str]] | None = None,
         dc_data_list: list[SourcesDataChannel | SinksDataChannel] | None = None,
+        record_fd: int | None = None
     ) -> None:
         """Initializes a new WebRTC instance.
 
@@ -130,12 +132,14 @@ class WebRTC:
             used for SinksAuto only for now).
         @param target_size: Expected size of the final sink stream. Mainly used by composer
             when ``merge_pip`` is set.
-            None to autodetect (no real autodetection implemeted yet, default to
+            None to autodetect (no real autodetection implemented yet, default to
             (1280,720)).
         @param call_start_cb: Called when call is started.
         @param dc_data_list: Data Channels to create.
             If a SourcesDataChannel is used as ``sources_data``, or a SinksDataChannel is
             used as ``sinks_data``, they will be automatically added to this list.
+        @param record_fd: File descriptor for recording the incoming stream.
+            If None, the stream will not be recorded.
         """
         self.main_loop = asyncio.get_event_loop()
         self.bridge = bridge
@@ -182,6 +186,9 @@ class WebRTC:
         else:
             self.desktop_portal = None
         self._gap_recovery_timeout = None
+        self._record_fd: int | None = record_fd
+        self._recording_elements: list[Gst.Element] = []
+        self._recording_muxer: Gst.Element | None = None
         self.reset_instance()
 
     @property
@@ -270,6 +277,31 @@ class WebRTC:
             # Also send any local candidates we buffered before this point.
             self._send_buffered_local_candidates()
 
+    @property
+    def record_fd(self) -> int|None:
+        return self._record_fd
+
+    @record_fd.setter
+    def record_fd(self, value: int|None) -> None:
+        """Set or unset the recording file descriptor.
+
+        When set during an active call, starts recording.
+        When unset during recording, stops recording.
+        """
+        if self._record_fd == value:
+            return
+
+        if value is not None:
+            raise NotImplementedError(
+                "Changing record_fd during a call is not supported yet"
+            )
+
+        self._record_fd = value
+
+        if value is None:
+            log.info("Stopping recording.")
+            self._remove_recording_branch()
+
     def bind(self, **kwargs: Callable) -> None:
         self.bindings.clear()
         for key, cb in kwargs.items():
@@ -471,6 +503,8 @@ class WebRTC:
 
     def reset_instance(self):
         """Inits or resets the instance variables to their default state."""
+        if self._recording_elements:
+            self._remove_recording_branch()
         self.role: str | None = None
         if self.pipeline is not None:
             self.set_pipeline_state(Gst.State.NULL)
@@ -1019,6 +1053,50 @@ class WebRTC:
         else:
             log.error("Failed to send GstForceKeyUnit event to decodebin.")
 
+    def _remove_recording_branch(self) -> None:
+        """Removes the recording branch from the pipeline."""
+        if not self.pipeline or not self._recording_elements:
+            log.warning("No recording branch to remove")
+            return
+
+        log.info("Removing recording branch from the pipeline.")
+
+        current_state = self.pipeline.get_state(0)[1]
+
+        if current_state == Gst.State.PLAYING:
+            self.set_pipeline_state(Gst.State.PAUSED)
+
+        try:
+            # Release muxer pads before removing elements
+            if self._recording_muxer:
+                for element in self._recording_elements:
+                    if "depay" in element.get_name():
+                        depay_src_pad = element.get_static_pad("src")
+                        if depay_src_pad:
+                            peer_pad = depay_src_pad.get_peer()
+                            if peer_pad and self._recording_muxer == peer_pad.get_parent():
+                                log.debug(f"Releasing muxer pad: {peer_pad.get_name()}")
+                                self._recording_muxer.release_request_pad(peer_pad)
+
+            # Remove elements in reverse order
+            elements_to_remove = list(self._recording_elements)
+            for element in reversed(elements_to_remove):
+                if element.get_parent() == self.pipeline:
+                    log.debug(f"Removing recording element: {element.get_name()}")
+                    element.set_state(Gst.State.NULL)
+                    self.pipeline.remove(element)
+                    self._recording_elements.remove(element)
+
+            self._recording_muxer = None
+            log.info("Recording branch successfully removed")
+        except Exception:
+            log.exception("Error removing recording branch")
+            self._recording_elements.clear()
+            self._recording_muxer = None
+        finally:
+            if current_state == Gst.State.PLAYING:
+                self.set_pipeline_state(Gst.State.PLAYING)
+
     def on_remote_decodebin_stream(self, decodebin, pad: Gst.Pad) -> None:
         """Handle the stream from the remote decodebin.
 
@@ -1211,46 +1289,215 @@ class WebRTC:
             log.warning(f"Ignoring unknown pad type from decodebin: {name}")
 
     def on_pad_added(self, webrtcbin, pad: Gst.Pad) -> None:
-        """Handle the addition of a new pad to the element.
+        """Handle the addition of a new pad from webrtcbin.
 
-        When a new source pad is added to the element, this method creates a decodebin,
+        Always creates a tee to allow dynamic recording branch addition.
-        connects it to handle the stream, and links the pad to the decodebin.
+        The tee splits the stream: one branch goes to decodebin for playback,
+        another can go to recording if enabled.
 
         @param webrtcbin: Signal source.
         @param pad: The newly added pad.
         """
         assert self.pipeline is not None
-        log.debug(f"Attempting to handle new source pad {pad.get_name()} from {webrtcbin.get_name()}")
+        log.debug(f"Handling new source pad {pad.get_name()}.")
+
         if pad.direction != Gst.PadDirection.SRC:
+            log.debug(f"Ignoring non-source pad: {pad.get_name()}")
             return
 
-        # Create the potential downstream element.
+        # Determine media type from caps.
-        decodebin = Gst.ElementFactory.make("decodebin")
+        caps = pad.get_current_caps() or pad.query_caps(None)
-        if not decodebin:
+        if not caps or caps.get_size() == 0:
-            log.error("Critical: Failed to create decodebin element")
+            log.warning("Empty caps on pad, ignoring.")
             return
 
-        decodebin.connect("pad-added", self.on_remote_decodebin_stream)
+        structure = caps.get_structure(0)
+        structure_name = structure.get_name()
+        log.debug(f"Pad caps structure: {structure_name}.")
 
-        self.pipeline.add(decodebin)
+        # We only handle RTP stream.
-        decodebin.sync_state_with_parent()
+        if structure_name != "application/x-rtp":
+            log.debug(f"Ignoring non-RTP pad with caps: {structure_name}.")
+            return
 
-        link_ret = pad.link(decodebin.get_static_pad("sink"))
+        # Extract media type.
+        media_type = structure.get_string("media")
+        if media_type not in ["video", "audio"]:
+            log.warning(f"Ignoring RTP pad with unexpected media type: {media_type}.")
+            return
 
-        if link_ret == Gst.PadLinkReturn.OK:
+        log.debug(f"Processing incoming {media_type} stream.")
-            log.debug(f"Successfully linked pad '{pad.get_name()}' to a new decodebin.")
+
-        elif link_ret == Gst.PadLinkReturn.WAS_LINKED:
+        current_state = self.pipeline.get_state(0)[1]
-            log.debug(f"Pad '{pad.get_name()}' was already linked. Cleaning up redundant decodebin.")
+        needs_state_restore = (current_state == Gst.State.PLAYING)
-            # The pad is already handled. We must clean up the decodebin we just created.
+
-            decodebin.set_state(Gst.State.NULL)
+        if needs_state_restore:
-            self.pipeline.remove(decodebin)
+            log.debug("Pausing pipeline for safe modification.")
-        else:
+            self.set_pipeline_state(Gst.State.PAUSED)
+
+        tee_name = f"{media_type}_tee"
+        decodebin_name = f"decodebin_{media_type}"
+
+        try:
+            # Create tee for this media stream (always present)
+            tee = Gst.ElementFactory.make("tee", tee_name)
+            if not tee:
+                log.error(f"Failed to create tee for {media_type}.")
+                return
+
+            self.pipeline.add(tee)
+
+            # Create decodebin for playback
+            decodebin = Gst.ElementFactory.make("decodebin", decodebin_name)
+            if not decodebin:
+                log.error(f"Failed to create decodebin for {media_type}.")
+                self.pipeline.remove(tee)
+                return
+
+            decodebin.connect("pad-added", self.on_remote_decodebin_stream)
+            self.pipeline.add(decodebin)
+
+            # Link: webrtcbin -> tee
+            if pad.link(tee.get_static_pad("sink")) != Gst.PadLinkReturn.OK:
+                log.error(f"Failed to link webrtcbin pad to tee for {media_type}.")
+                self.pipeline.remove(tee)
+                self.pipeline.remove(decodebin)
+                return
+
+            # Link tee -> decodebin (playback branch)
+            playback_pad = tee.get_request_pad("src_%u")
+            if playback_pad.link(decodebin.get_static_pad("sink")) != Gst.PadLinkReturn.OK:
+                log.error(f"Failed to link tee to decodebin for {media_type}.")
+                self.pipeline.remove(tee)
+                self.pipeline.remove(decodebin)
+                return
+
+            tee.sync_state_with_parent()
+            decodebin.sync_state_with_parent()
+
+            # If recording is active, add recording branch from tee
+            if self._record_fd is not None:
+                log.info(f"Recording is active, adding recording branch for {media_type}.")
+                recording_pad = tee.get_request_pad("src_%u")
+                if not self._add_rtp_recording_branch(recording_pad, media_type):
+                    log.warning(f"Failed to add recording branch for {media_type}")
+                    tee.release_request_pad(recording_pad)
+                else:
+                    log.info(f"Recording branch added for {media_type}.")
+
+            log.info(f"Successfully set up {media_type} pipeline with tee.")
+
+        except Exception:
+            log.exception(f"Error in on_pad_added for {media_type}.")
+            self._cleanup_elements([tee_name, decodebin_name])
+
+        finally:
+            if needs_state_restore:
+                log.debug("Restoring pipeline to PLAYING state.")
+                self.set_pipeline_state(Gst.State.PLAYING)
+            log.debug("Finished handling pad addition")
+
+    def _add_rtp_recording_branch(self, src_pad: Gst.Pad, media_type: str) -> bool:
+        """Add recording branch for RTP stream.
+
+        RTP stream are muxed and written as-is, without re-encoding.
+
+        @param src_pad: Source pad.
+        @param media_type: Type of the media ("video" or "audio").
+        @return: True is successful.
+        """
+        assert self.pipeline is not None
+        assert self._record_fd is not None
+
+        log.debug(f"Adding RTP recording branch for {media_type}")
+
+        queue = Gst.ElementFactory.make("queue", f"rec_queue_{media_type}")
+        if not queue:
+            log.error(f"Failed to create queue for {media_type} recording")
+            return False
+
+        depayloader_type = "rtpvp8depay" if media_type == "video" else "rtpopusdepay"
+        depayloader_name = f"{depayloader_type}_rec_{media_type}"
+        depayloader = Gst.ElementFactory.make(depayloader_type, depayloader_name)
+
+        if not depayloader:
+            log.error(f"Failed to create {depayloader_type}")
+            return False
+
+        # Create muxer and sink if they don't exist
+        if self._recording_muxer is None:
+            self._recording_muxer = Gst.ElementFactory.make("webmmux", "recording_muxer")
+            fdsink = Gst.ElementFactory.make("fdsink", "recording_sink")
+
+            if not self._recording_muxer or not fdsink:
+                log.error("Failed to create muxer or sink")
+                self._recording_muxer = None
+                return False
+
+            # Configure muxer for live streaming
+            self._recording_muxer.set_property("streamable", True)
+
+            fdsink.set_property("fd", self._record_fd)
+            fdsink.set_property("sync", False)
+            fdsink.set_property("async", False)
+
+            self.pipeline.add(self._recording_muxer)
+            self.pipeline.add(fdsink)
+
+            if not self._recording_muxer.link(fdsink):
+                log.error("Failed to link muxer to sink")
+                self._cleanup_elements([self._recording_muxer, fdsink], self._recording_elements)
+                self._recording_muxer = None
+                return False
+
+            self._recording_elements.extend([self._recording_muxer, fdsink])
+            self._recording_muxer.sync_state_with_parent()
+            fdsink.sync_state_with_parent()
+
+        self.pipeline.add(queue)
+        self.pipeline.add(depayloader)
+
+        # Link: src_pad -> queue -> depayloader
+        if src_pad.link(queue.get_static_pad("sink")) != Gst.PadLinkReturn.OK:
+            log.error(f"Failed to link tee src pad to queue for {media_type}")
+            self._cleanup_elements([queue, depayloader], self._recording_elements)
+            return False
+
+        if not queue.link(depayloader):
+            log.error(f"Failed to link queue to depayloader for {media_type}")
+            self._cleanup_elements([queue, depayloader], self._recording_elements)
+            return False
+
+        # Link depayloader to muxer
+        depayloader_src_pad = depayloader.get_static_pad("src")
+        muxer_pad_name = "video_%u" if media_type == "video" else "audio_%u"
+        muxer_sink_pad = self._recording_muxer.get_request_pad(muxer_pad_name)
+
+        if not muxer_sink_pad:
+            log.error(f"Failed to get request pad '{muxer_pad_name}' from muxer")
+            self._cleanup_elements([queue, depayloader], self._recording_elements)
+            return False
+
+        link_result = depayloader_src_pad.link(muxer_sink_pad)
+        if link_result != Gst.PadLinkReturn.OK:
+            if hasattr(link_result, "value_nick"):
+                reason = link_result.value_nick
+            else:
+                reason = link_result
             log.error(
-                f"Failed to link pad '{pad.get_name()}' for an unexpected reason: {link_ret}. "
+                f"Failed to link depayloader to muxer for {media_type} (reason: {reason})"
-                "Cleaning up."
             )
-            decodebin.set_state(Gst.State.NULL)
+            self._cleanup_elements([queue, depayloader], self._recording_elements)
-            self.pipeline.remove(decodebin)
+            self._recording_muxer.release_request_pad(muxer_sink_pad)
+            return False
+
+        queue.sync_state_with_parent()
+        depayloader.sync_state_with_parent()
+
+        self._recording_elements.extend([queue, depayloader])
+
+        log.debug(f"Successfully added {media_type} RTP recording branch")
+        return True
 
     def _setup_stdin_source(self) -> None:
         """Setup stdin source with decodebin for automatic format detection."""
@@ -1859,22 +2106,22 @@ class WebRTC:
         # then convert it to a more standard format that videorate can handle easily.
         if not desktop_src.link(video_scale):
             log.error("Failed to link desktop_src to videoscale")
-            self._cleanup_desktop_elements(desktop_elements)
+            self._cleanup_elements(desktop_elements)
             return
 
         if not video_scale.link(video_convert):
             log.error("Failed to link videoscale to videoconvert")
-            self._cleanup_desktop_elements(desktop_elements)
+            self._cleanup_elements(desktop_elements)
             return
 
         if not video_convert.link(video_rate):
             log.error("Failed to link videoconvert to videorate")
-            self._cleanup_desktop_elements(desktop_elements)
+            self._cleanup_elements(desktop_elements)
             return
 
         if not video_rate.link(queue):
             log.error("Failed to link videorate to queue")
-            self._cleanup_desktop_elements(desktop_elements)
+            self._cleanup_elements(desktop_elements)
             return
 
         sink_pad_template = self.video_selector.get_pad_template("sink_%u")
@@ -1883,7 +2130,7 @@ class WebRTC:
         )
         if not self.desktop_sink_pad:
             log.error("Failed to request a new sink pad from video_selector.")
-            self._cleanup_desktop_elements(desktop_elements)
+            self._cleanup_elements(desktop_elements)
             return
 
         log.debug(f"Requested and received pad {self.desktop_sink_pad.get_name()} from video_selector.")
@@ -1892,7 +2139,7 @@ class WebRTC:
         link_ret = queue_src_pad.link(self.desktop_sink_pad)
         if link_ret != Gst.PadLinkReturn.OK:
             log.error(f"Failed to link queue to video_selector: {link_ret}")
-            self._cleanup_desktop_elements(desktop_elements)
+            self._cleanup_elements(desktop_elements)
             self.video_selector.release_request_pad(self.desktop_sink_pad)
             self.desktop_sink_pad = None
             return
@@ -1903,15 +2150,27 @@ class WebRTC:
 
         log.debug("Desktop source pipeline successfully created and linked.")
 
-    def _cleanup_desktop_elements(self, elements: list[Gst.Element]) -> None:
+    def _cleanup_elements(
-        """Helper to cleanup elements on error.
+        self,
+        elements: list[Gst.Element],
+        elements_list: list[Gst.Element]|None = None
+    ) -> None:
+        """Helper to cleanup elements.
 
         @param elements: List of elements to remove from pipeline
         """
+        if self.pipeline is None:
+            return
+
         for element in elements:
-            if element:
+            try:
-                element.set_state(Gst.State.NULL)
+                if element:
-                self.pipeline.remove(element)
+                    element.set_state(Gst.State.NULL)
+                    self.pipeline.remove(element)
+                    if elements_list and element in elements_list:
+                        elements_list.remove(element)
+            except Exception as e:
+                log.warning(f"Error cleaning up element {element.get_name() if element else 'Unknown'}: {e}")
 
     def _remove_desktop_source(self, desktop_src: Gst.Element) -> None:
         """Remove the desktop source from the pipeline.
@@ -1928,10 +2187,7 @@ class WebRTC:
         # Unlink and remove in reverse order
         elements = [queue, capsfilter, video_scale, video_rate, video_convert, desktop_src]
 
-        for element in elements:
+        self._cleanup_elements(elements)
-            if element:
-                element.set_state(Gst.State.NULL)
-                self.pipeline.remove(element)
 
         # Release the pad associated with the desktop source
         if self.desktop_sink_pad: