gst/rtpmanager/gstrtpjitterbuffer.c: Remove private version of a function that is in -base now.

Original commit message from CVS:
* gst/rtpmanager/gstrtpjitterbuffer.c:
(gst_rtp_jitter_buffer_init), (gst_rtp_jitter_buffer_flush_stop),
(gst_rtp_jitter_buffer_src_event), (gst_rtp_jitter_buffer_chain),
(gst_rtp_jitter_buffer_loop):
Remove private version of a function that is in -base now.
Add src event handler.
Rework the jitterbuffer pushing loop so that it can quickly react to
lost packets and instruct the depayloader of them. This can then be used
to implement error concealment data.

1 files changed, 109 insertions, 37 deletions

diff --git a/gst/rtpmanager/gstrtpjitterbuffer.c b/gst/rtpmanager/gstrtpjitterbuffer.c
index 9bea78f2..2c070c21 100644
--- a/gst/rtpmanager/gstrtpjitterbuffer.c
+++ b/gst/rtpmanager/gstrtpjitterbuffer.c
@@ -146,6 +146,8 @@ struct _GstRtpJitterBufferPrivate
   guint32 last_popped_seqnum;
   /* the next expected seqnum */
   guint32 next_seqnum;
+  /* last output time */
+  GstClockTime last_out_time;
 
   /* state */
   gboolean eos;
@@ -219,6 +221,8 @@ static GstCaps *gst_rtp_jitter_buffer_getcaps (GstPad * pad);
 
 /* sinkpad overrides */
 static gboolean gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps);
+static gboolean gst_rtp_jitter_buffer_src_event (GstPad * pad,
+    GstEvent * event);
 static gboolean gst_rtp_jitter_buffer_sink_event (GstPad * pad,
     GstEvent * event);
 static GstFlowReturn gst_rtp_jitter_buffer_chain (GstPad * pad,
@@ -349,6 +353,8 @@ gst_rtp_jitter_buffer_init (GstRtpJitterBuffer * jitterbuffer,
       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_query));
   gst_pad_set_getcaps_function (priv->srcpad,
       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_getcaps));
+  gst_pad_set_event_function (priv->srcpad,
+      GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_event));
 
   priv->sinkpad =
       gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_sink_template,
@@ -541,6 +547,7 @@ gst_rtp_jitter_buffer_flush_stop (GstRtpJitterBuffer * jitterbuffer)
   priv->srcresult = GST_FLOW_OK;
   gst_segment_init (&priv->segment, GST_FORMAT_TIME);
   priv->last_popped_seqnum = -1;
+  priv->last_out_time = -1;
   priv->next_seqnum = -1;
   priv->clock_rate = -1;
   priv->eos = FALSE;
@@ -646,19 +653,26 @@ gst_rtp_jitter_buffer_change_state (GstElement * element,
   return ret;
 }
 
-/**
- * Performs comparison 'b - a' with check for overflows.
- */
-static inline gint
-priv_compare_rtp_seq_lt (guint16 a, guint16 b)
+static gboolean
+gst_rtp_jitter_buffer_src_event (GstPad * pad, GstEvent * event)
 {
-  /* check if diff more than half of the 16bit range */
-  if (abs (b - a) > (1 << 15)) {
-    /* one of a/b has wrapped */
-    return a - b;
-  } else {
-    return b - a;
+  gboolean ret = TRUE;
+  GstRtpJitterBuffer *jitterbuffer;
+  GstRtpJitterBufferPrivate *priv;
+
+  jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
+  priv = jitterbuffer->priv;
+
+  GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
+
+  switch (GST_EVENT_TYPE (event)) {
+    default:
+      ret = gst_pad_push_event (priv->sinkpad, event);
+      break;
   }
+  gst_object_unref (jitterbuffer);
+
+  return ret;
 }
 
 static gboolean
@@ -856,7 +870,8 @@ gst_rtp_jitter_buffer_chain (GstPad * pad, GstBuffer * buffer)
   /* let's check if this buffer is too late, we cannot accept packets with
    * bigger seqnum than the one we already pushed. */
   if (priv->last_popped_seqnum != -1) {
-    if (priv_compare_rtp_seq_lt (priv->last_popped_seqnum, seqnum) < 0)
+    /* FIXME. isn't this supposed to be <= ? */
+    if (gst_rtp_buffer_compare_seqnum (priv->last_popped_seqnum, seqnum) < 0)
       goto too_late;
   }
 
@@ -988,7 +1003,10 @@ gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer)
   GstBuffer *outbuf;
   GstFlowReturn result;
   guint16 seqnum;
+  guint32 next_seqnum;
   GstClockTime timestamp, out_time;
+  gboolean discont = FALSE;
+  gint gap;
 
   priv = jitterbuffer->priv;
 
@@ -996,10 +1014,9 @@ gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer)
 again:
   GST_DEBUG_OBJECT (jitterbuffer, "Peeking item");
   while (TRUE) {
-
     /* always wait if we are blocked */
     if (!priv->blocked) {
-      /* if we have a packet, we can grab it */
+      /* if we have a packet, we can exit the loop and grab it */
       if (rtp_jitter_buffer_num_packets (priv->jbuf) > 0)
         break;
       /* no packets but we are EOS, do eos logic */
@@ -1018,21 +1035,43 @@ again:
    * new buffer is available. The peeked buffer is valid for as long as we hold
    * the jitterbuffer lock. */
   outbuf = rtp_jitter_buffer_peek (priv->jbuf);
+
+  /* get the seqnum and the next expected seqnum */
   seqnum = gst_rtp_buffer_get_seq (outbuf);
+  next_seqnum = priv->next_seqnum;
 
   /* get the timestamp, this is already corrected for clock skew by the
    * jitterbuffer */
   timestamp = GST_BUFFER_TIMESTAMP (outbuf);
 
   GST_DEBUG_OBJECT (jitterbuffer,
-      "Peeked buffer #%d, timestamp %" GST_TIME_FORMAT ", now %d left",
-      seqnum, GST_TIME_ARGS (timestamp),
+      "Peeked buffer #%d, expect #%d, timestamp %" GST_TIME_FORMAT
+      ", now %d left", seqnum, next_seqnum, GST_TIME_ARGS (timestamp),
       rtp_jitter_buffer_num_packets (priv->jbuf));
 
   /* apply our timestamp offset to the incomming buffer, this will be our output
    * timestamp. */
   out_time = apply_offset (jitterbuffer, timestamp);
 
+  /* get the gap between this and the previous packet. If we don't know the
+   * previous packet seqnum assume no gap. */
+  if (next_seqnum != -1) {
+    gap = gst_rtp_buffer_compare_seqnum (next_seqnum, seqnum);
+
+    /* if we have a packet that we already pushed or considered dropped, pop it
+     * off and get the next packet */
+    if (gap < 0) {
+      GST_DEBUG_OBJECT (jitterbuffer, "Old packet #%d, next #%d dropping",
+          seqnum, next_seqnum);
+      outbuf = rtp_jitter_buffer_pop (priv->jbuf);
+      gst_buffer_unref (outbuf);
+      goto again;
+    }
+  } else {
+    GST_DEBUG_OBJECT (jitterbuffer, "no next seqnum known, first packet");
+    gap = -1;
+  }
+
   /* If we don't know what the next seqnum should be (== -1) we have to wait
    * because it might be possible that we are not receiving this buffer in-order,
    * a buffer with a lower seqnum could arrive later and we want to push that
@@ -1041,18 +1080,32 @@ again:
    * determine if we have missing a packet. If we have a missing packet (which
    * must be before this packet) we can wait for it until the deadline for this
    * packet expires. */
-  if ((priv->next_seqnum == -1 || priv->next_seqnum != seqnum)
-      && out_time != -1) {
+  if (gap != 0 && out_time != -1) {
     GstClockID id;
     GstClockTime sync_time;
     GstClockReturn ret;
     GstClock *clock;
+    GstClockTime duration = -1;
 
-    if (priv->next_seqnum != -1) {
-      /* we expected next_seqnum but received something else, that's a gap */
+    if (gap > 0) {
+      /* we have a gap */
       GST_WARNING_OBJECT (jitterbuffer,
-          "Sequence number GAP detected: expected %d instead of %d",
-          priv->next_seqnum, seqnum);
+          "Sequence number GAP detected: expected %d instead of %d (%d missing)",
+          next_seqnum, seqnum, gap);
+
+      if (priv->last_out_time != -1) {
+        GST_DEBUG_OBJECT (jitterbuffer,
+            "out_time %" GST_TIME_FORMAT ", last %" GST_TIME_FORMAT,
+            GST_TIME_ARGS (out_time), GST_TIME_ARGS (priv->last_out_time));
+        /* interpollate between the current time and the last time based on
+         * number of packets we are missing, this is the estimated duration
+         * for the missing packet based on equidistant packet spacing. */
+        duration = (out_time - priv->last_out_time) / (gap + 1);
+        GST_DEBUG_OBJECT (jitterbuffer, "duration %" GST_TIME_FORMAT,
+            GST_TIME_ARGS (duration));
+        /* add this duration to the timestamp of the last packet we pushed */
+        out_time = (priv->last_out_time + duration);
+      }
     } else {
       /* we don't know what the next_seqnum should be, wait for the last
        * possible moment to push this buffer, maybe we get an earlier seqnum
@@ -1104,27 +1157,45 @@ again:
           "Wait got unscheduled, will retry to push with new buffer");
       goto again;
     }
-    /* Get new timestamp, latency might have changed */
+
+    /* we now timed out, this means we lost a packet or finished synchronizing
+     * on the first buffer. */
+    if (gap > 0) {
+      GstEvent *event;
+
+      /* we had a gap and thus we lost a packet. Creat an event for this.  */
+      GST_DEBUG_OBJECT (jitterbuffer, "Packet #%d lost", next_seqnum);
+      priv->num_late++;
+      discont = TRUE;
+
+      /* create paket lost event */
+      event = gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM,
+          gst_structure_new ("GstRTPPacketLost",
+              "seqnum", G_TYPE_UINT, (guint) next_seqnum,
+              "timestamp", G_TYPE_UINT64, out_time,
+              "duration", G_TYPE_UINT64, duration, NULL));
+      gst_pad_push_event (priv->srcpad, event);
+
+      /* update our expected next packet */
+      priv->last_popped_seqnum = next_seqnum;
+      priv->last_out_time = out_time;
+      priv->next_seqnum = (next_seqnum + 1) & 0xffff;
+      /* look for next packet */
+      goto again;
+    }
+
+    /* there was no known gap,just the first packet, exit the loop and push */
+    GST_DEBUG_OBJECT (jitterbuffer, "First packet #%d synced", seqnum);
+
+    /* get new timestamp, latency might have changed */
     out_time = apply_offset (jitterbuffer, timestamp);
   }
 push_buffer:
+
   /* when we get here we are ready to pop and push the buffer */
   outbuf = rtp_jitter_buffer_pop (priv->jbuf);
 
-  /* check if we are pushing something unexpected */
-  if (priv->next_seqnum != -1 && priv->next_seqnum != seqnum) {
-    gint dropped;
-
-    /* calc number of missing packets, careful for wraparounds */
-    dropped = priv_compare_rtp_seq_lt (priv->next_seqnum, seqnum);
-
-    GST_DEBUG_OBJECT (jitterbuffer,
-        "Pushing DISCONT after dropping %d (%d to %d)", dropped,
-        priv->next_seqnum, seqnum);
-
-    /* update stats */
-    priv->num_late += dropped;
-
+  if (discont) {
     /* set DISCONT flag when we missed a packet. */
     outbuf = gst_buffer_make_metadata_writable (outbuf);
     GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_DISCONT);
@@ -1136,6 +1207,7 @@ push_buffer:
   /* 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;
+  priv->last_out_time = out_time;
   priv->next_seqnum = (seqnum + 1) & 0xffff;
   JBUF_UNLOCK (priv);