diff --git a/server/pkg/gst/gst.c b/server/pkg/gst/gst.c index f1258a8b..df7551dc 100644 --- a/server/pkg/gst/gst.c +++ b/server/pkg/gst/gst.c @@ -95,10 +95,31 @@ static GstFlowReturn gstreamer_send_new_sample_handler(GstElement *object, gpoin buffer = gst_sample_get_buffer(sample); if (buffer) { gst_buffer_extract_dup(buffer, 0, gst_buffer_get_size(buffer), ©, ©_size); - goHandlePipelineBuffer(ctx->pipelineId, copy, copy_size, - GST_BUFFER_DURATION(buffer), - GST_BUFFER_FLAG_IS_SET(buffer, GST_BUFFER_FLAG_DELTA_UNIT) - ); + + // Get the buffer PTS (presentation timestamp) + GstClockTime pts = GST_BUFFER_PTS(buffer); + + // Convert PTS to wallclock time by adding base time and subtracting pipeline time + GstClock *clock = gst_pipeline_get_clock(GST_PIPELINE(ctx->pipeline)); + if (clock) { + gst_object_ref(clock); + GstClockTime base_time = gst_element_get_base_time(ctx->pipeline); + GstClockTime wallclock_time = pts + base_time; + gst_object_unref(clock); + + goHandlePipelineBuffer(ctx->pipelineId, copy, copy_size, + wallclock_time, + GST_BUFFER_DURATION(buffer), + GST_BUFFER_FLAG_IS_SET(buffer, GST_BUFFER_FLAG_DELTA_UNIT) + ); + } else { + // Fallback to just PTS if clock is not available + goHandlePipelineBuffer(ctx->pipelineId, copy, copy_size, + pts, + GST_BUFFER_DURATION(buffer), + GST_BUFFER_FLAG_IS_SET(buffer, GST_BUFFER_FLAG_DELTA_UNIT) + ); + } } gst_sample_unref(sample); } diff --git a/server/pkg/gst/gst.go b/server/pkg/gst/gst.go index a9aa6709..c67195d2 100644 --- a/server/pkg/gst/gst.go +++ b/server/pkg/gst/gst.go @@ -200,7 +200,7 @@ func CheckPlugins(plugins []string) error { } //export goHandlePipelineBuffer -func goHandlePipelineBuffer(pipelineID C.int, buf C.gpointer, bufLen C.int, duration C.guint64, deltaUnit C.gboolean) { +func goHandlePipelineBuffer(pipelineID C.int, buf C.gpointer, bufLen C.int, pts C.guint64, duration C.guint64, deltaUnit C.gboolean) { defer C.g_free(buf) pipelinesLock.Lock() @@ -208,10 +208,14 @@ func goHandlePipelineBuffer(pipelineID C.int, buf C.gpointer, bufLen C.int, dura pipelinesLock.Unlock() if ok { + // Convert GStreamer PTS (in nanoseconds since epoch) to Go time.Time + // GStreamer timestamps are in nanoseconds + timestamp := time.Unix(0, int64(pts)) + pipeline.sample <- types.Sample{ Data: C.GoBytes(unsafe.Pointer(buf), bufLen), Length: int(bufLen), - Timestamp: time.Now(), + Timestamp: timestamp, Duration: time.Duration(duration), DeltaUnit: deltaUnit == C.TRUE, } diff --git a/server/pkg/gst/gst.h b/server/pkg/gst/gst.h index bdbd0347..a09a3b46 100644 --- a/server/pkg/gst/gst.h +++ b/server/pkg/gst/gst.h @@ -23,7 +23,7 @@ typedef struct GstPipelineCtx { GstElement *appsrc; } GstPipelineCtx; -extern void goHandlePipelineBuffer(int pipelineId, void *buffer, int bufferLen, guint64 duration, gboolean deltaUnit); +extern void goHandlePipelineBuffer(int pipelineId, void *buffer, int bufferLen, guint64 pts, guint64 duration, gboolean deltaUnit); extern void goPipelineLog(int pipelineId, char *level, char *msg); GstPipelineCtx *gstreamer_pipeline_create(char *pipelineStr, int pipelineId, GError **error);