gst/rtpmanager/gstrtpsession.c: Pass the running time to the session when processing RTP packets.

Original commit message from CVS:
* gst/rtpmanager/gstrtpsession.c: (get_current_times),
(rtcp_thread), (gst_rtp_session_chain_recv_rtp):
Pass the running time to the session when processing RTP packets.
Improve the time function to provide more info.
* gst/rtpmanager/rtpsession.c: (rtp_session_class_init),
(rtp_session_init), (update_arrival_stats),
(rtp_session_process_rtp), (rtp_session_process_sdes),
(rtp_session_process_rtcp), (session_start_rtcp),
(rtp_session_on_timeout):
* gst/rtpmanager/rtpsession.h:
Mark the internal source with a flag.
Use running_time instead of the more useless timestamp.
Validate a source when a valid SDES has been received.
Pass the current system time when processing SR packets.
* gst/rtpmanager/rtpsource.c: (rtp_source_class_init),
(rtp_source_init), (rtp_source_create_stats),
(rtp_source_get_property), (rtp_source_send_rtp),
(rtp_source_process_rb), (rtp_source_get_new_rb),
(rtp_source_get_last_rb):
* gst/rtpmanager/rtpsource.h:
Add property to get source stats.
Mark params as STATIC_STRINGS.
Calculate the bitrate at the sender SSRC.
Avoid negative values in the round trip time calculations.
* gst/rtpmanager/rtpstats.h:
Update some docs and change some variable name to more closely reflect
what it contains.

6 files changed, 213 insertions, 60 deletions

diff --git a/gst/rtpmanager/gstrtpsession.c b/gst/rtpmanager/gstrtpsession.c
index ec744a5bb..c8dfdd182 100644
--- a/gst/rtpmanager/gstrtpsession.c
+++ b/gst/rtpmanager/gstrtpsession.c
@@ -860,12 +860,13 @@ gst_rtp_session_get_property (GObject * object, guint prop_id,
   }
 }
 
-static guint64
-get_current_ntp_ns_time (GstRtpSession * rtpsession)
+static void
+get_current_times (GstRtpSession * rtpsession,
+    GstClockTime * running_time, guint64 * ntpnstime)
 {
-  guint64 ntpnstime;
+  guint64 ntpns;
   GstClock *clock;
-  GstClockTime base_time, ntpnsbase;
+  GstClockTime base_time, ntpnsbase, rt;
 
   GST_OBJECT_LOCK (rtpsession);
   if ((clock = GST_ELEMENT_CLOCK (rtpsession))) {
@@ -874,20 +875,21 @@ get_current_ntp_ns_time (GstRtpSession * rtpsession)
     gst_object_ref (clock);
     GST_OBJECT_UNLOCK (rtpsession);
 
-    /* get current NTP time */
-    ntpnstime = gst_clock_get_time (clock);
-    /* convert to running time */
-    ntpnstime -= base_time;
-    /* add NTP base offset */
-    ntpnstime += ntpnsbase;
+    /* get current clock time and convert to running time */
+    rt = gst_clock_get_time (clock) - base_time;
+    /* add NTP base offset to get NTP ns time */
+    ntpns = rt + ntpnsbase;
 
     gst_object_unref (clock);
   } else {
     GST_OBJECT_UNLOCK (rtpsession);
-    ntpnstime = -1;
+    rt = -1;
+    ntpns = -1;
   }
-
-  return ntpnstime;
+  if (running_time)
+    *running_time = rt;
+  if (ntpnstime)
+    *ntpnstime = ntpns;
 }
 
 static void
@@ -935,7 +937,7 @@ rtcp_thread (GstRtpSession * rtpsession)
     current_time = gst_clock_get_time (rtpsession->priv->sysclock);
 
     /* get current NTP time */
-    ntpnstime = get_current_ntp_ns_time (rtpsession);
+    get_current_times (rtpsession, NULL, &ntpnstime);
 
     /* we get unlocked because we need to perform reconsideration, don't perform
      * the timeout but get a new reporting estimate. */
@@ -1423,7 +1425,7 @@ gst_rtp_session_chain_recv_rtp (GstPad * pad, GstBuffer * buffer)
   GstRtpSession *rtpsession;
   GstRtpSessionPrivate *priv;
   GstFlowReturn ret;
-  GstClockTime current_time;
+  GstClockTime current_time, running_time;
   guint64 ntpnstime;
   GstClockTime timestamp;
 
@@ -1436,22 +1438,21 @@ gst_rtp_session_chain_recv_rtp (GstPad * pad, GstBuffer * buffer)
   timestamp = GST_BUFFER_TIMESTAMP (buffer);
   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
     /* convert to running time using the segment values */
-    ntpnstime =
+    running_time =
         gst_segment_to_running_time (&rtpsession->recv_rtp_seg, GST_FORMAT_TIME,
         timestamp);
     /* add constant to convert running time to NTP time */
-    ntpnstime += priv->ntpnsbase;
+    ntpnstime = running_time + priv->ntpnsbase;
   } else {
-    ntpnstime = get_current_ntp_ns_time (rtpsession);
+    get_current_times (rtpsession, &running_time, &ntpnstime);
   }
-
   current_time = gst_clock_get_time (priv->sysclock);
+
   ret = rtp_session_process_rtp (priv->session, buffer, current_time,
-      ntpnstime);
+      running_time, ntpnstime);
   if (ret != GST_FLOW_OK)
     goto push_error;
 
-
 done:
   gst_object_unref (rtpsession);
 
diff --git a/gst/rtpmanager/rtpsession.c b/gst/rtpmanager/rtpsession.c
index bf0a0aab9..ccb15e38b 100644
--- a/gst/rtpmanager/rtpsession.c
+++ b/gst/rtpmanager/rtpsession.c
@@ -329,6 +329,7 @@ rtp_session_init (RTPSession * sess)
   /* create an active SSRC for this session manager */
   sess->source = rtp_session_create_source (sess);
   sess->source->validated = TRUE;
+  sess->source->internal = TRUE;
   sess->stats.active_sources++;
 
   /* default UDP header length */
@@ -1329,11 +1330,11 @@ rtp_session_create_source (RTPSession * sess)
 static void
 update_arrival_stats (RTPSession * sess, RTPArrivalStats * arrival,
     gboolean rtp, GstBuffer * buffer, GstClockTime current_time,
-    guint64 ntpnstime)
+    GstClockTime running_time, guint64 ntpnstime)
 {
   /* get time of arrival */
   arrival->time = current_time;
-  arrival->timestamp = GST_BUFFER_TIMESTAMP (buffer);
+  arrival->running_time = running_time;
   arrival->ntpnstime = ntpnstime;
 
   /* get packet size including header overhead */
@@ -1368,7 +1369,7 @@ update_arrival_stats (RTPSession * sess, RTPArrivalStats * arrival,
  */
 GstFlowReturn
 rtp_session_process_rtp (RTPSession * sess, GstBuffer * buffer,
-    GstClockTime current_time, guint64 ntpnstime)
+    GstClockTime current_time, GstClockTime running_time, guint64 ntpnstime)
 {
   GstFlowReturn result;
   guint32 ssrc;
@@ -1385,7 +1386,8 @@ rtp_session_process_rtp (RTPSession * sess, GstBuffer * buffer,
 
   RTP_SESSION_LOCK (sess);
   /* update arrival stats */
-  update_arrival_stats (sess, &arrival, TRUE, buffer, current_time, ntpnstime);
+  update_arrival_stats (sess, &arrival, TRUE, buffer, current_time,
+      running_time, ntpnstime);
 
   /* ignore more RTP packets when we left the session */
   if (sess->source->received_bye)
@@ -1629,6 +1631,8 @@ rtp_session_process_sdes (RTPSession * sess, GstRTCPPacket * packet,
       j++;
     }
 
+    source->validated = TRUE;
+
     if (created)
       on_new_ssrc (sess, source);
     if (changed)
@@ -1759,7 +1763,7 @@ rtp_session_process_rtcp (RTPSession * sess, GstBuffer * buffer,
 
   RTP_SESSION_LOCK (sess);
   /* update arrival stats */
-  update_arrival_stats (sess, &arrival, FALSE, buffer, current_time, -1);
+  update_arrival_stats (sess, &arrival, FALSE, buffer, current_time, -1, -1);
 
   if (sess->sent_bye)
     goto ignore;
@@ -2097,8 +2101,8 @@ session_start_rtcp (RTPSession * sess, ReportData * data)
     rtp_source_get_new_sr (own, data->ntpnstime, &ntptime, &rtptime,
         &packet_count, &octet_count);
     /* store stats */
-    rtp_source_process_sr (own, data->ntpnstime, ntptime, rtptime, packet_count,
-        octet_count);
+    rtp_source_process_sr (own, data->current_time, ntptime, rtptime,
+        packet_count, octet_count);
 
     /* fill in sender report info */
     gst_rtcp_packet_sr_set_sender_info (packet, own->ssrc,
@@ -2331,6 +2335,7 @@ rtp_session_on_timeout (RTPSession * sess, GstClockTime current_time,
   GstFlowReturn result = GST_FLOW_OK;
   GList *item;
   ReportData data;
+  RTPSource *own;
 
   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
 
@@ -2344,6 +2349,8 @@ rtp_session_on_timeout (RTPSession * sess, GstClockTime current_time,
   data.is_bye = FALSE;
   data.has_sdes = FALSE;
 
+  own = sess->source;
+
   RTP_SESSION_LOCK (sess);
   /* get a new interval, we need this for various cleanups etc */
   data.interval = calculate_rtcp_interval (sess, TRUE, sess->first_rtcp);
@@ -2354,7 +2361,7 @@ rtp_session_on_timeout (RTPSession * sess, GstClockTime current_time,
 
   /* see if we need to generate SR or RR packets */
   if (is_rtcp_time (sess, current_time, &data)) {
-    if (sess->source->received_bye) {
+    if (own->received_bye) {
       /* generate BYE instead */
       GST_DEBUG ("generating BYE message");
       session_bye (sess, &data);
@@ -2401,21 +2408,21 @@ rtp_session_on_timeout (RTPSession * sess, GstClockTime current_time,
   }
 
   if (sess->change_ssrc) {
-    GST_DEBUG ("need to change our SSRC (%08x)", sess->source->ssrc);
+    GST_DEBUG ("need to change our SSRC (%08x)", own->ssrc);
     g_hash_table_steal (sess->ssrcs[sess->mask_idx],
-        GINT_TO_POINTER (sess->source->ssrc));
+        GINT_TO_POINTER (own->ssrc));
 
-    sess->source->ssrc = rtp_session_create_new_ssrc (sess);
-    rtp_source_reset (sess->source);
+    own->ssrc = rtp_session_create_new_ssrc (sess);
+    rtp_source_reset (own);
 
     g_hash_table_insert (sess->ssrcs[sess->mask_idx],
-        GINT_TO_POINTER (sess->source->ssrc), sess->source);
+        GINT_TO_POINTER (own->ssrc), own);
 
     g_free (sess->bye_reason);
     sess->bye_reason = NULL;
     sess->sent_bye = FALSE;
     sess->change_ssrc = FALSE;
-    GST_DEBUG ("changed our SSRC to %08x", sess->source->ssrc);
+    GST_DEBUG ("changed our SSRC to %08x", own->ssrc);
   }
   RTP_SESSION_UNLOCK (sess);
 
@@ -2426,7 +2433,7 @@ rtp_session_on_timeout (RTPSession * sess, GstClockTime current_time,
 
     GST_DEBUG ("sending packet");
     if (sess->callbacks.send_rtcp)
-      result = sess->callbacks.send_rtcp (sess, sess->source, data.rtcp,
+      result = sess->callbacks.send_rtcp (sess, own, data.rtcp,
           sess->sent_bye, sess->send_rtcp_user_data);
     else {
       GST_DEBUG ("freeing packet");
diff --git a/gst/rtpmanager/rtpsession.h b/gst/rtpmanager/rtpsession.h
index 823111938..f923959aa 100644
--- a/gst/rtpmanager/rtpsession.h
+++ b/gst/rtpmanager/rtpsession.h
@@ -282,7 +282,8 @@ RTPSource*      rtp_session_create_source          (RTPSession *sess);
 
 /* processing packets from receivers */
 GstFlowReturn   rtp_session_process_rtp            (RTPSession *sess, GstBuffer *buffer,
-                                                    GstClockTime current_time, guint64 ntpnstime);
+                                                    GstClockTime current_time,
+						    GstClockTime running_time, guint64 ntpnstime);
 GstFlowReturn   rtp_session_process_rtcp           (RTPSession *sess, GstBuffer *buffer,
                                                     GstClockTime current_time);
 
diff --git a/gst/rtpmanager/rtpsource.c b/gst/rtpmanager/rtpsource.c
index 5ee3cb313..fc2252710 100644
--- a/gst/rtpmanager/rtpsource.c
+++ b/gst/rtpmanager/rtpsource.c
@@ -60,6 +60,7 @@ enum
   PROP_SDES_LOCATION,
   PROP_SDES_TOOL,
   PROP_SDES_NOTE,
+  PROP_STATS,
   PROP_LAST
 };
 
@@ -89,55 +90,73 @@ rtp_source_class_init (RTPSourceClass * klass)
   g_object_class_install_property (gobject_class, PROP_SSRC,
       g_param_spec_uint ("ssrc", "SSRC",
           "The SSRC of this source", 0, G_MAXUINT,
-          DEFAULT_SSRC, G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY));
+          DEFAULT_SSRC,
+          G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_IS_CSRC,
       g_param_spec_boolean ("is-csrc", "Is CSRC",
           "If this SSRC is acting as a contributing source",
-          DEFAULT_IS_CSRC, G_PARAM_READABLE));
+          DEFAULT_IS_CSRC, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_IS_VALIDATED,
       g_param_spec_boolean ("is-validated", "Is Validated",
-          "If this SSRC is validated", DEFAULT_IS_VALIDATED, G_PARAM_READABLE));
+          "If this SSRC is validated", DEFAULT_IS_VALIDATED,
+          G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_IS_SENDER,
       g_param_spec_boolean ("is-sender", "Is Sender",
-          "If this SSRC is a sender", DEFAULT_IS_SENDER, G_PARAM_READABLE));
+          "If this SSRC is a sender", DEFAULT_IS_SENDER,
+          G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_SDES_CNAME,
       g_param_spec_string ("sdes-cname", "SDES CNAME",
           "The CNAME to put in SDES messages of this source",
-          DEFAULT_SDES_CNAME, G_PARAM_READWRITE));
+          DEFAULT_SDES_CNAME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_SDES_NAME,
       g_param_spec_string ("sdes-name", "SDES NAME",
           "The NAME to put in SDES messages of this source",
-          DEFAULT_SDES_NAME, G_PARAM_READWRITE));
+          DEFAULT_SDES_NAME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_SDES_EMAIL,
       g_param_spec_string ("sdes-email", "SDES EMAIL",
           "The EMAIL to put in SDES messages of this source",
-          DEFAULT_SDES_EMAIL, G_PARAM_READWRITE));
+          DEFAULT_SDES_EMAIL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_SDES_PHONE,
       g_param_spec_string ("sdes-phone", "SDES PHONE",
           "The PHONE to put in SDES messages of this source",
-          DEFAULT_SDES_PHONE, G_PARAM_READWRITE));
+          DEFAULT_SDES_PHONE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_SDES_LOCATION,
       g_param_spec_string ("sdes-location", "SDES LOCATION",
           "The LOCATION to put in SDES messages of this source",
-          DEFAULT_SDES_LOCATION, G_PARAM_READWRITE));
+          DEFAULT_SDES_LOCATION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_SDES_TOOL,
       g_param_spec_string ("sdes-tool", "SDES TOOL",
           "The TOOL to put in SDES messages of this source",
-          DEFAULT_SDES_TOOL, G_PARAM_READWRITE));
+          DEFAULT_SDES_TOOL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class, PROP_SDES_NOTE,
       g_param_spec_string ("sdes-note", "SDES NOTE",
           "The NOTE to put in SDES messages of this source",
-          DEFAULT_SDES_NOTE, G_PARAM_READWRITE));
+          DEFAULT_SDES_NOTE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
+  /**
+   * RTPSource::stats
+   *
+   * The statistics of the source. This property returns a GstStructure with
+   * name application/x-rtp-source-stats with the following fields:
+   *
+   *  
+   *
+   * 
+   */
+  g_object_class_install_property (gobject_class, PROP_STATS,
+      g_param_spec_boxed ("stats", "Stats",
+          "The stats of this source", GST_TYPE_STRUCTURE,
+          G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 
   GST_DEBUG_CATEGORY_INIT (rtp_source_debug, "rtpsource", 0, "RTP Source");
 }
@@ -166,6 +185,7 @@ rtp_source_init (RTPSource * src)
   /* sources are initialy on probation until we receive enough valid RTP
    * packets or a valid RTCP packet */
   src->validated = FALSE;
+  src->internal = FALSE;
   src->probation = RTP_DEFAULT_PROBATION;
 
   src->payload = 0;
@@ -200,6 +220,84 @@ rtp_source_finalize (GObject * object)
   G_OBJECT_CLASS (rtp_source_parent_class)->finalize (object);
 }
 
+static GstStructure *
+rtp_source_create_stats (RTPSource * src)
+{
+  GstStructure *s;
+  gboolean is_sender = src->is_sender;
+  gboolean internal = src->internal;
+
+  /* common data for all types of sources */
+  s = gst_structure_new ("application/x-rtp-source-stats",
+      "ssrc", G_TYPE_UINT, (guint) src->ssrc,
+      "internal", G_TYPE_BOOLEAN, internal,
+      "validated", G_TYPE_BOOLEAN, src->validated,
+      "received-bye", G_TYPE_BOOLEAN, src->received_bye,
+      "is-csrc", G_TYPE_BOOLEAN, src->is_csrc,
+      "is-sender", G_TYPE_BOOLEAN, is_sender, NULL);
+
+  if (internal) {
+    /* our internal source */
+    if (is_sender) {
+      /* if we are sending, report about how much we sent, other sources will
+       * have a RB with info on reception. */
+      gst_structure_set (s,
+          "octets-sent", G_TYPE_UINT64, src->stats.octets_sent,
+          "packets-sent", G_TYPE_UINT64, src->stats.packets_sent,
+          "bitrate", G_TYPE_UINT64, src->bitrate, NULL);
+    } else {
+      /* if we are not sending we have nothing more to report */
+    }
+  } else {
+    gboolean have_rb;
+    guint8 fractionlost = 0;
+    gint32 packetslost = 0;
+    guint32 exthighestseq = 0;
+    guint32 jitter = 0;
+    guint32 lsr = 0;
+    guint32 dlsr = 0;
+    guint32 round_trip = 0;
+
+    /* other sources */
+    if (is_sender) {
+      gboolean have_sr;
+      GstClockTime time = 0;
+      guint64 ntptime = 0;
+      guint32 rtptime = 0;
+      guint32 packet_count = 0;
+      guint32 octet_count = 0;
+
+      /* this source is sending to us, get the last SR. */
+      have_sr = rtp_source_get_last_sr (src, &time, &ntptime, &rtptime,
+          &packet_count, &octet_count);
+      gst_structure_set (s,
+          "octets-received", G_TYPE_UINT64, src->stats.octets_received,
+          "packets-received", G_TYPE_UINT64, src->stats.packets_received,
+          "have-sr", G_TYPE_BOOLEAN, have_sr,
+          "sr-ntptime", G_TYPE_UINT64, ntptime,
+          "sr-rtptime", G_TYPE_UINT, (guint) rtptime,
+          "sr-octet-count", G_TYPE_UINT, (guint) octet_count,
+          "sr-packet-count", G_TYPE_UINT, (guint) packet_count, NULL);
+    }
+    /* we might be sending to this SSRC so we report about how it is
+     * receiving our data */
+    have_rb = rtp_source_get_last_rb (src, &fractionlost, &packetslost,
+        &exthighestseq, &jitter, &lsr, &dlsr, &round_trip);
+
+    gst_structure_set (s,
+        "have-rb", G_TYPE_BOOLEAN, have_rb,
+        "rb-fractionlost", G_TYPE_UINT, (guint) fractionlost,
+        "rb-packetslost", G_TYPE_INT, (gint) packetslost,
+        "rb-exthighestseq", G_TYPE_UINT, (guint) exthighestseq,
+        "rb-jitter", G_TYPE_UINT, (guint) jitter,
+        "rb-lsr", G_TYPE_UINT, (guint) lsr,
+        "rb-dlsr", G_TYPE_UINT, (guint) dlsr,
+        "rb-round-trip", G_TYPE_UINT, (guint) round_trip, NULL);
+  }
+
+  return s;
+}
+
 static void
 rtp_source_set_property (GObject * object, guint prop_id,
     const GValue * value, GParamSpec * pspec)
@@ -295,6 +393,9 @@ rtp_source_get_property (GObject * object, guint prop_id,
       g_value_take_string (value, rtp_source_get_sdes_string (src,
               GST_RTCP_SDES_NOTE));
       break;
+    case PROP_STATS:
+      g_value_take_boxed (value, rtp_source_create_stats (src));
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -992,6 +1093,7 @@ rtp_source_send_rtp (RTPSource * src, GstBuffer * buffer, guint64 ntpnstime)
   guint32 rtptime;
   guint64 ext_rtptime;
   guint64 ntp_diff, rtp_diff;
+  guint64 elapsed;
 
   g_return_val_if_fail (RTP_IS_SOURCE (src), GST_FLOW_ERROR);
   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
@@ -1006,6 +1108,34 @@ rtp_source_send_rtp (RTPSource * src, GstBuffer * buffer, guint64 ntpnstime)
   /* update stats for the SR */
   src->stats.packets_sent++;
   src->stats.octets_sent += len;
+  src->bytes_sent += len;
+
+  if (src->prev_ntpnstime) {
+    elapsed = ntpnstime - src->prev_ntpnstime;
+
+    if (elapsed > (G_GINT64_CONSTANT (1) << 31)) {
+      guint64 rate;
+
+      rate =
+          gst_util_uint64_scale (src->bytes_sent, elapsed,
+          (G_GINT64_CONSTANT (1) << 29));
+
+      GST_LOG ("Elapsed %" G_GUINT64_FORMAT ", bytes %" G_GUINT64_FORMAT
+          ", rate %" G_GUINT64_FORMAT, elapsed, src->bytes_sent, rate);
+
+      if (src->bitrate == 0)
+        src->bitrate = rate;
+      else
+        src->bitrate = ((src->bitrate * 3) + rate) / 4;
+
+      src->prev_ntpnstime = ntpnstime;
+      src->bytes_sent = 0;
+    }
+  } else {
+    GST_LOG ("Reset bitrate measurement");
+    src->prev_ntpnstime = ntpnstime;
+    src->bitrate = 0;
+  }
 
   rtptime = gst_rtp_buffer_get_timestamp (buffer);
   ext_rtptime = src->last_rtptime;
@@ -1140,10 +1270,13 @@ rtp_source_process_rb (RTPSource * src, GstClockTime time, guint8 fractionlost,
   curr->lsr = lsr;
   curr->dlsr = dlsr;
 
-  /* calculate round trip */
-  ntp = (gst_rtcp_unix_to_ntp (time) >> 16) & 0xffffffff;
-  A = ntp - dlsr;
-  A -= lsr;
+  /* calculate round trip, round the time up */
+  ntp = ((gst_rtcp_unix_to_ntp (time) + 0xffff) >> 16) & 0xffffffff;
+  A = dlsr + lsr;
+  if (A > 0 && ntp > A)
+    A = ntp - A;
+  else
+    A = 0;
   curr->round_trip = A;
 
   GST_DEBUG ("NTP %04x:%04x, round trip %04x:%04x", ntp >> 16, ntp & 0xffff,
@@ -1229,7 +1362,7 @@ rtp_source_get_new_sr (RTPSource * src, guint64 ntpnstime,
 /**
  * rtp_source_get_new_rb:
  * @src: an #RTPSource
- * @ntpnstime: the current time in nanoseconds since 1970
+ * @time: the current time of the system clock
  * @fractionlost: fraction lost since last SR/RR
  * @packetslost: the cumululative number of packets lost
  * @exthighestseq: the extended last sequence number received
@@ -1242,7 +1375,7 @@ rtp_source_get_new_sr (RTPSource * src, guint64 ntpnstime,
  * Returns: %TRUE on success.
  */
 gboolean
-rtp_source_get_new_rb (RTPSource * src, guint64 ntpnstime,
+rtp_source_get_new_rb (RTPSource * src, GstClockTime time,
     guint8 * fractionlost, gint32 * packetslost, guint32 * exthighestseq,
     guint32 * jitter, guint32 * lsr, guint32 * dlsr)
 {
@@ -1288,7 +1421,7 @@ rtp_source_get_new_rb (RTPSource * src, guint64 ntpnstime,
 
     /* LSR is middle 32 bits of the last ntptime */
     LSR = (ntptime >> 16) & 0xffffffff;
-    diff = ntpnstime - sr_time;
+    diff = time - sr_time;
     GST_DEBUG ("last SR time diff %" GST_TIME_FORMAT, GST_TIME_ARGS (diff));
     /* DLSR, delay since last SR is expressed in 1/65536 second units */
     DLSR = gst_util_uint64_scale_int (diff, 65536, GST_SECOND);
@@ -1365,6 +1498,7 @@ rtp_source_get_last_sr (RTPSource * src, GstClockTime * time, guint64 * ntptime,
  * @jitter: the interarrival jitter
  * @lsr: the last SR packet from this source
  * @dlsr: the delay since last SR packet
+ * @round_trip: the round trip time
  *
  * Get the values of the last RB report set with rtp_source_process_rb().
  *
@@ -1373,7 +1507,7 @@ rtp_source_get_last_sr (RTPSource * src, GstClockTime * time, guint64 * ntptime,
 gboolean
 rtp_source_get_last_rb (RTPSource * src, guint8 * fractionlost,
     gint32 * packetslost, guint32 * exthighestseq, guint32 * jitter,
-    guint32 * lsr, guint32 * dlsr)
+    guint32 * lsr, guint32 * dlsr, guint32 * round_trip)
 {
   RTPReceiverReport *curr;
 
@@ -1395,6 +1529,8 @@ rtp_source_get_last_rb (RTPSource * src, guint8 * fractionlost,
     *lsr = curr->lsr;
   if (dlsr)
     *dlsr = curr->dlsr;
+  if (round_trip)
+    *round_trip = curr->round_trip;
 
   return TRUE;
 }
diff --git a/gst/rtpmanager/rtpsource.h b/gst/rtpmanager/rtpsource.h
index c4c23a8bf..0353a58b3 100644
--- a/gst/rtpmanager/rtpsource.h
+++ b/gst/rtpmanager/rtpsource.h
@@ -113,6 +113,7 @@ struct _RTPSource {
 
   gint          probation;
   gboolean      validated;
+  gboolean      internal;
   gboolean      is_csrc;
   gboolean      is_sender;
 
@@ -139,6 +140,11 @@ struct _RTPSource {
   GstClockTime  last_rtptime;
   GstClockTime  last_ntpnstime;
 
+  /* for bitrate estimation */
+  guint64       bitrate;
+  GstClockTime  prev_ntpnstime;
+  guint64       bytes_sent;
+
   GQueue       *packets;
 
   RTPSourceCallbacks callbacks;
@@ -198,7 +204,7 @@ void            rtp_source_process_rb          (RTPSource *src, GstClockTime tim
                                                 gint32 packetslost, guint32 exthighestseq, guint32 jitter,
                                                 guint32 lsr, guint32 dlsr);
 
-gboolean        rtp_source_get_new_sr          (RTPSource *src, GstClockTime time, guint64 *ntptime,
+gboolean        rtp_source_get_new_sr          (RTPSource *src, guint64 ntpnstime, guint64 *ntptime,
 		                                guint32 *rtptime, guint32 *packet_count,
 						guint32 *octet_count);
 gboolean        rtp_source_get_new_rb          (RTPSource *src, GstClockTime time, guint8 *fractionlost,
@@ -210,7 +216,7 @@ gboolean        rtp_source_get_last_sr         (RTPSource *src, GstClockTime *ti
 						guint32 *octet_count);
 gboolean        rtp_source_get_last_rb         (RTPSource *src, guint8 *fractionlost, gint32 *packetslost,
                                                 guint32 *exthighestseq, guint32 *jitter,
-                                                guint32 *lsr, guint32 *dlsr);
+                                                guint32 *lsr, guint32 *dlsr, guint32 *round_trip);
 
 void            rtp_source_reset               (RTPSource * src);
 
diff --git a/gst/rtpmanager/rtpstats.h b/gst/rtpmanager/rtpstats.h
index 3408300d3..e58243153 100644
--- a/gst/rtpmanager/rtpstats.h
+++ b/gst/rtpmanager/rtpstats.h
@@ -57,7 +57,9 @@ typedef struct {
 /**
  * RTPArrivalStats:
  * @time: arrival time of a packet according to the system clock
- * @timestamp: arrival time of a packet as buffer timestamp
+ * @running_time: arrival time of a packet as buffer running_time
+ * @ntpnstime: arrival time of a packet as NTP time in nanoseconds
+ * @have_address: if the @address field contains a valid address
  * @address: address of the sender of the packet
  * @bytes: bytes of the packet including lowlevel overhead
  * @payload_len: bytes of the RTP payload
@@ -66,7 +68,7 @@ typedef struct {
  */
 typedef struct {
   GstClockTime  time;
-  GstClockTime  timestamp;
+  GstClockTime  running_time;
   guint64       ntpnstime;
   gboolean      have_address;
   GstNetAddress address;