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  *  Copyright 2015 Kurento (http://kurento.org/)
   9  *   @author: Miguel París <mparisdiaz@gmail.com>
  10  *
  11  * This library is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU Library General Public
  13  * License as published by the Free Software Foundation; either
  14  * version 2 of the License, or (at your option) any later version.
  15  *
  16  * This library is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  19  * Library General Public License for more details.
  20  *
  21  * You should have received a copy of the GNU Library General Public
  22  * License along with this library; if not, write to the
  23  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  24  * Boston, MA 02110-1301, USA.
  25  *
  26  */
  27
  28 /**
  29  * SECTION:element-rtpjitterbuffer
  30  *
  31  * This element reorders and removes duplicate RTP packets as they are received
  32  * from a network source.
  33  *
  34  * The element needs the clock-rate of the RTP payload in order to estimate the
  35  * delay. This information is obtained either from the caps on the sink pad or,
  36  * when no caps are present, from the #GstRtpJitterBuffer::request-pt-map signal.
  37  * To clear the previous pt-map use the #GstRtpJitterBuffer::clear-pt-map signal.
  38  *
  39  * The rtpjitterbuffer will wait for missing packets up to a configurable time
  40  * limit using the #GstRtpJitterBuffer:latency property. Packets arriving too
  41  * late are considered to be lost packets. If the #GstRtpJitterBuffer:do-lost
  42  * property is set, lost packets will result in a custom serialized downstream
  43  * event of name GstRTPPacketLost. The lost packet events are usually used by a
  44  * depayloader or other element to create concealment data or some other logic
  45  * to gracefully handle the missing packets.
  46  *
  47  * The jitterbuffer will use the DTS (or PTS if no DTS is set) of the incomming
  48  * buffer and the rtptime inside the RTP packet to create a PTS on the outgoing
  49  * buffer.
  50  *
  51  * The jitterbuffer can also be configured to send early retransmission events
  52  * upstream by setting the #GstRtpJitterBuffer:do-retransmission property. In
  53  * this mode, the jitterbuffer tries to estimate when a packet should arrive and
  54  * sends a custom upstream event named GstRTPRetransmissionRequest when the
  55  * packet is considered late. The initial expected packet arrival time is
  56  * calculated as follows:
  57  *
  58  * - If seqnum N arrived at time T, seqnum N+1 is expected to arrive at
  59  *     T + packet-spacing + #GstRtpJitterBuffer:rtx-delay. The packet spacing is
  60  *     calculated from the DTS (or PTS is no DTS) of two consecutive RTP
  61  *     packets with different rtptime.
  62  *
  63  * - If seqnum N0 arrived at time T0 and seqnum Nm arrived at time Tm,
  64  *     seqnum Ni is expected at time Ti = T0 + i*(Tm - T0)/(Nm - N0). Any
  65  *     previously scheduled timeout is overwritten.
  66  *
  67  * - If seqnum N arrived, all seqnum older than
  68  *     N - #GstRtpJitterBuffer:rtx-delay-reorder are considered late
  69  *     immediately. This is to request fast feedback for abonormally reorder
  70  *     packets before any of the previous timeouts is triggered.
  71  *
  72  * A late packet triggers the GstRTPRetransmissionRequest custom upstream
  73  * event. After the initial timeout expires and the retransmission event is
  74  * sent, the timeout is scheduled for
  75  * T + #GstRtpJitterBuffer:rtx-retry-timeout. If the missing packet did not
  76  * arrive after #GstRtpJitterBuffer:rtx-retry-timeout, a new
  77  * GstRTPRetransmissionRequest is sent upstream and the timeout is rescheduled
  78  * again for T + #GstRtpJitterBuffer:rtx-retry-timeout. This repeats until
  79  * #GstRtpJitterBuffer:rtx-retry-period elapsed, at which point no further
  80  * retransmission requests are sent and the regular logic is performed to
  81  * schedule a lost packet as discussed above.
  82  *
  83  * This element acts as a live element and so adds #GstRtpJitterBuffer:latency
  84  * to the pipeline.
  85  *
  86  * This element will automatically be used inside rtpbin.
  87  *
  88  * <refsect2>
  89  * <title>Example pipelines</title>
  90  * |[
  91  * gst-launch-1.0 rtspsrc location=rtsp://192.168.1.133:8554/mpeg1or2AudioVideoTest ! rtpjitterbuffer ! rtpmpvdepay ! mpeg2dec ! xvimagesink
  92  * ]| Connect to a streaming server and decode the MPEG video. The jitterbuffer is
  93  * inserted into the pipeline to smooth out network jitter and to reorder the
  94  * out-of-order RTP packets.
  95  * </refsect2>
  96  */
  97
  98 #ifdef HAVE_CONFIG_H
  99 #include "config.h"
 100 #endif
 101
 102 #include <stdlib.h>
 103 #include <stdio.h>
 104 #include <string.h>
 105 #include <gst/rtp/gstrtpbuffer.h>
 106 #include <gst/net/net.h>
 107
 108 #include "gstrtpjitterbuffer.h"
 109 #include "rtpjitterbuffer.h"
 110 #include "rtpstats.h"
 111
 112 #include <gst/glib-compat-private.h>
 113
 114 GST_DEBUG_CATEGORY (rtpjitterbuffer_debug);
 115 #define GST_CAT_DEFAULT (rtpjitterbuffer_debug)
 116
 117 /* RTPJitterBuffer signals and args */
 118 enum
 119 {
 120   SIGNAL_REQUEST_PT_MAP,
 121   SIGNAL_CLEAR_PT_MAP,
 122   SIGNAL_HANDLE_SYNC,
 123   SIGNAL_ON_NPT_STOP,
 124   SIGNAL_SET_ACTIVE,
 125   LAST_SIGNAL
 126 };
 127
 128 #define DEFAULT_LATENCY_MS          200
 129 #define DEFAULT_DROP_ON_LATENCY     FALSE
 130 #define DEFAULT_TS_OFFSET           0
 131 #define DEFAULT_DO_LOST             FALSE
 132 #define DEFAULT_MODE                RTP_JITTER_BUFFER_MODE_SLAVE
 133 #define DEFAULT_PERCENT             0
 134 #define DEFAULT_DO_RETRANSMISSION   FALSE
 135 #define DEFAULT_RTX_NEXT_SEQNUM     TRUE
 136 #define DEFAULT_RTX_DELAY           -1
 137 #define DEFAULT_RTX_MIN_DELAY       0
 138 #define DEFAULT_RTX_DELAY_REORDER   3
 139 #define DEFAULT_RTX_RETRY_TIMEOUT   -1
 140 #define DEFAULT_RTX_MIN_RETRY_TIMEOUT   -1
 141 #define DEFAULT_RTX_RETRY_PERIOD    -1
 142 #define DEFAULT_RTX_MAX_RETRIES    -1
 143 #define DEFAULT_MAX_RTCP_RTP_TIME_DIFF 1000
 144 #define DEFAULT_MAX_DROPOUT_TIME    60000
 145 #define DEFAULT_MAX_MISORDER_TIME   2000
 146 #define DEFAULT_RFC7273_SYNC        FALSE
 147
 148 #define DEFAULT_AUTO_RTX_DELAY (20 * GST_MSECOND)
 149 #define DEFAULT_AUTO_RTX_TIMEOUT (40 * GST_MSECOND)
 150
 151 enum
 152 {
 153   PROP_0,
 154   PROP_LATENCY,
 155   PROP_DROP_ON_LATENCY,
 156   PROP_TS_OFFSET,
 157   PROP_DO_LOST,
 158   PROP_MODE,
 159   PROP_PERCENT,
 160   PROP_DO_RETRANSMISSION,
 161   PROP_RTX_NEXT_SEQNUM,
 162   PROP_RTX_DELAY,
 163   PROP_RTX_MIN_DELAY,
 164   PROP_RTX_DELAY_REORDER,
 165   PROP_RTX_RETRY_TIMEOUT,
 166   PROP_RTX_MIN_RETRY_TIMEOUT,
 167   PROP_RTX_RETRY_PERIOD,
 168   PROP_RTX_MAX_RETRIES,
 169   PROP_STATS,
 170   PROP_MAX_RTCP_RTP_TIME_DIFF,
 171   PROP_MAX_DROPOUT_TIME,
 172   PROP_MAX_MISORDER_TIME,
 173   PROP_RFC7273_SYNC
 174 };
 175
 176 #define JBUF_LOCK(priv)   G_STMT_START {                        \
 177     GST_TRACE("Locking from thread %p", g_thread_self());       \
 178     (g_mutex_lock (&(priv)->jbuf_lock));                        \
 179     GST_TRACE("Locked from thread %p", g_thread_self());        \
 180   } G_STMT_END
 181
 182 #define JBUF_LOCK_CHECK(priv,label) G_STMT_START {    \
 183   JBUF_LOCK (priv);                                   \
 184   if (G_UNLIKELY (priv->srcresult != GST_FLOW_OK))    \
 185     goto label;                                       \
 186 } G_STMT_END
 187 #define JBUF_UNLOCK(priv) G_STMT_START {                        \
 188     GST_TRACE ("Unlocking from thread %p", g_thread_self ());   \
 189     (g_mutex_unlock (&(priv)->jbuf_lock));                      \
 190 } G_STMT_END
 191
 192 #define JBUF_WAIT_TIMER(priv)   G_STMT_START {            \
 193   GST_DEBUG ("waiting timer");                            \
 194   (priv)->waiting_timer = TRUE;                           \
 195   g_cond_wait (&(priv)->jbuf_timer, &(priv)->jbuf_lock);  \
 196   (priv)->waiting_timer = FALSE;                          \
 197   GST_DEBUG ("waiting timer done");                       \
 198 } G_STMT_END
 199 #define JBUF_SIGNAL_TIMER(priv) G_STMT_START {            \
 200   if (G_UNLIKELY ((priv)->waiting_timer)) {               \
 201     GST_DEBUG ("signal timer");                           \
 202     g_cond_signal (&(priv)->jbuf_timer);                  \
 203   }                                                       \
 204 } G_STMT_END
 205
 206 #define JBUF_WAIT_EVENT(priv,label) G_STMT_START {       \
 207   GST_DEBUG ("waiting event");                           \
 208   (priv)->waiting_event = TRUE;                          \
 209   g_cond_wait (&(priv)->jbuf_event, &(priv)->jbuf_lock); \
 210   (priv)->waiting_event = FALSE;                         \
 211   GST_DEBUG ("waiting event done");                      \
 212   if (G_UNLIKELY (priv->srcresult != GST_FLOW_OK))       \
 213     goto label;                                          \
 214 } G_STMT_END
 215 #define JBUF_SIGNAL_EVENT(priv) G_STMT_START {           \
 216   if (G_UNLIKELY ((priv)->waiting_event)) {              \
 217     GST_DEBUG ("signal event");                          \
 218     g_cond_signal (&(priv)->jbuf_event);                 \
 219   }                                                      \
 220 } G_STMT_END
 221
 222 #define JBUF_WAIT_QUERY(priv,label) G_STMT_START {       \
 223   GST_DEBUG ("waiting query");                           \
 224   (priv)->waiting_query = TRUE;                          \
 225   g_cond_wait (&(priv)->jbuf_query, &(priv)->jbuf_lock); \
 226   (priv)->waiting_query = FALSE;                         \
 227   GST_DEBUG ("waiting query done");                      \
 228   if (G_UNLIKELY (priv->srcresult != GST_FLOW_OK))       \
 229     goto label;                                          \
 230 } G_STMT_END
 231 #define JBUF_SIGNAL_QUERY(priv,res) G_STMT_START {       \
 232   (priv)->last_query = res;                              \
 233   if (G_UNLIKELY ((priv)->waiting_query)) {              \
 234     GST_DEBUG ("signal query");                          \
 235     g_cond_signal (&(priv)->jbuf_query);                 \
 236   }                                                      \
 237 } G_STMT_END
 238
 239
 240 struct _GstRtpJitterBufferPrivate
 241 {
 242   GstPad *sinkpad, *srcpad;
 243   GstPad *rtcpsinkpad;
 244
 245   RTPJitterBuffer *jbuf;
 246   GMutex jbuf_lock;
 247   gboolean waiting_timer;
 248   GCond jbuf_timer;
 249   gboolean waiting_event;
 250   GCond jbuf_event;
 251   gboolean waiting_query;
 252   GCond jbuf_query;
 253   gboolean last_query;
 254   gboolean discont;
 255   gboolean ts_discont;
 256   gboolean active;
 257   guint64 out_offset;
 258
 259   gboolean timer_running;
 260   GThread *timer_thread;
 261
 262   /* properties */
 263   guint latency_ms;
 264   guint64 latency_ns;
 265   gboolean drop_on_latency;
 266   gint64 ts_offset;
 267   gboolean do_lost;
 268   gboolean do_retransmission;
 269   gboolean rtx_next_seqnum;
 270   gint rtx_delay;
 271   guint rtx_min_delay;
 272   gint rtx_delay_reorder;
 273   gint rtx_retry_timeout;
 274   gint rtx_min_retry_timeout;
 275   gint rtx_retry_period;
 276   gint rtx_max_retries;
 277   gint max_rtcp_rtp_time_diff;
 278   guint32 max_dropout_time;
 279   guint32 max_misorder_time;
 280
 281   /* the last seqnum we pushed out */
 282   guint32 last_popped_seqnum;
 283   /* the next expected seqnum we push */
 284   guint32 next_seqnum;
 285   /* seqnum-base, if known */
 286   guint32 seqnum_base;
 287   /* last output time */
 288   GstClockTime last_out_time;
 289   /* last valid input timestamp and rtptime pair */
 290   GstClockTime ips_dts;
 291   guint64 ips_rtptime;
 292   GstClockTime packet_spacing;
 293
 294   GQueue gap_packets;
 295
 296   /* the next expected seqnum we receive */
 297   GstClockTime last_in_dts;
 298   guint32 next_in_seqnum;
 299
 300   GArray *timers;
 301
 302   /* start and stop ranges */
 303   GstClockTime npt_start;
 304   GstClockTime npt_stop;
 305   guint64 ext_timestamp;
 306   guint64 last_elapsed;
 307   guint64 estimated_eos;
 308   GstClockID eos_id;
 309
 310   /* state */
 311   gboolean eos;
 312   guint last_percent;
 313
 314   /* clock rate and rtp timestamp offset */
 315   gint last_pt;
 316   gint32 clock_rate;
 317   gint64 clock_base;
 318   gint64 prev_ts_offset;
 319
 320   /* when we are shutting down */
 321   GstFlowReturn srcresult;
 322   gboolean blocked;
 323
 324   /* for sync */
 325   GstSegment segment;
 326   GstClockID clock_id;
 327   GstClockTime timer_timeout;
 328   guint16 timer_seqnum;
 329   /* the latency of the upstream peer, we have to take this into account when
 330    * synchronizing the buffers. */
 331   GstClockTime peer_latency;
 332   guint64 ext_rtptime;
 333   GstBuffer *last_sr;
 334
 335   /* some accounting */
 336   guint64 num_late;
 337   guint64 num_duplicates;
 338   guint64 num_rtx_requests;
 339   guint64 num_rtx_success;
 340   guint64 num_rtx_failed;
 341   gdouble avg_rtx_num;
 342   guint64 avg_rtx_rtt;
 343   RTPPacketRateCtx packet_rate_ctx;
 344
 345   /* for the jitter */
 346   GstClockTime last_dts;
 347   guint64 last_rtptime;
 348   GstClockTime avg_jitter;
 349 };
 350
 351 typedef enum
 352 {
 353   TIMER_TYPE_EXPECTED,
 354   TIMER_TYPE_LOST,
 355   TIMER_TYPE_DEADLINE,
 356   TIMER_TYPE_EOS
 357 } TimerType;
 358
 359 typedef struct
 360 {
 361   guint idx;
 362   guint16 seqnum;
 363   guint num;
 364   TimerType type;
 365   GstClockTime timeout;
 366   GstClockTime duration;
 367   GstClockTime rtx_base;
 368   GstClockTime rtx_delay;
 369   GstClockTime rtx_retry;
 370   GstClockTime rtx_last;
 371   guint num_rtx_retry;
 372 } TimerData;
 373
 374 #define GST_RTP_JITTER_BUFFER_GET_PRIVATE(o) \
 375   (G_TYPE_INSTANCE_GET_PRIVATE ((o), GST_TYPE_RTP_JITTER_BUFFER, \
 376                                 GstRtpJitterBufferPrivate))
 377
 378 static GstStaticPadTemplate gst_rtp_jitter_buffer_sink_template =
 379 GST_STATIC_PAD_TEMPLATE ("sink",
 380     GST_PAD_SINK,
 381     GST_PAD_ALWAYS,
 382     GST_STATIC_CAPS ("application/x-rtp"
 383         /* "clock-rate = (int) [ 1, 2147483647 ], "
 384          * "payload = (int) , "
 385          * "encoding-name = (string) "
 386          */ )
 387     );
 388
 389 static GstStaticPadTemplate gst_rtp_jitter_buffer_sink_rtcp_template =
 390 GST_STATIC_PAD_TEMPLATE ("sink_rtcp",
 391     GST_PAD_SINK,
 392     GST_PAD_REQUEST,
 393     GST_STATIC_CAPS ("application/x-rtcp")
 394     );
 395
 396 static GstStaticPadTemplate gst_rtp_jitter_buffer_src_template =
 397 GST_STATIC_PAD_TEMPLATE ("src",
 398     GST_PAD_SRC,
 399     GST_PAD_ALWAYS,
 400     GST_STATIC_CAPS ("application/x-rtp"
 401         /* "payload = (int) , "
 402          * "clock-rate = (int) , "
 403          * "encoding-name = (string) "
 404          */ )
 405     );
 406
 407 static guint gst_rtp_jitter_buffer_signals[LAST_SIGNAL] = { 0 };
 408
 409 #define gst_rtp_jitter_buffer_parent_class parent_class
 410 G_DEFINE_TYPE (GstRtpJitterBuffer, gst_rtp_jitter_buffer, GST_TYPE_ELEMENT);
 411
 412 /* object overrides */
 413 static void gst_rtp_jitter_buffer_set_property (GObject * object,
 414     guint prop_id, const GValue * value, GParamSpec * pspec);
 415 static void gst_rtp_jitter_buffer_get_property (GObject * object,
 416     guint prop_id, GValue * value, GParamSpec * pspec);
 417 static void gst_rtp_jitter_buffer_finalize (GObject * object);
 418
 419 /* element overrides */
 420 static GstStateChangeReturn gst_rtp_jitter_buffer_change_state (GstElement
 421     * element, GstStateChange transition);
 422 static GstPad *gst_rtp_jitter_buffer_request_new_pad (GstElement * element,
 423     GstPadTemplate * templ, const gchar * name, const GstCaps * filter);
 424 static void gst_rtp_jitter_buffer_release_pad (GstElement * element,
 425     GstPad * pad);
 426 static GstClock *gst_rtp_jitter_buffer_provide_clock (GstElement * element);
 427 static gboolean gst_rtp_jitter_buffer_set_clock (GstElement * element,
 428     GstClock * clock);
 429
 430 /* pad overrides */
 431 static GstCaps *gst_rtp_jitter_buffer_getcaps (GstPad * pad, GstCaps * filter);
 432 static GstIterator *gst_rtp_jitter_buffer_iterate_internal_links (GstPad * pad,
 433     GstObject * parent);
 434
 435 /* sinkpad overrides */
 436 static gboolean gst_rtp_jitter_buffer_sink_event (GstPad * pad,
 437     GstObject * parent, GstEvent * event);
 438 static GstFlowReturn gst_rtp_jitter_buffer_chain (GstPad * pad,
 439     GstObject * parent, GstBuffer * buffer);
 440
 441 static gboolean gst_rtp_jitter_buffer_sink_rtcp_event (GstPad * pad,
 442     GstObject * parent, GstEvent * event);
 443 static GstFlowReturn gst_rtp_jitter_buffer_chain_rtcp (GstPad * pad,
 444     GstObject * parent, GstBuffer * buffer);
 445
 446 static gboolean gst_rtp_jitter_buffer_sink_query (GstPad * pad,
 447     GstObject * parent, GstQuery * query);
 448
 449 /* srcpad overrides */
 450 static gboolean gst_rtp_jitter_buffer_src_event (GstPad * pad,
 451     GstObject * parent, GstEvent * event);
 452 static gboolean gst_rtp_jitter_buffer_src_activate_mode (GstPad * pad,
 453     GstObject * parent, GstPadMode mode, gboolean active);
 454 static void gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer);
 455 static gboolean gst_rtp_jitter_buffer_src_query (GstPad * pad,
 456     GstObject * parent, GstQuery * query);
 457
 458 static void
 459 gst_rtp_jitter_buffer_clear_pt_map (GstRtpJitterBuffer * jitterbuffer);
 460 static GstClockTime
 461 gst_rtp_jitter_buffer_set_active (GstRtpJitterBuffer * jitterbuffer,
 462     gboolean active, guint64 base_time);
 463 static void do_handle_sync (GstRtpJitterBuffer * jitterbuffer);
 464
 465 static void unschedule_current_timer (GstRtpJitterBuffer * jitterbuffer);
 466 static void remove_all_timers (GstRtpJitterBuffer * jitterbuffer);
 467
 468 static void wait_next_timeout (GstRtpJitterBuffer * jitterbuffer);
 469
 470 static GstStructure *gst_rtp_jitter_buffer_create_stats (GstRtpJitterBuffer *
 471     jitterbuffer);
 472
 473 static void
 474 gst_rtp_jitter_buffer_class_init (GstRtpJitterBufferClass * klass)
 475 {
 476   GObjectClass *gobject_class;
 477   GstElementClass *gstelement_class;
 478
 479   gobject_class = (GObjectClass *) klass;
 480   gstelement_class = (GstElementClass *) klass;
 481
 482   g_type_class_add_private (klass, sizeof (GstRtpJitterBufferPrivate));
 483
 484   gobject_class->finalize = gst_rtp_jitter_buffer_finalize;
 485
 486   gobject_class->set_property = gst_rtp_jitter_buffer_set_property;
 487   gobject_class->get_property = gst_rtp_jitter_buffer_get_property;
 488
 489   /**
 490    * GstRtpJitterBuffer:latency:
 491    *
 492    * The maximum latency of the jitterbuffer. Packets will be kept in the buffer
 493    * for at most this time.
 494    */
 495   g_object_class_install_property (gobject_class, PROP_LATENCY,
 496       g_param_spec_uint ("latency", "Buffer latency in ms",
 497           "Amount of ms to buffer", 0, G_MAXUINT, DEFAULT_LATENCY_MS,
 498           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 499   /**
 500    * GstRtpJitterBuffer:drop-on-latency:
 501    *
 502    * Drop oldest buffers when the queue is completely filled.
 503    */
 504   g_object_class_install_property (gobject_class, PROP_DROP_ON_LATENCY,
 505       g_param_spec_boolean ("drop-on-latency",
 506           "Drop buffers when maximum latency is reached",
 507           "Tells the jitterbuffer to never exceed the given latency in size",
 508           DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 509   /**
 510    * GstRtpJitterBuffer:ts-offset:
 511    *
 512    * Adjust GStreamer output buffer timestamps in the jitterbuffer with offset.
 513    * This is mainly used to ensure interstream synchronisation.
 514    */
 515   g_object_class_install_property (gobject_class, PROP_TS_OFFSET,
 516       g_param_spec_int64 ("ts-offset", "Timestamp Offset",
 517           "Adjust buffer timestamps with offset in nanoseconds", G_MININT64,
 518           G_MAXINT64, DEFAULT_TS_OFFSET,
 519           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 520
 521   /**
 522    * GstRtpJitterBuffer:do-lost:
 523    *
 524    * Send out a GstRTPPacketLost event downstream when a packet is considered
 525    * lost.
 526    */
 527   g_object_class_install_property (gobject_class, PROP_DO_LOST,
 528       g_param_spec_boolean ("do-lost", "Do Lost",
 529           "Send an event downstream when a packet is lost", DEFAULT_DO_LOST,
 530           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 531
 532   /**
 533    * GstRtpJitterBuffer:mode:
 534    *
 535    * Control the buffering and timestamping mode used by the jitterbuffer.
 536    */
 537   g_object_class_install_property (gobject_class, PROP_MODE,
 538       g_param_spec_enum ("mode", "Mode",
 539           "Control the buffering algorithm in use", RTP_TYPE_JITTER_BUFFER_MODE,
 540           DEFAULT_MODE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 541   /**
 542    * GstRtpJitterBuffer:percent:
 543    *
 544    * The percent of the jitterbuffer that is filled.
 545    */
 546   g_object_class_install_property (gobject_class, PROP_PERCENT,
 547       g_param_spec_int ("percent", "percent",
 548           "The buffer filled percent", 0, 100,
 549           0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 550   /**
 551    * GstRtpJitterBuffer:do-retransmission:
 552    *
 553    * Send out a GstRTPRetransmission event upstream when a packet is considered
 554    * late and should be retransmitted.
 555    *
 556    * Since: 1.2
 557    */
 558   g_object_class_install_property (gobject_class, PROP_DO_RETRANSMISSION,
 559       g_param_spec_boolean ("do-retransmission", "Do Retransmission",
 560           "Send retransmission events upstream when a packet is late",
 561           DEFAULT_DO_RETRANSMISSION,
 562           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 563
 564   /**
 565    * GstRtpJitterBuffer:rtx-next-seqnum
 566    *
 567    * Estimate when the next packet should arrive and schedule a retransmission
 568    * request for it.
 569    * This is, when packet N arrives, a GstRTPRetransmission event is schedule
 570    * for packet N+1. So it will be requested if it does not arrive at the expected time.
 571    * The expected time is calculated using the dts of N and the packet spacing.
 572    *
 573    * Since: 1.6
 574    */
 575   g_object_class_install_property (gobject_class, PROP_RTX_NEXT_SEQNUM,
 576       g_param_spec_boolean ("rtx-next-seqnum", "RTX next seqnum",
 577           "Estimate when the next packet should arrive and schedule a "
 578           "retransmission request for it.",
 579           DEFAULT_RTX_NEXT_SEQNUM, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 580
 581   /**
 582    * GstRtpJitterBuffer:rtx-delay:
 583    *
 584    * When a packet did not arrive at the expected time, wait this extra amount
 585    * of time before sending a retransmission event.
 586    *
 587    * When -1 is used, the max jitter will be used as extra delay.
 588    *
 589    * Since: 1.2
 590    */
 591   g_object_class_install_property (gobject_class, PROP_RTX_DELAY,
 592       g_param_spec_int ("rtx-delay", "RTX Delay",
 593           "Extra time in ms to wait before sending retransmission "
 594           "event (-1 automatic)", -1, G_MAXINT, DEFAULT_RTX_DELAY,
 595           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 596
 597   /**
 598    * GstRtpJitterBuffer:rtx-min-delay:
 599    *
 600    * When a packet did not arrive at the expected time, wait at least this extra amount
 601    * of time before sending a retransmission event.
 602    *
 603    * Since: 1.6
 604    */
 605   g_object_class_install_property (gobject_class, PROP_RTX_MIN_DELAY,
 606       g_param_spec_uint ("rtx-min-delay", "Minimum RTX Delay",
 607           "Minimum time in ms to wait before sending retransmission "
 608           "event", 0, G_MAXUINT, DEFAULT_RTX_MIN_DELAY,
 609           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 610   /**
 611    * GstRtpJitterBuffer:rtx-delay-reorder:
 612    *
 613    * Assume that a retransmission event should be sent when we see
 614    * this much packet reordering.
 615    *
 616    * When -1 is used, the value will be estimated based on observed packet
 617    * reordering.
 618    *
 619    * Since: 1.2
 620    */
 621   g_object_class_install_property (gobject_class, PROP_RTX_DELAY_REORDER,
 622       g_param_spec_int ("rtx-delay-reorder", "RTX Delay Reorder",
 623           "Sending retransmission event when this much reordering (-1 automatic)",
 624           -1, G_MAXINT, DEFAULT_RTX_DELAY_REORDER,
 625           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 626   /**
 627    * GstRtpJitterBuffer::rtx-retry-timeout:
 628    *
 629    * When no packet has been received after sending a retransmission event
 630    * for this time, retry sending a retransmission event.
 631    *
 632    * When -1 is used, the value will be estimated based on observed round
 633    * trip time.
 634    *
 635    * Since: 1.2
 636    */
 637   g_object_class_install_property (gobject_class, PROP_RTX_RETRY_TIMEOUT,
 638       g_param_spec_int ("rtx-retry-timeout", "RTX Retry Timeout",
 639           "Retry sending a transmission event after this timeout in "
 640           "ms (-1 automatic)", -1, G_MAXINT, DEFAULT_RTX_RETRY_TIMEOUT,
 641           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 642   /**
 643    * GstRtpJitterBuffer::rtx-min-retry-timeout:
 644    *
 645    * The minimum amount of time between retry timeouts. When
 646    * GstRtpJitterBuffer::rtx-retry-timeout is -1, this value ensures a
 647    * minimum interval between retry timeouts.
 648    *
 649    * When -1 is used, the value will be estimated based on the
 650    * packet spacing.
 651    *
 652    * Since: 1.6
 653    */
 654   g_object_class_install_property (gobject_class, PROP_RTX_MIN_RETRY_TIMEOUT,
 655       g_param_spec_int ("rtx-min-retry-timeout", "RTX Min Retry Timeout",
 656           "Minimum timeout between sending a transmission event in "
 657           "ms (-1 automatic)", -1, G_MAXINT, DEFAULT_RTX_MIN_RETRY_TIMEOUT,
 658           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 659   /**
 660    * GstRtpJitterBuffer:rtx-retry-period:
 661    *
 662    * The amount of time to try to get a retransmission.
 663    *
 664    * When -1 is used, the value will be estimated based on the jitterbuffer
 665    * latency and the observed round trip time.
 666    *
 667    * Since: 1.2
 668    */
 669   g_object_class_install_property (gobject_class, PROP_RTX_RETRY_PERIOD,
 670       g_param_spec_int ("rtx-retry-period", "RTX Retry Period",
 671           "Try to get a retransmission for this many ms "
 672           "(-1 automatic)", -1, G_MAXINT, DEFAULT_RTX_RETRY_PERIOD,
 673           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 674   /**
 675    * GstRtpJitterBuffer:rtx-max-retries:
 676    *
 677    * The maximum number of retries to request a retransmission.
 678    *
 679    * This implies that as maximum (rtx-max-retries + 1) retransmissions will be requested.
 680    * When -1 is used, the number of retransmission request will not be limited.
 681    *
 682    * Since: 1.6
 683    */
 684   g_object_class_install_property (gobject_class, PROP_RTX_MAX_RETRIES,
 685       g_param_spec_int ("rtx-max-retries", "RTX Max Retries",
 686           "The maximum number of retries to request a retransmission. "
 687           "(-1 not limited)", -1, G_MAXINT, DEFAULT_RTX_MAX_RETRIES,
 688           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 689
 690   g_object_class_install_property (gobject_class, PROP_MAX_DROPOUT_TIME,
 691       g_param_spec_uint ("max-dropout-time", "Max dropout time",
 692           "The maximum time (milliseconds) of missing packets tolerated.",
 693           0, G_MAXUINT, DEFAULT_MAX_DROPOUT_TIME,
 694           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 695
 696   g_object_class_install_property (gobject_class, PROP_MAX_MISORDER_TIME,
 697       g_param_spec_uint ("max-misorder-time", "Max misorder time",
 698           "The maximum time (milliseconds) of misordered packets tolerated.",
 699           0, G_MAXUINT, DEFAULT_MAX_MISORDER_TIME,
 700           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 701   /**
 702    * GstRtpJitterBuffer:stats:
 703    *
 704    * Various jitterbuffer statistics. This property returns a GstStructure
 705    * with name application/x-rtp-jitterbuffer-stats with the following fields:
 706    *
 707    * <itemizedlist>
 708    * <listitem>
 709    *   <para>
 710    *   #guint64
 711    *   <classname>&quot;rtx-count&quot;</classname>:
 712    *   the number of retransmissions requested.
 713    *   </para>
 714    * </listitem>
 715    * <listitem>
 716    *   <para>
 717    *   #guint64
 718    *   <classname>&quot;rtx-success-count&quot;</classname>:
 719    *   the number of successful retransmissions.
 720    *   </para>
 721    * </listitem>
 722    * <listitem>
 723    *   <para>
 724    *   #gdouble
 725    *   <classname>&quot;rtx-per-packet&quot;</classname>:
 726    *   average number of RTX per packet.
 727    *   </para>
 728    * </listitem>
 729    * <listitem>
 730    *   <para>
 731    *   #guint64
 732    *   <classname>&quot;rtx-rtt&quot;</classname>:
 733    *   average round trip time per RTX.
 734    *   </para>
 735    * </listitem>
 736    * </itemizedlist>
 737    *
 738    * Since: 1.4
 739    */
 740   g_object_class_install_property (gobject_class, PROP_STATS,
 741       g_param_spec_boxed ("stats", "Statistics",
 742           "Various statistics", GST_TYPE_STRUCTURE,
 743           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 744
 745   /**
 746    * GstRtpJitterBuffer:max-rtcp-rtp-time-diff
 747    *
 748    * The maximum amount of time in ms that the RTP time in the RTCP SRs
 749    * is allowed to be ahead of the last RTP packet we received. Use
 750    * -1 to disable ignoring of RTCP packets.
 751    *
 752    * Since: 1.8
 753    */
 754   g_object_class_install_property (gobject_class, PROP_MAX_RTCP_RTP_TIME_DIFF,
 755       g_param_spec_int ("max-rtcp-rtp-time-diff", "Max RTCP RTP Time Diff",
 756           "Maximum amount of time in ms that the RTP time in RTCP SRs "
 757           "is allowed to be ahead (-1 disabled)", -1, G_MAXINT,
 758           DEFAULT_MAX_RTCP_RTP_TIME_DIFF,
 759           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 760
 761   g_object_class_install_property (gobject_class, PROP_RFC7273_SYNC,
 762       g_param_spec_boolean ("rfc7273-sync", "Sync on RFC7273 clock",
 763           "Synchronize received streams to the RFC7273 clock "
 764           "(requires clock and offset to be provided)", DEFAULT_RFC7273_SYNC,
 765           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 766
 767   /**
 768    * GstRtpJitterBuffer::request-pt-map:
 769    * @buffer: the object which received the signal
 770    * @pt: the pt
 771    *
 772    * Request the payload type as #GstCaps for @pt.
 773    */
 774   gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP] =
 775       g_signal_new ("request-pt-map", G_TYPE_FROM_CLASS (klass),
 776       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
 777           request_pt_map), NULL, NULL, g_cclosure_marshal_generic,
 778       GST_TYPE_CAPS, 1, G_TYPE_UINT);
 779   /**
 780    * GstRtpJitterBuffer::handle-sync:
 781    * @buffer: the object which received the signal
 782    * @struct: a GstStructure containing sync values.
 783    *
 784    * Be notified of new sync values.
 785    */
 786   gst_rtp_jitter_buffer_signals[SIGNAL_HANDLE_SYNC] =
 787       g_signal_new ("handle-sync", G_TYPE_FROM_CLASS (klass),
 788       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
 789           handle_sync), NULL, NULL, g_cclosure_marshal_VOID__BOXED,
 790       G_TYPE_NONE, 1, GST_TYPE_STRUCTURE | G_SIGNAL_TYPE_STATIC_SCOPE);
 791
 792   /**
 793    * GstRtpJitterBuffer::on-npt-stop:
 794    * @buffer: the object which received the signal
 795    *
 796    * Signal that the jitterbufer has pushed the RTP packet that corresponds to
 797    * the npt-stop position.
 798    */
 799   gst_rtp_jitter_buffer_signals[SIGNAL_ON_NPT_STOP] =
 800       g_signal_new ("on-npt-stop", G_TYPE_FROM_CLASS (klass),
 801       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
 802           on_npt_stop), NULL, NULL, g_cclosure_marshal_VOID__VOID,
 803       G_TYPE_NONE, 0, G_TYPE_NONE);
 804
 805   /**
 806    * GstRtpJitterBuffer::clear-pt-map:
 807    * @buffer: the object which received the signal
 808    *
 809    * Invalidate the clock-rate as obtained with the
 810    * #GstRtpJitterBuffer::request-pt-map signal.
 811    */
 812   gst_rtp_jitter_buffer_signals[SIGNAL_CLEAR_PT_MAP] =
 813       g_signal_new ("clear-pt-map", G_TYPE_FROM_CLASS (klass),
 814       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 815       G_STRUCT_OFFSET (GstRtpJitterBufferClass, clear_pt_map), NULL, NULL,
 816       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0, G_TYPE_NONE);
 817
 818   /**
 819    * GstRtpJitterBuffer::set-active:
 820    * @buffer: the object which received the signal
 821    *
 822    * Start pushing out packets with the given base time. This signal is only
 823    * useful in buffering mode.
 824    *
 825    * Returns: the time of the last pushed packet.
 826    */
 827   gst_rtp_jitter_buffer_signals[SIGNAL_SET_ACTIVE] =
 828       g_signal_new ("set-active", G_TYPE_FROM_CLASS (klass),
 829       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 830       G_STRUCT_OFFSET (GstRtpJitterBufferClass, set_active), NULL, NULL,
 831       g_cclosure_marshal_generic, G_TYPE_UINT64, 2, G_TYPE_BOOLEAN,
 832       G_TYPE_UINT64);
 833
 834   gstelement_class->change_state =
 835       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_change_state);
 836   gstelement_class->request_new_pad =
 837       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_request_new_pad);
 838   gstelement_class->release_pad =
 839       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_release_pad);
 840   gstelement_class->provide_clock =
 841       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_provide_clock);
 842   gstelement_class->set_clock =
 843       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_set_clock);
 844
 845   gst_element_class_add_static_pad_template (gstelement_class,
 846       &gst_rtp_jitter_buffer_src_template);
 847   gst_element_class_add_static_pad_template (gstelement_class,
 848       &gst_rtp_jitter_buffer_sink_template);
 849   gst_element_class_add_static_pad_template (gstelement_class,
 850       &gst_rtp_jitter_buffer_sink_rtcp_template);
 851
 852   gst_element_class_set_static_metadata (gstelement_class,
 853       "RTP packet jitter-buffer", "Filter/Network/RTP",
 854       "A buffer that deals with network jitter and other transmission faults",
 855       "Philippe Kalaf <philippe.kalaf@collabora.co.uk>, "
 856       "Wim Taymans <wim.taymans@gmail.com>");
 857
 858   klass->clear_pt_map = GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_clear_pt_map);
 859   klass->set_active = GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_set_active);
 860
 861   GST_DEBUG_CATEGORY_INIT
 862       (rtpjitterbuffer_debug, "rtpjitterbuffer", 0, "RTP Jitter Buffer");
 863 }
 864
 865 static void
 866 gst_rtp_jitter_buffer_init (GstRtpJitterBuffer * jitterbuffer)
 867 {
 868   GstRtpJitterBufferPrivate *priv;
 869
 870   priv = GST_RTP_JITTER_BUFFER_GET_PRIVATE (jitterbuffer);
 871   jitterbuffer->priv = priv;
 872
 873   priv->latency_ms = DEFAULT_LATENCY_MS;
 874   priv->latency_ns = priv->latency_ms * GST_MSECOND;
 875   priv->drop_on_latency = DEFAULT_DROP_ON_LATENCY;
 876   priv->do_lost = DEFAULT_DO_LOST;
 877   priv->do_retransmission = DEFAULT_DO_RETRANSMISSION;
 878   priv->rtx_next_seqnum = DEFAULT_RTX_NEXT_SEQNUM;
 879   priv->rtx_delay = DEFAULT_RTX_DELAY;
 880   priv->rtx_min_delay = DEFAULT_RTX_MIN_DELAY;
 881   priv->rtx_delay_reorder = DEFAULT_RTX_DELAY_REORDER;
 882   priv->rtx_retry_timeout = DEFAULT_RTX_RETRY_TIMEOUT;
 883   priv->rtx_min_retry_timeout = DEFAULT_RTX_MIN_RETRY_TIMEOUT;
 884   priv->rtx_retry_period = DEFAULT_RTX_RETRY_PERIOD;
 885   priv->rtx_max_retries = DEFAULT_RTX_MAX_RETRIES;
 886   priv->max_rtcp_rtp_time_diff = DEFAULT_MAX_RTCP_RTP_TIME_DIFF;
 887   priv->max_dropout_time = DEFAULT_MAX_DROPOUT_TIME;
 888   priv->max_misorder_time = DEFAULT_MAX_MISORDER_TIME;
 889
 890   priv->last_dts = -1;
 891   priv->last_rtptime = -1;
 892   priv->avg_jitter = 0;
 893   priv->timers = g_array_new (FALSE, TRUE, sizeof (TimerData));
 894   priv->jbuf = rtp_jitter_buffer_new ();
 895   g_mutex_init (&priv->jbuf_lock);
 896   g_cond_init (&priv->jbuf_timer);
 897   g_cond_init (&priv->jbuf_event);
 898   g_cond_init (&priv->jbuf_query);
 899   g_queue_init (&priv->gap_packets);
 900
 901   /* reset skew detection initialy */
 902   rtp_jitter_buffer_reset_skew (priv->jbuf);
 903   rtp_jitter_buffer_set_delay (priv->jbuf, priv->latency_ns);
 904   rtp_jitter_buffer_set_buffering (priv->jbuf, FALSE);
 905   priv->active = TRUE;
 906
 907   priv->srcpad =
 908       gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_src_template,
 909       "src");
 910
 911   gst_pad_set_activatemode_function (priv->srcpad,
 912       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_activate_mode));
 913   gst_pad_set_query_function (priv->srcpad,
 914       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_query));
 915   gst_pad_set_event_function (priv->srcpad,
 916       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_event));
 917
 918   priv->sinkpad =
 919       gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_sink_template,
 920       "sink");
 921
 922   gst_pad_set_chain_function (priv->sinkpad,
 923       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_chain));
 924   gst_pad_set_event_function (priv->sinkpad,
 925       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_sink_event));
 926   gst_pad_set_query_function (priv->sinkpad,
 927       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_sink_query));
 928
 929   gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->srcpad);
 930   gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->sinkpad);
 931
 932   GST_OBJECT_FLAG_SET (jitterbuffer, GST_ELEMENT_FLAG_PROVIDE_CLOCK);
 933 }
 934
 935 #define IS_DROPABLE(it) (((it)->type == ITEM_TYPE_BUFFER) || ((it)->type == ITEM_TYPE_LOST))
 936
 937 #define ITEM_TYPE_BUFFER        0
 938 #define ITEM_TYPE_LOST          1
 939 #define ITEM_TYPE_EVENT         2
 940 #define ITEM_TYPE_QUERY         3
 941
 942 static RTPJitterBufferItem *
 943 alloc_item (gpointer data, guint type, GstClockTime dts, GstClockTime pts,
 944     guint seqnum, guint count, guint rtptime)
 945 {
 946   RTPJitterBufferItem *item;
 947
 948   item = g_slice_new (RTPJitterBufferItem);
 949   item->data = data;
 950   item->next = NULL;
 951   item->prev = NULL;
 952   item->type = type;
 953   item->dts = dts;
 954   item->pts = pts;
 955   item->seqnum = seqnum;
 956   item->count = count;
 957   item->rtptime = rtptime;
 958
 959   return item;
 960 }
 961
 962 static void
 963 free_item (RTPJitterBufferItem * item)
 964 {
 965   g_return_if_fail (item != NULL);
 966
 967   if (item->data && item->type != ITEM_TYPE_QUERY)
 968     gst_mini_object_unref (item->data);
 969   g_slice_free (RTPJitterBufferItem, item);
 970 }
 971
 972 static void
 973 free_item_and_retain_events (RTPJitterBufferItem * item, gpointer user_data)
 974 {
 975   GList **l = user_data;
 976
 977   if (item->data && item->type == ITEM_TYPE_EVENT
 978       && GST_EVENT_IS_STICKY (item->data)) {
 979     *l = g_list_prepend (*l, item->data);
 980   } else if (item->data && item->type != ITEM_TYPE_QUERY) {
 981     gst_mini_object_unref (item->data);
 982   }
 983   g_slice_free (RTPJitterBufferItem, item);
 984 }
 985
 986 static void
 987 gst_rtp_jitter_buffer_finalize (GObject * object)
 988 {
 989   GstRtpJitterBuffer *jitterbuffer;
 990   GstRtpJitterBufferPrivate *priv;
 991
 992   jitterbuffer = GST_RTP_JITTER_BUFFER (object);
 993   priv = jitterbuffer->priv;
 994
 995   g_array_free (priv->timers, TRUE);
 996   g_mutex_clear (&priv->jbuf_lock);
 997   g_cond_clear (&priv->jbuf_timer);
 998   g_cond_clear (&priv->jbuf_event);
 999   g_cond_clear (&priv->jbuf_query);
1000
1001   rtp_jitter_buffer_flush (priv->jbuf, (GFunc) free_item, NULL);
1002   g_queue_foreach (&priv->gap_packets, (GFunc) gst_buffer_unref, NULL);
1003   g_queue_clear (&priv->gap_packets);
1004   g_object_unref (priv->jbuf);
1005
1006   G_OBJECT_CLASS (parent_class)->finalize (object);
1007 }
1008
1009 static GstIterator *
1010 gst_rtp_jitter_buffer_iterate_internal_links (GstPad * pad, GstObject * parent)
1011 {
1012   GstRtpJitterBuffer *jitterbuffer;
1013   GstPad *otherpad = NULL;
1014   GstIterator *it = NULL;
1015   GValue val = { 0, };
1016
1017   jitterbuffer = GST_RTP_JITTER_BUFFER_CAST (parent);
1018
1019   if (pad == jitterbuffer->priv->sinkpad) {
1020     otherpad = jitterbuffer->priv->srcpad;
1021   } else if (pad == jitterbuffer->priv->srcpad) {
1022     otherpad = jitterbuffer->priv->sinkpad;
1023   } else if (pad == jitterbuffer->priv->rtcpsinkpad) {
1024     it = gst_iterator_new_single (GST_TYPE_PAD, NULL);
1025   }
1026
1027   if (it == NULL) {
1028     g_value_init (&val, GST_TYPE_PAD);
1029     g_value_set_object (&val, otherpad);
1030     it = gst_iterator_new_single (GST_TYPE_PAD, &val);
1031     g_value_unset (&val);
1032   }
1033
1034   return it;
1035 }
1036
1037 static GstPad *
1038 create_rtcp_sink (GstRtpJitterBuffer * jitterbuffer)
1039 {
1040   GstRtpJitterBufferPrivate *priv;
1041
1042   priv = jitterbuffer->priv;
1043
1044   GST_DEBUG_OBJECT (jitterbuffer, "creating RTCP sink pad");
1045
1046   priv->rtcpsinkpad =
1047       gst_pad_new_from_static_template
1048       (&gst_rtp_jitter_buffer_sink_rtcp_template, "sink_rtcp");
1049   gst_pad_set_chain_function (priv->rtcpsinkpad,
1050       gst_rtp_jitter_buffer_chain_rtcp);
1051   gst_pad_set_event_function (priv->rtcpsinkpad,
1052       (GstPadEventFunction) gst_rtp_jitter_buffer_sink_rtcp_event);
1053   gst_pad_set_iterate_internal_links_function (priv->rtcpsinkpad,
1054       gst_rtp_jitter_buffer_iterate_internal_links);
1055   gst_pad_set_active (priv->rtcpsinkpad, TRUE);
1056   gst_element_add_pad (GST_ELEMENT_CAST (jitterbuffer), priv->rtcpsinkpad);
1057
1058   return priv->rtcpsinkpad;
1059 }
1060
1061 static void
1062 remove_rtcp_sink (GstRtpJitterBuffer * jitterbuffer)
1063 {
1064   GstRtpJitterBufferPrivate *priv;
1065
1066   priv = jitterbuffer->priv;
1067
1068   GST_DEBUG_OBJECT (jitterbuffer, "removing RTCP sink pad");
1069
1070   gst_pad_set_active (priv->rtcpsinkpad, FALSE);
1071
1072   gst_element_remove_pad (GST_ELEMENT_CAST (jitterbuffer), priv->rtcpsinkpad);
1073   priv->rtcpsinkpad = NULL;
1074 }
1075
1076 static GstPad *
1077 gst_rtp_jitter_buffer_request_new_pad (GstElement * element,
1078     GstPadTemplate * templ, const gchar * name, const GstCaps * filter)
1079 {
1080   GstRtpJitterBuffer *jitterbuffer;
1081   GstElementClass *klass;
1082   GstPad *result;
1083   GstRtpJitterBufferPrivate *priv;
1084
1085   g_return_val_if_fail (templ != NULL, NULL);
1086   g_return_val_if_fail (GST_IS_RTP_JITTER_BUFFER (element), NULL);
1087
1088   jitterbuffer = GST_RTP_JITTER_BUFFER_CAST (element);
1089   priv = jitterbuffer->priv;
1090   klass = GST_ELEMENT_GET_CLASS (element);
1091
1092   GST_DEBUG_OBJECT (element, "requesting pad %s", GST_STR_NULL (name));
1093
1094   /* figure out the template */
1095   if (templ == gst_element_class_get_pad_template (klass, "sink_rtcp")) {
1096     if (priv->rtcpsinkpad != NULL)
1097       goto exists;
1098
1099     result = create_rtcp_sink (jitterbuffer);
1100   } else
1101     goto wrong_template;
1102
1103   return result;
1104
1105   /* ERRORS */
1106 wrong_template:
1107   {
1108     g_warning ("rtpjitterbuffer: this is not our template");
1109     return NULL;
1110   }
1111 exists:
1112   {
1113     g_warning ("rtpjitterbuffer: pad already requested");
1114     return NULL;
1115   }
1116 }
1117
1118 static void
1119 gst_rtp_jitter_buffer_release_pad (GstElement * element, GstPad * pad)
1120 {
1121   GstRtpJitterBuffer *jitterbuffer;
1122   GstRtpJitterBufferPrivate *priv;
1123
1124   g_return_if_fail (GST_IS_RTP_JITTER_BUFFER (element));
1125   g_return_if_fail (GST_IS_PAD (pad));
1126
1127   jitterbuffer = GST_RTP_JITTER_BUFFER_CAST (element);
1128   priv = jitterbuffer->priv;
1129
1130   GST_DEBUG_OBJECT (element, "releasing pad %s:%s", GST_DEBUG_PAD_NAME (pad));
1131
1132   if (priv->rtcpsinkpad == pad) {
1133     remove_rtcp_sink (jitterbuffer);
1134   } else
1135     goto wrong_pad;
1136
1137   return;
1138
1139   /* ERRORS */
1140 wrong_pad:
1141   {
1142     g_warning ("gstjitterbuffer: asked to release an unknown pad");
1143     return;
1144   }
1145 }
1146
1147 static GstClock *
1148 gst_rtp_jitter_buffer_provide_clock (GstElement * element)
1149 {
1150   return gst_system_clock_obtain ();
1151 }
1152
1153 static gboolean
1154 gst_rtp_jitter_buffer_set_clock (GstElement * element, GstClock * clock)
1155 {
1156   GstRtpJitterBuffer *jitterbuffer = GST_RTP_JITTER_BUFFER (element);
1157
1158   rtp_jitter_buffer_set_pipeline_clock (jitterbuffer->priv->jbuf, clock);
1159
1160   return GST_ELEMENT_CLASS (parent_class)->set_clock (element, clock);
1161 }
1162
1163 static void
1164 gst_rtp_jitter_buffer_clear_pt_map (GstRtpJitterBuffer * jitterbuffer)
1165 {
1166   GstRtpJitterBufferPrivate *priv;
1167
1168   priv = jitterbuffer->priv;
1169
1170   /* this will trigger a new pt-map request signal, FIXME, do something better. */
1171
1172   JBUF_LOCK (priv);
1173   priv->clock_rate = -1;
1174   /* do not clear current content, but refresh state for new arrival */
1175   GST_DEBUG_OBJECT (jitterbuffer, "reset jitterbuffer");
1176   rtp_jitter_buffer_reset_skew (priv->jbuf);
1177   JBUF_UNLOCK (priv);
1178 }
1179
1180 static GstClockTime
1181 gst_rtp_jitter_buffer_set_active (GstRtpJitterBuffer * jbuf, gboolean active,
1182     guint64 offset)
1183 {
1184   GstRtpJitterBufferPrivate *priv;
1185   GstClockTime last_out;
1186   RTPJitterBufferItem *item;
1187
1188   priv = jbuf->priv;
1189
1190   JBUF_LOCK (priv);
1191   GST_DEBUG_OBJECT (jbuf, "setting active %d with offset %" GST_TIME_FORMAT,
1192       active, GST_TIME_ARGS (offset));
1193
1194   if (active != priv->active) {
1195     /* add the amount of time spent in paused to the output offset. All
1196      * outgoing buffers will have this offset applied to their timestamps in
1197      * order to make them arrive in time in the sink. */
1198     priv->out_offset = offset;
1199     GST_DEBUG_OBJECT (jbuf, "out offset %" GST_TIME_FORMAT,
1200         GST_TIME_ARGS (priv->out_offset));
1201     priv->active = active;
1202     JBUF_SIGNAL_EVENT (priv);
1203   }
1204   if (!active) {
1205     rtp_jitter_buffer_set_buffering (priv->jbuf, TRUE);
1206   }
1207   if ((item = rtp_jitter_buffer_peek (priv->jbuf))) {
1208     /* head buffer timestamp and offset gives our output time */
1209     last_out = item->dts + priv->ts_offset;
1210   } else {
1211     /* use last known time when the buffer is empty */
1212     last_out = priv->last_out_time;
1213   }
1214   JBUF_UNLOCK (priv);
1215
1216   return last_out;
1217 }
1218
1219 static GstCaps *
1220 gst_rtp_jitter_buffer_getcaps (GstPad * pad, GstCaps * filter)
1221 {
1222   GstRtpJitterBuffer *jitterbuffer;
1223   GstRtpJitterBufferPrivate *priv;
1224   GstPad *other;
1225   GstCaps *caps;
1226   GstCaps *templ;
1227
1228   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
1229   priv = jitterbuffer->priv;
1230
1231   other = (pad == priv->srcpad ? priv->sinkpad : priv->srcpad);
1232
1233   caps = gst_pad_peer_query_caps (other, filter);
1234
1235   templ = gst_pad_get_pad_template_caps (pad);
1236   if (caps == NULL) {
1237     GST_DEBUG_OBJECT (jitterbuffer, "use template");
1238     caps = templ;
1239   } else {
1240     GstCaps *intersect;
1241
1242     GST_DEBUG_OBJECT (jitterbuffer, "intersect with template");
1243
1244     intersect = gst_caps_intersect (caps, templ);
1245     gst_caps_unref (caps);
1246     gst_caps_unref (templ);
1247
1248     caps = intersect;
1249   }
1250   gst_object_unref (jitterbuffer);
1251
1252   return caps;
1253 }
1254
1255 /*
1256  * Must be called with JBUF_LOCK held
1257  */
1258
1259 static gboolean
1260 gst_jitter_buffer_sink_parse_caps (GstRtpJitterBuffer * jitterbuffer,
1261     GstCaps * caps)
1262 {
1263   GstRtpJitterBufferPrivate *priv;
1264   GstStructure *caps_struct;
1265   guint val;
1266   GstClockTime tval;
1267   const gchar *ts_refclk, *mediaclk;
1268
1269   priv = jitterbuffer->priv;
1270
1271   /* first parse the caps */
1272   caps_struct = gst_caps_get_structure (caps, 0);
1273
1274   GST_DEBUG_OBJECT (jitterbuffer, "got caps");
1275
1276   /* we need a clock-rate to convert the rtp timestamps to GStreamer time and to
1277    * measure the amount of data in the buffer */
1278   if (!gst_structure_get_int (caps_struct, "clock-rate", &priv->clock_rate))
1279     goto error;
1280
1281   if (priv->clock_rate <= 0)
1282     goto wrong_rate;
1283
1284   GST_DEBUG_OBJECT (jitterbuffer, "got clock-rate %d", priv->clock_rate);
1285
1286   rtp_jitter_buffer_set_clock_rate (priv->jbuf, priv->clock_rate);
1287
1288   /* The clock base is the RTP timestamp corrsponding to the npt-start value. We
1289    * can use this to track the amount of time elapsed on the sender. */
1290   if (gst_structure_get_uint (caps_struct, "clock-base", &val))
1291     priv->clock_base = val;
1292   else
1293     priv->clock_base = -1;
1294
1295   priv->ext_timestamp = priv->clock_base;
1296
1297   GST_DEBUG_OBJECT (jitterbuffer, "got clock-base %" G_GINT64_FORMAT,
1298       priv->clock_base);
1299
1300   if (gst_structure_get_uint (caps_struct, "seqnum-base", &val)) {
1301     /* first expected seqnum, only update when we didn't have a previous base. */
1302     if (priv->next_in_seqnum == -1)
1303       priv->next_in_seqnum = val;
1304     if (priv->next_seqnum == -1) {
1305       priv->next_seqnum = val;
1306       JBUF_SIGNAL_EVENT (priv);
1307     }
1308     priv->seqnum_base = val;
1309   } else {
1310     priv->seqnum_base = -1;
1311   }
1312
1313   GST_DEBUG_OBJECT (jitterbuffer, "got seqnum-base %d", priv->next_in_seqnum);
1314
1315   /* the start and stop times. The seqnum-base corresponds to the start time. We
1316    * will keep track of the seqnums on the output and when we reach the one
1317    * corresponding to npt-stop, we emit the npt-stop-reached signal */
1318   if (gst_structure_get_clock_time (caps_struct, "npt-start", &tval))
1319     priv->npt_start = tval;
1320   else
1321     priv->npt_start = 0;
1322
1323   if (gst_structure_get_clock_time (caps_struct, "npt-stop", &tval))
1324     priv->npt_stop = tval;
1325   else
1326     priv->npt_stop = -1;
1327
1328   GST_DEBUG_OBJECT (jitterbuffer,
1329       "npt start/stop: %" GST_TIME_FORMAT "-%" GST_TIME_FORMAT,
1330       GST_TIME_ARGS (priv->npt_start), GST_TIME_ARGS (priv->npt_stop));
1331
1332   if ((ts_refclk = gst_structure_get_string (caps_struct, "a-ts-refclk"))) {
1333     GstClock *clock = NULL;
1334     guint64 clock_offset = -1;
1335
1336     GST_DEBUG_OBJECT (jitterbuffer, "Have timestamp reference clock %s",
1337         ts_refclk);
1338
1339     if (g_str_has_prefix (ts_refclk, "ntp=")) {
1340       if (g_str_has_prefix (ts_refclk, "ntp=/traceable/")) {
1341         GST_FIXME_OBJECT (jitterbuffer, "Can't handle traceable NTP clocks");
1342       } else {
1343         const gchar *host, *portstr;
1344         gchar *hostname;
1345         guint port;
1346
1347         host = ts_refclk + sizeof ("ntp=") - 1;
1348         if (host[0] == '[') {
1349           /* IPv6 */
1350           portstr = strchr (host, ']');
1351           if (portstr && portstr[1] == ':')
1352             portstr = portstr + 1;
1353           else
1354             portstr = NULL;
1355         } else {
1356           portstr = strrchr (host, ':');
1357         }
1358
1359
1360         if (!portstr || sscanf (portstr, ":%u", &port) != 1)
1361           port = 123;
1362
1363         if (portstr)
1364           hostname = g_strndup (host, (portstr - host));
1365         else
1366           hostname = g_strdup (host);
1367
1368         clock = gst_ntp_clock_new (NULL, hostname, port, 0);
1369         g_free (hostname);
1370       }
1371     } else if (g_str_has_prefix (ts_refclk, "ptp=IEEE1588-2008:")) {
1372       const gchar *domainstr =
1373           ts_refclk + sizeof ("ptp=IEEE1588-2008:XX-XX-XX-XX-XX-XX-XX-XX") - 1;
1374       guint domain;
1375
1376       if (domainstr[0] != ':' || sscanf (domainstr, ":%u", &domain) != 1)
1377         domain = 0;
1378
1379       clock = gst_ptp_clock_new (NULL, domain);
1380     } else {
1381       GST_FIXME_OBJECT (jitterbuffer, "Unsupported timestamp reference clock");
1382     }
1383
1384     if ((mediaclk = gst_structure_get_string (caps_struct, "a-mediaclk"))) {
1385       GST_DEBUG_OBJECT (jitterbuffer, "Got media clock %s", mediaclk);
1386
1387       if (!g_str_has_prefix (mediaclk, "direct=")
1388           || sscanf (mediaclk, "direct=%" G_GUINT64_FORMAT, &clock_offset) != 1)
1389         GST_FIXME_OBJECT (jitterbuffer, "Unsupported media clock");
1390       if (strstr (mediaclk, "rate=") != NULL) {
1391         GST_FIXME_OBJECT (jitterbuffer, "Rate property not supported");
1392         clock_offset = -1;
1393       }
1394     }
1395
1396     rtp_jitter_buffer_set_media_clock (priv->jbuf, clock, clock_offset);
1397   } else {
1398     rtp_jitter_buffer_set_media_clock (priv->jbuf, NULL, -1);
1399   }
1400
1401   return TRUE;
1402
1403   /* ERRORS */
1404 error:
1405   {
1406     GST_DEBUG_OBJECT (jitterbuffer, "No clock-rate in caps!");
1407     return FALSE;
1408   }
1409 wrong_rate:
1410   {
1411     GST_DEBUG_OBJECT (jitterbuffer, "Invalid clock-rate %d", priv->clock_rate);
1412     return FALSE;
1413   }
1414 }
1415
1416 static void
1417 gst_rtp_jitter_buffer_flush_start (GstRtpJitterBuffer * jitterbuffer)
1418 {
1419   GstRtpJitterBufferPrivate *priv;
1420
1421   priv = jitterbuffer->priv;
1422
1423   JBUF_LOCK (priv);
1424   /* mark ourselves as flushing */
1425   priv->srcresult = GST_FLOW_FLUSHING;
1426   GST_DEBUG_OBJECT (jitterbuffer, "Disabling pop on queue");
1427   /* this unblocks any waiting pops on the src pad task */
1428   JBUF_SIGNAL_EVENT (priv);
1429   JBUF_SIGNAL_QUERY (priv, FALSE);
1430   JBUF_UNLOCK (priv);
1431 }
1432
1433 static void
1434 gst_rtp_jitter_buffer_flush_stop (GstRtpJitterBuffer * jitterbuffer)
1435 {
1436   GstRtpJitterBufferPrivate *priv;
1437
1438   priv = jitterbuffer->priv;
1439
1440   JBUF_LOCK (priv);
1441   GST_DEBUG_OBJECT (jitterbuffer, "Enabling pop on queue");
1442   /* Mark as non flushing */
1443   priv->srcresult = GST_FLOW_OK;
1444   gst_segment_init (&priv->segment, GST_FORMAT_TIME);
1445   priv->last_popped_seqnum = -1;
1446   priv->last_out_time = -1;
1447   priv->next_seqnum = -1;
1448   priv->seqnum_base = -1;
1449   priv->ips_rtptime = -1;
1450   priv->ips_dts = GST_CLOCK_TIME_NONE;
1451   priv->packet_spacing = 0;
1452   priv->next_in_seqnum = -1;
1453   priv->clock_rate = -1;
1454   priv->last_pt = -1;
1455   priv->eos = FALSE;
1456   priv->estimated_eos = -1;
1457   priv->last_elapsed = 0;
1458   priv->ext_timestamp = -1;
1459   priv->avg_jitter = 0;
1460   priv->last_dts = -1;
1461   priv->last_rtptime = -1;
1462   priv->last_in_dts = 0;
1463   GST_DEBUG_OBJECT (jitterbuffer, "flush and reset jitterbuffer");
1464   rtp_jitter_buffer_flush (priv->jbuf, (GFunc) free_item, NULL);
1465   rtp_jitter_buffer_disable_buffering (priv->jbuf, FALSE);
1466   rtp_jitter_buffer_reset_skew (priv->jbuf);
1467   remove_all_timers (jitterbuffer);
1468   g_queue_foreach (&priv->gap_packets, (GFunc) gst_buffer_unref, NULL);
1469   g_queue_clear (&priv->gap_packets);
1470   JBUF_UNLOCK (priv);
1471 }
1472
1473 static gboolean
1474 gst_rtp_jitter_buffer_src_activate_mode (GstPad * pad, GstObject * parent,
1475     GstPadMode mode, gboolean active)
1476 {
1477   gboolean result;
1478   GstRtpJitterBuffer *jitterbuffer = NULL;
1479
1480   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
1481
1482   switch (mode) {
1483     case GST_PAD_MODE_PUSH:
1484       if (active) {
1485         /* allow data processing */
1486         gst_rtp_jitter_buffer_flush_stop (jitterbuffer);
1487
1488         /* start pushing out buffers */
1489         GST_DEBUG_OBJECT (jitterbuffer, "Starting task on srcpad");
1490         result = gst_pad_start_task (jitterbuffer->priv->srcpad,
1491             (GstTaskFunction) gst_rtp_jitter_buffer_loop, jitterbuffer, NULL);
1492       } else {
1493         /* make sure all data processing stops ASAP */
1494         gst_rtp_jitter_buffer_flush_start (jitterbuffer);
1495
1496         /* NOTE this will hardlock if the state change is called from the src pad
1497          * task thread because we will _join() the thread. */
1498         GST_DEBUG_OBJECT (jitterbuffer, "Stopping task on srcpad");
1499         result = gst_pad_stop_task (pad);
1500       }
1501       break;
1502     default:
1503       result = FALSE;
1504       break;
1505   }
1506   return result;
1507 }
1508
1509 static GstStateChangeReturn
1510 gst_rtp_jitter_buffer_change_state (GstElement * element,
1511     GstStateChange transition)
1512 {
1513   GstRtpJitterBuffer *jitterbuffer;
1514   GstRtpJitterBufferPrivate *priv;
1515   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
1516
1517   jitterbuffer = GST_RTP_JITTER_BUFFER (element);
1518   priv = jitterbuffer->priv;
1519
1520   switch (transition) {
1521     case GST_STATE_CHANGE_NULL_TO_READY:
1522       break;
1523     case GST_STATE_CHANGE_READY_TO_PAUSED:
1524       JBUF_LOCK (priv);
1525       /* reset negotiated values */
1526       priv->clock_rate = -1;
1527       priv->clock_base = -1;
1528       priv->peer_latency = 0;
1529       priv->last_pt = -1;
1530       /* block until we go to PLAYING */
1531       priv->blocked = TRUE;
1532       priv->timer_running = TRUE;
1533       priv->timer_thread =
1534           g_thread_new ("timer", (GThreadFunc) wait_next_timeout, jitterbuffer);
1535       JBUF_UNLOCK (priv);
1536       break;
1537     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1538       JBUF_LOCK (priv);
1539       /* unblock to allow streaming in PLAYING */
1540       priv->blocked = FALSE;
1541       JBUF_SIGNAL_EVENT (priv);
1542       JBUF_SIGNAL_TIMER (priv);
1543       JBUF_UNLOCK (priv);
1544       break;
1545     default:
1546       break;
1547   }
1548
1549   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1550
1551   switch (transition) {
1552     case GST_STATE_CHANGE_READY_TO_PAUSED:
1553       /* we are a live element because we sync to the clock, which we can only
1554        * do in the PLAYING state */
1555       if (ret != GST_STATE_CHANGE_FAILURE)
1556         ret = GST_STATE_CHANGE_NO_PREROLL;
1557       break;
1558     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1559       JBUF_LOCK (priv);
1560       /* block to stop streaming when PAUSED */
1561       priv->blocked = TRUE;
1562       unschedule_current_timer (jitterbuffer);
1563       JBUF_UNLOCK (priv);
1564       if (ret != GST_STATE_CHANGE_FAILURE)
1565         ret = GST_STATE_CHANGE_NO_PREROLL;
1566       break;
1567     case GST_STATE_CHANGE_PAUSED_TO_READY:
1568       JBUF_LOCK (priv);
1569       gst_buffer_replace (&priv->last_sr, NULL);
1570       priv->timer_running = FALSE;
1571       unschedule_current_timer (jitterbuffer);
1572       JBUF_SIGNAL_TIMER (priv);
1573       JBUF_SIGNAL_QUERY (priv, FALSE);
1574       JBUF_UNLOCK (priv);
1575       g_thread_join (priv->timer_thread);
1576       priv->timer_thread = NULL;
1577       break;
1578     case GST_STATE_CHANGE_READY_TO_NULL:
1579       break;
1580     default:
1581       break;
1582   }
1583
1584   return ret;
1585 }
1586
1587 static gboolean
1588 gst_rtp_jitter_buffer_src_event (GstPad * pad, GstObject * parent,
1589     GstEvent * event)
1590 {
1591   gboolean ret = TRUE;
1592   GstRtpJitterBuffer *jitterbuffer;
1593   GstRtpJitterBufferPrivate *priv;
1594
1595   jitterbuffer = GST_RTP_JITTER_BUFFER_CAST (parent);
1596   priv = jitterbuffer->priv;
1597
1598   GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
1599
1600   switch (GST_EVENT_TYPE (event)) {
1601     case GST_EVENT_LATENCY:
1602     {
1603       GstClockTime latency;
1604
1605       gst_event_parse_latency (event, &latency);
1606
1607       GST_DEBUG_OBJECT (jitterbuffer,
1608           "configuring latency of %" GST_TIME_FORMAT, GST_TIME_ARGS (latency));
1609
1610       JBUF_LOCK (priv);
1611       /* adjust the overall buffer delay to the total pipeline latency in
1612        * buffering mode because if downstream consumes too fast (because of
1613        * large latency or queues, we would start rebuffering again. */
1614       if (rtp_jitter_buffer_get_mode (priv->jbuf) ==
1615           RTP_JITTER_BUFFER_MODE_BUFFER) {
1616         rtp_jitter_buffer_set_delay (priv->jbuf, latency);
1617       }
1618       JBUF_UNLOCK (priv);
1619
1620       ret = gst_pad_push_event (priv->sinkpad, event);
1621       break;
1622     }
1623     default:
1624       ret = gst_pad_push_event (priv->sinkpad, event);
1625       break;
1626   }
1627
1628   return ret;
1629 }
1630
1631 /* handles and stores the event in the jitterbuffer, must be called with
1632  * LOCK */
1633 static gboolean
1634 queue_event (GstRtpJitterBuffer * jitterbuffer, GstEvent * event)
1635 {
1636   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1637   RTPJitterBufferItem *item;
1638   gboolean head;
1639
1640   switch (GST_EVENT_TYPE (event)) {
1641     case GST_EVENT_CAPS:
1642     {
1643       GstCaps *caps;
1644
1645       gst_event_parse_caps (event, &caps);
1646       gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
1647       break;
1648     }
1649     case GST_EVENT_SEGMENT:
1650       gst_event_copy_segment (event, &priv->segment);
1651
1652       /* we need time for now */
1653       if (priv->segment.format != GST_FORMAT_TIME)
1654         goto newseg_wrong_format;
1655
1656       GST_DEBUG_OBJECT (jitterbuffer,
1657           "segment:  %" GST_SEGMENT_FORMAT, &priv->segment);
1658       break;
1659     case GST_EVENT_EOS:
1660       priv->eos = TRUE;
1661       rtp_jitter_buffer_disable_buffering (priv->jbuf, TRUE);
1662       break;
1663     default:
1664       break;
1665   }
1666
1667
1668   GST_DEBUG_OBJECT (jitterbuffer, "adding event");
1669   item = alloc_item (event, ITEM_TYPE_EVENT, -1, -1, -1, 0, -1);
1670   rtp_jitter_buffer_insert (priv->jbuf, item, &head, NULL, -1);
1671   if (head)
1672     JBUF_SIGNAL_EVENT (priv);
1673
1674   return TRUE;
1675
1676   /* ERRORS */
1677 newseg_wrong_format:
1678   {
1679     GST_DEBUG_OBJECT (jitterbuffer, "received non TIME newsegment");
1680     gst_event_unref (event);
1681     return FALSE;
1682   }
1683 }
1684
1685 static gboolean
1686 gst_rtp_jitter_buffer_sink_event (GstPad * pad, GstObject * parent,
1687     GstEvent * event)
1688 {
1689   gboolean ret = TRUE;
1690   GstRtpJitterBuffer *jitterbuffer;
1691   GstRtpJitterBufferPrivate *priv;
1692
1693   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
1694   priv = jitterbuffer->priv;
1695
1696   GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
1697
1698   switch (GST_EVENT_TYPE (event)) {
1699     case GST_EVENT_FLUSH_START:
1700       ret = gst_pad_push_event (priv->srcpad, event);
1701       gst_rtp_jitter_buffer_flush_start (jitterbuffer);
1702       /* wait for the loop to go into PAUSED */
1703       gst_pad_pause_task (priv->srcpad);
1704       break;
1705     case GST_EVENT_FLUSH_STOP:
1706       ret = gst_pad_push_event (priv->srcpad, event);
1707       ret =
1708           gst_rtp_jitter_buffer_src_activate_mode (priv->srcpad, parent,
1709           GST_PAD_MODE_PUSH, TRUE);
1710       break;
1711     default:
1712       if (GST_EVENT_IS_SERIALIZED (event)) {
1713         /* serialized events go in the queue */
1714         JBUF_LOCK (priv);
1715         if (priv->srcresult != GST_FLOW_OK) {
1716           /* Errors in sticky event pushing are no problem and ignored here
1717            * as they will cause more meaningful errors during data flow.
1718            * For EOS events, that are not followed by data flow, we still
1719            * return FALSE here though.
1720            */
1721           if (!GST_EVENT_IS_STICKY (event) ||
1722               GST_EVENT_TYPE (event) == GST_EVENT_EOS)
1723             goto out_flow_error;
1724         }
1725         /* refuse more events on EOS */
1726         if (priv->eos)
1727           goto out_eos;
1728         ret = queue_event (jitterbuffer, event);
1729         JBUF_UNLOCK (priv);
1730       } else {
1731         /* non-serialized events are forwarded downstream immediately */
1732         ret = gst_pad_push_event (priv->srcpad, event);
1733       }
1734       break;
1735   }
1736   return ret;
1737
1738   /* ERRORS */
1739 out_flow_error:
1740   {
1741     GST_DEBUG_OBJECT (jitterbuffer,
1742         "refusing event, we have a downstream flow error: %s",
1743         gst_flow_get_name (priv->srcresult));
1744     JBUF_UNLOCK (priv);
1745     gst_event_unref (event);
1746     return FALSE;
1747   }
1748 out_eos:
1749   {
1750     GST_DEBUG_OBJECT (jitterbuffer, "refusing event, we are EOS");
1751     JBUF_UNLOCK (priv);
1752     gst_event_unref (event);
1753     return FALSE;
1754   }
1755 }
1756
1757 static gboolean
1758 gst_rtp_jitter_buffer_sink_rtcp_event (GstPad * pad, GstObject * parent,
1759     GstEvent * event)
1760 {
1761   gboolean ret = TRUE;
1762   GstRtpJitterBuffer *jitterbuffer;
1763
1764   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
1765
1766   GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
1767
1768   switch (GST_EVENT_TYPE (event)) {
1769     case GST_EVENT_FLUSH_START:
1770       gst_event_unref (event);
1771       break;
1772     case GST_EVENT_FLUSH_STOP:
1773       gst_event_unref (event);
1774       break;
1775     default:
1776       ret = gst_pad_event_default (pad, parent, event);
1777       break;
1778   }
1779
1780   return ret;
1781 }
1782
1783 /*
1784  * Must be called with JBUF_LOCK held, will release the LOCK when emiting the
1785  * signal. The function returns GST_FLOW_ERROR when a parsing error happened and
1786  * GST_FLOW_FLUSHING when the element is shutting down. On success
1787  * GST_FLOW_OK is returned.
1788  */
1789 static GstFlowReturn
1790 gst_rtp_jitter_buffer_get_clock_rate (GstRtpJitterBuffer * jitterbuffer,
1791     guint8 pt)
1792 {
1793   GValue ret = { 0 };
1794   GValue args[2] = { {0}, {0} };
1795   GstCaps *caps;
1796   gboolean res;
1797
1798   g_value_init (&args[0], GST_TYPE_ELEMENT);
1799   g_value_set_object (&args[0], jitterbuffer);
1800   g_value_init (&args[1], G_TYPE_UINT);
1801   g_value_set_uint (&args[1], pt);
1802
1803   g_value_init (&ret, GST_TYPE_CAPS);
1804   g_value_set_boxed (&ret, NULL);
1805
1806   JBUF_UNLOCK (jitterbuffer->priv);
1807   g_signal_emitv (args, gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP], 0,
1808       &ret);
1809   JBUF_LOCK_CHECK (jitterbuffer->priv, out_flushing);
1810
1811   g_value_unset (&args[0]);
1812   g_value_unset (&args[1]);
1813   caps = (GstCaps *) g_value_dup_boxed (&ret);
1814   g_value_unset (&ret);
1815   if (!caps)
1816     goto no_caps;
1817
1818   res = gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
1819   gst_caps_unref (caps);
1820
1821   if (G_UNLIKELY (!res))
1822     goto parse_failed;
1823
1824   return GST_FLOW_OK;
1825
1826   /* ERRORS */
1827 no_caps:
1828   {
1829     GST_DEBUG_OBJECT (jitterbuffer, "could not get caps");
1830     return GST_FLOW_ERROR;
1831   }
1832 out_flushing:
1833   {
1834     GST_DEBUG_OBJECT (jitterbuffer, "we are flushing");
1835     return GST_FLOW_FLUSHING;
1836   }
1837 parse_failed:
1838   {
1839     GST_DEBUG_OBJECT (jitterbuffer, "parse failed");
1840     return GST_FLOW_ERROR;
1841   }
1842 }
1843
1844 /* call with jbuf lock held */
1845 static GstMessage *
1846 check_buffering_percent (GstRtpJitterBuffer * jitterbuffer, gint percent)
1847 {
1848   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1849   GstMessage *message = NULL;
1850
1851   if (percent == -1)
1852     return NULL;
1853
1854   /* Post a buffering message */
1855   if (priv->last_percent != percent) {
1856     priv->last_percent = percent;
1857     message =
1858         gst_message_new_buffering (GST_OBJECT_CAST (jitterbuffer), percent);
1859     gst_message_set_buffering_stats (message, GST_BUFFERING_LIVE, -1, -1, -1);
1860   }
1861
1862   return message;
1863 }
1864
1865 static GstClockTime
1866 apply_offset (GstRtpJitterBuffer * jitterbuffer, GstClockTime timestamp)
1867 {
1868   GstRtpJitterBufferPrivate *priv;
1869
1870   priv = jitterbuffer->priv;
1871
1872   if (timestamp == -1)
1873     return -1;
1874
1875   /* apply the timestamp offset, this is used for inter stream sync */
1876   timestamp += priv->ts_offset;
1877   /* add the offset, this is used when buffering */
1878   timestamp += priv->out_offset;
1879
1880   return timestamp;
1881 }
1882
1883 static TimerData *
1884 find_timer (GstRtpJitterBuffer * jitterbuffer, TimerType type, guint16 seqnum)
1885 {
1886   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1887   TimerData *timer = NULL;
1888   gint i, len;
1889
1890   len = priv->timers->len;
1891   for (i = 0; i < len; i++) {
1892     TimerData *test = &g_array_index (priv->timers, TimerData, i);
1893     if (test->seqnum == seqnum && test->type == type) {
1894       timer = test;
1895       break;
1896     }
1897   }
1898   return timer;
1899 }
1900
1901 static void
1902 unschedule_current_timer (GstRtpJitterBuffer * jitterbuffer)
1903 {
1904   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1905
1906   if (priv->clock_id) {
1907     GST_DEBUG_OBJECT (jitterbuffer, "unschedule current timer");
1908     gst_clock_id_unschedule (priv->clock_id);
1909     priv->clock_id = NULL;
1910   }
1911 }
1912
1913 static GstClockTime
1914 get_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer)
1915 {
1916   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1917   GstClockTime test_timeout;
1918
1919   if ((test_timeout = timer->timeout) == -1)
1920     return -1;
1921
1922   if (timer->type != TIMER_TYPE_EXPECTED) {
1923     /* add our latency and offset to get output times. */
1924     test_timeout = apply_offset (jitterbuffer, test_timeout);
1925     test_timeout += priv->latency_ns;
1926   }
1927   return test_timeout;
1928 }
1929
1930 static void
1931 recalculate_timer (GstRtpJitterBuffer * jitterbuffer, TimerData * timer)
1932 {
1933   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1934
1935   if (priv->clock_id) {
1936     GstClockTime timeout = get_timeout (jitterbuffer, timer);
1937
1938     GST_DEBUG ("%" GST_TIME_FORMAT " <> %" GST_TIME_FORMAT,
1939         GST_TIME_ARGS (timeout), GST_TIME_ARGS (priv->timer_timeout));
1940
1941     if (timeout == -1 || timeout < priv->timer_timeout)
1942       unschedule_current_timer (jitterbuffer);
1943   }
1944 }
1945
1946 static TimerData *
1947 add_timer (GstRtpJitterBuffer * jitterbuffer, TimerType type,
1948     guint16 seqnum, guint num, GstClockTime timeout, GstClockTime delay,
1949     GstClockTime duration)
1950 {
1951   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1952   TimerData *timer;
1953   gint len;
1954
1955   GST_DEBUG_OBJECT (jitterbuffer,
1956       "add timer %d for seqnum %d to %" GST_TIME_FORMAT ", delay %"
1957       GST_TIME_FORMAT, type, seqnum, GST_TIME_ARGS (timeout),
1958       GST_TIME_ARGS (delay));
1959
1960   len = priv->timers->len;
1961   g_array_set_size (priv->timers, len + 1);
1962   timer = &g_array_index (priv->timers, TimerData, len);
1963   timer->idx = len;
1964   timer->type = type;
1965   timer->seqnum = seqnum;
1966   timer->num = num;
1967   timer->timeout = timeout + delay;
1968   timer->duration = duration;
1969   if (type == TIMER_TYPE_EXPECTED) {
1970     timer->rtx_base = timeout;
1971     timer->rtx_delay = delay;
1972     timer->rtx_retry = 0;
1973   }
1974   timer->num_rtx_retry = 0;
1975   recalculate_timer (jitterbuffer, timer);
1976   JBUF_SIGNAL_TIMER (priv);
1977
1978   return timer;
1979 }
1980
1981 static void
1982 reschedule_timer (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
1983     guint16 seqnum, GstClockTime timeout, GstClockTime delay, gboolean reset)
1984 {
1985   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1986   gboolean seqchange, timechange;
1987   guint16 oldseq;
1988
1989   seqchange = timer->seqnum != seqnum;
1990   timechange = timer->timeout != timeout;
1991
1992   if (!seqchange && !timechange)
1993     return;
1994
1995   oldseq = timer->seqnum;
1996
1997   GST_DEBUG_OBJECT (jitterbuffer,
1998       "replace timer for seqnum %d->%d to %" GST_TIME_FORMAT,
1999       oldseq, seqnum, GST_TIME_ARGS (timeout + delay));
2000
2001   timer->timeout = timeout + delay;
2002   timer->seqnum = seqnum;
2003   if (reset) {
2004     timer->rtx_base = timeout;
2005     timer->rtx_delay = delay;
2006     timer->rtx_retry = 0;
2007   }
2008   if (seqchange)
2009     timer->num_rtx_retry = 0;
2010
2011   if (priv->clock_id) {
2012     /* we changed the seqnum and there is a timer currently waiting with this
2013      * seqnum, unschedule it */
2014     if (seqchange && priv->timer_seqnum == oldseq)
2015       unschedule_current_timer (jitterbuffer);
2016     /* we changed the time, check if it is earlier than what we are waiting
2017      * for and unschedule if so */
2018     else if (timechange)
2019       recalculate_timer (jitterbuffer, timer);
2020   }
2021 }
2022
2023 static TimerData *
2024 set_timer (GstRtpJitterBuffer * jitterbuffer, TimerType type,
2025     guint16 seqnum, GstClockTime timeout)
2026 {
2027   TimerData *timer;
2028
2029   /* find the seqnum timer */
2030   timer = find_timer (jitterbuffer, type, seqnum);
2031   if (timer == NULL) {
2032     timer = add_timer (jitterbuffer, type, seqnum, 0, timeout, 0, -1);
2033   } else {
2034     reschedule_timer (jitterbuffer, timer, seqnum, timeout, 0, FALSE);
2035   }
2036   return timer;
2037 }
2038
2039 static void
2040 remove_timer (GstRtpJitterBuffer * jitterbuffer, TimerData * timer)
2041 {
2042   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2043   guint idx;
2044
2045   if (priv->clock_id && priv->timer_seqnum == timer->seqnum)
2046     unschedule_current_timer (jitterbuffer);
2047
2048   idx = timer->idx;
2049   GST_DEBUG_OBJECT (jitterbuffer, "removed index %d", idx);
2050   g_array_remove_index_fast (priv->timers, idx);
2051   timer->idx = idx;
2052 }
2053
2054 static void
2055 remove_all_timers (GstRtpJitterBuffer * jitterbuffer)
2056 {
2057   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2058   GST_DEBUG_OBJECT (jitterbuffer, "removed all timers");
2059   g_array_set_size (priv->timers, 0);
2060   unschedule_current_timer (jitterbuffer);
2061 }
2062
2063 /* get the extra delay to wait before sending RTX */
2064 static GstClockTime
2065 get_rtx_delay (GstRtpJitterBufferPrivate * priv)
2066 {
2067   GstClockTime delay;
2068
2069   if (priv->rtx_delay == -1) {
2070     if (priv->avg_jitter == 0 && priv->packet_spacing == 0) {
2071       delay = DEFAULT_AUTO_RTX_DELAY;
2072     } else {
2073       /* jitter is in nanoseconds, maximum of 2x jitter and half the
2074        * packet spacing is a good margin */
2075       delay = MAX (priv->avg_jitter * 2, priv->packet_spacing / 2);
2076     }
2077   } else {
2078     delay = priv->rtx_delay * GST_MSECOND;
2079   }
2080   if (priv->rtx_min_delay > 0)
2081     delay = MAX (delay, priv->rtx_min_delay * GST_MSECOND);
2082
2083   return delay;
2084 }
2085
2086 /* we just received a packet with seqnum and dts.
2087  *
2088  * First check for old seqnum that we are still expecting. If the gap with the
2089  * current seqnum is too big, unschedule the timeouts.
2090  *
2091  * If we have a valid packet spacing estimate we can set a timer for when we
2092  * should receive the next packet.
2093  * If we don't have a valid estimate, we remove any timer we might have
2094  * had for this packet.
2095  */
2096 static void
2097 update_timers (GstRtpJitterBuffer * jitterbuffer, guint16 seqnum,
2098     GstClockTime dts, gboolean do_next_seqnum)
2099 {
2100   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2101   TimerData *timer = NULL;
2102   gint i, len;
2103
2104   /* go through all timers and unschedule the ones with a large gap, also find
2105    * the timer for the seqnum */
2106   len = priv->timers->len;
2107   for (i = 0; i < len; i++) {
2108     TimerData *test = &g_array_index (priv->timers, TimerData, i);
2109     gint gap;
2110
2111     gap = gst_rtp_buffer_compare_seqnum (test->seqnum, seqnum);
2112
2113     GST_DEBUG_OBJECT (jitterbuffer, "%d, %d, #%d<->#%d gap %d", i,
2114         test->type, test->seqnum, seqnum, gap);
2115
2116     if (gap == 0) {
2117       GST_DEBUG ("found timer for current seqnum");
2118       /* the timer for the current seqnum */
2119       timer = test;
2120       /* when no retransmission, we can stop now, we only need to find the
2121        * timer for the current seqnum */
2122       if (!priv->do_retransmission)
2123         break;
2124     } else if (gap > priv->rtx_delay_reorder) {
2125       /* max gap, we exceeded the max reorder distance and we don't expect the
2126        * missing packet to be this reordered */
2127       if (test->num_rtx_retry == 0 && test->type == TIMER_TYPE_EXPECTED)
2128         reschedule_timer (jitterbuffer, test, test->seqnum, -1, 0, FALSE);
2129     }
2130   }
2131
2132   do_next_seqnum = do_next_seqnum && priv->packet_spacing > 0
2133       && priv->do_retransmission && priv->rtx_next_seqnum;
2134
2135   if (timer && timer->type != TIMER_TYPE_DEADLINE) {
2136     if (timer->num_rtx_retry > 0) {
2137       GstClockTime rtx_last, delay;
2138
2139       /* we scheduled a retry for this packet and now we have it */
2140       priv->num_rtx_success++;
2141       /* all the previous retry attempts failed */
2142       priv->num_rtx_failed += timer->num_rtx_retry - 1;
2143       /* number of retries before receiving the packet */
2144       if (priv->avg_rtx_num == 0.0)
2145         priv->avg_rtx_num = timer->num_rtx_retry;
2146       else
2147         priv->avg_rtx_num = (timer->num_rtx_retry + 7 * priv->avg_rtx_num) / 8;
2148       /* calculate the delay between retransmission request and receiving this
2149        * packet, start with when we scheduled this timeout last */
2150       rtx_last = timer->rtx_last;
2151       if (dts != GST_CLOCK_TIME_NONE && dts > rtx_last) {
2152         /* we have a valid delay if this packet arrived after we scheduled the
2153          * request */
2154         delay = dts - rtx_last;
2155         if (priv->avg_rtx_rtt == 0)
2156           priv->avg_rtx_rtt = delay;
2157         else
2158           priv->avg_rtx_rtt = (delay + 7 * priv->avg_rtx_rtt) / 8;
2159       } else
2160         delay = 0;
2161
2162       GST_LOG_OBJECT (jitterbuffer,
2163           "RTX success %" G_GUINT64_FORMAT ", failed %" G_GUINT64_FORMAT
2164           ", requests %" G_GUINT64_FORMAT ", dups %" G_GUINT64_FORMAT
2165           ", avg-num %g, delay %" GST_TIME_FORMAT ", avg-rtt %" GST_TIME_FORMAT,
2166           priv->num_rtx_success, priv->num_rtx_failed, priv->num_rtx_requests,
2167           priv->num_duplicates, priv->avg_rtx_num, GST_TIME_ARGS (delay),
2168           GST_TIME_ARGS (priv->avg_rtx_rtt));
2169
2170       /* don't try to estimate the next seqnum because this is a retransmitted
2171        * packet and it probably did not arrive with the expected packet
2172        * spacing. */
2173       do_next_seqnum = FALSE;
2174     }
2175   }
2176
2177   if (do_next_seqnum && dts != GST_CLOCK_TIME_NONE) {
2178     GstClockTime expected, delay;
2179
2180     /* calculate expected arrival time of the next seqnum */
2181     expected = dts + priv->packet_spacing;
2182
2183     delay = get_rtx_delay (priv);
2184
2185     /* and update/install timer for next seqnum */
2186     if (timer) {
2187       reschedule_timer (jitterbuffer, timer, priv->next_in_seqnum, expected,
2188           delay, TRUE);
2189     } else {
2190       add_timer (jitterbuffer, TIMER_TYPE_EXPECTED, priv->next_in_seqnum, 0,
2191           expected, delay, priv->packet_spacing);
2192     }
2193   } else if (timer && timer->type != TIMER_TYPE_DEADLINE) {
2194     /* if we had a timer, remove it, we don't know when to expect the next
2195      * packet. */
2196     remove_timer (jitterbuffer, timer);
2197   }
2198 }
2199
2200 static void
2201 calculate_packet_spacing (GstRtpJitterBuffer * jitterbuffer, guint32 rtptime,
2202     GstClockTime dts)
2203 {
2204   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2205
2206   /* we need consecutive seqnums with a different
2207    * rtptime to estimate the packet spacing. */
2208   if (priv->ips_rtptime != rtptime) {
2209     /* rtptime changed, check dts diff */
2210     if (priv->ips_dts != -1 && dts != -1 && dts > priv->ips_dts) {
2211       GstClockTime new_packet_spacing = dts - priv->ips_dts;
2212       GstClockTime old_packet_spacing = priv->packet_spacing;
2213
2214       /* Biased towards bigger packet spacings to prevent
2215        * too many unneeded retransmission requests for next
2216        * packets that just arrive a little later than we would
2217        * expect */
2218       if (old_packet_spacing > new_packet_spacing)
2219         priv->packet_spacing =
2220             (new_packet_spacing + 3 * old_packet_spacing) / 4;
2221       else if (old_packet_spacing > 0)
2222         priv->packet_spacing =
2223             (3 * new_packet_spacing + old_packet_spacing) / 4;
2224       else
2225         priv->packet_spacing = new_packet_spacing;
2226
2227       GST_DEBUG_OBJECT (jitterbuffer,
2228           "new packet spacing %" GST_TIME_FORMAT
2229           " old packet spacing %" GST_TIME_FORMAT
2230           " combined to %" GST_TIME_FORMAT,
2231           GST_TIME_ARGS (new_packet_spacing),
2232           GST_TIME_ARGS (old_packet_spacing),
2233           GST_TIME_ARGS (priv->packet_spacing));
2234     }
2235     priv->ips_rtptime = rtptime;
2236     priv->ips_dts = dts;
2237   }
2238 }
2239
2240 static void
2241 calculate_expected (GstRtpJitterBuffer * jitterbuffer, guint32 expected,
2242     guint16 seqnum, GstClockTime dts, gint gap)
2243 {
2244   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2245   GstClockTime total_duration, duration, expected_dts;
2246   TimerType type;
2247
2248   GST_DEBUG_OBJECT (jitterbuffer,
2249       "dts %" GST_TIME_FORMAT ", last %" GST_TIME_FORMAT,
2250       GST_TIME_ARGS (dts), GST_TIME_ARGS (priv->last_in_dts));
2251
2252   if (dts == GST_CLOCK_TIME_NONE) {
2253     GST_WARNING_OBJECT (jitterbuffer, "Have no DTS");
2254     return;
2255   }
2256
2257   /* the total duration spanned by the missing packets */
2258   if (dts >= priv->last_in_dts)
2259     total_duration = dts - priv->last_in_dts;
2260   else
2261     total_duration = 0;
2262
2263   /* interpolate between the current time and the last time based on
2264    * number of packets we are missing, this is the estimated duration
2265    * for the missing packet based on equidistant packet spacing. */
2266   duration = total_duration / (gap + 1);
2267
2268   GST_DEBUG_OBJECT (jitterbuffer, "duration %" GST_TIME_FORMAT,
2269       GST_TIME_ARGS (duration));
2270
2271   if (total_duration > priv->latency_ns) {
2272     GstClockTime gap_time;
2273     guint lost_packets;
2274
2275     if (duration > 0) {
2276       GstClockTime gap_dur = gap * duration;
2277       if (gap_dur > priv->latency_ns)
2278         gap_time = gap_dur - priv->latency_ns;
2279       else
2280         gap_time = 0;
2281       lost_packets = gap_time / duration;
2282     } else {
2283       gap_time = total_duration - priv->latency_ns;
2284       lost_packets = gap;
2285     }
2286
2287     /* too many lost packets, some of the missing packets are already
2288      * too late and we can generate lost packet events for them. */
2289     GST_DEBUG_OBJECT (jitterbuffer,
2290         "lost packets (%d, #%d->#%d) duration too large %" GST_TIME_FORMAT
2291         " > %" GST_TIME_FORMAT ", consider %u lost (%" GST_TIME_FORMAT ")",
2292         gap, expected, seqnum, GST_TIME_ARGS (total_duration),
2293         GST_TIME_ARGS (priv->latency_ns), lost_packets,
2294         GST_TIME_ARGS (gap_time));
2295
2296     /* this timer will fire immediately and the lost event will be pushed from
2297      * the timer thread */
2298     if (lost_packets > 0) {
2299       add_timer (jitterbuffer, TIMER_TYPE_LOST, expected, lost_packets,
2300           priv->last_in_dts + duration, 0, gap_time);
2301       expected += lost_packets;
2302       priv->last_in_dts += gap_time;
2303     }
2304   }
2305
2306   expected_dts = priv->last_in_dts + duration;
2307
2308   if (priv->do_retransmission) {
2309     TimerData *timer;
2310
2311     type = TIMER_TYPE_EXPECTED;
2312     /* if we had a timer for the first missing packet, update it. */
2313     if ((timer = find_timer (jitterbuffer, type, expected))) {
2314       GstClockTime timeout = timer->timeout;
2315
2316       timer->duration = duration;
2317       if (timeout > (expected_dts + timer->rtx_retry)) {
2318         GstClockTime delay = timeout - expected_dts - timer->rtx_retry;
2319         reschedule_timer (jitterbuffer, timer, timer->seqnum, expected_dts,
2320             delay, TRUE);
2321       }
2322       expected++;
2323       expected_dts += duration;
2324     }
2325   } else {
2326     type = TIMER_TYPE_LOST;
2327   }
2328
2329   while (gst_rtp_buffer_compare_seqnum (expected, seqnum) > 0) {
2330     add_timer (jitterbuffer, type, expected, 0, expected_dts, 0, duration);
2331     expected_dts += duration;
2332     expected++;
2333   }
2334 }
2335
2336 static void
2337 calculate_jitter (GstRtpJitterBuffer * jitterbuffer, GstClockTime dts,
2338     guint rtptime)
2339 {
2340   gint32 rtpdiff;
2341   GstClockTimeDiff dtsdiff, rtpdiffns, diff;
2342   GstRtpJitterBufferPrivate *priv;
2343
2344   priv = jitterbuffer->priv;
2345
2346   if (G_UNLIKELY (dts == GST_CLOCK_TIME_NONE) || priv->clock_rate <= 0)
2347     goto no_time;
2348
2349   if (priv->last_dts != -1)
2350     dtsdiff = dts - priv->last_dts;
2351   else
2352     dtsdiff = 0;
2353
2354   if (priv->last_rtptime != -1)
2355     rtpdiff = rtptime - (guint32) priv->last_rtptime;
2356   else
2357     rtpdiff = 0;
2358
2359   priv->last_dts = dts;
2360   priv->last_rtptime = rtptime;
2361
2362   if (rtpdiff > 0)
2363     rtpdiffns =
2364         gst_util_uint64_scale_int (rtpdiff, GST_SECOND, priv->clock_rate);
2365   else
2366     rtpdiffns =
2367         -gst_util_uint64_scale_int (-rtpdiff, GST_SECOND, priv->clock_rate);
2368
2369   diff = ABS (dtsdiff - rtpdiffns);
2370
2371   /* jitter is stored in nanoseconds */
2372   priv->avg_jitter = (diff + (15 * priv->avg_jitter)) >> 4;
2373
2374   GST_LOG_OBJECT (jitterbuffer,
2375       "dtsdiff %" GST_TIME_FORMAT " rtptime %" GST_TIME_FORMAT
2376       ", clock-rate %d, diff %" GST_TIME_FORMAT ", jitter: %" GST_TIME_FORMAT,
2377       GST_TIME_ARGS (dtsdiff), GST_TIME_ARGS (rtpdiffns), priv->clock_rate,
2378       GST_TIME_ARGS (diff), GST_TIME_ARGS (priv->avg_jitter));
2379
2380   return;
2381
2382   /* ERRORS */
2383 no_time:
2384   {
2385     GST_DEBUG_OBJECT (jitterbuffer,
2386         "no dts or no clock-rate, can't calculate jitter");
2387     return;
2388   }
2389 }
2390
2391 static gint
2392 compare_buffer_seqnum (GstBuffer * a, GstBuffer * b, gpointer user_data)
2393 {
2394   GstRTPBuffer rtp_a = GST_RTP_BUFFER_INIT;
2395   GstRTPBuffer rtp_b = GST_RTP_BUFFER_INIT;
2396   guint seq_a, seq_b;
2397
2398   gst_rtp_buffer_map (a, GST_MAP_READ, &rtp_a);
2399   seq_a = gst_rtp_buffer_get_seq (&rtp_a);
2400   gst_rtp_buffer_unmap (&rtp_a);
2401
2402   gst_rtp_buffer_map (b, GST_MAP_READ, &rtp_b);
2403   seq_b = gst_rtp_buffer_get_seq (&rtp_b);
2404   gst_rtp_buffer_unmap (&rtp_b);
2405
2406   return gst_rtp_buffer_compare_seqnum (seq_b, seq_a);
2407 }
2408
2409 static gboolean
2410 handle_big_gap_buffer (GstRtpJitterBuffer * jitterbuffer, gboolean future,
2411     GstBuffer * buffer, guint8 pt, guint16 seqnum, gint gap, guint max_dropout,
2412     guint max_misorder)
2413 {
2414   GstRtpJitterBufferPrivate *priv;
2415   guint gap_packets_length;
2416   gboolean reset = FALSE;
2417
2418   priv = jitterbuffer->priv;
2419
2420   if ((gap_packets_length = g_queue_get_length (&priv->gap_packets)) > 0) {
2421     GList *l;
2422     guint32 prev_gap_seq = -1;
2423     gboolean all_consecutive = TRUE;
2424
2425     g_queue_insert_sorted (&priv->gap_packets, buffer,
2426         (GCompareDataFunc) compare_buffer_seqnum, NULL);
2427
2428     for (l = priv->gap_packets.head; l; l = l->next) {
2429       GstBuffer *gap_buffer = l->data;
2430       GstRTPBuffer gap_rtp = GST_RTP_BUFFER_INIT;
2431       guint32 gap_seq;
2432
2433       gst_rtp_buffer_map (gap_buffer, GST_MAP_READ, &gap_rtp);
2434
2435       all_consecutive = (gst_rtp_buffer_get_payload_type (&gap_rtp) == pt);
2436
2437       gap_seq = gst_rtp_buffer_get_seq (&gap_rtp);
2438       if (prev_gap_seq == -1)
2439         prev_gap_seq = gap_seq;
2440       else if (gst_rtp_buffer_compare_seqnum (gap_seq, prev_gap_seq) != -1)
2441         all_consecutive = FALSE;
2442       else
2443         prev_gap_seq = gap_seq;
2444
2445       gst_rtp_buffer_unmap (&gap_rtp);
2446       if (!all_consecutive)
2447         break;
2448     }
2449
2450     if (all_consecutive && gap_packets_length > 3) {
2451       GST_DEBUG_OBJECT (jitterbuffer,
2452           "buffer too %s %d < %d, got 5 consecutive ones - reset",
2453           (future ? "new" : "old"), gap,
2454           (future ? max_dropout : -max_misorder));
2455       reset = TRUE;
2456     } else if (!all_consecutive) {
2457       g_queue_foreach (&priv->gap_packets, (GFunc) gst_buffer_unref, NULL);
2458       g_queue_clear (&priv->gap_packets);
2459       GST_DEBUG_OBJECT (jitterbuffer,
2460           "buffer too %s %d < %d, got no 5 consecutive ones - dropping",
2461           (future ? "new" : "old"), gap,
2462           (future ? max_dropout : -max_misorder));
2463       buffer = NULL;
2464     } else {
2465       GST_DEBUG_OBJECT (jitterbuffer,
2466           "buffer too %s %d < %d, got %u consecutive ones - waiting",
2467           (future ? "new" : "old"), gap,
2468           (future ? max_dropout : -max_misorder), gap_packets_length + 1);
2469       buffer = NULL;
2470     }
2471   } else {
2472     GST_DEBUG_OBJECT (jitterbuffer,
2473         "buffer too %s %d < %d, first one - waiting", (future ? "new" : "old"),
2474         gap, -max_misorder);
2475     g_queue_push_tail (&priv->gap_packets, buffer);
2476     buffer = NULL;
2477   }
2478
2479   return reset;
2480 }
2481
2482 static GstClockTime
2483 get_current_running_time (GstRtpJitterBuffer * jitterbuffer)
2484 {
2485   GstClock *clock = gst_element_get_clock (GST_ELEMENT_CAST (jitterbuffer));
2486   GstClockTime running_time = GST_CLOCK_TIME_NONE;
2487
2488   if (clock) {
2489     GstClockTime base_time =
2490         gst_element_get_base_time (GST_ELEMENT_CAST (jitterbuffer));
2491     GstClockTime clock_time = gst_clock_get_time (clock);
2492
2493     if (clock_time > base_time)
2494       running_time = clock_time - base_time;
2495     else
2496       running_time = 0;
2497
2498     gst_object_unref (clock);
2499   }
2500
2501   return running_time;
2502 }
2503
2504 static GstFlowReturn
2505 gst_rtp_jitter_buffer_chain (GstPad * pad, GstObject * parent,
2506     GstBuffer * buffer)
2507 {
2508   GstRtpJitterBuffer *jitterbuffer;
2509   GstRtpJitterBufferPrivate *priv;
2510   guint16 seqnum;
2511   guint32 expected, rtptime;
2512   GstFlowReturn ret = GST_FLOW_OK;
2513   GstClockTime dts, pts;
2514   guint64 latency_ts;
2515   gboolean head;
2516   gint percent = -1;
2517   guint8 pt;
2518   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
2519   gboolean do_next_seqnum = FALSE;
2520   RTPJitterBufferItem *item;
2521   GstMessage *msg = NULL;
2522   gboolean estimated_dts = FALSE;
2523   guint32 packet_rate, max_dropout, max_misorder;
2524
2525   jitterbuffer = GST_RTP_JITTER_BUFFER_CAST (parent);
2526
2527   priv = jitterbuffer->priv;
2528
2529   if (G_UNLIKELY (!gst_rtp_buffer_map (buffer, GST_MAP_READ, &rtp)))
2530     goto invalid_buffer;
2531
2532   pt = gst_rtp_buffer_get_payload_type (&rtp);
2533   seqnum = gst_rtp_buffer_get_seq (&rtp);
2534   rtptime = gst_rtp_buffer_get_timestamp (&rtp);
2535   gst_rtp_buffer_unmap (&rtp);
2536
2537   /* make sure we have PTS and DTS set */
2538   pts = GST_BUFFER_PTS (buffer);
2539   dts = GST_BUFFER_DTS (buffer);
2540   if (dts == -1)
2541     dts = pts;
2542   else if (pts == -1)
2543     pts = dts;
2544
2545   if (dts == -1) {
2546     /* If we have no DTS here, i.e. no capture time, get one from the
2547      * clock now to have something to calculate with in the future. */
2548     dts = get_current_running_time (jitterbuffer);
2549     pts = dts;
2550
2551     /* Remember that we estimated the DTS if we are running already
2552      * and this is not our first packet (or first packet after a reset).
2553      * If it's the first packet, we somehow must generate a timestamp for
2554      * everything, otherwise we can't calculate any times
2555      */
2556     estimated_dts = (priv->next_in_seqnum != -1);
2557   } else {
2558     /* take the DTS of the buffer. This is the time when the packet was
2559      * received and is used to calculate jitter and clock skew. We will adjust
2560      * this DTS with the smoothed value after processing it in the
2561      * jitterbuffer and assign it as the PTS. */
2562     /* bring to running time */
2563     dts = gst_segment_to_running_time (&priv->segment, GST_FORMAT_TIME, dts);
2564   }
2565
2566   GST_DEBUG_OBJECT (jitterbuffer,
2567       "Received packet #%d at time %" GST_TIME_FORMAT ", discont %d", seqnum,
2568       GST_TIME_ARGS (dts), GST_BUFFER_IS_DISCONT (buffer));
2569
2570   JBUF_LOCK_CHECK (priv, out_flushing);
2571
2572   if (G_UNLIKELY (priv->last_pt != pt)) {
2573     GstCaps *caps;
2574
2575     GST_DEBUG_OBJECT (jitterbuffer, "pt changed from %u to %u", priv->last_pt,
2576         pt);
2577
2578     priv->last_pt = pt;
2579     /* reset clock-rate so that we get a new one */
2580     priv->clock_rate = -1;
2581
2582     /* Try to get the clock-rate from the caps first if we can. If there are no
2583      * caps we must fire the signal to get the clock-rate. */
2584     if ((caps = gst_pad_get_current_caps (pad))) {
2585       gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
2586       gst_caps_unref (caps);
2587     }
2588   }
2589
2590   if (G_UNLIKELY (priv->clock_rate == -1)) {
2591     /* no clock rate given on the caps, try to get one with the signal */
2592     if (gst_rtp_jitter_buffer_get_clock_rate (jitterbuffer,
2593             pt) == GST_FLOW_FLUSHING)
2594       goto out_flushing;
2595
2596     if (G_UNLIKELY (priv->clock_rate == -1))
2597       goto no_clock_rate;
2598   }
2599
2600   /* don't accept more data on EOS */
2601   if (G_UNLIKELY (priv->eos))
2602     goto have_eos;
2603
2604   calculate_jitter (jitterbuffer, dts, rtptime);
2605
2606   if (priv->seqnum_base != -1) {
2607     gint gap;
2608
2609     gap = gst_rtp_buffer_compare_seqnum (priv->seqnum_base, seqnum);
2610
2611     if (gap < 0) {
2612       GST_DEBUG_OBJECT (jitterbuffer,
2613           "packet seqnum #%d before seqnum-base #%d", seqnum,
2614           priv->seqnum_base);
2615       gst_buffer_unref (buffer);
2616       ret = GST_FLOW_OK;
2617       goto finished;
2618     } else if (gap > 16384) {
2619       /* From now on don't compare against the seqnum base anymore as
2620        * at some point in the future we will wrap around and also that
2621        * much reordering is very unlikely */
2622       priv->seqnum_base = -1;
2623     }
2624   }
2625
2626   expected = priv->next_in_seqnum;
2627
2628   packet_rate =
2629       gst_rtp_packet_rate_ctx_update (&priv->packet_rate_ctx, seqnum, rtptime);
2630   max_dropout =
2631       gst_rtp_packet_rate_ctx_get_max_dropout (&priv->packet_rate_ctx,
2632       priv->max_dropout_time);
2633   max_misorder =
2634       gst_rtp_packet_rate_ctx_get_max_misorder (&priv->packet_rate_ctx,
2635       priv->max_misorder_time);
2636   GST_TRACE_OBJECT (jitterbuffer,
2637       "packet_rate: %d, max_dropout: %d, max_misorder: %d", packet_rate,
2638       max_dropout, max_misorder);
2639
2640   /* now check against our expected seqnum */
2641   if (G_LIKELY (expected != -1)) {
2642     gint gap;
2643
2644     /* now calculate gap */
2645     gap = gst_rtp_buffer_compare_seqnum (expected, seqnum);
2646
2647     GST_DEBUG_OBJECT (jitterbuffer, "expected #%d, got #%d, gap of %d",
2648         expected, seqnum, gap);
2649
2650     if (G_LIKELY (gap == 0)) {
2651       /* packet is expected */
2652       calculate_packet_spacing (jitterbuffer, rtptime, dts);
2653       do_next_seqnum = TRUE;
2654     } else {
2655       gboolean reset = FALSE;
2656
2657       if (gap < 0) {
2658         /* we received an old packet */
2659         if (G_UNLIKELY (gap != -1 && gap < -max_misorder)) {
2660           reset =
2661               handle_big_gap_buffer (jitterbuffer, FALSE, buffer, pt, seqnum,
2662               gap, max_dropout, max_misorder);
2663           buffer = NULL;
2664         } else {
2665           GST_DEBUG_OBJECT (jitterbuffer, "old packet received");
2666         }
2667       } else {
2668         /* new packet, we are missing some packets */
2669         if (G_UNLIKELY (priv->timers->len >= max_dropout)) {
2670           /* If we have timers for more than RTP_MAX_DROPOUT packets
2671            * pending this means that we have a huge gap overall. We can
2672            * reset the jitterbuffer at this point because there's
2673            * just too much data missing to be able to do anything
2674            * sensible with the past data. Just try again from the
2675            * next packet */
2676           GST_WARNING_OBJECT (jitterbuffer,
2677               "%d pending timers > %d - resetting", priv->timers->len,
2678               max_dropout);
2679           reset = TRUE;
2680           gst_buffer_unref (buffer);
2681           buffer = NULL;
2682         } else if (G_UNLIKELY (gap >= max_dropout)) {
2683           reset =
2684               handle_big_gap_buffer (jitterbuffer, TRUE, buffer, pt, seqnum,
2685               gap, max_dropout, max_misorder);
2686           buffer = NULL;
2687         } else {
2688           GST_DEBUG_OBJECT (jitterbuffer, "%d missing packets", gap);
2689           /* fill in the gap with EXPECTED timers */
2690           calculate_expected (jitterbuffer, expected, seqnum, dts, gap);
2691
2692           do_next_seqnum = TRUE;
2693         }
2694       }
2695       if (G_UNLIKELY (reset)) {
2696         GList *events = NULL, *l;
2697         GList *buffers;
2698
2699         GST_DEBUG_OBJECT (jitterbuffer, "flush and reset jitterbuffer");
2700         rtp_jitter_buffer_flush (priv->jbuf,
2701             (GFunc) free_item_and_retain_events, &events);
2702         rtp_jitter_buffer_reset_skew (priv->jbuf);
2703         remove_all_timers (jitterbuffer);
2704         priv->discont = TRUE;
2705         priv->last_popped_seqnum = -1;
2706
2707         if (priv->gap_packets.head) {
2708           GstBuffer *gap_buffer = priv->gap_packets.head->data;
2709           GstRTPBuffer gap_rtp = GST_RTP_BUFFER_INIT;
2710
2711           gst_rtp_buffer_map (gap_buffer, GST_MAP_READ, &gap_rtp);
2712           priv->next_seqnum = gst_rtp_buffer_get_seq (&gap_rtp);
2713           gst_rtp_buffer_unmap (&gap_rtp);
2714         } else {
2715           priv->next_seqnum = seqnum;
2716         }
2717
2718         priv->last_in_dts = -1;
2719         priv->next_in_seqnum = -1;
2720
2721         /* Insert all sticky events again in order, otherwise we would
2722          * potentially loose STREAM_START, CAPS or SEGMENT events
2723          */
2724         events = g_list_reverse (events);
2725         for (l = events; l; l = l->next) {
2726           RTPJitterBufferItem *item;
2727
2728           item = alloc_item (l->data, ITEM_TYPE_EVENT, -1, -1, -1, 0, -1);
2729           rtp_jitter_buffer_insert (priv->jbuf, item, &head, NULL, -1);
2730         }
2731         g_list_free (events);
2732
2733         JBUF_SIGNAL_EVENT (priv);
2734
2735         /* reset spacing estimation when gap */
2736         priv->ips_rtptime = -1;
2737         priv->ips_dts = GST_CLOCK_TIME_NONE;
2738
2739         buffers = g_list_copy (priv->gap_packets.head);
2740         g_queue_clear (&priv->gap_packets);
2741
2742         priv->ips_rtptime = -1;
2743         priv->ips_dts = GST_CLOCK_TIME_NONE;
2744         JBUF_UNLOCK (jitterbuffer->priv);
2745
2746         for (l = buffers; l; l = l->next) {
2747           ret = gst_rtp_jitter_buffer_chain (pad, parent, l->data);
2748           l->data = NULL;
2749           if (ret != GST_FLOW_OK)
2750             break;
2751         }
2752         for (; l; l = l->next)
2753           gst_buffer_unref (l->data);
2754         g_list_free (buffers);
2755
2756         return ret;
2757       }
2758       /* reset spacing estimation when gap */
2759       priv->ips_rtptime = -1;
2760       priv->ips_dts = GST_CLOCK_TIME_NONE;
2761     }
2762   } else {
2763     GST_DEBUG_OBJECT (jitterbuffer, "First buffer #%d", seqnum);
2764
2765     /* we don't know what the next_in_seqnum should be, wait for the last
2766      * possible moment to push this buffer, maybe we get an earlier seqnum
2767      * while we wait */
2768     set_timer (jitterbuffer, TIMER_TYPE_DEADLINE, seqnum, dts);
2769     do_next_seqnum = TRUE;
2770     /* take rtptime and dts to calculate packet spacing */
2771     priv->ips_rtptime = rtptime;
2772     priv->ips_dts = dts;
2773   }
2774
2775   /* We had no huge gap, let's drop all the gap packets */
2776   if (buffer != NULL) {
2777     GST_DEBUG_OBJECT (jitterbuffer, "Clearing gap packets");
2778     g_queue_foreach (&priv->gap_packets, (GFunc) gst_buffer_unref, NULL);
2779     g_queue_clear (&priv->gap_packets);
2780   } else {
2781     GST_DEBUG_OBJECT (jitterbuffer,
2782         "Had big gap, waiting for more consecutive packets");
2783     JBUF_UNLOCK (jitterbuffer->priv);
2784     return GST_FLOW_OK;
2785   }
2786
2787   if (do_next_seqnum) {
2788     priv->last_in_dts = dts;
2789     priv->next_in_seqnum = (seqnum + 1) & 0xffff;
2790   }
2791
2792   /* let's check if this buffer is too late, we can only accept packets with
2793    * bigger seqnum than the one we last pushed. */
2794   if (G_LIKELY (priv->last_popped_seqnum != -1)) {
2795     gint gap;
2796
2797     gap = gst_rtp_buffer_compare_seqnum (priv->last_popped_seqnum, seqnum);
2798
2799     /* priv->last_popped_seqnum >= seqnum, we're too late. */
2800     if (G_UNLIKELY (gap <= 0))
2801       goto too_late;
2802   }
2803
2804   /* let's drop oldest packet if the queue is already full and drop-on-latency
2805    * is set. We can only do this when there actually is a latency. When no
2806    * latency is set, we just pump it in the queue and let the other end push it
2807    * out as fast as possible. */
2808   if (priv->latency_ms && priv->drop_on_latency) {
2809     latency_ts =
2810         gst_util_uint64_scale_int (priv->latency_ms, priv->clock_rate, 1000);
2811
2812     if (G_UNLIKELY (rtp_jitter_buffer_get_ts_diff (priv->jbuf) >= latency_ts)) {
2813       RTPJitterBufferItem *old_item;
2814
2815       old_item = rtp_jitter_buffer_peek (priv->jbuf);
2816
2817       if (IS_DROPABLE (old_item)) {
2818         old_item = rtp_jitter_buffer_pop (priv->jbuf, &percent);
2819         GST_DEBUG_OBJECT (jitterbuffer, "Queue full, dropping old packet %p",
2820             old_item);
2821         priv->next_seqnum = (old_item->seqnum + 1) & 0xffff;
2822         free_item (old_item);
2823       }
2824       /* we might have removed some head buffers, signal the pushing thread to
2825        * see if it can push now */
2826       JBUF_SIGNAL_EVENT (priv);
2827     }
2828   }
2829
2830   /* If we estimated the DTS, don't consider it in the clock skew calculations
2831    * later. The code above always sets dts to pts or the other way around if
2832    * any of those is valid in the buffer, so we know that if we estimated the
2833    * dts that both are unknown */
2834   if (estimated_dts)
2835     item =
2836         alloc_item (buffer, ITEM_TYPE_BUFFER, GST_CLOCK_TIME_NONE,
2837         GST_CLOCK_TIME_NONE, seqnum, 1, rtptime);
2838   else
2839     item = alloc_item (buffer, ITEM_TYPE_BUFFER, dts, pts, seqnum, 1, rtptime);
2840
2841   /* now insert the packet into the queue in sorted order. This function returns
2842    * FALSE if a packet with the same seqnum was already in the queue, meaning we
2843    * have a duplicate. */
2844   if (G_UNLIKELY (!rtp_jitter_buffer_insert (priv->jbuf, item,
2845               &head, &percent,
2846               gst_element_get_base_time (GST_ELEMENT_CAST (jitterbuffer)))))
2847     goto duplicate;
2848
2849   /* update timers */
2850   update_timers (jitterbuffer, seqnum, dts, do_next_seqnum);
2851
2852   /* we had an unhandled SR, handle it now */
2853   if (priv->last_sr)
2854     do_handle_sync (jitterbuffer);
2855
2856   if (G_UNLIKELY (head)) {
2857     /* signal addition of new buffer when the _loop is waiting. */
2858     if (G_LIKELY (priv->active))
2859       JBUF_SIGNAL_EVENT (priv);
2860
2861     /* let's unschedule and unblock any waiting buffers. We only want to do this
2862      * when the head buffer changed */
2863     if (G_UNLIKELY (priv->clock_id)) {
2864       GST_DEBUG_OBJECT (jitterbuffer, "Unscheduling waiting new buffer");
2865       unschedule_current_timer (jitterbuffer);
2866     }
2867   }
2868
2869   GST_DEBUG_OBJECT (jitterbuffer,
2870       "Pushed packet #%d, now %d packets, head: %d, " "percent %d", seqnum,
2871       rtp_jitter_buffer_num_packets (priv->jbuf), head, percent);
2872
2873   msg = check_buffering_percent (jitterbuffer, percent);
2874
2875 finished:
2876   JBUF_UNLOCK (priv);
2877
2878   if (msg)
2879     gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer), msg);
2880
2881   return ret;
2882
2883   /* ERRORS */
2884 invalid_buffer:
2885   {
2886     /* this is not fatal but should be filtered earlier */
2887     GST_ELEMENT_WARNING (jitterbuffer, STREAM, DECODE, (NULL),
2888         ("Received invalid RTP payload, dropping"));
2889     gst_buffer_unref (buffer);
2890     return GST_FLOW_OK;
2891   }
2892 no_clock_rate:
2893   {
2894     GST_WARNING_OBJECT (jitterbuffer,
2895         "No clock-rate in caps!, dropping buffer");
2896     gst_buffer_unref (buffer);
2897     goto finished;
2898   }
2899 out_flushing:
2900   {
2901     ret = priv->srcresult;
2902     GST_DEBUG_OBJECT (jitterbuffer, "flushing %s", gst_flow_get_name (ret));
2903     gst_buffer_unref (buffer);
2904     goto finished;
2905   }
2906 have_eos:
2907   {
2908     ret = GST_FLOW_EOS;
2909     GST_WARNING_OBJECT (jitterbuffer, "we are EOS, refusing buffer");
2910     gst_buffer_unref (buffer);
2911     goto finished;
2912   }
2913 too_late:
2914   {
2915     GST_DEBUG_OBJECT (jitterbuffer, "Packet #%d too late as #%d was already"
2916         " popped, dropping", seqnum, priv->last_popped_seqnum);
2917     priv->num_late++;
2918     gst_buffer_unref (buffer);
2919     goto finished;
2920   }
2921 duplicate:
2922   {
2923     GST_DEBUG_OBJECT (jitterbuffer, "Duplicate packet #%d detected, dropping",
2924         seqnum);
2925     priv->num_duplicates++;
2926     free_item (item);
2927     goto finished;
2928   }
2929 }
2930
2931 static GstClockTime
2932 compute_elapsed (GstRtpJitterBuffer * jitterbuffer, RTPJitterBufferItem * item)
2933 {
2934   guint64 ext_time, elapsed;
2935   guint32 rtp_time;
2936   GstRtpJitterBufferPrivate *priv;
2937
2938   priv = jitterbuffer->priv;
2939   rtp_time = item->rtptime;
2940
2941   GST_LOG_OBJECT (jitterbuffer, "rtp %" G_GUINT32_FORMAT ", ext %"
2942       G_GUINT64_FORMAT, rtp_time, priv->ext_timestamp);
2943
2944   ext_time = priv->ext_timestamp;
2945   ext_time = gst_rtp_buffer_ext_timestamp (&ext_time, rtp_time);
2946   if (ext_time < priv->ext_timestamp) {
2947     ext_time = priv->ext_timestamp;
2948   } else {
2949     priv->ext_timestamp = ext_time;
2950   }
2951
2952   if (ext_time > priv->clock_base)
2953     elapsed = ext_time - priv->clock_base;
2954   else
2955     elapsed = 0;
2956
2957   elapsed = gst_util_uint64_scale_int (elapsed, GST_SECOND, priv->clock_rate);
2958   return elapsed;
2959 }
2960
2961 static void
2962 update_estimated_eos (GstRtpJitterBuffer * jitterbuffer,
2963     RTPJitterBufferItem * item)
2964 {
2965   guint64 total, elapsed, left, estimated;
2966   GstClockTime out_time;
2967   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2968
2969   if (priv->npt_stop == -1 || priv->ext_timestamp == -1
2970       || priv->clock_base == -1 || priv->clock_rate <= 0)
2971     return;
2972
2973   /* compute the elapsed time */
2974   elapsed = compute_elapsed (jitterbuffer, item);
2975
2976   /* do nothing if elapsed time doesn't increment */
2977   if (priv->last_elapsed && elapsed <= priv->last_elapsed)
2978     return;
2979
2980   priv->last_elapsed = elapsed;
2981
2982   /* this is the total time we need to play */
2983   total = priv->npt_stop - priv->npt_start;
2984   GST_LOG_OBJECT (jitterbuffer, "total %" GST_TIME_FORMAT,
2985       GST_TIME_ARGS (total));
2986
2987   /* this is how much time there is left */
2988   if (total > elapsed)
2989     left = total - elapsed;
2990   else
2991     left = 0;
2992
2993   /* if we have less time left that the size of the buffer, we will not
2994    * be able to keep it filled, disabled buffering then */
2995   if (left < rtp_jitter_buffer_get_delay (priv->jbuf)) {
2996     GST_DEBUG_OBJECT (jitterbuffer, "left %" GST_TIME_FORMAT
2997         ", disable buffering close to EOS", GST_TIME_ARGS (left));
2998     rtp_jitter_buffer_disable_buffering (priv->jbuf, TRUE);
2999   }
3000
3001   /* this is the current time as running-time */
3002   out_time = item->dts;
3003
3004   if (elapsed > 0)
3005     estimated = gst_util_uint64_scale (out_time, total, elapsed);
3006   else {
3007     /* if there is almost nothing left,
3008      * we may never advance enough to end up in the above case */
3009     if (total < GST_SECOND)
3010       estimated = GST_SECOND;
3011     else
3012       estimated = -1;
3013   }
3014   GST_LOG_OBJECT (jitterbuffer, "elapsed %" GST_TIME_FORMAT ", estimated %"
3015       GST_TIME_FORMAT, GST_TIME_ARGS (elapsed), GST_TIME_ARGS (estimated));
3016
3017   if (estimated != -1 && priv->estimated_eos != estimated) {
3018     set_timer (jitterbuffer, TIMER_TYPE_EOS, -1, estimated);
3019     priv->estimated_eos = estimated;
3020   }
3021 }
3022
3023 /* take a buffer from the queue and push it */
3024 static GstFlowReturn
3025 pop_and_push_next (GstRtpJitterBuffer * jitterbuffer, guint seqnum)
3026 {
3027   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
3028   GstFlowReturn result = GST_FLOW_OK;
3029   RTPJitterBufferItem *item;
3030   GstBuffer *outbuf = NULL;
3031   GstEvent *outevent = NULL;
3032   GstQuery *outquery = NULL;
3033   GstClockTime dts, pts;
3034   gint percent = -1;
3035   gboolean do_push = TRUE;
3036   guint type;
3037   GstMessage *msg;
3038
3039   /* when we get here we are ready to pop and push the buffer */
3040   item = rtp_jitter_buffer_pop (priv->jbuf, &percent);
3041   type = item->type;
3042
3043   switch (type) {
3044     case ITEM_TYPE_BUFFER:
3045
3046       /* we need to make writable to change the flags and timestamps */
3047       outbuf = gst_buffer_make_writable (item->data);
3048
3049       if (G_UNLIKELY (priv->discont)) {
3050         /* set DISCONT flag when we missed a packet. We pushed the buffer writable
3051          * into the jitterbuffer so we can modify now. */
3052         GST_DEBUG_OBJECT (jitterbuffer, "mark output buffer discont");
3053         GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_DISCONT);
3054         priv->discont = FALSE;
3055       }
3056       if (G_UNLIKELY (priv->ts_discont)) {
3057         GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_RESYNC);
3058         priv->ts_discont = FALSE;
3059       }
3060
3061       dts =
3062           gst_segment_position_from_running_time (&priv->segment,
3063           GST_FORMAT_TIME, item->dts);
3064       pts =
3065           gst_segment_position_from_running_time (&priv->segment,
3066           GST_FORMAT_TIME, item->pts);
3067
3068       /* apply timestamp with offset to buffer now */
3069       GST_BUFFER_DTS (outbuf) = apply_offset (jitterbuffer, dts);
3070       GST_BUFFER_PTS (outbuf) = apply_offset (jitterbuffer, pts);
3071
3072       /* update the elapsed time when we need to check against the npt stop time. */
3073       update_estimated_eos (jitterbuffer, item);
3074
3075       priv->last_out_time = GST_BUFFER_PTS (outbuf);
3076       break;
3077     case ITEM_TYPE_LOST:
3078       priv->discont = TRUE;
3079       if (!priv->do_lost)
3080         do_push = FALSE;
3081       /* FALLTHROUGH */
3082     case ITEM_TYPE_EVENT:
3083       outevent = item->data;
3084       break;
3085     case ITEM_TYPE_QUERY:
3086       outquery = item->data;
3087       break;
3088   }
3089
3090   /* now we are ready to push the buffer. Save the seqnum and release the lock
3091    * so the other end can push stuff in the queue again. */
3092   if (seqnum != -1) {
3093     priv->last_popped_seqnum = seqnum;
3094     priv->next_seqnum = (seqnum + item->count) & 0xffff;
3095   }
3096   msg = check_buffering_percent (jitterbuffer, percent);
3097   JBUF_UNLOCK (priv);
3098
3099   item->data = NULL;
3100   free_item (item);
3101
3102   if (msg)
3103     gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer), msg);
3104
3105   switch (type) {
3106     case ITEM_TYPE_BUFFER:
3107       /* push buffer */
3108       GST_DEBUG_OBJECT (jitterbuffer,
3109           "Pushing buffer %d, dts %" GST_TIME_FORMAT ", pts %" GST_TIME_FORMAT,
3110           seqnum, GST_TIME_ARGS (GST_BUFFER_DTS (outbuf)),
3111           GST_TIME_ARGS (GST_BUFFER_PTS (outbuf)));
3112       result = gst_pad_push (priv->srcpad, outbuf);
3113
3114       JBUF_LOCK_CHECK (priv, out_flushing);
3115       break;
3116     case ITEM_TYPE_LOST:
3117     case ITEM_TYPE_EVENT:
3118       /* We got not enough consecutive packets with a huge gap, we can
3119        * as well just drop them here now on EOS */
3120       if (GST_EVENT_TYPE (outevent) == GST_EVENT_EOS) {
3121         GST_DEBUG_OBJECT (jitterbuffer, "Clearing gap packets on EOS");
3122         g_queue_foreach (&priv->gap_packets, (GFunc) gst_buffer_unref, NULL);
3123         g_queue_clear (&priv->gap_packets);
3124       }
3125
3126       GST_DEBUG_OBJECT (jitterbuffer, "%sPushing event %" GST_PTR_FORMAT
3127           ", seqnum %d", do_push ? "" : "NOT ", outevent, seqnum);
3128
3129       if (do_push)
3130         gst_pad_push_event (priv->srcpad, outevent);
3131       else
3132         gst_event_unref (outevent);
3133
3134       result = GST_FLOW_OK;
3135
3136       JBUF_LOCK_CHECK (priv, out_flushing);
3137       break;
3138     case ITEM_TYPE_QUERY:
3139     {
3140       gboolean res;
3141
3142       res = gst_pad_peer_query (priv->srcpad, outquery);
3143
3144       JBUF_LOCK_CHECK (priv, out_flushing);
3145       result = GST_FLOW_OK;
3146       GST_LOG_OBJECT (jitterbuffer, "did query %p, return %d", outquery, res);
3147       JBUF_SIGNAL_QUERY (priv, res);
3148       break;
3149     }
3150   }
3151   return result;
3152
3153   /* ERRORS */
3154 out_flushing:
3155   {
3156     return priv->srcresult;
3157   }
3158 }
3159
3160 #define GST_FLOW_WAIT GST_FLOW_CUSTOM_SUCCESS
3161
3162 /* Peek a buffer and compare the seqnum to the expected seqnum.
3163  * If all is fine, the buffer is pushed.
3164  * If something is wrong, we wait for some event
3165  */
3166 static GstFlowReturn
3167 handle_next_buffer (GstRtpJitterBuffer * jitterbuffer)
3168 {
3169   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
3170   GstFlowReturn result;
3171   RTPJitterBufferItem *item;
3172   guint seqnum;
3173   guint32 next_seqnum;
3174
3175   /* only push buffers when PLAYING and active and not buffering */
3176   if (priv->blocked || !priv->active ||
3177       rtp_jitter_buffer_is_buffering (priv->jbuf)) {
3178     return GST_FLOW_WAIT;
3179   }
3180
3181   /* peek a buffer, we're just looking at the sequence number.
3182    * If all is fine, we'll pop and push it. If the sequence number is wrong we
3183    * wait for a timeout or something to change.
3184    * The peeked buffer is valid for as long as we hold the jitterbuffer lock. */
3185   item = rtp_jitter_buffer_peek (priv->jbuf);
3186   if (item == NULL) {
3187     goto wait;
3188   }
3189
3190   /* get the seqnum and the next expected seqnum */
3191   seqnum = item->seqnum;
3192   if (seqnum == -1) {
3193     return pop_and_push_next (jitterbuffer, seqnum);
3194   }
3195
3196   next_seqnum = priv->next_seqnum;
3197
3198   /* get the gap between this and the previous packet. If we don't know the
3199    * previous packet seqnum assume no gap. */
3200   if (G_UNLIKELY (next_seqnum == -1)) {
3201     GST_DEBUG_OBJECT (jitterbuffer, "First buffer #%d", seqnum);
3202     /* we don't know what the next_seqnum should be, the chain function should
3203      * have scheduled a DEADLINE timer that will increment next_seqnum when it
3204      * fires, so wait for that */
3205     result = GST_FLOW_WAIT;
3206   } else {
3207     gint gap = gst_rtp_buffer_compare_seqnum (next_seqnum, seqnum);
3208
3209     if (G_LIKELY (gap == 0)) {
3210       /* no missing packet, pop and push */
3211       result = pop_and_push_next (jitterbuffer, seqnum);
3212     } else if (G_UNLIKELY (gap < 0)) {
3213       /* if we have a packet that we already pushed or considered dropped, pop it
3214        * off and get the next packet */
3215       GST_DEBUG_OBJECT (jitterbuffer, "Old packet #%d, next #%d dropping",
3216           seqnum, next_seqnum);
3217       item = rtp_jitter_buffer_pop (priv->jbuf, NULL);
3218       free_item (item);
3219       result = GST_FLOW_OK;
3220     } else {
3221       /* the chain function has scheduled timers to request retransmission or
3222        * when to consider the packet lost, wait for that */
3223       GST_DEBUG_OBJECT (jitterbuffer,
3224           "Sequence number GAP detected: expected %d instead of %d (%d missing)",
3225           next_seqnum, seqnum, gap);
3226       result = GST_FLOW_WAIT;
3227     }
3228   }
3229
3230   return result;
3231
3232 wait:
3233   {
3234     GST_DEBUG_OBJECT (jitterbuffer, "no buffer, going to wait");
3235     if (priv->eos) {
3236       return GST_FLOW_EOS;
3237     } else {
3238       return GST_FLOW_WAIT;
3239     }
3240   }
3241 }
3242
3243 static GstClockTime
3244 get_rtx_retry_timeout (GstRtpJitterBufferPrivate * priv)
3245 {
3246   GstClockTime rtx_retry_timeout;
3247   GstClockTime rtx_min_retry_timeout;
3248
3249   if (priv->rtx_retry_timeout == -1) {
3250     if (priv->avg_rtx_rtt == 0)
3251       rtx_retry_timeout = DEFAULT_AUTO_RTX_TIMEOUT;
3252     else
3253       /* we want to ask for a retransmission after we waited for a
3254        * complete RTT and the additional jitter */
3255       rtx_retry_timeout = priv->avg_rtx_rtt + priv->avg_jitter * 2;
3256   } else {
3257     rtx_retry_timeout = priv->rtx_retry_timeout * GST_MSECOND;
3258   }
3259   /* make sure we don't retry too often. On very low latency networks,
3260    * the RTT and jitter can be very low. */
3261   if (priv->rtx_min_retry_timeout == -1) {
3262     rtx_min_retry_timeout = priv->packet_spacing;
3263   } else {
3264     rtx_min_retry_timeout = priv->rtx_min_retry_timeout * GST_MSECOND;
3265   }
3266   rtx_retry_timeout = MAX (rtx_retry_timeout, rtx_min_retry_timeout);
3267
3268   return rtx_retry_timeout;
3269 }
3270
3271 static GstClockTime
3272 get_rtx_retry_period (GstRtpJitterBufferPrivate * priv,
3273     GstClockTime rtx_retry_timeout)
3274 {
3275   GstClockTime rtx_retry_period;
3276
3277   if (priv->rtx_retry_period == -1) {
3278     /* we retry up to the configured jitterbuffer size but leaving some
3279      * room for the retransmission to arrive in time */
3280     if (rtx_retry_timeout > priv->latency_ns) {
3281       rtx_retry_period = 0;
3282     } else {
3283       rtx_retry_period = priv->latency_ns - rtx_retry_timeout;
3284     }
3285   } else {
3286     rtx_retry_period = priv->rtx_retry_period * GST_MSECOND;
3287   }
3288   return rtx_retry_period;
3289 }
3290
3291 /* the timeout for when we expected a packet expired */
3292 static gboolean
3293 do_expected_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
3294     GstClockTime now)
3295 {
3296   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
3297   GstEvent *event;
3298   guint delay, delay_ms, avg_rtx_rtt_ms;
3299   guint rtx_retry_timeout_ms, rtx_retry_period_ms;
3300   GstClockTime rtx_retry_period;
3301   GstClockTime rtx_retry_timeout;
3302   GstClock *clock;
3303
3304   GST_DEBUG_OBJECT (jitterbuffer, "expected %d didn't arrive, now %"
3305       GST_TIME_FORMAT, timer->seqnum, GST_TIME_ARGS (now));
3306
3307   rtx_retry_timeout = get_rtx_retry_timeout (priv);
3308   rtx_retry_period = get_rtx_retry_period (priv, rtx_retry_timeout);
3309
3310   GST_DEBUG_OBJECT (jitterbuffer, "timeout %" GST_TIME_FORMAT ", period %"
3311       GST_TIME_FORMAT, GST_TIME_ARGS (rtx_retry_timeout),
3312       GST_TIME_ARGS (rtx_retry_period));
3313
3314   delay = timer->rtx_delay + timer->rtx_retry;
3315
3316   delay_ms = GST_TIME_AS_MSECONDS (delay);
3317   rtx_retry_timeout_ms = GST_TIME_AS_MSECONDS (rtx_retry_timeout);
3318   rtx_retry_period_ms = GST_TIME_AS_MSECONDS (rtx_retry_period);
3319   avg_rtx_rtt_ms = GST_TIME_AS_MSECONDS (priv->avg_rtx_rtt);
3320
3321   event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
3322       gst_structure_new ("GstRTPRetransmissionRequest",
3323           "seqnum", G_TYPE_UINT, (guint) timer->seqnum,
3324           "running-time", G_TYPE_UINT64, timer->rtx_base,
3325           "delay", G_TYPE_UINT, delay_ms,
3326           "retry", G_TYPE_UINT, timer->num_rtx_retry,
3327           "frequency", G_TYPE_UINT, rtx_retry_timeout_ms,
3328           "period", G_TYPE_UINT, rtx_retry_period_ms,
3329           "deadline", G_TYPE_UINT, priv->latency_ms,
3330           "packet-spacing", G_TYPE_UINT64, priv->packet_spacing,
3331           "avg-rtt", G_TYPE_UINT, avg_rtx_rtt_ms, NULL));
3332
3333   priv->num_rtx_requests++;
3334   timer->num_rtx_retry++;
3335
3336   GST_OBJECT_LOCK (jitterbuffer);
3337   if ((clock = GST_ELEMENT_CLOCK (jitterbuffer))) {
3338     timer->rtx_last = gst_clock_get_time (clock);
3339     timer->rtx_last -= GST_ELEMENT_CAST (jitterbuffer)->base_time;
3340   } else {
3341     timer->rtx_last = now;
3342   }
3343   GST_OBJECT_UNLOCK (jitterbuffer);
3344
3345   /* calculate the timeout for the next retransmission attempt */
3346   timer->rtx_retry += rtx_retry_timeout;
3347   GST_DEBUG_OBJECT (jitterbuffer, "base %" GST_TIME_FORMAT ", delay %"
3348       GST_TIME_FORMAT ", retry %" GST_TIME_FORMAT ", num_retry %u",
3349       GST_TIME_ARGS (timer->rtx_base), GST_TIME_ARGS (timer->rtx_delay),
3350       GST_TIME_ARGS (timer->rtx_retry), timer->num_rtx_retry);
3351   if ((priv->rtx_max_retries != -1
3352           && timer->num_rtx_retry >= priv->rtx_max_retries)
3353       || (timer->rtx_retry + timer->rtx_delay > rtx_retry_period)) {
3354     GST_DEBUG_OBJECT (jitterbuffer, "reschedule as LOST timer");
3355     /* too many retransmission request, we now convert the timer
3356      * to a lost timer, leave the num_rtx_retry as it is for stats */
3357     timer->type = TIMER_TYPE_LOST;
3358     timer->rtx_delay = 0;
3359     timer->rtx_retry = 0;
3360   }
3361   reschedule_timer (jitterbuffer, timer, timer->seqnum,
3362       timer->rtx_base + timer->rtx_retry, timer->rtx_delay, FALSE);
3363
3364   JBUF_UNLOCK (priv);
3365   gst_pad_push_event (priv->sinkpad, event);
3366   JBUF_LOCK (priv);
3367
3368   return FALSE;
3369 }
3370
3371 /* a packet is lost */
3372 static gboolean
3373 do_lost_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
3374     GstClockTime now)
3375 {
3376   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
3377   GstClockTime duration, timestamp;
3378   guint seqnum, lost_packets, num_rtx_retry, next_in_seqnum;
3379   gboolean head;
3380   GstEvent *event;
3381   RTPJitterBufferItem *item;
3382
3383   seqnum = timer->seqnum;
3384   timestamp = apply_offset (jitterbuffer, timer->timeout);
3385   duration = timer->duration;
3386   if (duration == GST_CLOCK_TIME_NONE && priv->packet_spacing > 0)
3387     duration = priv->packet_spacing;
3388   lost_packets = MAX (timer->num, 1);
3389   num_rtx_retry = timer->num_rtx_retry;
3390
3391   /* we had a gap and thus we lost some packets. Create an event for this.  */
3392   if (lost_packets > 1)
3393     GST_DEBUG_OBJECT (jitterbuffer, "Packets #%d -> #%d lost", seqnum,
3394         seqnum + lost_packets - 1);
3395   else
3396     GST_DEBUG_OBJECT (jitterbuffer, "Packet #%d lost", seqnum);
3397
3398   priv->num_late += lost_packets;
3399   priv->num_rtx_failed += num_rtx_retry;
3400
3401   next_in_seqnum = (seqnum + lost_packets) & 0xffff;
3402
3403   /* we now only accept seqnum bigger than this */
3404   if (gst_rtp_buffer_compare_seqnum (priv->next_in_seqnum, next_in_seqnum) > 0)
3405     priv->next_in_seqnum = next_in_seqnum;
3406
3407   /* create paket lost event */
3408   event = gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM,
3409       gst_structure_new ("GstRTPPacketLost",
3410           "seqnum", G_TYPE_UINT, (guint) seqnum,
3411           "timestamp", G_TYPE_UINT64, timestamp,
3412           "duration", G_TYPE_UINT64, duration,
3413           "retry", G_TYPE_UINT, num_rtx_retry, NULL));
3414
3415   item = alloc_item (event, ITEM_TYPE_LOST, -1, -1, seqnum, lost_packets, -1);
3416   rtp_jitter_buffer_insert (priv->jbuf, item, &head, NULL, -1);
3417
3418   /* remove timer now */
3419   remove_timer (jitterbuffer, timer);
3420   if (head)
3421     JBUF_SIGNAL_EVENT (priv);
3422
3423   return TRUE;
3424 }
3425
3426 static gboolean
3427 do_eos_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
3428     GstClockTime now)
3429 {
3430   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
3431
3432   GST_INFO_OBJECT (jitterbuffer, "got the NPT timeout");
3433   remove_timer (jitterbuffer, timer);
3434   if (!priv->eos) {
3435     /* there was no EOS in the buffer, put one in there now */
3436     queue_event (jitterbuffer, gst_event_new_eos ());
3437   }
3438   JBUF_SIGNAL_EVENT (priv);
3439
3440   return TRUE;
3441 }
3442
3443 static gboolean
3444 do_deadline_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
3445     GstClockTime now)
3446 {
3447   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
3448
3449   GST_INFO_OBJECT (jitterbuffer, "got deadline timeout");
3450
3451   /* timer seqnum might have been obsoleted by caps seqnum-base,
3452    * only mess with current ongoing seqnum if still unknown */
3453   if (priv->next_seqnum == -1)
3454     priv->next_seqnum = timer->seqnum;
3455   remove_timer (jitterbuffer, timer);
3456   JBUF_SIGNAL_EVENT (priv);
3457
3458   return TRUE;
3459 }
3460
3461 static gboolean
3462 do_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
3463     GstClockTime now)
3464 {
3465   gboolean removed = FALSE;
3466
3467   switch (timer->type) {
3468     case TIMER_TYPE_EXPECTED:
3469       removed = do_expected_timeout (jitterbuffer, timer, now);
3470       break;
3471     case TIMER_TYPE_LOST:
3472       removed = do_lost_timeout (jitterbuffer, timer, now);
3473       break;
3474     case TIMER_TYPE_DEADLINE:
3475       removed = do_deadline_timeout (jitterbuffer, timer, now);
3476       break;
3477     case TIMER_TYPE_EOS:
3478       removed = do_eos_timeout (jitterbuffer, timer, now);
3479       break;
3480   }
3481   return removed;
3482 }
3483
3484 /* called when we need to wait for the next timeout.
3485  *
3486  * We loop over the array of recorded timeouts and wait for the earliest one.
3487  * When it timed out, do the logic associated with the timer.
3488  *
3489  * If there are no timers, we wait on a gcond until something new happens.
3490  */
3491 static void
3492 wait_next_timeout (GstRtpJitterBuffer * jitterbuffer)
3493 {
3494   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
3495   GstClockTime now = 0;
3496
3497   JBUF_LOCK (priv);
3498   while (priv->timer_running) {
3499     TimerData *timer = NULL;
3500     GstClockTime timer_timeout = -1;
3501     gint i, len;
3502
3503     /* If we have a clock, update "now" now with the very
3504      * latest running time we have. If timers are unscheduled below we
3505      * otherwise wouldn't update now (it's only updated when timers
3506      * expire), and also for the very first loop iteration now would
3507      * otherwise always be 0
3508      */
3509     GST_OBJECT_LOCK (jitterbuffer);
3510     if (GST_ELEMENT_CLOCK (jitterbuffer)) {
3511       now =
3512           gst_clock_get_time (GST_ELEMENT_CLOCK (jitterbuffer)) -
3513           GST_ELEMENT_CAST (jitterbuffer)->base_time;
3514     }
3515     GST_OBJECT_UNLOCK (jitterbuffer);
3516
3517     GST_DEBUG_OBJECT (jitterbuffer, "now %" GST_TIME_FORMAT,
3518         GST_TIME_ARGS (now));
3519
3520     len = priv->timers->len;
3521     for (i = 0; i < len; i++) {
3522       TimerData *test = &g_array_index (priv->timers, TimerData, i);
3523       GstClockTime test_timeout = get_timeout (jitterbuffer, test);
3524       gboolean save_best = FALSE;
3525
3526       GST_DEBUG_OBJECT (jitterbuffer, "%d, %d, %d, %" GST_TIME_FORMAT,
3527           i, test->type, test->seqnum, GST_TIME_ARGS (test_timeout));
3528
3529       /* find the smallest timeout */
3530       if (timer == NULL) {
3531         save_best = TRUE;
3532       } else if (timer_timeout == -1) {
3533         /* we already have an immediate timeout, the new timer must be an
3534          * immediate timer with smaller seqnum to become the best */
3535         if (test_timeout == -1
3536             && (gst_rtp_buffer_compare_seqnum (test->seqnum,
3537                     timer->seqnum) > 0))
3538           save_best = TRUE;
3539       } else if (test_timeout == -1) {
3540         /* first immediate timer */
3541         save_best = TRUE;
3542       } else if (test_timeout < timer_timeout) {
3543         /* earlier timer */
3544         save_best = TRUE;
3545       } else if (test_timeout == timer_timeout
3546           && (gst_rtp_buffer_compare_seqnum (test->seqnum,
3547                   timer->seqnum) > 0)) {
3548         /* same timer, smaller seqnum */
3549         save_best = TRUE;
3550       }
3551       if (save_best) {
3552         GST_DEBUG_OBJECT (jitterbuffer, "new best %d", i);
3553         timer = test;
3554         timer_timeout = test_timeout;
3555       }
3556     }
3557     if (timer && !priv->blocked) {
3558       GstClock *clock;
3559       GstClockTime sync_time;
3560       GstClockID id;
3561       GstClockReturn ret;
3562       GstClockTimeDiff clock_jitter;
3563
3564       if (timer_timeout == -1 || timer_timeout <= now) {
3565         do_timeout (jitterbuffer, timer, now);
3566         /* check here, do_timeout could have released the lock */
3567         if (!priv->timer_running)
3568           break;
3569         continue;
3570       }
3571
3572       GST_OBJECT_LOCK (jitterbuffer);
3573       clock = GST_ELEMENT_CLOCK (jitterbuffer);
3574       if (!clock) {
3575         GST_OBJECT_UNLOCK (jitterbuffer);
3576         /* let's just push if there is no clock */
3577         GST_DEBUG_OBJECT (jitterbuffer, "No clock, timeout right away");
3578         now = timer_timeout;
3579         continue;
3580       }
3581
3582       /* prepare for sync against clock */
3583       sync_time = timer_timeout + GST_ELEMENT_CAST (jitterbuffer)->base_time;
3584       /* add latency of peer to get input time */
3585       sync_time += priv->peer_latency;
3586
3587       GST_DEBUG_OBJECT (jitterbuffer, "sync to timestamp %" GST_TIME_FORMAT
3588           " with sync time %" GST_TIME_FORMAT,
3589           GST_TIME_ARGS (timer_timeout), GST_TIME_ARGS (sync_time));
3590
3591       /* create an entry for the clock */
3592       id = priv->clock_id = gst_clock_new_single_shot_id (clock, sync_time);
3593       priv->timer_timeout = timer_timeout;
3594       priv->timer_seqnum = timer->seqnum;
3595       GST_OBJECT_UNLOCK (jitterbuffer);
3596
3597       /* release the lock so that the other end can push stuff or unlock */
3598       JBUF_UNLOCK (priv);
3599
3600       ret = gst_clock_id_wait (id, &clock_jitter);
3601
3602       JBUF_LOCK (priv);
3603       if (!priv->timer_running) {
3604         gst_clock_id_unref (id);
3605         priv->clock_id = NULL;
3606         break;
3607       }
3608
3609       if (ret != GST_CLOCK_UNSCHEDULED) {
3610         now = timer_timeout + MAX (clock_jitter, 0);
3611         GST_DEBUG_OBJECT (jitterbuffer,
3612             "sync done, %d, #%d, %" GST_STIME_FORMAT, ret, priv->timer_seqnum,
3613             GST_STIME_ARGS (clock_jitter));
3614       } else {
3615         GST_DEBUG_OBJECT (jitterbuffer, "sync unscheduled");
3616       }
3617       /* and free the entry */
3618       gst_clock_id_unref (id);
3619       priv->clock_id = NULL;
3620     } else {
3621       /* no timers, wait for activity */
3622       JBUF_WAIT_TIMER (priv);
3623     }
3624   }
3625   JBUF_UNLOCK (priv);
3626
3627   GST_DEBUG_OBJECT (jitterbuffer, "we are stopping");
3628   return;
3629 }
3630
3631 /*
3632  * This funcion implements the main pushing loop on the source pad.
3633  *
3634  * It first tries to push as many buffers as possible. If there is a seqnum
3635  * mismatch, we wait for the next timeouts.
3636  */
3637 static void
3638 gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer)
3639 {
3640   GstRtpJitterBufferPrivate *priv;
3641   GstFlowReturn result = GST_FLOW_OK;
3642
3643   priv = jitterbuffer->priv;
3644
3645   JBUF_LOCK_CHECK (priv, flushing);
3646   do {
3647     result = handle_next_buffer (jitterbuffer);
3648     if (G_LIKELY (result == GST_FLOW_WAIT)) {
3649       /* now wait for the next event */
3650       JBUF_WAIT_EVENT (priv, flushing);
3651       result = GST_FLOW_OK;
3652     }
3653   } while (result == GST_FLOW_OK);
3654   /* store result for upstream */
3655   priv->srcresult = result;
3656   /* if we get here we need to pause */
3657   goto pause;
3658
3659   /* ERRORS */
3660 flushing:
3661   {
3662     result = priv->srcresult;
3663     goto pause;
3664   }
3665 pause:
3666   {
3667     GstEvent *event;
3668
3669     JBUF_SIGNAL_QUERY (priv, FALSE);
3670     JBUF_UNLOCK (priv);
3671
3672     GST_DEBUG_OBJECT (jitterbuffer, "pausing task, reason %s",
3673         gst_flow_get_name (result));
3674     gst_pad_pause_task (priv->srcpad);
3675     if (result == GST_FLOW_EOS) {
3676       event = gst_event_new_eos ();
3677       gst_pad_push_event (priv->srcpad, event);
3678     }
3679     return;
3680   }
3681 }
3682
3683 /* collect the info from the lastest RTCP packet and the jitterbuffer sync, do
3684  * some sanity checks and then emit the handle-sync signal with the parameters.
3685  * This function must be called with the LOCK */
3686 static void
3687 do_handle_sync (GstRtpJitterBuffer * jitterbuffer)
3688 {
3689   GstRtpJitterBufferPrivate *priv;
3690   guint64 base_rtptime, base_time;
3691   guint32 clock_rate;
3692   guint64 last_rtptime;
3693   guint64 clock_base;
3694   guint64 ext_rtptime, diff;
3695   gboolean valid = TRUE, keep = FALSE;
3696
3697   priv = jitterbuffer->priv;
3698
3699   /* get the last values from the jitterbuffer */
3700   rtp_jitter_buffer_get_sync (priv->jbuf, &base_rtptime, &base_time,
3701       &clock_rate, &last_rtptime);
3702
3703   clock_base = priv->clock_base;
3704   ext_rtptime = priv->ext_rtptime;
3705
3706   GST_DEBUG_OBJECT (jitterbuffer, "ext SR %" G_GUINT64_FORMAT ", base %"
3707       G_GUINT64_FORMAT ", clock-rate %" G_GUINT32_FORMAT
3708       ", clock-base %" G_GUINT64_FORMAT ", last-rtptime %" G_GUINT64_FORMAT,
3709       ext_rtptime, base_rtptime, clock_rate, clock_base, last_rtptime);
3710
3711   if (base_rtptime == -1 || clock_rate == -1 || base_time == -1) {
3712     /* we keep this SR packet for later. When we get a valid RTP packet the
3713      * above values will be set and we can try to use the SR packet */
3714     GST_DEBUG_OBJECT (jitterbuffer, "keeping for later, no RTP values");
3715     keep = TRUE;
3716   } else {
3717     /* we can't accept anything that happened before we did the last resync */
3718     if (base_rtptime > ext_rtptime) {
3719       GST_DEBUG_OBJECT (jitterbuffer, "dropping, older than base time");
3720       valid = FALSE;
3721     } else {
3722       /* the SR RTP timestamp must be something close to what we last observed
3723        * in the jitterbuffer */
3724       if (ext_rtptime > last_rtptime) {
3725         /* check how far ahead it is to our RTP timestamps */
3726         diff = ext_rtptime - last_rtptime;
3727         /* if bigger than 1 second, we drop it */
3728         if (jitterbuffer->priv->max_rtcp_rtp_time_diff != -1 &&
3729             diff >
3730             gst_util_uint64_scale (jitterbuffer->priv->max_rtcp_rtp_time_diff,
3731                 clock_rate, 1000)) {
3732           GST_DEBUG_OBJECT (jitterbuffer, "too far ahead");
3733           /* should drop this, but some RTSP servers end up with bogus
3734            * way too ahead RTCP packet when repeated PAUSE/PLAY,
3735            * so still trigger rptbin sync but invalidate RTCP data
3736            * (sync might use other methods) */
3737           ext_rtptime = -1;
3738         }
3739         GST_DEBUG_OBJECT (jitterbuffer, "ext last %" G_GUINT64_FORMAT ", diff %"
3740             G_GUINT64_FORMAT, last_rtptime, diff);
3741       }
3742     }
3743   }
3744
3745   if (keep) {
3746     GST_DEBUG_OBJECT (jitterbuffer, "keeping RTCP packet for later");
3747   } else if (valid) {
3748     GstStructure *s;
3749
3750     s = gst_structure_new ("application/x-rtp-sync",
3751         "base-rtptime", G_TYPE_UINT64, base_rtptime,
3752         "base-time", G_TYPE_UINT64, base_time,
3753         "clock-rate", G_TYPE_UINT, clock_rate,
3754         "clock-base", G_TYPE_UINT64, clock_base,
3755         "sr-ext-rtptime", G_TYPE_UINT64, ext_rtptime,
3756         "sr-buffer", GST_TYPE_BUFFER, priv->last_sr, NULL);
3757
3758     GST_DEBUG_OBJECT (jitterbuffer, "signaling sync");
3759     gst_buffer_replace (&priv->last_sr, NULL);
3760     JBUF_UNLOCK (priv);
3761     g_signal_emit (jitterbuffer,
3762         gst_rtp_jitter_buffer_signals[SIGNAL_HANDLE_SYNC], 0, s);
3763     JBUF_LOCK (priv);
3764     gst_structure_free (s);
3765   } else {
3766     GST_DEBUG_OBJECT (jitterbuffer, "dropping RTCP packet");
3767     gst_buffer_replace (&priv->last_sr, NULL);
3768   }
3769 }
3770
3771 static GstFlowReturn
3772 gst_rtp_jitter_buffer_chain_rtcp (GstPad * pad, GstObject * parent,
3773     GstBuffer * buffer)
3774 {
3775   GstRtpJitterBuffer *jitterbuffer;
3776   GstRtpJitterBufferPrivate *priv;
3777   GstFlowReturn ret = GST_FLOW_OK;
3778   guint32 ssrc;
3779   GstRTCPPacket packet;
3780   guint64 ext_rtptime;
3781   guint32 rtptime;
3782   GstRTCPBuffer rtcp = { NULL, };
3783
3784   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
3785
3786   if (G_UNLIKELY (!gst_rtcp_buffer_validate_reduced (buffer)))
3787     goto invalid_buffer;
3788
3789   priv = jitterbuffer->priv;
3790
3791   gst_rtcp_buffer_map (buffer, GST_MAP_READ, &rtcp);
3792
3793   if (!gst_rtcp_buffer_get_first_packet (&rtcp, &packet))
3794     goto empty_buffer;
3795
3796   /* first packet must be SR or RR or else the validate would have failed */
3797   switch (gst_rtcp_packet_get_type (&packet)) {
3798     case GST_RTCP_TYPE_SR:
3799       gst_rtcp_packet_sr_get_sender_info (&packet, &ssrc, NULL, &rtptime,
3800           NULL, NULL);
3801       break;
3802     default:
3803       goto ignore_buffer;
3804   }
3805   gst_rtcp_buffer_unmap (&rtcp);
3806
3807   GST_DEBUG_OBJECT (jitterbuffer, "received RTCP of SSRC %08x", ssrc);
3808
3809   JBUF_LOCK (priv);
3810   /* convert the RTP timestamp to our extended timestamp, using the same offset
3811    * we used in the jitterbuffer */
3812   ext_rtptime = priv->jbuf->ext_rtptime;
3813   ext_rtptime = gst_rtp_buffer_ext_timestamp (&ext_rtptime, rtptime);
3814
3815   priv->ext_rtptime = ext_rtptime;
3816   gst_buffer_replace (&priv->last_sr, buffer);
3817
3818   do_handle_sync (jitterbuffer);
3819   JBUF_UNLOCK (priv);
3820
3821 done:
3822   gst_buffer_unref (buffer);
3823
3824   return ret;
3825
3826 invalid_buffer:
3827   {
3828     /* this is not fatal but should be filtered earlier */
3829     GST_ELEMENT_WARNING (jitterbuffer, STREAM, DECODE, (NULL),
3830         ("Received invalid RTCP payload, dropping"));
3831     ret = GST_FLOW_OK;
3832     goto done;
3833   }
3834 empty_buffer:
3835   {
3836     /* this is not fatal but should be filtered earlier */
3837     GST_ELEMENT_WARNING (jitterbuffer, STREAM, DECODE, (NULL),
3838         ("Received empty RTCP payload, dropping"));
3839     gst_rtcp_buffer_unmap (&rtcp);
3840     ret = GST_FLOW_OK;
3841     goto done;
3842   }
3843 ignore_buffer:
3844   {
3845     GST_DEBUG_OBJECT (jitterbuffer, "ignoring RTCP packet");
3846     gst_rtcp_buffer_unmap (&rtcp);
3847     ret = GST_FLOW_OK;
3848     goto done;
3849   }
3850 }
3851
3852 static gboolean
3853 gst_rtp_jitter_buffer_sink_query (GstPad * pad, GstObject * parent,
3854     GstQuery * query)
3855 {
3856   gboolean res = FALSE;
3857   GstRtpJitterBuffer *jitterbuffer;
3858   GstRtpJitterBufferPrivate *priv;
3859
3860   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
3861   priv = jitterbuffer->priv;
3862
3863   switch (GST_QUERY_TYPE (query)) {
3864     case GST_QUERY_CAPS:
3865     {
3866       GstCaps *filter, *caps;
3867
3868       gst_query_parse_caps (query, &filter);
3869       caps = gst_rtp_jitter_buffer_getcaps (pad, filter);
3870       gst_query_set_caps_result (query, caps);
3871       gst_caps_unref (caps);
3872       res = TRUE;
3873       break;
3874     }
3875     default:
3876       if (GST_QUERY_IS_SERIALIZED (query)) {
3877         RTPJitterBufferItem *item;
3878         gboolean head;
3879
3880         JBUF_LOCK_CHECK (priv, out_flushing);
3881         if (rtp_jitter_buffer_get_mode (priv->jbuf) !=
3882             RTP_JITTER_BUFFER_MODE_BUFFER) {
3883           GST_DEBUG_OBJECT (jitterbuffer, "adding serialized query");
3884           item = alloc_item (query, ITEM_TYPE_QUERY, -1, -1, -1, 0, -1);
3885           rtp_jitter_buffer_insert (priv->jbuf, item, &head, NULL, -1);
3886           if (head)
3887             JBUF_SIGNAL_EVENT (priv);
3888           JBUF_WAIT_QUERY (priv, out_flushing);
3889           res = priv->last_query;
3890         } else {
3891           GST_DEBUG_OBJECT (jitterbuffer, "refusing query, we are buffering");
3892           res = FALSE;
3893         }
3894         JBUF_UNLOCK (priv);
3895       } else {
3896         res = gst_pad_query_default (pad, parent, query);
3897       }
3898       break;
3899   }
3900   return res;
3901   /* ERRORS */
3902 out_flushing:
3903   {
3904     GST_DEBUG_OBJECT (jitterbuffer, "we are flushing");
3905     JBUF_UNLOCK (priv);
3906     return FALSE;
3907   }
3908
3909 }
3910
3911 static gboolean
3912 gst_rtp_jitter_buffer_src_query (GstPad * pad, GstObject * parent,
3913     GstQuery * query)
3914 {
3915   GstRtpJitterBuffer *jitterbuffer;
3916   GstRtpJitterBufferPrivate *priv;
3917   gboolean res = FALSE;
3918
3919   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
3920   priv = jitterbuffer->priv;
3921
3922   switch (GST_QUERY_TYPE (query)) {
3923     case GST_QUERY_LATENCY:
3924     {
3925       /* We need to send the query upstream and add the returned latency to our
3926        * own */
3927       GstClockTime min_latency, max_latency;
3928       gboolean us_live;
3929       GstClockTime our_latency;
3930
3931       if ((res = gst_pad_peer_query (priv->sinkpad, query))) {
3932         gst_query_parse_latency (query, &us_live, &min_latency, &max_latency);
3933
3934         GST_DEBUG_OBJECT (jitterbuffer, "Peer latency: min %"
3935             GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
3936             GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
3937
3938         /* store this so that we can safely sync on the peer buffers. */
3939         JBUF_LOCK (priv);
3940         priv->peer_latency = min_latency;
3941         our_latency = priv->latency_ns;
3942         JBUF_UNLOCK (priv);
3943
3944         GST_DEBUG_OBJECT (jitterbuffer, "Our latency: %" GST_TIME_FORMAT,
3945             GST_TIME_ARGS (our_latency));
3946
3947         /* we add some latency but can buffer an infinite amount of time */
3948         min_latency += our_latency;
3949         max_latency = -1;
3950
3951         GST_DEBUG_OBJECT (jitterbuffer, "Calculated total latency : min %"
3952             GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
3953             GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
3954
3955         gst_query_set_latency (query, TRUE, min_latency, max_latency);
3956       }
3957       break;
3958     }
3959     case GST_QUERY_POSITION:
3960     {
3961       GstClockTime start, last_out;
3962       GstFormat fmt;
3963
3964       gst_query_parse_position (query, &fmt, NULL);
3965       if (fmt != GST_FORMAT_TIME) {
3966         res = gst_pad_query_default (pad, parent, query);
3967         break;
3968       }
3969
3970       JBUF_LOCK (priv);
3971       start = priv->npt_start;
3972       last_out = priv->last_out_time;
3973       JBUF_UNLOCK (priv);
3974
3975       GST_DEBUG_OBJECT (jitterbuffer, "npt start %" GST_TIME_FORMAT
3976           ", last out %" GST_TIME_FORMAT, GST_TIME_ARGS (start),
3977           GST_TIME_ARGS (last_out));
3978
3979       if (GST_CLOCK_TIME_IS_VALID (start) && GST_CLOCK_TIME_IS_VALID (last_out)) {
3980         /* bring 0-based outgoing time to stream time */
3981         gst_query_set_position (query, GST_FORMAT_TIME, start + last_out);
3982         res = TRUE;
3983       } else {
3984         res = gst_pad_query_default (pad, parent, query);
3985       }
3986       break;
3987     }
3988     case GST_QUERY_CAPS:
3989     {
3990       GstCaps *filter, *caps;
3991
3992       gst_query_parse_caps (query, &filter);
3993       caps = gst_rtp_jitter_buffer_getcaps (pad, filter);
3994       gst_query_set_caps_result (query, caps);
3995       gst_caps_unref (caps);
3996       res = TRUE;
3997       break;
3998     }
3999     default:
4000       res = gst_pad_query_default (pad, parent, query);
4001       break;
4002   }
4003
4004   return res;
4005 }
4006
4007 static void
4008 gst_rtp_jitter_buffer_set_property (GObject * object,
4009     guint prop_id, const GValue * value, GParamSpec * pspec)
4010 {
4011   GstRtpJitterBuffer *jitterbuffer;
4012   GstRtpJitterBufferPrivate *priv;
4013
4014   jitterbuffer = GST_RTP_JITTER_BUFFER (object);
4015   priv = jitterbuffer->priv;
4016
4017   switch (prop_id) {
4018     case PROP_LATENCY:
4019     {
4020       guint new_latency, old_latency;
4021
4022       new_latency = g_value_get_uint (value);
4023
4024       JBUF_LOCK (priv);
4025       old_latency = priv->latency_ms;
4026       priv->latency_ms = new_latency;
4027       priv->latency_ns = priv->latency_ms * GST_MSECOND;
4028       rtp_jitter_buffer_set_delay (priv->jbuf, priv->latency_ns);
4029       JBUF_UNLOCK (priv);
4030
4031       /* post message if latency changed, this will inform the parent pipeline
4032        * that a latency reconfiguration is possible/needed. */
4033       if (new_latency != old_latency) {
4034         GST_DEBUG_OBJECT (jitterbuffer, "latency changed to: %" GST_TIME_FORMAT,
4035             GST_TIME_ARGS (new_latency * GST_MSECOND));
4036
4037         gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer),
4038             gst_message_new_latency (GST_OBJECT_CAST (jitterbuffer)));
4039       }
4040       break;
4041     }
4042     case PROP_DROP_ON_LATENCY:
4043       JBUF_LOCK (priv);
4044       priv->drop_on_latency = g_value_get_boolean (value);
4045       JBUF_UNLOCK (priv);
4046       break;
4047     case PROP_TS_OFFSET:
4048       JBUF_LOCK (priv);
4049       priv->ts_offset = g_value_get_int64 (value);
4050       priv->ts_discont = TRUE;
4051       JBUF_UNLOCK (priv);
4052       break;
4053     case PROP_DO_LOST:
4054       JBUF_LOCK (priv);
4055       priv->do_lost = g_value_get_boolean (value);
4056       JBUF_UNLOCK (priv);
4057       break;
4058     case PROP_MODE:
4059       JBUF_LOCK (priv);
4060       rtp_jitter_buffer_set_mode (priv->jbuf, g_value_get_enum (value));
4061       JBUF_UNLOCK (priv);
4062       break;
4063     case PROP_DO_RETRANSMISSION:
4064       JBUF_LOCK (priv);
4065       priv->do_retransmission = g_value_get_boolean (value);
4066       JBUF_UNLOCK (priv);
4067       break;
4068     case PROP_RTX_NEXT_SEQNUM:
4069       JBUF_LOCK (priv);
4070       priv->rtx_next_seqnum = g_value_get_boolean (value);
4071       JBUF_UNLOCK (priv);
4072       break;
4073     case PROP_RTX_DELAY:
4074       JBUF_LOCK (priv);
4075       priv->rtx_delay = g_value_get_int (value);
4076       JBUF_UNLOCK (priv);
4077       break;
4078     case PROP_RTX_MIN_DELAY:
4079       JBUF_LOCK (priv);
4080       priv->rtx_min_delay = g_value_get_uint (value);
4081       JBUF_UNLOCK (priv);
4082       break;
4083     case PROP_RTX_DELAY_REORDER:
4084       JBUF_LOCK (priv);
4085       priv->rtx_delay_reorder = g_value_get_int (value);
4086       JBUF_UNLOCK (priv);
4087       break;
4088     case PROP_RTX_RETRY_TIMEOUT:
4089       JBUF_LOCK (priv);
4090       priv->rtx_retry_timeout = g_value_get_int (value);
4091       JBUF_UNLOCK (priv);
4092       break;
4093     case PROP_RTX_MIN_RETRY_TIMEOUT:
4094       JBUF_LOCK (priv);
4095       priv->rtx_min_retry_timeout = g_value_get_int (value);
4096       JBUF_UNLOCK (priv);
4097       break;
4098     case PROP_RTX_RETRY_PERIOD:
4099       JBUF_LOCK (priv);
4100       priv->rtx_retry_period = g_value_get_int (value);
4101       JBUF_UNLOCK (priv);
4102       break;
4103     case PROP_RTX_MAX_RETRIES:
4104       JBUF_LOCK (priv);
4105       priv->rtx_max_retries = g_value_get_int (value);
4106       JBUF_UNLOCK (priv);
4107       break;
4108     case PROP_MAX_RTCP_RTP_TIME_DIFF:
4109       JBUF_LOCK (priv);
4110       priv->max_rtcp_rtp_time_diff = g_value_get_int (value);
4111       JBUF_UNLOCK (priv);
4112       break;
4113     case PROP_MAX_DROPOUT_TIME:
4114       JBUF_LOCK (priv);
4115       priv->max_dropout_time = g_value_get_uint (value);
4116       JBUF_UNLOCK (priv);
4117       break;
4118     case PROP_MAX_MISORDER_TIME:
4119       JBUF_LOCK (priv);
4120       priv->max_misorder_time = g_value_get_uint (value);
4121       JBUF_UNLOCK (priv);
4122       break;
4123     case PROP_RFC7273_SYNC:
4124       JBUF_LOCK (priv);
4125       rtp_jitter_buffer_set_rfc7273_sync (priv->jbuf,
4126           g_value_get_boolean (value));
4127       JBUF_UNLOCK (priv);
4128       break;
4129     default:
4130       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
4131       break;
4132   }
4133 }
4134
4135 static void
4136 gst_rtp_jitter_buffer_get_property (GObject * object,
4137     guint prop_id, GValue * value, GParamSpec * pspec)
4138 {
4139   GstRtpJitterBuffer *jitterbuffer;
4140   GstRtpJitterBufferPrivate *priv;
4141
4142   jitterbuffer = GST_RTP_JITTER_BUFFER (object);
4143   priv = jitterbuffer->priv;
4144
4145   switch (prop_id) {
4146     case PROP_LATENCY:
4147       JBUF_LOCK (priv);
4148       g_value_set_uint (value, priv->latency_ms);
4149       JBUF_UNLOCK (priv);
4150       break;
4151     case PROP_DROP_ON_LATENCY:
4152       JBUF_LOCK (priv);
4153       g_value_set_boolean (value, priv->drop_on_latency);
4154       JBUF_UNLOCK (priv);
4155       break;
4156     case PROP_TS_OFFSET:
4157       JBUF_LOCK (priv);
4158       g_value_set_int64 (value, priv->ts_offset);
4159       JBUF_UNLOCK (priv);
4160       break;
4161     case PROP_DO_LOST:
4162       JBUF_LOCK (priv);
4163       g_value_set_boolean (value, priv->do_lost);
4164       JBUF_UNLOCK (priv);
4165       break;
4166     case PROP_MODE:
4167       JBUF_LOCK (priv);
4168       g_value_set_enum (value, rtp_jitter_buffer_get_mode (priv->jbuf));
4169       JBUF_UNLOCK (priv);
4170       break;
4171     case PROP_PERCENT:
4172     {
4173       gint percent;
4174
4175       JBUF_LOCK (priv);
4176       if (priv->srcresult != GST_FLOW_OK)
4177         percent = 100;
4178       else
4179         percent = rtp_jitter_buffer_get_percent (priv->jbuf);
4180
4181       g_value_set_int (value, percent);
4182       JBUF_UNLOCK (priv);
4183       break;
4184     }
4185     case PROP_DO_RETRANSMISSION:
4186       JBUF_LOCK (priv);
4187       g_value_set_boolean (value, priv->do_retransmission);
4188       JBUF_UNLOCK (priv);
4189       break;
4190     case PROP_RTX_NEXT_SEQNUM:
4191       JBUF_LOCK (priv);
4192       g_value_set_boolean (value, priv->rtx_next_seqnum);
4193       JBUF_UNLOCK (priv);
4194       break;
4195     case PROP_RTX_DELAY:
4196       JBUF_LOCK (priv);
4197       g_value_set_int (value, priv->rtx_delay);
4198       JBUF_UNLOCK (priv);
4199       break;
4200     case PROP_RTX_MIN_DELAY:
4201       JBUF_LOCK (priv);
4202       g_value_set_uint (value, priv->rtx_min_delay);
4203       JBUF_UNLOCK (priv);
4204       break;
4205     case PROP_RTX_DELAY_REORDER:
4206       JBUF_LOCK (priv);
4207       g_value_set_int (value, priv->rtx_delay_reorder);
4208       JBUF_UNLOCK (priv);
4209       break;
4210     case PROP_RTX_RETRY_TIMEOUT:
4211       JBUF_LOCK (priv);
4212       g_value_set_int (value, priv->rtx_retry_timeout);
4213       JBUF_UNLOCK (priv);
4214       break;
4215     case PROP_RTX_MIN_RETRY_TIMEOUT:
4216       JBUF_LOCK (priv);
4217       g_value_set_int (value, priv->rtx_min_retry_timeout);
4218       JBUF_UNLOCK (priv);
4219       break;
4220     case PROP_RTX_RETRY_PERIOD:
4221       JBUF_LOCK (priv);
4222       g_value_set_int (value, priv->rtx_retry_period);
4223       JBUF_UNLOCK (priv);
4224       break;
4225     case PROP_RTX_MAX_RETRIES:
4226       JBUF_LOCK (priv);
4227       g_value_set_int (value, priv->rtx_max_retries);
4228       JBUF_UNLOCK (priv);
4229       break;
4230     case PROP_STATS:
4231       g_value_take_boxed (value,
4232           gst_rtp_jitter_buffer_create_stats (jitterbuffer));
4233       break;
4234     case PROP_MAX_RTCP_RTP_TIME_DIFF:
4235       JBUF_LOCK (priv);
4236       g_value_set_int (value, priv->max_rtcp_rtp_time_diff);
4237       JBUF_UNLOCK (priv);
4238       break;
4239     case PROP_MAX_DROPOUT_TIME:
4240       JBUF_LOCK (priv);
4241       g_value_set_uint (value, priv->max_dropout_time);
4242       JBUF_UNLOCK (priv);
4243       break;
4244     case PROP_MAX_MISORDER_TIME:
4245       JBUF_LOCK (priv);
4246       g_value_set_uint (value, priv->max_misorder_time);
4247       JBUF_UNLOCK (priv);
4248       break;
4249     case PROP_RFC7273_SYNC:
4250       JBUF_LOCK (priv);
4251       g_value_set_boolean (value,
4252           rtp_jitter_buffer_get_rfc7273_sync (priv->jbuf));
4253       JBUF_UNLOCK (priv);
4254       break;
4255     default:
4256       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
4257       break;
4258   }
4259 }
4260
4261 static GstStructure *
4262 gst_rtp_jitter_buffer_create_stats (GstRtpJitterBuffer * jbuf)
4263 {
4264   GstStructure *s;
4265
4266   JBUF_LOCK (jbuf->priv);
4267   s = gst_structure_new ("application/x-rtp-jitterbuffer-stats",
4268       "rtx-count", G_TYPE_UINT64, jbuf->priv->num_rtx_requests,
4269       "rtx-success-count", G_TYPE_UINT64, jbuf->priv->num_rtx_success,
4270       "rtx-per-packet", G_TYPE_DOUBLE, jbuf->priv->avg_rtx_num,
4271       "rtx-rtt", G_TYPE_UINT64, jbuf->priv->avg_rtx_rtt, NULL);
4272   JBUF_UNLOCK (jbuf->priv);
4273
4274   return s;
4275 }