gst/rtpmanager/: Updated example pipelines in docs.

Original commit message from CVS:
* gst/rtpmanager/gstrtpbin-marshal.list:
* gst/rtpmanager/gstrtpbin.c: (gst_rtp_bin_get_client),
(gst_rtp_bin_associate), (gst_rtp_bin_sync_chain), (create_stream),
(gst_rtp_bin_init), (caps_changed), (new_ssrc_pad_found),
(create_recv_rtp), (create_recv_rtcp), (create_send_rtp):
* gst/rtpmanager/gstrtpbin.h:
Updated example pipelines in docs.
Handle sync_rtcp buffers from the SSRC demuxer to perform lip-sync.
Set the default latency correctly.
Add some more points where we can get caps.
* gst/rtpmanager/gstrtpjitterbuffer.c:
(gst_rtp_jitter_buffer_class_init),
(gst_jitter_buffer_sink_parse_caps), (gst_rtp_jitter_buffer_loop),
(gst_rtp_jitter_buffer_query),
(gst_rtp_jitter_buffer_set_property),
(gst_rtp_jitter_buffer_get_property):
Add ts-offset property to control timestamping.
* gst/rtpmanager/gstrtpsession.c: (gst_rtp_session_class_init),
(gst_rtp_session_init), (gst_rtp_session_set_property),
(gst_rtp_session_get_property), (get_current_ntp_ns_time),
(rtcp_thread), (stop_rtcp_thread), (gst_rtp_session_change_state),
(gst_rtp_session_send_rtcp), (gst_rtp_session_sync_rtcp),
(gst_rtp_session_cache_caps), (gst_rtp_session_clock_rate),
(gst_rtp_session_sink_setcaps), (gst_rtp_session_chain_recv_rtp),
(gst_rtp_session_event_send_rtp_sink),
(gst_rtp_session_chain_send_rtp), (create_recv_rtp_sink),
(create_recv_rtcp_sink), (create_send_rtp_sink),
(create_send_rtcp_src):
Various cleanups.
Feed rtpsession manager with NTP time based on pipeline clock when
handling RTP packets and RTCP timeouts.
Perform all RTCP with the system clock.
Set caps on RTCP outgoing buffers.
* gst/rtpmanager/gstrtpssrcdemux.c: (find_demux_pad_for_ssrc),
(create_demux_pad_for_ssrc), (gst_rtp_ssrc_demux_base_init),
(gst_rtp_ssrc_demux_init), (gst_rtp_ssrc_demux_sink_event),
(gst_rtp_ssrc_demux_rtcp_sink_event), (gst_rtp_ssrc_demux_chain),
(gst_rtp_ssrc_demux_rtcp_chain):
* gst/rtpmanager/gstrtpssrcdemux.h:
Also demux RTCP messages.
* gst/rtpmanager/rtpsession.c: (rtp_session_set_callbacks),
(update_arrival_stats), (rtp_session_process_rtp),
(rtp_session_process_rb), (rtp_session_process_sr),
(rtp_session_process_rr), (rtp_session_process_rtcp),
(rtp_session_send_rtp), (rtp_session_send_bye),
(session_start_rtcp), (session_report_blocks), (session_cleanup),
(rtp_session_on_timeout):
* gst/rtpmanager/rtpsession.h:
Remove the get_time callback, the GStreamer part will feed us with
enough timing information.
Split sync timing and RTCP timing information.
Factor out common RB handling for SR and RR.
Send out SR RTCP packets for lip-sync.
Move SR and RR packet info generation to the source.
* gst/rtpmanager/rtpsource.c: (rtp_source_init),
(rtp_source_update_caps), (get_clock_rate), (calculate_jitter),
(rtp_source_process_rtp), (rtp_source_send_rtp),
(rtp_source_process_sr), (rtp_source_process_rb),
(rtp_source_get_new_sr), (rtp_source_get_new_rb),
(rtp_source_get_last_sr):
* gst/rtpmanager/rtpsource.h:
* gst/rtpmanager/rtpstats.h:
Use caps on incomming buffers to get timing information when they are
there.
Calculate clock scew of the receiver compared to the sender and adjust
the rtp timestamps.
Calculate the round trip in sources.
Do SR and RR calculations in the source.

1 files changed, 96 insertions, 25 deletions

diff --git a/gst/rtpmanager/gstrtpjitterbuffer.c b/gst/rtpmanager/gstrtpjitterbuffer.c
index bd1c07ea..a23fbb87 100644
--- a/gst/rtpmanager/gstrtpjitterbuffer.c
+++ b/gst/rtpmanager/gstrtpjitterbuffer.c
@@ -99,12 +99,14 @@ enum
 
 #define DEFAULT_LATENCY_MS      200
 #define DEFAULT_DROP_ON_LATENCY FALSE
+#define DEFAULT_TS_OFFSET       0
 
 enum
 {
   PROP_0,
   PROP_LATENCY,
-  PROP_DROP_ON_LATENCY
+  PROP_DROP_ON_LATENCY,
+  PROP_TS_OFFSET
 };
 
 #define JBUF_LOCK(priv)   (g_mutex_lock ((priv)->jbuf_lock))
@@ -137,6 +139,7 @@ struct _GstRtpJitterBufferPrivate
   /* properties */
   guint latency_ms;
   gboolean drop_on_latency;
+  gint64 ts_offset;
 
   /* the last seqnum we pushed out */
   guint32 last_popped_seqnum;
@@ -150,6 +153,7 @@ struct _GstRtpJitterBufferPrivate
   gint32 clock_rate;
   gint64 clock_base;
   guint64 exttimestamp;
+  gint64 prev_ts_offset;
 
   /* when we are shutting down */
   GstFlowReturn srcresult;
@@ -278,6 +282,16 @@ gst_rtp_jitter_buffer_class_init (GstRtpJitterBufferClass * klass)
           "Tells the jitterbuffer to never exceed the given latency in size",
           DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE));
   /**
+   * GstRtpJitterBuffer::ts-offset:
+   * 
+   * Adjust RTP timestamps in the jitterbuffer with offset.
+   */
+  g_object_class_install_property (gobject_class, PROP_TS_OFFSET,
+      g_param_spec_int64 ("ts-offset",
+          "Timestamp Offset",
+          "Adjust buffer RTP timestamps with offset in nanoseconds", G_MININT64,
+          G_MAXINT64, DEFAULT_TS_OFFSET, G_PARAM_READWRITE));
+  /**
    * GstRtpJitterBuffer::request-pt-map:
    * @buffer: the object which received the signal
    * @pt: the pt
@@ -421,7 +435,7 @@ gst_jitter_buffer_sink_parse_caps (GstRtpJitterBuffer * jitterbuffer,
 {
   GstRtpJitterBufferPrivate *priv;
   GstStructure *caps_struct;
-  const GValue *value;
+  guint val;
 
   priv = jitterbuffer->priv;
 
@@ -443,22 +457,22 @@ gst_jitter_buffer_sink_parse_caps (GstRtpJitterBuffer * jitterbuffer,
   /* gah, clock-base is uint. If we don't have a base, we will use the first
    * buffer timestamp as the base time. This will screw up sync but it's better
    * than nothing. */
-  value = gst_structure_get_value (caps_struct, "clock-base");
-  if (value && G_VALUE_HOLDS_UINT (value)) {
-    priv->clock_base = g_value_get_uint (value);
-    GST_DEBUG_OBJECT (jitterbuffer, "got clock-base %" G_GINT64_FORMAT,
-        priv->clock_base);
-  } else
+  if (gst_structure_get_uint (caps_struct, "clock-base", &val))
+    priv->clock_base = val;
+  else
     priv->clock_base = -1;
 
+  GST_DEBUG_OBJECT (jitterbuffer, "got clock-base %" G_GINT64_FORMAT,
+      priv->clock_base);
+
   /* first expected seqnum */
-  value = gst_structure_get_value (caps_struct, "seqnum-base");
-  if (value && G_VALUE_HOLDS_UINT (value)) {
-    priv->next_seqnum = g_value_get_uint (value);
-    GST_DEBUG_OBJECT (jitterbuffer, "got seqnum-base %d", priv->next_seqnum);
-  } else
+  if (gst_structure_get_uint (caps_struct, "seqnum-base", &val))
+    priv->next_seqnum = val;
+  else
     priv->next_seqnum = -1;
 
+  GST_DEBUG_OBJECT (jitterbuffer, "got seqnum-base %d", priv->next_seqnum);
+
   return TRUE;
 
   /* ERRORS */
@@ -929,6 +943,7 @@ gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer)
   GstClockTime timestamp;
   gint64 running_time;
   guint64 exttimestamp;
+  gint ts_offset_rtp;
 
   priv = jitterbuffer->priv;
 
@@ -996,8 +1011,11 @@ again:
         exttimestamp, priv->clock_base);
 
     /* if no clock_base was given, take first ts as base */
-    if (priv->clock_base == -1)
+    if (priv->clock_base == -1) {
+      GST_DEBUG_OBJECT (jitterbuffer,
+          "no clock base, using exttimestamp %" G_GUINT64_FORMAT, exttimestamp);
       priv->clock_base = exttimestamp;
+    }
 
     /* take rtp timestamp offset into account, this can wrap around */
     exttimestamp -= priv->clock_base;
@@ -1089,6 +1107,34 @@ push_buffer:
     outbuf = gst_buffer_make_metadata_writable (outbuf);
     GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_DISCONT);
   }
+
+  /* apply the timestamp offset */
+  if (priv->ts_offset > 0)
+    ts_offset_rtp =
+        gst_util_uint64_scale_int (priv->ts_offset, priv->clock_rate,
+        GST_SECOND);
+  else if (priv->ts_offset < 0)
+    ts_offset_rtp =
+        -gst_util_uint64_scale_int (-priv->ts_offset, priv->clock_rate,
+        GST_SECOND);
+  else
+    ts_offset_rtp = 0;
+
+  if (ts_offset_rtp != 0) {
+    guint32 timestamp;
+
+    /* if the offset changed, mark with discont */
+    if (priv->ts_offset != priv->prev_ts_offset) {
+      GST_DEBUG_OBJECT (jitterbuffer, "changing offset to %d", ts_offset_rtp);
+      GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_DISCONT);
+      priv->prev_ts_offset = priv->ts_offset;
+    }
+
+    timestamp = gst_rtp_buffer_get_timestamp (outbuf);
+    timestamp += ts_offset_rtp;
+    gst_rtp_buffer_set_timestamp (outbuf, timestamp);
+  }
+
   /* now we are ready to push the buffer. Save the seqnum and release the lock
    * so the other end can push stuff in the queue again. */
   priv->last_popped_seqnum = seqnum;
@@ -1158,6 +1204,7 @@ gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query)
       GstClockTime min_latency, max_latency;
       gboolean us_live;
       GstPad *peer;
+      GstClockTime our_latency;
 
       if ((peer = gst_pad_get_peer (priv->sinkpad))) {
         if ((res = gst_pad_query (peer, query))) {
@@ -1172,11 +1219,16 @@ gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query)
           priv->peer_latency = min_latency;
           JBUF_UNLOCK (priv);
 
-          min_latency += priv->latency_ms * GST_MSECOND;
+          our_latency = ((guint64) priv->latency_ms) * GST_MSECOND;
+
+          GST_DEBUG_OBJECT (jitterbuffer, "Our latency: %" GST_TIME_FORMAT,
+              GST_TIME_ARGS (our_latency));
+
+          min_latency += our_latency;
           /* max_latency can be -1, meaning there is no upper limit for the
            * latency. */
           if (max_latency != -1)
-            max_latency += priv->latency_ms * GST_MSECOND;
+            max_latency += our_latency * GST_MSECOND;
 
           GST_DEBUG_OBJECT (jitterbuffer, "Calculated total latency : min %"
               GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
@@ -1199,7 +1251,11 @@ static void
 gst_rtp_jitter_buffer_set_property (GObject * object,
     guint prop_id, const GValue * value, GParamSpec * pspec)
 {
-  GstRtpJitterBuffer *jitterbuffer = GST_RTP_JITTER_BUFFER (object);
+  GstRtpJitterBuffer *jitterbuffer;
+  GstRtpJitterBufferPrivate *priv;
+
+  jitterbuffer = GST_RTP_JITTER_BUFFER (object);
+  priv = jitterbuffer->priv;
 
   switch (prop_id) {
     case PROP_LATENCY:
@@ -1208,23 +1264,29 @@ gst_rtp_jitter_buffer_set_property (GObject * object,
 
       /* FIXME, not threadsafe */
       new_latency = g_value_get_uint (value);
-      old_latency = jitterbuffer->priv->latency_ms;
+      old_latency = priv->latency_ms;
 
-      jitterbuffer->priv->latency_ms = new_latency;
+      priv->latency_ms = new_latency;
 
       /* post message if latency changed, this will inform the parent pipeline
        * that a latency reconfiguration is possible/needed. */
       if (new_latency != old_latency) {
+        GST_DEBUG_OBJECT (jitterbuffer, "latency changed to: %" GST_TIME_FORMAT,
+            GST_TIME_ARGS (new_latency * GST_MSECOND));
+
         gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer),
             gst_message_new_latency (GST_OBJECT_CAST (jitterbuffer)));
       }
       break;
     }
     case PROP_DROP_ON_LATENCY:
-    {
-      jitterbuffer->priv->drop_on_latency = g_value_get_boolean (value);
+      priv->drop_on_latency = g_value_get_boolean (value);
+      break;
+    case PROP_TS_OFFSET:
+      JBUF_LOCK (priv);
+      priv->ts_offset = g_value_get_int64 (value);
+      JBUF_UNLOCK (priv);
       break;
-    }
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -1235,14 +1297,23 @@ static void
 gst_rtp_jitter_buffer_get_property (GObject * object,
     guint prop_id, GValue * value, GParamSpec * pspec)
 {
-  GstRtpJitterBuffer *jitterbuffer = GST_RTP_JITTER_BUFFER (object);
+  GstRtpJitterBuffer *jitterbuffer;
+  GstRtpJitterBufferPrivate *priv;
+
+  jitterbuffer = GST_RTP_JITTER_BUFFER (object);
+  priv = jitterbuffer->priv;
 
   switch (prop_id) {
     case PROP_LATENCY:
-      g_value_set_uint (value, jitterbuffer->priv->latency_ms);
+      g_value_set_uint (value, priv->latency_ms);
       break;
     case PROP_DROP_ON_LATENCY:
-      g_value_set_boolean (value, jitterbuffer->priv->drop_on_latency);
+      g_value_set_boolean (value, priv->drop_on_latency);
+      break;
+    case PROP_TS_OFFSET:
+      JBUF_LOCK (priv);
+      g_value_set_int64 (value, priv->ts_offset);
+      JBUF_UNLOCK (priv);
       break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);