gst/rtpmanager/gstrtpjitterbuffer.c

   1 /*
   2  * Farsight Voice+Video library
   3  *
   4  *  Copyright 2007 Collabora Ltd,
   5  *  Copyright 2007 Nokia Corporation
   6  *   @author: Philippe Kalaf <philippe.kalaf@collabora.co.uk>.
   7  *  Copyright 2007 Wim Taymans <wim.taymans@gmail.com>
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Library General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Library General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Library General Public
  20  * License along with this library; if not, write to the
  21  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  22  * Boston, MA 02111-1307, USA.
  23  *
  24  */
  25
  26 /**
  27  * SECTION:element-gstrtpjitterbuffer
  28  *
  29  * This element reorders and removes duplicate RTP packets as they are received
  30  * from a network source. It will also wait for missing packets up to a
  31  * configurable time limit using the #GstRtpJitterBuffer:latency property.
  32  * Packets arriving too late are considered to be lost packets.
  33  *
  34  * This element acts as a live element and so adds #GstRtpJitterBuffer:latency
  35  * to the pipeline.
  36  *
  37  * The element needs the clock-rate of the RTP payload in order to estimate the
  38  * delay. This information is obtained either from the caps on the sink pad or,
  39  * when no caps are present, from the #GstRtpJitterBuffer::request-pt-map signal.
  40  * To clear the previous pt-map use the #GstRtpJitterBuffer::clear-pt-map signal.
  41  *
  42  * This element will automatically be used inside gstrtpbin.
  43  *
  44  * <refsect2>
  45  * <title>Example pipelines</title>
  46  * |[
  47  * gst-launch rtspsrc location=rtsp://192.168.1.133:8554/mpeg1or2AudioVideoTest ! gstrtpjitterbuffer ! rtpmpvdepay ! mpeg2dec ! xvimagesink
  48  * ]| Connect to a streaming server and decode the MPEG video. The jitterbuffer is
  49  * inserted into the pipeline to smooth out network jitter and to reorder the
  50  * out-of-order RTP packets.
  51  * </refsect2>
  52  *
  53  * Last reviewed on 2007-05-28 (0.10.5)
  54  */
  55
  56 #ifdef HAVE_CONFIG_H
  57 #include "config.h"
  58 #endif
  59
  60 #include <stdlib.h>
  61 #include <string.h>
  62 #include <gst/rtp/gstrtpbuffer.h>
  63
  64 #include "gstrtpbin-marshal.h"
  65
  66 #include "gstrtpjitterbuffer.h"
  67 #include "rtpjitterbuffer.h"
  68
  69 GST_DEBUG_CATEGORY (rtpjitterbuffer_debug);
  70 #define GST_CAT_DEFAULT (rtpjitterbuffer_debug)
  71
  72 /* low and high threshold tell the queue when to start and stop buffering */
  73 #define LOW_THRESHOLD 0.2
  74 #define HIGH_THRESHOLD 0.8
  75
  76 /* elementfactory information */
  77 static const GstElementDetails gst_rtp_jitter_buffer_details =
  78 GST_ELEMENT_DETAILS ("RTP packet jitter-buffer",
  79     "Filter/Network/RTP",
  80     "A buffer that deals with network jitter and other transmission faults",
  81     "Philippe Kalaf <philippe.kalaf@collabora.co.uk>, "
  82     "Wim Taymans <wim.taymans@gmail.com>");
  83
  84 /* RTPJitterBuffer signals and args */
  85 enum
  86 {
  87   SIGNAL_REQUEST_PT_MAP,
  88   SIGNAL_CLEAR_PT_MAP,
  89   LAST_SIGNAL
  90 };
  91
  92 #define DEFAULT_LATENCY_MS      200
  93 #define DEFAULT_DROP_ON_LATENCY FALSE
  94 #define DEFAULT_TS_OFFSET       0
  95 #define DEFAULT_DO_LOST         FALSE
  96
  97 enum
  98 {
  99   PROP_0,
 100   PROP_LATENCY,
 101   PROP_DROP_ON_LATENCY,
 102   PROP_TS_OFFSET,
 103   PROP_DO_LOST,
 104   PROP_LAST
 105 };
 106
 107 #define JBUF_LOCK(priv)   (g_mutex_lock ((priv)->jbuf_lock))
 108
 109 #define JBUF_LOCK_CHECK(priv,label) G_STMT_START {    \
 110   JBUF_LOCK (priv);                                   \
 111   if (priv->srcresult != GST_FLOW_OK)                 \
 112     goto label;                                       \
 113 } G_STMT_END
 114
 115 #define JBUF_UNLOCK(priv) (g_mutex_unlock ((priv)->jbuf_lock))
 116 #define JBUF_WAIT(priv)   (g_cond_wait ((priv)->jbuf_cond, (priv)->jbuf_lock))
 117
 118 #define JBUF_WAIT_CHECK(priv,label) G_STMT_START {    \
 119   JBUF_WAIT(priv);                                    \
 120   if (priv->srcresult != GST_FLOW_OK)                 \
 121     goto label;                                       \
 122 } G_STMT_END
 123
 124 #define JBUF_SIGNAL(priv) (g_cond_signal ((priv)->jbuf_cond))
 125
 126 struct _GstRtpJitterBufferPrivate
 127 {
 128   GstPad *sinkpad, *srcpad;
 129
 130   RTPJitterBuffer *jbuf;
 131   GMutex *jbuf_lock;
 132   GCond *jbuf_cond;
 133   gboolean waiting;
 134   gboolean discont;
 135
 136   /* properties */
 137   guint latency_ms;
 138   gboolean drop_on_latency;
 139   gint64 ts_offset;
 140   gboolean do_lost;
 141
 142   /* the last seqnum we pushed out */
 143   guint32 last_popped_seqnum;
 144   /* the next expected seqnum */
 145   guint32 next_seqnum;
 146   /* last output time */
 147   GstClockTime last_out_time;
 148
 149   /* state */
 150   gboolean eos;
 151
 152   /* clock rate and rtp timestamp offset */
 153   gint last_pt;
 154   gint32 clock_rate;
 155   gint64 clock_base;
 156   gint64 prev_ts_offset;
 157
 158   /* when we are shutting down */
 159   GstFlowReturn srcresult;
 160   gboolean blocked;
 161
 162   /* for sync */
 163   GstSegment segment;
 164   GstClockID clock_id;
 165   /* the latency of the upstream peer, we have to take this into account when
 166    * synchronizing the buffers. */
 167   GstClockTime peer_latency;
 168
 169   /* some accounting */
 170   guint64 num_late;
 171   guint64 num_duplicates;
 172 };
 173
 174 #define GST_RTP_JITTER_BUFFER_GET_PRIVATE(o) \
 175   (G_TYPE_INSTANCE_GET_PRIVATE ((o), GST_TYPE_RTP_JITTER_BUFFER, \
 176                                 GstRtpJitterBufferPrivate))
 177
 178 static GstStaticPadTemplate gst_rtp_jitter_buffer_sink_template =
 179 GST_STATIC_PAD_TEMPLATE ("sink",
 180     GST_PAD_SINK,
 181     GST_PAD_ALWAYS,
 182     GST_STATIC_CAPS ("application/x-rtp, "
 183         "clock-rate = (int) [ 1, 2147483647 ]"
 184         /* "payload = (int) , "
 185          * "encoding-name = (string) "
 186          */ )
 187     );
 188
 189 static GstStaticPadTemplate gst_rtp_jitter_buffer_src_template =
 190 GST_STATIC_PAD_TEMPLATE ("src",
 191     GST_PAD_SRC,
 192     GST_PAD_ALWAYS,
 193     GST_STATIC_CAPS ("application/x-rtp"
 194         /* "payload = (int) , "
 195          * "clock-rate = (int) , "
 196          * "encoding-name = (string) "
 197          */ )
 198     );
 199
 200 static guint gst_rtp_jitter_buffer_signals[LAST_SIGNAL] = { 0 };
 201
 202 GST_BOILERPLATE (GstRtpJitterBuffer, gst_rtp_jitter_buffer, GstElement,
 203     GST_TYPE_ELEMENT);
 204
 205 /* object overrides */
 206 static void gst_rtp_jitter_buffer_set_property (GObject * object,
 207     guint prop_id, const GValue * value, GParamSpec * pspec);
 208 static void gst_rtp_jitter_buffer_get_property (GObject * object,
 209     guint prop_id, GValue * value, GParamSpec * pspec);
 210 static void gst_rtp_jitter_buffer_finalize (GObject * object);
 211
 212 /* element overrides */
 213 static GstStateChangeReturn gst_rtp_jitter_buffer_change_state (GstElement
 214     * element, GstStateChange transition);
 215
 216 /* pad overrides */
 217 static GstCaps *gst_rtp_jitter_buffer_getcaps (GstPad * pad);
 218
 219 /* sinkpad overrides */
 220 static gboolean gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps);
 221 static gboolean gst_rtp_jitter_buffer_src_event (GstPad * pad,
 222     GstEvent * event);
 223 static gboolean gst_rtp_jitter_buffer_sink_event (GstPad * pad,
 224     GstEvent * event);
 225 static GstFlowReturn gst_rtp_jitter_buffer_chain (GstPad * pad,
 226     GstBuffer * buffer);
 227
 228 /* srcpad overrides */
 229 static gboolean
 230 gst_rtp_jitter_buffer_src_activate_push (GstPad * pad, gboolean active);
 231 static void gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer);
 232 static gboolean gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query);
 233
 234 static void
 235 gst_rtp_jitter_buffer_clear_pt_map (GstRtpJitterBuffer * jitterbuffer);
 236
 237 static void
 238 gst_rtp_jitter_buffer_base_init (gpointer klass)
 239 {
 240   GstElementClass *element_class = GST_ELEMENT_CLASS (klass);
 241
 242   gst_element_class_add_pad_template (element_class,
 243       gst_static_pad_template_get (&gst_rtp_jitter_buffer_src_template));
 244   gst_element_class_add_pad_template (element_class,
 245       gst_static_pad_template_get (&gst_rtp_jitter_buffer_sink_template));
 246   gst_element_class_set_details (element_class, &gst_rtp_jitter_buffer_details);
 247 }
 248
 249 static void
 250 gst_rtp_jitter_buffer_class_init (GstRtpJitterBufferClass * klass)
 251 {
 252   GObjectClass *gobject_class;
 253   GstElementClass *gstelement_class;
 254
 255   gobject_class = (GObjectClass *) klass;
 256   gstelement_class = (GstElementClass *) klass;
 257
 258   g_type_class_add_private (klass, sizeof (GstRtpJitterBufferPrivate));
 259
 260   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_finalize);
 261
 262   gobject_class->set_property = gst_rtp_jitter_buffer_set_property;
 263   gobject_class->get_property = gst_rtp_jitter_buffer_get_property;
 264
 265   /**
 266    * GstRtpJitterBuffer::latency:
 267    *
 268    * The maximum latency of the jitterbuffer. Packets will be kept in the buffer
 269    * for at most this time.
 270    */
 271   g_object_class_install_property (gobject_class, PROP_LATENCY,
 272       g_param_spec_uint ("latency", "Buffer latency in ms",
 273           "Amount of ms to buffer", 0, G_MAXUINT, DEFAULT_LATENCY_MS,
 274           G_PARAM_READWRITE));
 275   /**
 276    * GstRtpJitterBuffer::drop-on-latency:
 277    *
 278    * Drop oldest buffers when the queue is completely filled.
 279    */
 280   g_object_class_install_property (gobject_class, PROP_DROP_ON_LATENCY,
 281       g_param_spec_boolean ("drop-on-latency",
 282           "Drop buffers when maximum latency is reached",
 283           "Tells the jitterbuffer to never exceed the given latency in size",
 284           DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE));
 285   /**
 286    * GstRtpJitterBuffer::ts-offset:
 287    *
 288    * Adjust GStreamer output buffer timestamps in the jitterbuffer with offset.
 289    * This is mainly used to ensure interstream synchronisation.
 290    */
 291   g_object_class_install_property (gobject_class, PROP_TS_OFFSET,
 292       g_param_spec_int64 ("ts-offset", "Timestamp Offset",
 293           "Adjust buffer timestamps with offset in nanoseconds", G_MININT64,
 294           G_MAXINT64, DEFAULT_TS_OFFSET,
 295           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 296
 297   /**
 298    * GstRtpJitterBuffer::do-lost:
 299    *
 300    * Send out a GstRTPPacketLost event downstream when a packet is considered
 301    * lost.
 302    */
 303   g_object_class_install_property (gobject_class, PROP_DO_LOST,
 304       g_param_spec_boolean ("do-lost", "Do Lost",
 305           "Send an event downstream when a packet is lost", DEFAULT_DO_LOST,
 306           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 307   /**
 308    * GstRtpJitterBuffer::request-pt-map:
 309    * @buffer: the object which received the signal
 310    * @pt: the pt
 311    *
 312    * Request the payload type as #GstCaps for @pt.
 313    */
 314   gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP] =
 315       g_signal_new ("request-pt-map", G_TYPE_FROM_CLASS (klass),
 316       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
 317           request_pt_map), NULL, NULL, gst_rtp_bin_marshal_BOXED__UINT,
 318       GST_TYPE_CAPS, 1, G_TYPE_UINT);
 319   /**
 320    * GstRtpJitterBuffer::clear-pt-map:
 321    * @buffer: the object which received the signal
 322    *
 323    * Invalidate the clock-rate as obtained with the
 324    * #GstRtpJitterBuffer::request-pt-map signal.
 325    */
 326   gst_rtp_jitter_buffer_signals[SIGNAL_CLEAR_PT_MAP] =
 327       g_signal_new ("clear-pt-map", G_TYPE_FROM_CLASS (klass),
 328       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
 329           clear_pt_map), NULL, NULL, g_cclosure_marshal_VOID__VOID,
 330       G_TYPE_NONE, 0, G_TYPE_NONE);
 331
 332   gstelement_class->change_state = gst_rtp_jitter_buffer_change_state;
 333
 334   klass->clear_pt_map = GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_clear_pt_map);
 335
 336   GST_DEBUG_CATEGORY_INIT
 337       (rtpjitterbuffer_debug, "gstrtpjitterbuffer", 0, "RTP Jitter Buffer");
 338 }
 339
 340 static void
 341 gst_rtp_jitter_buffer_init (GstRtpJitterBuffer * jitterbuffer,
 342     GstRtpJitterBufferClass * klass)
 343 {
 344   GstRtpJitterBufferPrivate *priv;
 345
 346   priv = GST_RTP_JITTER_BUFFER_GET_PRIVATE (jitterbuffer);
 347   jitterbuffer->priv = priv;
 348
 349   priv->latency_ms = DEFAULT_LATENCY_MS;
 350   priv->drop_on_latency = DEFAULT_DROP_ON_LATENCY;
 351   priv->do_lost = DEFAULT_DO_LOST;
 352
 353   priv->jbuf = rtp_jitter_buffer_new ();
 354   priv->jbuf_lock = g_mutex_new ();
 355   priv->jbuf_cond = g_cond_new ();
 356
 357   priv->srcpad =
 358       gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_src_template,
 359       "src");
 360
 361   gst_pad_set_activatepush_function (priv->srcpad,
 362       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_activate_push));
 363   gst_pad_set_query_function (priv->srcpad,
 364       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_query));
 365   gst_pad_set_getcaps_function (priv->srcpad,
 366       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_getcaps));
 367   gst_pad_set_event_function (priv->srcpad,
 368       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_event));
 369
 370   priv->sinkpad =
 371       gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_sink_template,
 372       "sink");
 373
 374   gst_pad_set_chain_function (priv->sinkpad,
 375       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_chain));
 376   gst_pad_set_event_function (priv->sinkpad,
 377       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_sink_event));
 378   gst_pad_set_setcaps_function (priv->sinkpad,
 379       GST_DEBUG_FUNCPTR (gst_jitter_buffer_sink_setcaps));
 380   gst_pad_set_getcaps_function (priv->sinkpad,
 381       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_getcaps));
 382
 383   gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->srcpad);
 384   gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->sinkpad);
 385 }
 386
 387 static void
 388 gst_rtp_jitter_buffer_finalize (GObject * object)
 389 {
 390   GstRtpJitterBuffer *jitterbuffer;
 391
 392   jitterbuffer = GST_RTP_JITTER_BUFFER (object);
 393
 394   g_mutex_free (jitterbuffer->priv->jbuf_lock);
 395   g_cond_free (jitterbuffer->priv->jbuf_cond);
 396
 397   g_object_unref (jitterbuffer->priv->jbuf);
 398
 399   G_OBJECT_CLASS (parent_class)->finalize (object);
 400 }
 401
 402 static void
 403 gst_rtp_jitter_buffer_clear_pt_map (GstRtpJitterBuffer * jitterbuffer)
 404 {
 405   GstRtpJitterBufferPrivate *priv;
 406
 407   priv = jitterbuffer->priv;
 408
 409   /* this will trigger a new pt-map request signal, FIXME, do something better. */
 410   priv->clock_rate = -1;
 411 }
 412
 413 static GstCaps *
 414 gst_rtp_jitter_buffer_getcaps (GstPad * pad)
 415 {
 416   GstRtpJitterBuffer *jitterbuffer;
 417   GstRtpJitterBufferPrivate *priv;
 418   GstPad *other;
 419   GstCaps *caps;
 420   const GstCaps *templ;
 421
 422   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
 423   priv = jitterbuffer->priv;
 424
 425   other = (pad == priv->srcpad ? priv->sinkpad : priv->srcpad);
 426
 427   caps = gst_pad_peer_get_caps (other);
 428
 429   templ = gst_pad_get_pad_template_caps (pad);
 430   if (caps == NULL) {
 431     GST_DEBUG_OBJECT (jitterbuffer, "copy template");
 432     caps = gst_caps_copy (templ);
 433   } else {
 434     GstCaps *intersect;
 435
 436     GST_DEBUG_OBJECT (jitterbuffer, "intersect with template");
 437
 438     intersect = gst_caps_intersect (caps, templ);
 439     gst_caps_unref (caps);
 440
 441     caps = intersect;
 442   }
 443   gst_object_unref (jitterbuffer);
 444
 445   return caps;
 446 }
 447
 448 static gboolean
 449 gst_jitter_buffer_sink_parse_caps (GstRtpJitterBuffer * jitterbuffer,
 450     GstCaps * caps)
 451 {
 452   GstRtpJitterBufferPrivate *priv;
 453   GstStructure *caps_struct;
 454   guint val;
 455
 456   priv = jitterbuffer->priv;
 457
 458   /* first parse the caps */
 459   caps_struct = gst_caps_get_structure (caps, 0);
 460
 461   GST_DEBUG_OBJECT (jitterbuffer, "got caps");
 462
 463   /* we need a clock-rate to convert the rtp timestamps to GStreamer time and to
 464    * measure the amount of data in the buffer */
 465   if (!gst_structure_get_int (caps_struct, "clock-rate", &priv->clock_rate))
 466     goto error;
 467
 468   if (priv->clock_rate <= 0)
 469     goto wrong_rate;
 470
 471   GST_DEBUG_OBJECT (jitterbuffer, "got clock-rate %d", priv->clock_rate);
 472
 473   /* gah, clock-base is uint. If we don't have a base, we will use the first
 474    * buffer timestamp as the base time. This will screw up sync but it's better
 475    * than nothing. */
 476   if (gst_structure_get_uint (caps_struct, "clock-base", &val))
 477     priv->clock_base = val;
 478   else
 479     priv->clock_base = -1;
 480
 481   GST_DEBUG_OBJECT (jitterbuffer, "got clock-base %" G_GINT64_FORMAT,
 482       priv->clock_base);
 483
 484   /* first expected seqnum */
 485   if (gst_structure_get_uint (caps_struct, "seqnum-base", &val))
 486     priv->next_seqnum = val;
 487   else
 488     priv->next_seqnum = -1;
 489
 490   GST_DEBUG_OBJECT (jitterbuffer, "got seqnum-base %d", priv->next_seqnum);
 491
 492   return TRUE;
 493
 494   /* ERRORS */
 495 error:
 496   {
 497     GST_DEBUG_OBJECT (jitterbuffer, "No clock-rate in caps!");
 498     return FALSE;
 499   }
 500 wrong_rate:
 501   {
 502     GST_DEBUG_OBJECT (jitterbuffer, "Invalid clock-rate %d", priv->clock_rate);
 503     return FALSE;
 504   }
 505 }
 506
 507 static gboolean
 508 gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps)
 509 {
 510   GstRtpJitterBuffer *jitterbuffer;
 511   GstRtpJitterBufferPrivate *priv;
 512   gboolean res;
 513
 514   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
 515   priv = jitterbuffer->priv;
 516
 517   res = gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
 518
 519   /* set same caps on srcpad on success */
 520   if (res)
 521     gst_pad_set_caps (priv->srcpad, caps);
 522
 523   gst_object_unref (jitterbuffer);
 524
 525   return res;
 526 }
 527
 528 static void
 529 gst_rtp_jitter_buffer_flush_start (GstRtpJitterBuffer * jitterbuffer)
 530 {
 531   GstRtpJitterBufferPrivate *priv;
 532
 533   priv = jitterbuffer->priv;
 534
 535   JBUF_LOCK (priv);
 536   /* mark ourselves as flushing */
 537   priv->srcresult = GST_FLOW_WRONG_STATE;
 538   GST_DEBUG_OBJECT (jitterbuffer, "Disabling pop on queue");
 539   /* this unblocks any waiting pops on the src pad task */
 540   JBUF_SIGNAL (priv);
 541   /* unlock clock, we just unschedule, the entry will be released by the
 542    * locking streaming thread. */
 543   if (priv->clock_id)
 544     gst_clock_id_unschedule (priv->clock_id);
 545   JBUF_UNLOCK (priv);
 546 }
 547
 548 static void
 549 gst_rtp_jitter_buffer_flush_stop (GstRtpJitterBuffer * jitterbuffer)
 550 {
 551   GstRtpJitterBufferPrivate *priv;
 552
 553   priv = jitterbuffer->priv;
 554
 555   JBUF_LOCK (priv);
 556   GST_DEBUG_OBJECT (jitterbuffer, "Enabling pop on queue");
 557   /* Mark as non flushing */
 558   priv->srcresult = GST_FLOW_OK;
 559   gst_segment_init (&priv->segment, GST_FORMAT_TIME);
 560   priv->last_popped_seqnum = -1;
 561   priv->last_out_time = -1;
 562   priv->next_seqnum = -1;
 563   priv->clock_rate = -1;
 564   priv->eos = FALSE;
 565   rtp_jitter_buffer_flush (priv->jbuf);
 566   rtp_jitter_buffer_reset_skew (priv->jbuf);
 567   JBUF_UNLOCK (priv);
 568 }
 569
 570 static gboolean
 571 gst_rtp_jitter_buffer_src_activate_push (GstPad * pad, gboolean active)
 572 {
 573   gboolean result = TRUE;
 574   GstRtpJitterBuffer *jitterbuffer = NULL;
 575
 576   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
 577
 578   if (active) {
 579     /* allow data processing */
 580     gst_rtp_jitter_buffer_flush_stop (jitterbuffer);
 581
 582     /* start pushing out buffers */
 583     GST_DEBUG_OBJECT (jitterbuffer, "Starting task on srcpad");
 584     gst_pad_start_task (jitterbuffer->priv->srcpad,
 585         (GstTaskFunction) gst_rtp_jitter_buffer_loop, jitterbuffer);
 586   } else {
 587     /* make sure all data processing stops ASAP */
 588     gst_rtp_jitter_buffer_flush_start (jitterbuffer);
 589
 590     /* NOTE this will hardlock if the state change is called from the src pad
 591      * task thread because we will _join() the thread. */
 592     GST_DEBUG_OBJECT (jitterbuffer, "Stopping task on srcpad");
 593     result = gst_pad_stop_task (pad);
 594   }
 595
 596   gst_object_unref (jitterbuffer);
 597
 598   return result;
 599 }
 600
 601 static GstStateChangeReturn
 602 gst_rtp_jitter_buffer_change_state (GstElement * element,
 603     GstStateChange transition)
 604 {
 605   GstRtpJitterBuffer *jitterbuffer;
 606   GstRtpJitterBufferPrivate *priv;
 607   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
 608
 609   jitterbuffer = GST_RTP_JITTER_BUFFER (element);
 610   priv = jitterbuffer->priv;
 611
 612   switch (transition) {
 613     case GST_STATE_CHANGE_NULL_TO_READY:
 614       break;
 615     case GST_STATE_CHANGE_READY_TO_PAUSED:
 616       JBUF_LOCK (priv);
 617       /* reset negotiated values */
 618       priv->clock_rate = -1;
 619       priv->clock_base = -1;
 620       priv->peer_latency = 0;
 621       priv->last_pt = -1;
 622       /* block until we go to PLAYING */
 623       priv->blocked = TRUE;
 624       /* reset skew detection initialy */
 625       rtp_jitter_buffer_reset_skew (priv->jbuf);
 626       JBUF_UNLOCK (priv);
 627       break;
 628     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
 629       JBUF_LOCK (priv);
 630       /* unblock to allow streaming in PLAYING */
 631       priv->blocked = FALSE;
 632       JBUF_SIGNAL (priv);
 633       JBUF_UNLOCK (priv);
 634       break;
 635     default:
 636       break;
 637   }
 638
 639   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 640
 641   switch (transition) {
 642     case GST_STATE_CHANGE_READY_TO_PAUSED:
 643       /* we are a live element because we sync to the clock, which we can only
 644        * do in the PLAYING state */
 645       if (ret != GST_STATE_CHANGE_FAILURE)
 646         ret = GST_STATE_CHANGE_NO_PREROLL;
 647       break;
 648     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
 649       JBUF_LOCK (priv);
 650       /* block to stop streaming when PAUSED */
 651       priv->blocked = TRUE;
 652       JBUF_UNLOCK (priv);
 653       if (ret != GST_STATE_CHANGE_FAILURE)
 654         ret = GST_STATE_CHANGE_NO_PREROLL;
 655       break;
 656     case GST_STATE_CHANGE_PAUSED_TO_READY:
 657       break;
 658     case GST_STATE_CHANGE_READY_TO_NULL:
 659       break;
 660     default:
 661       break;
 662   }
 663
 664   return ret;
 665 }
 666
 667 static gboolean
 668 gst_rtp_jitter_buffer_src_event (GstPad * pad, GstEvent * event)
 669 {
 670   gboolean ret = TRUE;
 671   GstRtpJitterBuffer *jitterbuffer;
 672   GstRtpJitterBufferPrivate *priv;
 673
 674   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
 675   priv = jitterbuffer->priv;
 676
 677   GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
 678
 679   switch (GST_EVENT_TYPE (event)) {
 680     default:
 681       ret = gst_pad_push_event (priv->sinkpad, event);
 682       break;
 683   }
 684   gst_object_unref (jitterbuffer);
 685
 686   return ret;
 687 }
 688
 689 static gboolean
 690 gst_rtp_jitter_buffer_sink_event (GstPad * pad, GstEvent * event)
 691 {
 692   gboolean ret = TRUE;
 693   GstRtpJitterBuffer *jitterbuffer;
 694   GstRtpJitterBufferPrivate *priv;
 695
 696   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
 697   priv = jitterbuffer->priv;
 698
 699   GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
 700
 701   switch (GST_EVENT_TYPE (event)) {
 702     case GST_EVENT_NEWSEGMENT:
 703     {
 704       GstFormat format;
 705       gdouble rate, arate;
 706       gint64 start, stop, time;
 707       gboolean update;
 708
 709       gst_event_parse_new_segment_full (event, &update, &rate, &arate, &format,
 710           &start, &stop, &time);
 711
 712       /* we need time for now */
 713       if (format != GST_FORMAT_TIME)
 714         goto newseg_wrong_format;
 715
 716       GST_DEBUG_OBJECT (jitterbuffer,
 717           "newsegment: update %d, rate %g, arate %g, start %" GST_TIME_FORMAT
 718           ", stop %" GST_TIME_FORMAT ", time %" GST_TIME_FORMAT,
 719           update, rate, arate, GST_TIME_ARGS (start), GST_TIME_ARGS (stop),
 720           GST_TIME_ARGS (time));
 721
 722       /* now configure the values, we need these to time the release of the
 723        * buffers on the srcpad. */
 724       gst_segment_set_newsegment_full (&priv->segment, update,
 725           rate, arate, format, start, stop, time);
 726
 727       /* FIXME, push SEGMENT in the queue. Sorting order might be difficult. */
 728       ret = gst_pad_push_event (priv->srcpad, event);
 729       break;
 730     }
 731     case GST_EVENT_FLUSH_START:
 732       gst_rtp_jitter_buffer_flush_start (jitterbuffer);
 733       ret = gst_pad_push_event (priv->srcpad, event);
 734       break;
 735     case GST_EVENT_FLUSH_STOP:
 736       ret = gst_pad_push_event (priv->srcpad, event);
 737       ret = gst_rtp_jitter_buffer_src_activate_push (priv->srcpad, TRUE);
 738       break;
 739     case GST_EVENT_EOS:
 740     {
 741       /* push EOS in queue. We always push it at the head */
 742       JBUF_LOCK (priv);
 743       /* check for flushing, we need to discard the event and return FALSE when
 744        * we are flushing */
 745       ret = priv->srcresult == GST_FLOW_OK;
 746       if (ret && !priv->eos) {
 747         GST_DEBUG_OBJECT (jitterbuffer, "queuing EOS");
 748         priv->eos = TRUE;
 749         JBUF_SIGNAL (priv);
 750       } else if (priv->eos) {
 751         GST_DEBUG_OBJECT (jitterbuffer, "dropping EOS, we are already EOS");
 752       } else {
 753         GST_DEBUG_OBJECT (jitterbuffer, "dropping EOS, reason %s",
 754             gst_flow_get_name (priv->srcresult));
 755       }
 756       JBUF_UNLOCK (priv);
 757       gst_event_unref (event);
 758       break;
 759     }
 760     default:
 761       ret = gst_pad_push_event (priv->srcpad, event);
 762       break;
 763   }
 764
 765 done:
 766   gst_object_unref (jitterbuffer);
 767
 768   return ret;
 769
 770   /* ERRORS */
 771 newseg_wrong_format:
 772   {
 773     GST_DEBUG_OBJECT (jitterbuffer, "received non TIME newsegment");
 774     ret = FALSE;
 775     goto done;
 776   }
 777 }
 778
 779 static gboolean
 780 gst_rtp_jitter_buffer_get_clock_rate (GstRtpJitterBuffer * jitterbuffer,
 781     guint8 pt)
 782 {
 783   GValue ret = { 0 };
 784   GValue args[2] = { {0}, {0} };
 785   GstCaps *caps;
 786   gboolean res;
 787
 788   g_value_init (&args[0], GST_TYPE_ELEMENT);
 789   g_value_set_object (&args[0], jitterbuffer);
 790   g_value_init (&args[1], G_TYPE_UINT);
 791   g_value_set_uint (&args[1], pt);
 792
 793   g_value_init (&ret, GST_TYPE_CAPS);
 794   g_value_set_boxed (&ret, NULL);
 795
 796   g_signal_emitv (args, gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP], 0,
 797       &ret);
 798
 799   g_value_unset (&args[0]);
 800   g_value_unset (&args[1]);
 801   caps = (GstCaps *) g_value_dup_boxed (&ret);
 802   g_value_unset (&ret);
 803   if (!caps)
 804     goto no_caps;
 805
 806   res = gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
 807
 808   gst_caps_unref (caps);
 809
 810   return res;
 811
 812   /* ERRORS */
 813 no_caps:
 814   {
 815     GST_DEBUG_OBJECT (jitterbuffer, "could not get caps");
 816     return FALSE;
 817   }
 818 }
 819
 820 static GstFlowReturn
 821 gst_rtp_jitter_buffer_chain (GstPad * pad, GstBuffer * buffer)
 822 {
 823   GstRtpJitterBuffer *jitterbuffer;
 824   GstRtpJitterBufferPrivate *priv;
 825   guint16 seqnum;
 826   GstFlowReturn ret = GST_FLOW_OK;
 827   GstClockTime timestamp;
 828   guint64 latency_ts;
 829   gboolean tail;
 830
 831   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
 832
 833   if (!gst_rtp_buffer_validate (buffer))
 834     goto invalid_buffer;
 835
 836   priv = jitterbuffer->priv;
 837
 838   if (priv->last_pt != gst_rtp_buffer_get_payload_type (buffer)) {
 839     GstCaps *caps;
 840
 841     priv->last_pt = gst_rtp_buffer_get_payload_type (buffer);
 842     /* reset clock-rate so that we get a new one */
 843     priv->clock_rate = -1;
 844     /* Try to get the clock-rate from the caps first if we can. If there are no
 845      * caps we must fire the signal to get the clock-rate. */
 846     if ((caps = GST_BUFFER_CAPS (buffer))) {
 847       gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
 848     }
 849   }
 850
 851   if (priv->clock_rate == -1) {
 852     guint8 pt;
 853
 854     /* no clock rate given on the caps, try to get one with the signal */
 855     pt = gst_rtp_buffer_get_payload_type (buffer);
 856
 857     gst_rtp_jitter_buffer_get_clock_rate (jitterbuffer, pt);
 858     if (priv->clock_rate == -1)
 859       goto not_negotiated;
 860   }
 861
 862   /* take the timestamp of the buffer. This is the time when the packet was
 863    * received and is used to calculate jitter and clock skew. We will adjust
 864    * this timestamp with the smoothed value after processing it in the
 865    * jitterbuffer. */
 866   timestamp = GST_BUFFER_TIMESTAMP (buffer);
 867   /* bring to running time */
 868   timestamp = gst_segment_to_running_time (&priv->segment, GST_FORMAT_TIME,
 869       timestamp);
 870
 871   seqnum = gst_rtp_buffer_get_seq (buffer);
 872   GST_DEBUG_OBJECT (jitterbuffer,
 873       "Received packet #%d at time %" GST_TIME_FORMAT, seqnum,
 874       GST_TIME_ARGS (timestamp));
 875
 876   JBUF_LOCK_CHECK (priv, out_flushing);
 877   /* don't accept more data on EOS */
 878   if (priv->eos)
 879     goto have_eos;
 880
 881   /* let's check if this buffer is too late, we can only accept packets with
 882    * bigger seqnum than the one we last pushed. */
 883   if (priv->last_popped_seqnum != -1) {
 884     gint gap;
 885
 886     gap = gst_rtp_buffer_compare_seqnum (priv->last_popped_seqnum, seqnum);
 887
 888     if (gap <= 0) {
 889       /* priv->last_popped_seqnum >= seqnum, this packet is too late or the
 890        * sender might have been restarted with different seqnum. */
 891       if (gap < -100) {
 892         GST_DEBUG_OBJECT (jitterbuffer, "reset: buffer too old %d", gap);
 893         priv->last_popped_seqnum = -1;
 894         priv->next_seqnum = -1;
 895       } else {
 896         goto too_late;
 897       }
 898     } else {
 899       /* priv->last_popped_seqnum < seqnum, this is a new packet */
 900       if (gap > 3000) {
 901         GST_DEBUG_OBJECT (jitterbuffer, "reset: too many dropped packets %d",
 902             gap);
 903         priv->last_popped_seqnum = -1;
 904         priv->next_seqnum = -1;
 905       }
 906     }
 907   }
 908
 909   /* let's drop oldest packet if the queue is already full and drop-on-latency
 910    * is set. We can only do this when there actually is a latency. When no
 911    * latency is set, we just pump it in the queue and let the other end push it
 912    * out as fast as possible. */
 913   if (priv->latency_ms && priv->drop_on_latency) {
 914
 915     latency_ts =
 916         gst_util_uint64_scale_int (priv->latency_ms, priv->clock_rate, 1000);
 917
 918     if (rtp_jitter_buffer_get_ts_diff (priv->jbuf) >= latency_ts) {
 919       GstBuffer *old_buf;
 920
 921       old_buf = rtp_jitter_buffer_pop (priv->jbuf);
 922
 923       GST_DEBUG_OBJECT (jitterbuffer, "Queue full, dropping old packet #%d",
 924           gst_rtp_buffer_get_seq (old_buf));
 925
 926       gst_buffer_unref (old_buf);
 927     }
 928   }
 929
 930   /* we need to make the metadata writable before pushing it in the jitterbuffer
 931    * because the jitterbuffer will update the timestamp */
 932   buffer = gst_buffer_make_metadata_writable (buffer);
 933
 934   /* now insert the packet into the queue in sorted order. This function returns
 935    * FALSE if a packet with the same seqnum was already in the queue, meaning we
 936    * have a duplicate. */
 937   if (!rtp_jitter_buffer_insert (priv->jbuf, buffer, timestamp,
 938           priv->clock_rate, &tail))
 939     goto duplicate;
 940
 941   /* signal addition of new buffer when the _loop is waiting. */
 942   if (priv->waiting)
 943     JBUF_SIGNAL (priv);
 944
 945   /* let's unschedule and unblock any waiting buffers. We only want to do this
 946    * when the tail buffer changed */
 947   if (priv->clock_id && tail) {
 948     GST_DEBUG_OBJECT (jitterbuffer,
 949         "Unscheduling waiting buffer, new tail buffer");
 950     gst_clock_id_unschedule (priv->clock_id);
 951   }
 952
 953   GST_DEBUG_OBJECT (jitterbuffer, "Pushed packet #%d, now %d packets",
 954       seqnum, rtp_jitter_buffer_num_packets (priv->jbuf));
 955
 956 finished:
 957   JBUF_UNLOCK (priv);
 958
 959   gst_object_unref (jitterbuffer);
 960
 961   return ret;
 962
 963   /* ERRORS */
 964 invalid_buffer:
 965   {
 966     /* this is not fatal but should be filtered earlier */
 967     GST_ELEMENT_WARNING (jitterbuffer, STREAM, DECODE, (NULL),
 968         ("Received invalid RTP payload, dropping"));
 969     gst_buffer_unref (buffer);
 970     gst_object_unref (jitterbuffer);
 971     return GST_FLOW_OK;
 972   }
 973 not_negotiated:
 974   {
 975     GST_WARNING_OBJECT (jitterbuffer, "No clock-rate in caps!");
 976     gst_buffer_unref (buffer);
 977     gst_object_unref (jitterbuffer);
 978     return GST_FLOW_OK;
 979   }
 980 out_flushing:
 981   {
 982     ret = priv->srcresult;
 983     GST_DEBUG_OBJECT (jitterbuffer, "flushing %s", gst_flow_get_name (ret));
 984     gst_buffer_unref (buffer);
 985     goto finished;
 986   }
 987 have_eos:
 988   {
 989     ret = GST_FLOW_UNEXPECTED;
 990     GST_WARNING_OBJECT (jitterbuffer, "we are EOS, refusing buffer");
 991     gst_buffer_unref (buffer);
 992     goto finished;
 993   }
 994 too_late:
 995   {
 996     GST_WARNING_OBJECT (jitterbuffer, "Packet #%d too late as #%d was already"
 997         " popped, dropping", seqnum, priv->last_popped_seqnum);
 998     priv->num_late++;
 999     gst_buffer_unref (buffer);
1000     goto finished;
1001   }
1002 duplicate:
1003   {
1004     GST_WARNING_OBJECT (jitterbuffer, "Duplicate packet #%d detected, dropping",
1005         seqnum);
1006     priv->num_duplicates++;
1007     gst_buffer_unref (buffer);
1008     goto finished;
1009   }
1010 }
1011
1012 static GstClockTime
1013 apply_offset (GstRtpJitterBuffer * jitterbuffer, GstClockTime timestamp)
1014 {
1015   GstRtpJitterBufferPrivate *priv;
1016
1017   priv = jitterbuffer->priv;
1018
1019   if (timestamp == -1)
1020     return -1;
1021
1022   /* apply the timestamp offset */
1023   timestamp += priv->ts_offset;
1024
1025   return timestamp;
1026 }
1027
1028 /**
1029  * This funcion will push out buffers on the source pad.
1030  *
1031  * For each pushed buffer, the seqnum is recorded, if the next buffer B has a
1032  * different seqnum (missing packets before B), this function will wait for the
1033  * missing packet to arrive up to the timestamp of buffer B.
1034  */
1035 static void
1036 gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer)
1037 {
1038   GstRtpJitterBufferPrivate *priv;
1039   GstBuffer *outbuf;
1040   GstFlowReturn result;
1041   guint16 seqnum;
1042   guint32 next_seqnum;
1043   GstClockTime timestamp, out_time;
1044   gboolean discont = FALSE;
1045   gint gap;
1046
1047   priv = jitterbuffer->priv;
1048
1049   JBUF_LOCK_CHECK (priv, flushing);
1050 again:
1051   GST_DEBUG_OBJECT (jitterbuffer, "Peeking item");
1052   while (TRUE) {
1053     /* always wait if we are blocked */
1054     if (!priv->blocked) {
1055       /* if we have a packet, we can exit the loop and grab it */
1056       if (rtp_jitter_buffer_num_packets (priv->jbuf) > 0)
1057         break;
1058       /* no packets but we are EOS, do eos logic */
1059       if (priv->eos)
1060         goto do_eos;
1061     }
1062     /* underrun, wait for packets or flushing now */
1063     priv->waiting = TRUE;
1064     JBUF_WAIT_CHECK (priv, flushing);
1065     priv->waiting = FALSE;
1066   }
1067
1068   /* peek a buffer, we're just looking at the timestamp and the sequence number.
1069    * If all is fine, we'll pop and push it. If the sequence number is wrong we
1070    * wait on the timestamp. In the chain function we will unlock the wait when a
1071    * new buffer is available. The peeked buffer is valid for as long as we hold
1072    * the jitterbuffer lock. */
1073   outbuf = rtp_jitter_buffer_peek (priv->jbuf);
1074
1075   /* get the seqnum and the next expected seqnum */
1076   seqnum = gst_rtp_buffer_get_seq (outbuf);
1077   next_seqnum = priv->next_seqnum;
1078
1079   /* get the timestamp, this is already corrected for clock skew by the
1080    * jitterbuffer */
1081   timestamp = GST_BUFFER_TIMESTAMP (outbuf);
1082
1083   GST_DEBUG_OBJECT (jitterbuffer,
1084       "Peeked buffer #%d, expect #%d, timestamp %" GST_TIME_FORMAT
1085       ", now %d left", seqnum, next_seqnum, GST_TIME_ARGS (timestamp),
1086       rtp_jitter_buffer_num_packets (priv->jbuf));
1087
1088   /* apply our timestamp offset to the incomming buffer, this will be our output
1089    * timestamp. */
1090   out_time = apply_offset (jitterbuffer, timestamp);
1091
1092   /* get the gap between this and the previous packet. If we don't know the
1093    * previous packet seqnum assume no gap. */
1094   if (next_seqnum != -1) {
1095     gap = gst_rtp_buffer_compare_seqnum (next_seqnum, seqnum);
1096
1097     /* if we have a packet that we already pushed or considered dropped, pop it
1098      * off and get the next packet */
1099     if (gap < 0) {
1100       GST_DEBUG_OBJECT (jitterbuffer, "Old packet #%d, next #%d dropping",
1101           seqnum, next_seqnum);
1102       outbuf = rtp_jitter_buffer_pop (priv->jbuf);
1103       gst_buffer_unref (outbuf);
1104       goto again;
1105     }
1106   } else {
1107     GST_DEBUG_OBJECT (jitterbuffer, "no next seqnum known, first packet");
1108     gap = -1;
1109   }
1110
1111   /* If we don't know what the next seqnum should be (== -1) we have to wait
1112    * because it might be possible that we are not receiving this buffer in-order,
1113    * a buffer with a lower seqnum could arrive later and we want to push that
1114    * earlier buffer before this buffer then.
1115    * If we know the expected seqnum, we can compare it to the current seqnum to
1116    * determine if we have missing a packet. If we have a missing packet (which
1117    * must be before this packet) we can wait for it until the deadline for this
1118    * packet expires. */
1119   if (gap != 0 && out_time != -1) {
1120     GstClockID id;
1121     GstClockTime sync_time;
1122     GstClockReturn ret;
1123     GstClock *clock;
1124     GstClockTime duration = GST_CLOCK_TIME_NONE;
1125
1126     if (gap > 0) {
1127       /* we have a gap */
1128       GST_WARNING_OBJECT (jitterbuffer,
1129           "Sequence number GAP detected: expected %d instead of %d (%d missing)",
1130           next_seqnum, seqnum, gap);
1131
1132       if (priv->last_out_time != -1) {
1133         GST_DEBUG_OBJECT (jitterbuffer,
1134             "out_time %" GST_TIME_FORMAT ", last %" GST_TIME_FORMAT,
1135             GST_TIME_ARGS (out_time), GST_TIME_ARGS (priv->last_out_time));
1136         /* interpolate between the current time and the last time based on
1137          * number of packets we are missing, this is the estimated duration
1138          * for the missing packet based on equidistant packet spacing. Also make
1139          * sure we never go negative. */
1140         if (out_time > priv->last_out_time)
1141           duration = (out_time - priv->last_out_time) / (gap + 1);
1142         else
1143           goto lost;
1144
1145         GST_DEBUG_OBJECT (jitterbuffer, "duration %" GST_TIME_FORMAT,
1146             GST_TIME_ARGS (duration));
1147         /* add this duration to the timestamp of the last packet we pushed */
1148         out_time = (priv->last_out_time + duration);
1149       }
1150     } else {
1151       /* we don't know what the next_seqnum should be, wait for the last
1152        * possible moment to push this buffer, maybe we get an earlier seqnum
1153        * while we wait */
1154       GST_DEBUG_OBJECT (jitterbuffer, "First buffer %d, do sync", seqnum);
1155     }
1156
1157     GST_OBJECT_LOCK (jitterbuffer);
1158     clock = GST_ELEMENT_CLOCK (jitterbuffer);
1159     if (!clock) {
1160       GST_OBJECT_UNLOCK (jitterbuffer);
1161       /* let's just push if there is no clock */
1162       goto push_buffer;
1163     }
1164
1165     GST_DEBUG_OBJECT (jitterbuffer, "sync to timestamp %" GST_TIME_FORMAT,
1166         GST_TIME_ARGS (out_time));
1167
1168     /* prepare for sync against clock */
1169     sync_time = out_time + GST_ELEMENT_CAST (jitterbuffer)->base_time;
1170     /* add latency, this includes our own latency and the peer latency. */
1171     sync_time += (priv->latency_ms * GST_MSECOND);
1172     sync_time += priv->peer_latency;
1173
1174     /* create an entry for the clock */
1175     id = priv->clock_id = gst_clock_new_single_shot_id (clock, sync_time);
1176     GST_OBJECT_UNLOCK (jitterbuffer);
1177
1178     /* release the lock so that the other end can push stuff or unlock */
1179     JBUF_UNLOCK (priv);
1180
1181     ret = gst_clock_id_wait (id, NULL);
1182
1183     JBUF_LOCK (priv);
1184     /* and free the entry */
1185     gst_clock_id_unref (id);
1186     priv->clock_id = NULL;
1187
1188     /* at this point, the clock could have been unlocked by a timeout, a new
1189      * tail element was added to the queue or because we are shutting down. Check
1190      * for shutdown first. */
1191     if (priv->srcresult != GST_FLOW_OK)
1192       goto flushing;
1193
1194     /* if we got unscheduled and we are not flushing, it's because a new tail
1195      * element became available in the queue. Grab it and try to push or sync. */
1196     if (ret == GST_CLOCK_UNSCHEDULED) {
1197       GST_DEBUG_OBJECT (jitterbuffer,
1198           "Wait got unscheduled, will retry to push with new buffer");
1199       goto again;
1200     }
1201
1202   lost:
1203     /* we now timed out, this means we lost a packet or finished synchronizing
1204      * on the first buffer. */
1205     if (gap > 0) {
1206       GstEvent *event;
1207
1208       /* we had a gap and thus we lost a packet. Create an event for this.  */
1209       GST_DEBUG_OBJECT (jitterbuffer, "Packet #%d lost", next_seqnum);
1210       priv->num_late++;
1211       discont = TRUE;
1212
1213       if (priv->do_lost) {
1214         /* create paket lost event */
1215         event = gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM,
1216             gst_structure_new ("GstRTPPacketLost",
1217                 "seqnum", G_TYPE_UINT, (guint) next_seqnum,
1218                 "timestamp", G_TYPE_UINT64, out_time,
1219                 "duration", G_TYPE_UINT64, duration, NULL));
1220         gst_pad_push_event (priv->srcpad, event);
1221       }
1222
1223       /* update our expected next packet */
1224       priv->last_popped_seqnum = next_seqnum;
1225       priv->last_out_time = out_time;
1226       priv->next_seqnum = (next_seqnum + 1) & 0xffff;
1227       /* look for next packet */
1228       goto again;
1229     }
1230
1231     /* there was no known gap,just the first packet, exit the loop and push */
1232     GST_DEBUG_OBJECT (jitterbuffer, "First packet #%d synced", seqnum);
1233
1234     /* get new timestamp, latency might have changed */
1235     out_time = apply_offset (jitterbuffer, timestamp);
1236   }
1237 push_buffer:
1238
1239   /* when we get here we are ready to pop and push the buffer */
1240   outbuf = rtp_jitter_buffer_pop (priv->jbuf);
1241
1242   if (discont || priv->discont) {
1243     /* set DISCONT flag when we missed a packet. We pushed the buffer writable
1244      * into the jitterbuffer so we can modify now. */
1245     GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_DISCONT);
1246     priv->discont = FALSE;
1247   }
1248
1249   /* apply timestamp with offset to buffer now */
1250   GST_BUFFER_TIMESTAMP (outbuf) = out_time;
1251
1252   /* now we are ready to push the buffer. Save the seqnum and release the lock
1253    * so the other end can push stuff in the queue again. */
1254   priv->last_popped_seqnum = seqnum;
1255   priv->last_out_time = out_time;
1256   priv->next_seqnum = (seqnum + 1) & 0xffff;
1257   JBUF_UNLOCK (priv);
1258
1259   /* push buffer */
1260   GST_DEBUG_OBJECT (jitterbuffer,
1261       "Pushing buffer %d, timestamp %" GST_TIME_FORMAT, seqnum,
1262       GST_TIME_ARGS (out_time));
1263   result = gst_pad_push (priv->srcpad, outbuf);
1264   if (result != GST_FLOW_OK)
1265     goto pause;
1266
1267   return;
1268
1269   /* ERRORS */
1270 do_eos:
1271   {
1272     /* store result, we are flushing now */
1273     GST_DEBUG_OBJECT (jitterbuffer, "We are EOS, pushing EOS downstream");
1274     priv->srcresult = GST_FLOW_UNEXPECTED;
1275     gst_pad_pause_task (priv->srcpad);
1276     gst_pad_push_event (priv->srcpad, gst_event_new_eos ());
1277     JBUF_UNLOCK (priv);
1278     return;
1279   }
1280 flushing:
1281   {
1282     GST_DEBUG_OBJECT (jitterbuffer, "we are flushing");
1283     gst_pad_pause_task (priv->srcpad);
1284     JBUF_UNLOCK (priv);
1285     return;
1286   }
1287 pause:
1288   {
1289     const gchar *reason = gst_flow_get_name (result);
1290
1291     GST_DEBUG_OBJECT (jitterbuffer, "pausing task, reason %s", reason);
1292
1293     JBUF_LOCK (priv);
1294     /* store result */
1295     priv->srcresult = result;
1296     /* we don't post errors or anything because upstream will do that for us
1297      * when we pass the return value upstream. */
1298     gst_pad_pause_task (priv->srcpad);
1299     JBUF_UNLOCK (priv);
1300     return;
1301   }
1302 }
1303
1304 static gboolean
1305 gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query)
1306 {
1307   GstRtpJitterBuffer *jitterbuffer;
1308   GstRtpJitterBufferPrivate *priv;
1309   gboolean res = FALSE;
1310
1311   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
1312   priv = jitterbuffer->priv;
1313
1314   switch (GST_QUERY_TYPE (query)) {
1315     case GST_QUERY_LATENCY:
1316     {
1317       /* We need to send the query upstream and add the returned latency to our
1318        * own */
1319       GstClockTime min_latency, max_latency;
1320       gboolean us_live;
1321       GstClockTime our_latency;
1322
1323       if ((res = gst_pad_peer_query (priv->sinkpad, query))) {
1324         gst_query_parse_latency (query, &us_live, &min_latency, &max_latency);
1325
1326         GST_DEBUG_OBJECT (jitterbuffer, "Peer latency: min %"
1327             GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
1328             GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
1329
1330         /* store this so that we can safely sync on the peer buffers. */
1331         JBUF_LOCK (priv);
1332         priv->peer_latency = min_latency;
1333         our_latency = ((guint64) priv->latency_ms) * GST_MSECOND;
1334         JBUF_UNLOCK (priv);
1335
1336         GST_DEBUG_OBJECT (jitterbuffer, "Our latency: %" GST_TIME_FORMAT,
1337             GST_TIME_ARGS (our_latency));
1338
1339         /* we add some latency but can buffer an infinite amount of time */
1340         min_latency += our_latency;
1341         max_latency = -1;
1342
1343         GST_DEBUG_OBJECT (jitterbuffer, "Calculated total latency : min %"
1344             GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
1345             GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
1346
1347         gst_query_set_latency (query, TRUE, min_latency, max_latency);
1348       }
1349       break;
1350     }
1351     default:
1352       res = gst_pad_query_default (pad, query);
1353       break;
1354   }
1355
1356   gst_object_unref (jitterbuffer);
1357
1358   return res;
1359 }
1360
1361 static void
1362 gst_rtp_jitter_buffer_set_property (GObject * object,
1363     guint prop_id, const GValue * value, GParamSpec * pspec)
1364 {
1365   GstRtpJitterBuffer *jitterbuffer;
1366   GstRtpJitterBufferPrivate *priv;
1367
1368   jitterbuffer = GST_RTP_JITTER_BUFFER (object);
1369   priv = jitterbuffer->priv;
1370
1371   switch (prop_id) {
1372     case PROP_LATENCY:
1373     {
1374       guint new_latency, old_latency;
1375
1376       new_latency = g_value_get_uint (value);
1377
1378       JBUF_LOCK (priv);
1379       old_latency = priv->latency_ms;
1380       priv->latency_ms = new_latency;
1381       JBUF_UNLOCK (priv);
1382
1383       /* post message if latency changed, this will inform the parent pipeline
1384        * that a latency reconfiguration is possible/needed. */
1385       if (new_latency != old_latency) {
1386         GST_DEBUG_OBJECT (jitterbuffer, "latency changed to: %" GST_TIME_FORMAT,
1387             GST_TIME_ARGS (new_latency * GST_MSECOND));
1388
1389         gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer),
1390             gst_message_new_latency (GST_OBJECT_CAST (jitterbuffer)));
1391       }
1392       break;
1393     }
1394     case PROP_DROP_ON_LATENCY:
1395       JBUF_LOCK (priv);
1396       priv->drop_on_latency = g_value_get_boolean (value);
1397       JBUF_UNLOCK (priv);
1398       break;
1399     case PROP_TS_OFFSET:
1400       JBUF_LOCK (priv);
1401       priv->ts_offset = g_value_get_int64 (value);
1402       /* FIXME, we don't really have a method for signaling a timestamp
1403        * DISCONT without also making this a data discont. */
1404       /* priv->discont = TRUE; */
1405       JBUF_UNLOCK (priv);
1406       break;
1407     case PROP_DO_LOST:
1408       JBUF_LOCK (priv);
1409       priv->do_lost = g_value_get_boolean (value);
1410       JBUF_UNLOCK (priv);
1411       break;
1412     default:
1413       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1414       break;
1415   }
1416 }
1417
1418 static void
1419 gst_rtp_jitter_buffer_get_property (GObject * object,
1420     guint prop_id, GValue * value, GParamSpec * pspec)
1421 {
1422   GstRtpJitterBuffer *jitterbuffer;
1423   GstRtpJitterBufferPrivate *priv;
1424
1425   jitterbuffer = GST_RTP_JITTER_BUFFER (object);
1426   priv = jitterbuffer->priv;
1427
1428   switch (prop_id) {
1429     case PROP_LATENCY:
1430       JBUF_LOCK (priv);
1431       g_value_set_uint (value, priv->latency_ms);
1432       JBUF_UNLOCK (priv);
1433       break;
1434     case PROP_DROP_ON_LATENCY:
1435       JBUF_LOCK (priv);
1436       g_value_set_boolean (value, priv->drop_on_latency);
1437       JBUF_UNLOCK (priv);
1438       break;
1439     case PROP_TS_OFFSET:
1440       JBUF_LOCK (priv);
1441       g_value_set_int64 (value, priv->ts_offset);
1442       JBUF_UNLOCK (priv);
1443       break;
1444     case PROP_DO_LOST:
1445       JBUF_LOCK (priv);
1446       g_value_set_boolean (value, priv->do_lost);
1447       JBUF_UNLOCK (priv);
1448       break;
1449     default:
1450       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1451       break;
1452   }
1453 }