gst/rtpmanager/gstrtpbin.c: Emit pt map requests and cache results.

Original commit message from CVS:
* gst/rtpmanager/gstrtpbin.c: (create_session), (get_pt_map),
(create_stream), (gst_rtp_bin_class_init), (pt_map_requested):
Emit pt map requests and cache results.
* gst/rtpmanager/gstrtpjitterbuffer.c:
(gst_rtp_jitter_buffer_class_init),
(gst_jitter_buffer_sink_parse_caps),
(gst_jitter_buffer_sink_setcaps),
(gst_rtp_jitter_buffer_get_clock_rate),
(gst_rtp_jitter_buffer_chain), (gst_rtp_jitter_buffer_loop):
* gst/rtpmanager/gstrtpjitterbuffer.h:
* gst/rtpmanager/gstrtpptdemux.c: (gst_rtp_pt_demux_chain):
Emit request-pt-map signals.

4 files changed, 162 insertions, 91 deletions

diff --git a/gst/rtpmanager/gstrtpbin.c b/gst/rtpmanager/gstrtpbin.c
index 39e1fc857..b0882b67e 100644
--- a/gst/rtpmanager/gstrtpbin.c
+++ b/gst/rtpmanager/gstrtpbin.c
@@ -125,6 +125,9 @@ typedef struct _GstRTPBinClient GstRTPBinClient;
 
 static guint gst_rtp_bin_signals[LAST_SIGNAL] = { 0 };
 
+static GstCaps *pt_map_requested (GstElement * element, guint pt,
+    GstRTPBinStream * stream);
+
 /* Manages the RTP stream for one SSRC.
  *
  * We pipe the stream (comming from the SSRC demuxer) into a jitterbuffer.
@@ -217,7 +220,7 @@ create_session (GstRTPBin * rtpbin, gint id)
   sess->bin = rtpbin;
   sess->session = elem;
   sess->demux = demux;
-  sess->ptmap = g_hash_table_new (g_int_hash, g_int_equal);
+  sess->ptmap = g_hash_table_new (NULL, NULL);
   rtpbin->sessions = g_slist_prepend (rtpbin->sessions, sess);
 
   gst_bin_add (GST_BIN_CAST (rtpbin), elem);
@@ -263,24 +266,40 @@ get_pt_map (GstRTPBinSession * session, guint pt)
 {
   GstCaps *caps = NULL;
   GstRTPBin *bin;
+  GValue ret = { 0 };
+  GValue args[3] = { {0}, {0}, {0} };
 
-  GST_DEBUG ("finding pt %d in cache", pt);
+  GST_DEBUG ("searching pt %d in cache", pt);
 
   /* first look in the cache */
   caps = g_hash_table_lookup (session->ptmap, GINT_TO_POINTER (pt));
-  if (caps)
+  if (caps) {
     goto done;
+  }
 
   bin = session->bin;
 
   GST_DEBUG ("emiting signal for pt %d in session %d", pt, session->id);
 
   /* not in cache, send signal to request caps */
-  g_signal_emit (G_OBJECT (bin), gst_rtp_bin_signals[SIGNAL_REQUEST_PT_MAP], 0,
-      session->id, pt, &caps);
+  g_value_init (&args[0], GST_TYPE_ELEMENT);
+  g_value_set_object (&args[0], bin);
+  g_value_init (&args[1], G_TYPE_UINT);
+  g_value_set_uint (&args[1], session->id);
+  g_value_init (&args[2], G_TYPE_UINT);
+  g_value_set_uint (&args[2], pt);
+
+  g_value_init (&ret, GST_TYPE_CAPS);
+  g_value_set_boxed (&ret, NULL);
+
+  g_signal_emitv (args, gst_rtp_bin_signals[SIGNAL_REQUEST_PT_MAP], 0, &ret);
+
+  caps = (GstCaps *) g_value_get_boxed (&ret);
   if (!caps)
     goto no_caps;
 
+  GST_DEBUG ("caching pt %d as %" GST_PTR_FORMAT, pt, caps);
+
   /* store in cache */
   g_hash_table_insert (session->ptmap, GINT_TO_POINTER (pt), caps);
 
@@ -295,42 +314,6 @@ no_caps:
   }
 }
 
-/* callback from the jitterbuffer when it needs to know the clockrate of a
- * specific payload type. */
-static guint32
-clock_rate_request (GstElement * buffer, guint pt, GstRTPBinStream * stream)
-{
-  GstCaps *caps;
-  GstRTPBinSession *session;
-  GstStructure *s;
-  gint32 clock_rate;
-
-  session = stream->session;
-
-  caps = get_pt_map (session, pt);
-  if (!caps)
-    goto no_map;
-
-  s = gst_caps_get_structure (caps, 0);
-
-  if (!gst_structure_get_int (s, "clock-rate", &clock_rate))
-    goto no_clock_rate;
-
-  return clock_rate;
-
-  /* ERRORS */
-no_map:
-  {
-    GST_DEBUG ("no pt map found");
-    return 0;
-  }
-no_clock_rate:
-  {
-    GST_DEBUG ("no clock-rate in caps found");
-    return 0;
-  }
-}
-
 static GstRTPBinStream *
 create_stream (GstRTPBinSession * session, guint32 ssrc)
 {
@@ -352,8 +335,8 @@ create_stream (GstRTPBinSession * session, guint32 ssrc)
   session->streams = g_slist_prepend (session->streams, stream);
 
   /* provide clock_rate to the jitterbuffer when needed */
-  g_signal_connect (buffer, "request-clock-rate",
-      (GCallback) clock_rate_request, stream);
+  g_signal_connect (buffer, "request-pt-map",
+      (GCallback) pt_map_requested, stream);
 
   gst_bin_add (GST_BIN_CAST (session->bin), buffer);
   gst_element_set_state (buffer, GST_STATE_PLAYING);
@@ -457,7 +440,7 @@ gst_rtp_bin_class_init (GstRTPBinClass * klass)
   gst_rtp_bin_signals[SIGNAL_REQUEST_PT_MAP] =
       g_signal_new ("request-pt-map", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPBinClass, request_pt_map),
-      NULL, NULL, gst_rtp_bin_marshal_BOXED__UINT_UINT, GST_TYPE_CAPS, 1,
+      NULL, NULL, gst_rtp_bin_marshal_BOXED__UINT_UINT, GST_TYPE_CAPS, 2,
       G_TYPE_UINT, G_TYPE_UINT);
 
   gstelement_class->provide_clock =
@@ -593,18 +576,26 @@ pt_map_requested (GstElement * element, guint pt, GstRTPBinStream * stream)
 {
   GstRTPBin *rtpbin;
   GstRTPBinSession *session;
-  gint id;
+  GstCaps *caps;
 
   rtpbin = stream->bin;
   session = stream->session;
 
-  /* find session id */
-  id = session->id;
-
   GST_DEBUG_OBJECT (rtpbin, "payload map requested for pt %d in session %d", pt,
-      id);
+      session->id);
 
-  return NULL;
+  caps = get_pt_map (session, pt);
+  if (!caps)
+    goto no_caps;
+
+  return caps;
+
+  /* ERRORS */
+no_caps:
+  {
+    GST_DEBUG_OBJECT (rtpbin, "could not get caps");
+    return NULL;
+  }
 }
 
 /* a new pad (SSRC) was created in @session */
diff --git a/gst/rtpmanager/gstrtpjitterbuffer.c b/gst/rtpmanager/gstrtpjitterbuffer.c
index 39b476d92..1973e9b43 100644
--- a/gst/rtpmanager/gstrtpjitterbuffer.c
+++ b/gst/rtpmanager/gstrtpjitterbuffer.c
@@ -83,7 +83,7 @@ GST_ELEMENT_DETAILS ("RTP packet jitter-buffer",
 enum
 {
   /* FILL ME */
-  SIGNAL_REQUEST_CLOCK_RATE,
+  SIGNAL_REQUEST_PT_MAP,
   LAST_SIGNAL
 };
 
@@ -227,17 +227,17 @@ gst_rtp_jitter_buffer_class_init (GstRTPJitterBufferClass * klass)
           DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE));
 
   /**
-   * GstRTPJitterBuffer::request-clock-rate:
+   * GstRTPJitterBuffer::request-pt-map:
    * @buffer: the object which received the signal
    * @pt: the pt
    *
    * Request the payload type as #GstCaps for @pt.
    */
-  gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_CLOCK_RATE] =
-      g_signal_new ("request-clock-rate", G_TYPE_FROM_CLASS (klass),
+  gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP] =
+      g_signal_new ("request-pt-map", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPJitterBufferClass,
-          request_clock_rate), NULL, NULL, gst_rtp_bin_marshal_UINT__UINT,
-      G_TYPE_UINT, 1, G_TYPE_UINT);
+          request_pt_map), NULL, NULL, gst_rtp_bin_marshal_BOXED__UINT,
+      GST_TYPE_CAPS, 1, G_TYPE_UINT);
 
   gstelement_class->change_state = gst_rtp_jitter_buffer_change_state;
 
@@ -341,14 +341,13 @@ gst_rtp_jitter_buffer_getcaps (GstPad * pad)
 }
 
 static gboolean
-gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps)
+gst_jitter_buffer_sink_parse_caps (GstRTPJitterBuffer * jitterbuffer,
+    GstCaps * caps)
 {
-  GstRTPJitterBuffer *jitterbuffer;
   GstRTPJitterBufferPrivate *priv;
   GstStructure *caps_struct;
   const GValue *value;
 
-  jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
   priv = jitterbuffer->priv;
 
   /* first parse the caps */
@@ -387,28 +386,42 @@ gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps)
   async_jitter_queue_set_max_queue_length (priv->queue,
       priv->latency_ms * priv->clock_rate / 1000);
 
-  /* set same caps on srcpad */
-  gst_pad_set_caps (priv->srcpad, caps);
-
-  gst_object_unref (jitterbuffer);
-
   return TRUE;
 
   /* ERRORS */
 error:
   {
     GST_DEBUG_OBJECT (jitterbuffer, "No clock-rate in caps!");
-    gst_object_unref (jitterbuffer);
     return FALSE;
   }
 wrong_rate:
   {
     GST_DEBUG_OBJECT (jitterbuffer, "Invalid clock-rate %d", priv->clock_rate);
-    gst_object_unref (jitterbuffer);
     return FALSE;
   }
 }
 
+static gboolean
+gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps)
+{
+  GstRTPJitterBuffer *jitterbuffer;
+  GstRTPJitterBufferPrivate *priv;
+  gboolean res;
+
+  jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
+  priv = jitterbuffer->priv;
+
+  res = gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
+
+  /* set same caps on srcpad on success */
+  if (res)
+    gst_pad_set_caps (priv->srcpad, caps);
+
+  gst_object_unref (jitterbuffer);
+
+  return res;
+}
+
 static void
 free_func (gpointer data, GstRTPJitterBuffer * user_data)
 {
@@ -670,6 +683,42 @@ newseg_wrong_format:
   }
 }
 
+static gboolean
+gst_rtp_jitter_buffer_get_clock_rate (GstRTPJitterBuffer * jitterbuffer,
+    guint8 pt)
+{
+  GValue ret = { 0 };
+  GValue args[2] = { {0}, {0} };
+  GstCaps *caps;
+  gboolean res;
+
+  g_value_init (&args[0], GST_TYPE_ELEMENT);
+  g_value_set_object (&args[0], jitterbuffer);
+  g_value_init (&args[1], G_TYPE_UINT);
+  g_value_set_uint (&args[1], pt);
+
+  g_value_init (&ret, GST_TYPE_CAPS);
+  g_value_set_boxed (&ret, NULL);
+
+  g_signal_emitv (args, gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP], 0,
+      &ret);
+
+  caps = (GstCaps *) g_value_get_boxed (&ret);
+  if (!caps)
+    goto no_caps;
+
+  res = gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
+
+  return res;
+
+  /* ERRORS */
+no_caps:
+  {
+    GST_DEBUG_OBJECT (jitterbuffer, "could not get caps");
+    return FALSE;
+  }
+}
+
 static GstFlowReturn
 gst_rtp_jitter_buffer_chain (GstPad * pad, GstBuffer * buffer)
 {
@@ -678,14 +727,23 @@ gst_rtp_jitter_buffer_chain (GstPad * pad, GstBuffer * buffer)
   guint16 seqnum;
   GstFlowReturn ret;
 
+  jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
 
-  g_return_val_if_fail (gst_rtp_buffer_validate (buffer), GST_FLOW_ERROR);
+  if (!gst_rtp_buffer_validate (buffer))
+    goto invalid_buffer;
 
-  jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
   priv = jitterbuffer->priv;
 
-  if (priv->clock_rate == -1)
-    goto not_negotiated;
+  if (priv->clock_rate == -1) {
+    guint8 pt;
+
+    /* no clock rate given on the caps, try to get one with the signal */
+    pt = gst_rtp_buffer_get_payload_type (buffer);
+
+    gst_rtp_jitter_buffer_get_clock_rate (jitterbuffer, pt);
+    if (priv->clock_rate == -1)
+      goto not_negotiated;
+  }
 
   seqnum = gst_rtp_buffer_get_seq (buffer);
   GST_DEBUG_OBJECT (jitterbuffer, "Received packet #%d", seqnum);
@@ -743,6 +801,15 @@ finished:
   return ret;
 
   /* ERRORS */
+invalid_buffer:
+  {
+    /* this is fatal and should be filtered earlier */
+    GST_ELEMENT_ERROR (jitterbuffer, STREAM, DECODE, (NULL),
+        ("Received invalid RTP payload"));
+    gst_buffer_unref (buffer);
+    gst_object_unref (jitterbuffer);
+    return GST_FLOW_ERROR;
+  }
 not_negotiated:
   {
     GST_DEBUG_OBJECT (jitterbuffer, "No clock-rate in caps!");
diff --git a/gst/rtpmanager/gstrtpjitterbuffer.h b/gst/rtpmanager/gstrtpjitterbuffer.h
index 215d8b70f..e101039a7 100644
--- a/gst/rtpmanager/gstrtpjitterbuffer.h
+++ b/gst/rtpmanager/gstrtpjitterbuffer.h
@@ -63,7 +63,7 @@ struct _GstRTPJitterBufferClass
 {
   GstElementClass parent_class;
 
-  guint (*request_clock_rate) (GstRTPJitterBuffer *buffer, guint pt);
+  GstCaps* (*request_pt_map) (GstRTPJitterBuffer *buffer, guint pt);
 
   /*< private > */
   gpointer _gst_reserved[GST_PADDING];
diff --git a/gst/rtpmanager/gstrtpptdemux.c b/gst/rtpmanager/gstrtpptdemux.c
index 7ac6e90c9..247df148f 100644
--- a/gst/rtpmanager/gstrtpptdemux.c
+++ b/gst/rtpmanager/gstrtpptdemux.c
@@ -97,7 +97,6 @@ static void gst_rtp_pt_demux_release (GstElement * element);
 static gboolean gst_rtp_pt_demux_setup (GstElement * element);
 
 static GstFlowReturn gst_rtp_pt_demux_chain (GstPad * pad, GstBuffer * buf);
-static GstCaps *gst_rtp_pt_demux_getcaps (GstPad * pad);
 static GstStateChangeReturn gst_rtp_pt_demux_change_state (GstElement * element,
     GstStateChange transition);
 
@@ -234,6 +233,11 @@ gst_rtp_pt_demux_chain (GstPad * pad, GstBuffer * buf)
     gchar *padname;
     GstCaps *caps;
     GstRTPPtDemuxPad *rtpdemuxpad;
+    GValue ret = { 0 };
+    GValue args[2] = { {0}
+    , {0}
+    };
+
 
     klass = GST_ELEMENT_GET_CLASS (rtpdemux);
     templ = gst_element_class_get_pad_template (klass, "src_%d");
@@ -241,22 +245,35 @@ gst_rtp_pt_demux_chain (GstPad * pad, GstBuffer * buf)
     srcpad = gst_pad_new_from_template (templ, padname);
     g_free (padname);
 
-    caps = gst_buffer_get_caps (buf);
+    /* figure out the caps */
+    g_value_init (&args[0], GST_TYPE_ELEMENT);
+    g_value_set_object (&args[0], rtpdemux);
+    g_value_init (&args[1], G_TYPE_UINT);
+    g_value_set_uint (&args[1], pt);
+
+    g_value_init (&ret, GST_TYPE_CAPS);
+    g_value_set_boxed (&ret, NULL);
+
+    g_signal_emitv (args, gst_rtp_pt_demux_signals[SIGNAL_REQUEST_PT_MAP], 0,
+        &ret);
+
+    caps = g_value_get_boxed (&ret);
+    if (!caps)
+      goto no_caps;
+
     caps = gst_caps_make_writable (caps);
     gst_caps_set_simple (caps, "payload", G_TYPE_INT, pt, NULL);
     gst_pad_set_caps (srcpad, caps);
 
-    /* XXX: set _link () function */
-    gst_pad_set_getcaps_function (srcpad, gst_rtp_pt_demux_getcaps);
-    gst_pad_set_active (srcpad, TRUE);
-    gst_element_add_pad (element, srcpad);
-
     GST_DEBUG ("Adding pt=%d to the list.", pt);
     rtpdemuxpad = g_new0 (GstRTPPtDemuxPad, 1);
     rtpdemuxpad->pt = pt;
     rtpdemuxpad->pad = srcpad;
     rtpdemux->srcpads = g_slist_append (rtpdemux->srcpads, rtpdemuxpad);
 
+    gst_pad_set_active (srcpad, TRUE);
+    gst_element_add_pad (element, srcpad);
+
     GST_DEBUG ("emitting new-payload_type for pt %d", pt);
     g_signal_emit (G_OBJECT (rtpdemux),
         gst_rtp_pt_demux_signals[SIGNAL_NEW_PAYLOAD_TYPE], 0, pt, srcpad);
@@ -272,6 +289,8 @@ gst_rtp_pt_demux_chain (GstPad * pad, GstBuffer * buf)
         gst_rtp_pt_demux_signals[SIGNAL_PAYLOAD_TYPE_CHANGE], 0, emit_pt);
   }
 
+  gst_buffer_set_caps (buf, GST_PAD_CAPS (srcpad));
+
   /* push to srcpad */
   if (srcpad)
     ret = gst_pad_push (srcpad, GST_BUFFER (buf));
@@ -287,19 +306,13 @@ invalid_buffer:
     gst_buffer_unref (buf);
     return GST_FLOW_ERROR;
   }
-}
-
-static GstCaps *
-gst_rtp_pt_demux_getcaps (GstPad * pad)
-{
-  GstCaps *caps;
-
-  GST_OBJECT_LOCK (pad);
-  if ((caps = GST_PAD_CAPS (pad)))
-    caps = gst_caps_ref (caps);
-  GST_OBJECT_UNLOCK (pad);
-
-  return caps;
+no_caps:
+  {
+    GST_ELEMENT_ERROR (rtpdemux, STREAM, DECODE, (NULL),
+        ("Could not get caps for payload"));
+    gst_buffer_unref (buf);
+    return GST_FLOW_ERROR;
+  }
 }
 
 static GstPad *