jitterbuffer: do more buffering implementation

Add callback for buffering stats.
Configure the latency in the jitterbuffer instead of passing it with _insert.
Calculate buffering levels when pushing and popping
Post buffering messages.

gst/rtpmanager/gstrtpjitterbuffer.c

@@ -285,6 +285,9 @@ static gboolean gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query);
 
 static void
 gst_rtp_jitter_buffer_clear_pt_map (GstRtpJitterBuffer * jitterbuffer);
+static void
+do_stats_cb (RTPJitterBuffer * jbuf, guint percent,
+    GstRtpJitterBuffer * jitterbuffer);
 
 static void
 gst_rtp_jitter_buffer_base_init (gpointer klass)
@@ -448,6 +451,9 @@ gst_rtp_jitter_buffer_init (GstRtpJitterBuffer * jitterbuffer,
   priv->do_lost = DEFAULT_DO_LOST;
 
   priv->jbuf = rtp_jitter_buffer_new ();
+  rtp_jitter_buffer_set_delay (priv->jbuf, priv->latency_ns);
+  rtp_jitter_buffer_set_stats_cb (priv->jbuf, (RTPBufferingStats) do_stats_cb,
+      jitterbuffer);
   priv->jbuf_lock = g_mutex_new ();
   priv->jbuf_cond = g_cond_new ();
 
@@ -1132,6 +1138,19 @@ parse_failed:
   }
 }
 
+static void
+do_stats_cb (RTPJitterBuffer * jbuf, guint percent,
+    GstRtpJitterBuffer * jitterbuffer)
+{
+  GstMessage *message;
+
+  /* Post a buffering message */
+  message = gst_message_new_buffering (GST_OBJECT_CAST (jitterbuffer), percent);
+  gst_message_set_buffering_stats (message, GST_BUFFERING_LIVE, -1, -1, -1);
+
+  gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer), message);
+}
+
 static GstFlowReturn
 gst_rtp_jitter_buffer_chain (GstPad * pad, GstBuffer * buffer)
 {
@@ -1438,8 +1457,8 @@ again:
     id = NULL;
     /* always wait if we are blocked */
     if (G_LIKELY (!priv->blocked)) {
-      /* we're buffering, wait */
+      /* we're buffering but not EOS, wait. */
-      if (rtp_jitter_buffer_is_buffering (priv->jbuf))
+      if (!priv->eos && rtp_jitter_buffer_is_buffering (priv->jbuf))
         goto do_wait;
       /* if we have a packet, we can exit the loop and grab it */
       if (rtp_jitter_buffer_num_packets (priv->jbuf) > 0)
@@ -1969,6 +1988,7 @@ gst_rtp_jitter_buffer_set_property (GObject * object,
       old_latency = priv->latency_ms;
       priv->latency_ms = new_latency;
       priv->latency_ns = priv->latency_ms * GST_MSECOND;
+      rtp_jitter_buffer_set_delay (priv->jbuf, priv->latency_ns);
       JBUF_UNLOCK (priv);
 
       /* post message if latency changed, this will inform the parent pipeline

gst/rtpmanager/rtpjitterbuffer.c

@@ -100,6 +100,24 @@ rtp_jitter_buffer_new (void)
   return jbuf;
 }
 
+/**
+ * rtp_jitter_buffer_set_stats_cb:
+ * @jbuf: an #RTPJitterBuffer
+ * @stats: the stats callback
+ * @user_data: user data passed to the callback
+ *
+ * Install a callbacl that will be called when the buffering state of @jbuf
+ * changed.
+ */
+void
+rtp_jitter_buffer_set_stats_cb (RTPJitterBuffer * jbuf,
+    RTPBufferingStats stats_cb, gpointer user_data)
+{
+  jbuf->stats_cb = stats_cb;
+  jbuf->stats_data = user_data;
+}
+
+
 /**
  * rtp_jitter_buffer_get_mode:
  * @jbuf: an #RTPJitterBuffer
@@ -125,14 +143,26 @@ void
 rtp_jitter_buffer_set_mode (RTPJitterBuffer * jbuf, RTPJitterBufferMode mode)
 {
   jbuf->mode = mode;
-
-  if (mode == RTP_JITTER_BUFFER_MODE_BUFFER) {
-    /* when we're in buffering mode, always set our state to buffering, we'll
-     * get out of it when we push the next buffer */
-    jbuf->buffering = TRUE;
-  }
 }
 
+GstClockTime
+rtp_jitter_buffer_get_delay (RTPJitterBuffer * jbuf)
+{
+  return jbuf->delay;
+}
+
+void
+rtp_jitter_buffer_set_delay (RTPJitterBuffer * jbuf, GstClockTime delay)
+{
+  jbuf->delay = delay;
+  jbuf->low_level = (delay * 15) / 100;
+  jbuf->high_level = (delay * 90) / 100;
+  GST_DEBUG ("delay %" GST_TIME_FORMAT ", min %" GST_TIME_FORMAT ", max %"
+      GST_TIME_FORMAT, GST_TIME_ARGS (jbuf->level),
+      GST_TIME_ARGS (jbuf->low_level), GST_TIME_ARGS (jbuf->high_level));
+}
+
+
 /**
  * rtp_jitter_buffer_reset_skew:
  * @jbuf: an #RTPJitterBuffer
@@ -175,6 +205,55 @@ rtp_jitter_buffer_resync (RTPJitterBuffer * jbuf, GstClockTime time,
   }
 }
 
+static void
+update_buffer_level (RTPJitterBuffer * jbuf)
+{
+  GstBuffer *high_buf, *low_buf;
+  gboolean post = FALSE;
+
+  high_buf = g_queue_peek_head (jbuf->packets);
+  low_buf = g_queue_peek_tail (jbuf->packets);
+
+  if (!high_buf || !low_buf || high_buf == low_buf) {
+    jbuf->level = 0;
+  } else {
+    guint64 high_ts, low_ts;
+
+    high_ts = GST_BUFFER_TIMESTAMP (high_buf);
+    low_ts = GST_BUFFER_TIMESTAMP (low_buf);
+
+    if (high_ts > low_ts)
+      jbuf->level = high_ts - low_ts;
+    else
+      jbuf->level = 0;
+  }
+  GST_DEBUG ("buffer level %" GST_TIME_FORMAT, GST_TIME_ARGS (jbuf->level));
+
+  if (jbuf->buffering) {
+    post = TRUE;
+    if (jbuf->level > jbuf->high_level) {
+      GST_DEBUG ("buffering finished");
+      jbuf->buffering = FALSE;
+    }
+  } else {
+    if (jbuf->level < jbuf->low_level) {
+      GST_DEBUG ("buffering started");
+      jbuf->buffering = TRUE;
+      post = TRUE;
+    }
+  }
+  if (post) {
+    gint percent = (jbuf->level * 100 / jbuf->delay);
+
+    percent = MIN (percent, 100);
+
+    if (jbuf->stats_cb)
+      jbuf->stats_cb (jbuf, percent, jbuf->stats_data);
+
+    GST_DEBUG ("buffering %d", percent);
+  }
+}
+
 /* For the clock skew we use a windowed low point averaging algorithm as can be
  * found in http://www.grame.fr/pub/TR-050601.pdf. The idea is that the jitter is
  * composed of:
@@ -431,14 +510,13 @@ no_skew:
         out_time = jbuf->prev_out_time;
       }
     }
-
+    if (time != -1 && out_time + jbuf->delay < time) {
-    if (time != -1 && out_time + max_delay < time) {
       /* if we are going to produce a timestamp that is later than the input
        * timestamp, we need to reset the jitterbuffer. Likely the server paused
        * temporarily */
       GST_DEBUG ("out %" GST_TIME_FORMAT " + %" G_GUINT64_FORMAT " < time %"
           GST_TIME_FORMAT ", reset jitterbuffer", GST_TIME_ARGS (out_time),
-          max_delay, GST_TIME_ARGS (time));
+          jbuf->delay, GST_TIME_ARGS (time));
       rtp_jitter_buffer_resync (jbuf, time, gstrtptime, ext_rtptime, TRUE);
       out_time = time;
       send_diff = 0;
@@ -516,10 +594,6 @@ rtp_jitter_buffer_insert (RTPJitterBuffer * jbuf, GstBuffer * buf,
         time = 0;
       else
         time = -1;
-      time = calculate_skew (jbuf, rtptime, time, clock_rate, max_delay);
-      break;
-    case RTP_JITTER_BUFFER_MODE_SLAVE:
-      time = calculate_skew (jbuf, rtptime, time, clock_rate, max_delay);
       break;
     case RTP_JITTER_BUFFER_MODE_BUFFER:
       /* send -1 for all timestamps except the first one. This will make the
@@ -528,12 +602,12 @@ rtp_jitter_buffer_insert (RTPJitterBuffer * jbuf, GstBuffer * buf,
        * buffered */
       if (jbuf->base_time != -1)
         time = -1;
-      time = calculate_skew (jbuf, rtptime, time, clock_rate, max_delay);
       break;
+    case RTP_JITTER_BUFFER_MODE_SLAVE:
     default:
       break;
   }
-
+  time = calculate_skew (jbuf, rtptime, time, clock_rate);
   GST_BUFFER_TIMESTAMP (buf) = time;
 
   /* It's more likely that the packet was inserted in the front of the buffer */
@@ -542,6 +616,10 @@ rtp_jitter_buffer_insert (RTPJitterBuffer * jbuf, GstBuffer * buf,
   else
     g_queue_push_tail (jbuf->packets, buf);
 
+  /* buffering mode, update buffer stats */
+  if (jbuf->mode == RTP_JITTER_BUFFER_MODE_BUFFER)
+    update_buffer_level (jbuf);
+
   /* tail was changed when we did not find a previous packet, we set the return
    * flag when requested. */
   if (G_LIKELY (tail))
@@ -576,6 +654,10 @@ rtp_jitter_buffer_pop (RTPJitterBuffer * jbuf)
 
   buf = g_queue_pop_tail (jbuf->packets);
 
+  /* buffering mode, update buffer stats */
+  if (jbuf->mode == RTP_JITTER_BUFFER_MODE_BUFFER)
+    update_buffer_level (jbuf);
+
   return buf;
 }
 

gst/rtpmanager/rtpjitterbuffer.h

@@ -56,13 +56,14 @@ typedef enum {
 } RTPJitterBufferMode;
 
 /**
- * RTPTailChanged:
+ * RTPBufferingStats:
  * @jbuf: an #RTPJitterBuffer
+ * @percent: the buffering percent
  * @user_data: user data specified when registering
  *
- * This callback will be called when the tail buffer of @jbuf changed.
+ * Called when buffering is going on in @jbuf.
  */
-typedef void (*RTPTailChanged) (RTPJitterBuffer *jbuf, gpointer user_data);
+typedef void (*RTPBufferingStats) (RTPJitterBuffer *jbuf, guint percent, gpointer user_data);
 
 #define RTP_JITTER_BUFFER_MAX_WINDOW 512
 /**
@@ -77,7 +78,15 @@ struct _RTPJitterBuffer {
 
   RTPJitterBufferMode mode;
 
-  gboolean       buffering;
+  GstClockTime   delay;
+
+  /* for buffering */
+  gboolean          buffering;
+  guint64           level;
+  guint64           low_level;
+  guint64           high_level;
+  RTPBufferingStats stats_cb;
+  gpointer          stats_data;
 
   /* for calculating skew */
   GstClockTime   base_time;
@@ -105,9 +114,15 @@ GType rtp_jitter_buffer_get_type (void);
 /* managing lifetime */
 RTPJitterBuffer*      rtp_jitter_buffer_new              (void);
 
+void                  rtp_jitter_buffer_set_stats_cb     (RTPJitterBuffer *jbuf, RTPBufferingStats stats_cb,
+                                                          gpointer user_data);
+
 RTPJitterBufferMode   rtp_jitter_buffer_get_mode         (RTPJitterBuffer *jbuf);
 void                  rtp_jitter_buffer_set_mode         (RTPJitterBuffer *jbuf, RTPJitterBufferMode mode);
 
+GstClockTime          rtp_jitter_buffer_get_delay        (RTPJitterBuffer *jbuf);
+void                  rtp_jitter_buffer_set_delay        (RTPJitterBuffer *jbuf, GstClockTime delay);
+
 void                  rtp_jitter_buffer_reset_skew       (RTPJitterBuffer *jbuf);
 
 gboolean              rtp_jitter_buffer_insert           (RTPJitterBuffer *jbuf, GstBuffer *buf,