gst/pipewiresrc: Improve base_time handling

It can not generically assumed that the gstreamer clock (and therefore
the base_time) is based on CLOCK_MONOTONIC.

It was tried to use the logic provided by
GstBaseSrc::gst_base_src_do_sync() in commit 004206db37
("gst/pipewiresrc: Let GstBaseSrc handle pseudo-live calculations").
This has the downside, that a potential jitter on the first buffer is
included in the calculated time offset. In gstreamer pipelines with
multiple pipewiresrc elements and big jitter on the first buffer the
streams will stay out of sync.

Improve that by checking if the gstreamer clock is provided by pipewire
and therefore known to be CLOCK_MONOTONIC or if it is provided by
gstreamer and we need to manually calculate the base_time.

Signed-off-by: Stefan Klug <stefan.klug@ideasonboard.com>

src/gst/gstpipewiresrc.c

@@ -756,7 +756,7 @@ static GstBuffer *dequeue_buffer(GstPipeWireSrc *pwsrc)
     GST_LOG_OBJECT (pwsrc, "pts %" G_GUINT64_FORMAT ", dts_offset %" G_GUINT64_FORMAT, h->pts, h->dts_offset);
 
     if (GST_CLOCK_TIME_IS_VALID (h->pts)) {
-      GST_BUFFER_PTS (buf) = h->pts + GST_PIPEWIRE_CLOCK (pwsrc->clock)->time_offset;
+      GST_BUFFER_PTS (buf) = h->pts;
       if (GST_BUFFER_PTS (buf) + h->dts_offset > 0)
         GST_BUFFER_DTS (buf) = GST_BUFFER_PTS (buf) + h->dts_offset;
     }
@@ -1525,6 +1525,7 @@ gst_pipewire_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
   GstPipeWireSrc *pwsrc;
   GstClockTime pts, dts, base_time;
   const char *error = NULL;
+  GstClock *clock;
   GstBuffer *buf;
   gboolean update_time = FALSE, timeout = FALSE;
   GstCaps *caps = NULL;
@@ -1611,25 +1612,38 @@ gst_pipewire_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
   pw_thread_loop_unlock (pwsrc->stream->core->loop);
 
   *buffer = buf;
-
+  clock = gst_element_get_clock (GST_ELEMENT_CAST (pwsrc));
-  if (pwsrc->is_live)
-    base_time = GST_ELEMENT_CAST (psrc)->base_time;
-  else
-    base_time = 0;
 
   if (update_time) {
-    GstClock *clock = gst_element_get_clock (GST_ELEMENT_CAST (pwsrc));
     if (clock != NULL) {
       pts = dts = gst_clock_get_time (clock);
       gst_object_unref (clock);
     } else {
       pts = dts = GST_CLOCK_TIME_NONE;
     }
+
+    GST_LOG_OBJECT (pwsrc, "Sending keepalive buffer");
   } else {
     pts = GST_BUFFER_PTS (*buffer);
     dts = GST_BUFFER_DTS (*buffer);
   }
 
+  /*
+   * We need to map the pipwire time to gstreamer time. If the gstreamer clock
+   * is provided by us, we can safely use the base_time of the element.
+   * Otherwise we can not assume that the gstreamer clock is CLOCK_MONOTONIC and
+   * must therefore fall back to our own base_time. This might introduce a bit
+   * of jitter.
+   */
+  base_time = 0;
+  if (pwsrc->is_live) {
+    if (clock == pwsrc->stream->clock) {
+      base_time = gst_element_get_base_time (GST_ELEMENT_CAST (pwsrc));
+    } else {
+      base_time = pwsrc->pw_base_time;
+    }
+  }
+
   if (GST_CLOCK_TIME_IS_VALID (pts))
     pts = (pts >= base_time ? pts - base_time : 0);
   if (GST_CLOCK_TIME_IS_VALID (dts))
@@ -1741,6 +1755,7 @@ gst_pipewire_src_change_state (GstElement * element, GstStateChange transition)
       GST_DEBUG_OBJECT (this, "activating stream");
 
       pw_thread_loop_lock (this->stream->core->loop);
+      this->pw_base_time = pw_stream_get_nsec (this->stream->pwstream);
       pw_stream_set_active (this->stream->pwstream, true);
       /* if state have been paused for longer time, the underlying node might
        * be moved from idle to suspended, which would mean format cleared via

src/gst/gstpipewiresrc.h

@@ -79,6 +79,7 @@ struct _GstPipeWireSrc {
 
   gboolean is_live;
   int64_t delay;
+  uint64_t pw_base_time;
   GstClockTime min_latency;
   GstClockTime max_latency;