source: send latency to client

Send the latency to the client and use this to configure the pinossrc
element latency correctly.

pinos/gst/gstpinossrc.c

@@ -86,6 +86,7 @@ static gboolean gst_pinos_src_unlock_stop (GstBaseSrc * basesrc);
 static gboolean gst_pinos_src_start (GstBaseSrc * basesrc);
 static gboolean gst_pinos_src_stop (GstBaseSrc * basesrc);
 static gboolean gst_pinos_src_event (GstBaseSrc * src, GstEvent * event);
+static gboolean gst_pinos_src_query (GstBaseSrc * src, GstQuery * query);
 
 static void
 gst_pinos_src_set_property (GObject * object, guint prop_id,
@@ -248,6 +249,7 @@ gst_pinos_src_class_init (GstPinosSrcClass * klass)
   gstbasesrc_class->start = gst_pinos_src_start;
   gstbasesrc_class->stop = gst_pinos_src_stop;
   gstbasesrc_class->event = gst_pinos_src_event;
+  gstbasesrc_class->query = gst_pinos_src_query;
   gstpushsrc_class->create = gst_pinos_src_create;
 
   GST_DEBUG_CATEGORY_INIT (pinos_src_debug, "pinossrc", 0,
@@ -489,6 +491,44 @@ on_stream_notify (GObject    *gobject,
   pinos_main_loop_signal (pinossrc->loop, FALSE);
 }
 
+static void
+parse_clock_info (GstPinosSrc *pinossrc)
+{
+  PinosProperties *props;
+  const gchar *var;
+
+  g_object_get (pinossrc->stream, "properties", &props, NULL);
+
+  var = pinos_properties_get (props, "pinos.clock.type");
+  if (var != NULL) {
+    GST_DEBUG_OBJECT (pinossrc, "got clock type %s", var);
+    if (strcmp (var, "gst.net.time.provider") == 0) {
+      const gchar *address;
+      gint port;
+
+      address = pinos_properties_get (props, "pinos.clock.address");
+      port = atoi (pinos_properties_get (props, "pinos.clock.port"));
+
+      GST_DEBUG_OBJECT (pinossrc, "making net clock for %s:%d", address, port);
+      if (pinossrc->clock)
+        gst_object_unref (pinossrc->clock);
+      pinossrc->clock = gst_net_client_clock_new ("pinosclock", address, port, 0);
+    }
+  }
+  var = pinos_properties_get (props, "pinos.latency.is-live");
+  pinossrc->is_live = var ? (atoi (var) == 1) : FALSE;
+  gst_base_src_set_live (GST_BASE_SRC (pinossrc), pinossrc->is_live);
+
+  var = pinos_properties_get (props, "pinos.latency.max");
+  pinossrc->max_latency = var ? (GstClockTime) atoi (var) : GST_CLOCK_TIME_NONE;
+
+  var = pinos_properties_get (props, "pinos.latency.min");
+  pinossrc->min_latency = var ? (GstClockTime) atoi (var) : 0;
+
+  pinos_properties_free (props);
+}
+
+
 static gboolean
 gst_pinos_src_stream_start (GstPinosSrc *pinossrc, GstCaps * caps)
 {
@@ -512,6 +552,9 @@ gst_pinos_src_stream_start (GstPinosSrc *pinossrc, GstCaps * caps)
 
     pinos_main_loop_wait (pinossrc->loop);
   }
+
+  parse_clock_info (pinossrc);
+
   pinos_main_loop_unlock (pinossrc->loop);
 
   return res;
@@ -523,36 +566,6 @@ start_error:
   }
 }
 
-static void
-parse_clock_info (GstPinosSrc *pinossrc)
-{
-  PinosProperties *props;
-  const gchar *var;
-
-  g_object_get (pinossrc->stream, "properties", &props, NULL);
-
-  var = pinos_properties_get (props, "pinos.clock.type");
-  if (var == NULL) {
-    pinos_properties_free (props);
-    return;
-  }
-
-  GST_DEBUG_OBJECT (pinossrc, "got clock type %s", var);
-  if (strcmp (var, "gst.net.time.provider") == 0) {
-    const gchar *address;
-    gint port;
-
-    address = pinos_properties_get (props, "pinos.clock.address");
-    port = atoi (pinos_properties_get (props, "pinos.clock.port"));
-
-    GST_DEBUG_OBJECT (pinossrc, "making net clock for %s:%d", address, port);
-    if (pinossrc->clock)
-      gst_object_unref (pinossrc->clock);
-    pinossrc->clock = gst_net_client_clock_new ("pinosclock", address, port, 0);
-  }
-  pinos_properties_free (props);
-}
-
 static gboolean
 gst_pinos_src_negotiate (GstBaseSrc * basesrc)
 {
@@ -775,6 +788,26 @@ gst_pinos_src_event (GstBaseSrc * src, GstEvent * event)
   return res;
 }
 
+static gboolean
+gst_pinos_src_query (GstBaseSrc * src, GstQuery * query)
+{
+  gboolean res = FALSE;
+  GstPinosSrc *pinossrc;
+
+  pinossrc = GST_PINOS_SRC (src);
+
+  switch (GST_QUERY_TYPE (query)) {
+    case GST_QUERY_LATENCY:
+      gst_query_set_latency (query, pinossrc->is_live, pinossrc->min_latency, pinossrc->max_latency);
+      res = TRUE;
+      break;
+    default:
+      res = GST_BASE_SRC_CLASS (parent_class)->query (src, query);
+      break;
+  }
+  return res;
+}
+
 static GstFlowReturn
 gst_pinos_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
 {

pinos/gst/gstpinossrc.h

@@ -58,6 +58,10 @@ struct _GstPinosSrc {
   gboolean negotiated;
   gboolean flushing;
 
+  gboolean is_live;
+  GstClockTime min_latency;
+  GstClockTime max_latency;
+
   GMainContext *context;
   PinosMainLoop *loop;
   PinosContext *ctx;

pinos/modules/gst/gst-source.c

@@ -243,9 +243,31 @@ set_state (PinosSource      *source,
       break;
 
     case PINOS_SOURCE_STATE_RUNNING:
-      gst_element_set_state (priv->pipeline, GST_STATE_PLAYING);
+    {
-      break;
+      GstQuery *query;
 
+      gst_element_set_state (priv->pipeline, GST_STATE_PLAYING);
+      gst_element_get_state (priv->pipeline, NULL, NULL, -1);
+
+      query = gst_query_new_latency ();
+      if (gst_element_query (GST_ELEMENT_CAST (priv->pipeline), query)) {
+        gboolean live;
+        GstClockTime min_latency, max_latency;
+
+        gst_query_parse_latency (query, &live, &min_latency, &max_latency);
+
+        GST_DEBUG_OBJECT (priv->pipeline,
+            "got min latency %" GST_TIME_FORMAT ", max latency %"
+            GST_TIME_FORMAT ", live %d", GST_TIME_ARGS (min_latency),
+            GST_TIME_ARGS (max_latency), live);
+
+        pinos_properties_setf (priv->props, "pinos.latency.is-live", "%d", live);
+        pinos_properties_setf (priv->props, "pinos.latency.min", "%"G_GUINT64_FORMAT, min_latency);
+        pinos_properties_setf (priv->props, "pinos.latency.max", "%"G_GUINT64_FORMAT, max_latency);
+      }
+      gst_query_unref (query);
+      break;
+    }
     case PINOS_SOURCE_STATE_ERROR:
       break;
   }
@@ -324,6 +346,9 @@ on_socket_notify (GObject    *gobject,
   GstCaps *caps;
   GBytes *requested_format, *format = NULL;
   gchar *str;
+  gpointer state = NULL;
+  const gchar *key, *val;
+  PinosProperties *props;
 
   g_object_get (gobject, "socket", &socket, NULL);
 
@@ -376,6 +401,14 @@ on_socket_notify (GObject    *gobject,
     pinos_source_update_possible_formats (PINOS_SOURCE (source), format);
     g_bytes_unref (format);
   }
+
+  g_object_get (gobject, "properties", &props, NULL);
+  while ((key = pinos_properties_iterate (priv->props, &state))) {
+    val = pinos_properties_get (priv->props, key);
+    pinos_properties_set (props, key, val);
+  }
+  g_object_set (gobject, "properties", props, NULL);
+  pinos_properties_free (props);
 }
 
 static PinosSourceOutput *
@@ -390,7 +423,7 @@ create_source_output (PinosSource     *source,
   PinosGstSourcePrivate *priv = s->priv;
   PinosSourceOutput *output;
   gpointer state = NULL;
-  const char *key, *val;
+  const gchar *key, *val;
 
   if (priv->n_outputs == 0) {
     if (!start_pipeline (s, error))