gst/rtpmanager/gstrtpjitterbuffer.c

   1 /*
   2  * Farsight Voice+Video library
   3  *
   4  *  Copyright 2007 Collabora Ltd,
   5  *  Copyright 2007 Nokia Corporation
   6  *   @author: Philippe Kalaf <philippe.kalaf@collabora.co.uk>.
   7  *  Copyright 2007 Wim Taymans <wim.taymans@gmail.com>
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Library General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Library General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Library General Public
  20  * License along with this library; if not, write to the
  21  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  22  * Boston, MA 02110-1301, USA.
  23  *
  24  */
  25
  26 /**
  27  * SECTION:element-rtpjitterbuffer
  28  *
  29  * This element reorders and removes duplicate RTP packets as they are received
  30  * from a network source.
  31  *
  32  * The element needs the clock-rate of the RTP payload in order to estimate the
  33  * delay. This information is obtained either from the caps on the sink pad or,
  34  * when no caps are present, from the #GstRtpJitterBuffer::request-pt-map signal.
  35  * To clear the previous pt-map use the #GstRtpJitterBuffer::clear-pt-map signal.
  36  *
  37  * The rtpjitterbuffer will wait for missing packets up to a configurable time
  38  * limit using the #GstRtpJitterBuffer:latency property. Packets arriving too
  39  * late are considered to be lost packets. If the #GstRtpJitterBuffer:do-lost
  40  * property is set, lost packets will result in a custom serialized downstream
  41  * event of name GstRTPPacketLost. The lost packet events are usually used by a
  42  * depayloader or other element to create concealment data or some other logic
  43  * to gracefully handle the missing packets.
  44  *
  45  * The jitterbuffer will use the DTS (or PTS if no DTS is set) of the incomming
  46  * buffer and the rtptime inside the RTP packet to create a PTS on the outgoing
  47  * buffer.
  48  *
  49  * The jitterbuffer can also be configured to send early retransmission events
  50  * upstream by setting the #GstRtpJitterBuffer:do-retransmission property. In
  51  * this mode, the jitterbuffer tries to estimate when a packet should arrive and
  52  * sends a custom upstream event named GstRTPRetransmissionRequest when the
  53  * packet is considered late. The initial expected packet arrival time is
  54  * calculated as follows:
  55  *
  56  * - If seqnum N arrived at time T, seqnum N+1 is expected to arrive at
  57  *     T + packet-spacing + #GstRtpJitterBuffer:rtx-delay. The packet spacing is
  58  *     calculated from the DTS (or PTS is no DTS) of two consecutive RTP
  59  *     packets with different rtptime.
  60  *
  61  * - If seqnum N0 arrived at time T0 and seqnum Nm arrived at time Tm,
  62  *     seqnum Ni is expected at time Ti = T0 + i*(Tm - T0)/(Nm - N0). Any
  63  *     previously scheduled timeout is overwritten.
  64  *
  65  * - If seqnum N arrived, all seqnum older than
  66  *     N - #GstRtpJitterBuffer:rtx-delay-reorder are considered late
  67  *     immediately. This is to request fast feedback for abonormally reorder
  68  *     packets before any of the previous timeouts is triggered.
  69  *
  70  * A late packet triggers the GstRTPRetransmissionRequest custom upstream
  71  * event. After the initial timeout expires and the retransmission event is
  72  * sent, the timeout is scheduled for
  73  * T + #GstRtpJitterBuffer:rtx-retry-timeout. If the missing packet did not
  74  * arrive after #GstRtpJitterBuffer:rtx-retry-timeout, a new
  75  * GstRTPRetransmissionRequest is sent upstream and the timeout is rescheduled
  76  * again for T + #GstRtpJitterBuffer:rtx-retry-timeout. This repeats until
  77  * #GstRtpJitterBuffer:rtx-retry-period elapsed, at which point no further
  78  * retransmission requests are sent and the regular logic is performed to
  79  * schedule a lost packet as discussed above.
  80  *
  81  * This element acts as a live element and so adds #GstRtpJitterBuffer:latency
  82  * to the pipeline.
  83  *
  84  * This element will automatically be used inside rtpbin.
  85  *
  86  * <refsect2>
  87  * <title>Example pipelines</title>
  88  * |[
  89  * gst-launch-1.0 rtspsrc location=rtsp://192.168.1.133:8554/mpeg1or2AudioVideoTest ! rtpjitterbuffer ! rtpmpvdepay ! mpeg2dec ! xvimagesink
  90  * ]| Connect to a streaming server and decode the MPEG video. The jitterbuffer is
  91  * inserted into the pipeline to smooth out network jitter and to reorder the
  92  * out-of-order RTP packets.
  93  * </refsect2>
  94  *
  95  * Last reviewed on 2007-05-28 (0.10.5)
  96  */
  97
  98 #ifdef HAVE_CONFIG_H
  99 #include "config.h"
 100 #endif
 101
 102 #include <stdlib.h>
 103 #include <string.h>
 104 #include <gst/rtp/gstrtpbuffer.h>
 105
 106 #include "gstrtpjitterbuffer.h"
 107 #include "rtpjitterbuffer.h"
 108 #include "rtpstats.h"
 109
 110 #include <gst/glib-compat-private.h>
 111
 112 GST_DEBUG_CATEGORY (rtpjitterbuffer_debug);
 113 #define GST_CAT_DEFAULT (rtpjitterbuffer_debug)
 114
 115 /* RTPJitterBuffer signals and args */
 116 enum
 117 {
 118   SIGNAL_REQUEST_PT_MAP,
 119   SIGNAL_CLEAR_PT_MAP,
 120   SIGNAL_HANDLE_SYNC,
 121   SIGNAL_ON_NPT_STOP,
 122   SIGNAL_SET_ACTIVE,
 123   LAST_SIGNAL
 124 };
 125
 126 #define DEFAULT_LATENCY_MS          200
 127 #define DEFAULT_DROP_ON_LATENCY     FALSE
 128 #define DEFAULT_TS_OFFSET           0
 129 #define DEFAULT_DO_LOST             FALSE
 130 #define DEFAULT_MODE                RTP_JITTER_BUFFER_MODE_SLAVE
 131 #define DEFAULT_PERCENT             0
 132 #define DEFAULT_DO_RETRANSMISSION   FALSE
 133 #define DEFAULT_RTX_DELAY           20
 134 #define DEFAULT_RTX_DELAY_REORDER   3
 135 #define DEFAULT_RTX_RETRY_TIMEOUT   40
 136 #define DEFAULT_RTX_RETRY_PERIOD    160
 137
 138 enum
 139 {
 140   PROP_0,
 141   PROP_LATENCY,
 142   PROP_DROP_ON_LATENCY,
 143   PROP_TS_OFFSET,
 144   PROP_DO_LOST,
 145   PROP_MODE,
 146   PROP_PERCENT,
 147   PROP_DO_RETRANSMISSION,
 148   PROP_RTX_DELAY,
 149   PROP_RTX_DELAY_REORDER,
 150   PROP_RTX_RETRY_TIMEOUT,
 151   PROP_RTX_RETRY_PERIOD,
 152   PROP_STATS,
 153   PROP_LAST
 154 };
 155
 156 #define JBUF_LOCK(priv)   (g_mutex_lock (&(priv)->jbuf_lock))
 157
 158 #define JBUF_LOCK_CHECK(priv,label) G_STMT_START {    \
 159   JBUF_LOCK (priv);                                   \
 160   if (G_UNLIKELY (priv->srcresult != GST_FLOW_OK))    \
 161     goto label;                                       \
 162 } G_STMT_END
 163 #define JBUF_UNLOCK(priv) (g_mutex_unlock (&(priv)->jbuf_lock))
 164
 165 #define JBUF_WAIT_TIMER(priv)   G_STMT_START {            \
 166   GST_DEBUG ("waiting timer");                            \
 167   (priv)->waiting_timer = TRUE;                           \
 168   g_cond_wait (&(priv)->jbuf_timer, &(priv)->jbuf_lock);  \
 169   (priv)->waiting_timer = FALSE;                          \
 170   GST_DEBUG ("waiting timer done");                       \
 171 } G_STMT_END
 172 #define JBUF_SIGNAL_TIMER(priv) G_STMT_START {            \
 173   if (G_UNLIKELY ((priv)->waiting_timer)) {               \
 174     GST_DEBUG ("signal timer");                           \
 175     g_cond_signal (&(priv)->jbuf_timer);                  \
 176   }                                                       \
 177 } G_STMT_END
 178
 179 #define JBUF_WAIT_EVENT(priv,label) G_STMT_START {       \
 180   GST_DEBUG ("waiting event");                           \
 181   (priv)->waiting_event = TRUE;                          \
 182   g_cond_wait (&(priv)->jbuf_event, &(priv)->jbuf_lock); \
 183   (priv)->waiting_event = FALSE;                         \
 184   GST_DEBUG ("waiting event done");                      \
 185   if (G_UNLIKELY (priv->srcresult != GST_FLOW_OK))       \
 186     goto label;                                          \
 187 } G_STMT_END
 188 #define JBUF_SIGNAL_EVENT(priv) G_STMT_START {           \
 189   if (G_UNLIKELY ((priv)->waiting_event)) {              \
 190     GST_DEBUG ("signal event");                          \
 191     g_cond_signal (&(priv)->jbuf_event);                 \
 192   }                                                      \
 193 } G_STMT_END
 194
 195 struct _GstRtpJitterBufferPrivate
 196 {
 197   GstPad *sinkpad, *srcpad;
 198   GstPad *rtcpsinkpad;
 199
 200   RTPJitterBuffer *jbuf;
 201   GMutex jbuf_lock;
 202   gboolean waiting_timer;
 203   GCond jbuf_timer;
 204   gboolean waiting_event;
 205   GCond jbuf_event;
 206   gboolean discont;
 207   gboolean ts_discont;
 208   gboolean active;
 209   guint64 out_offset;
 210
 211   gboolean timer_running;
 212   GThread *timer_thread;
 213
 214   /* properties */
 215   guint latency_ms;
 216   guint64 latency_ns;
 217   gboolean drop_on_latency;
 218   gint64 ts_offset;
 219   gboolean do_lost;
 220   gboolean do_retransmission;
 221   gint rtx_delay;
 222   gint rtx_delay_reorder;
 223   gint rtx_retry_timeout;
 224   gint rtx_retry_period;
 225
 226   /* the last seqnum we pushed out */
 227   guint32 last_popped_seqnum;
 228   /* the next expected seqnum we push */
 229   guint32 next_seqnum;
 230   /* last output time */
 231   GstClockTime last_out_time;
 232   /* last valid input timestamp and rtptime pair */
 233   GstClockTime ips_dts;
 234   guint64 ips_rtptime;
 235   GstClockTime packet_spacing;
 236
 237   /* the next expected seqnum we receive */
 238   GstClockTime last_in_dts;
 239   guint32 last_in_seqnum;
 240   guint32 next_in_seqnum;
 241
 242   GArray *timers;
 243
 244   /* start and stop ranges */
 245   GstClockTime npt_start;
 246   GstClockTime npt_stop;
 247   guint64 ext_timestamp;
 248   guint64 last_elapsed;
 249   guint64 estimated_eos;
 250   GstClockID eos_id;
 251
 252   /* state */
 253   gboolean eos;
 254
 255   /* clock rate and rtp timestamp offset */
 256   gint last_pt;
 257   gint32 clock_rate;
 258   gint64 clock_base;
 259   gint64 prev_ts_offset;
 260
 261   /* when we are shutting down */
 262   GstFlowReturn srcresult;
 263   gboolean blocked;
 264
 265   /* for sync */
 266   GstSegment segment;
 267   GstClockID clock_id;
 268   GstClockTime timer_timeout;
 269   guint16 timer_seqnum;
 270   /* the latency of the upstream peer, we have to take this into account when
 271    * synchronizing the buffers. */
 272   GstClockTime peer_latency;
 273   guint64 ext_rtptime;
 274   GstBuffer *last_sr;
 275
 276   /* some accounting */
 277   guint64 num_late;
 278   guint64 num_duplicates;
 279   guint64 num_rtx_requests;
 280   guint64 num_rtx_success;
 281   guint64 num_rtx_failed;
 282   gdouble avg_rtx_num;
 283   guint64 avg_rtx_rtt;
 284 };
 285
 286 typedef enum
 287 {
 288   TIMER_TYPE_EXPECTED,
 289   TIMER_TYPE_LOST,
 290   TIMER_TYPE_DEADLINE,
 291   TIMER_TYPE_EOS
 292 } TimerType;
 293
 294 typedef struct
 295 {
 296   guint idx;
 297   guint16 seqnum;
 298   guint num;
 299   TimerType type;
 300   GstClockTime timeout;
 301   GstClockTime duration;
 302   GstClockTime rtx_base;
 303   GstClockTime rtx_delay;
 304   GstClockTime rtx_retry;
 305   GstClockTime rtx_last;
 306   guint num_rtx_retry;
 307 } TimerData;
 308
 309 #define GST_RTP_JITTER_BUFFER_GET_PRIVATE(o) \
 310   (G_TYPE_INSTANCE_GET_PRIVATE ((o), GST_TYPE_RTP_JITTER_BUFFER, \
 311                                 GstRtpJitterBufferPrivate))
 312
 313 static GstStaticPadTemplate gst_rtp_jitter_buffer_sink_template =
 314 GST_STATIC_PAD_TEMPLATE ("sink",
 315     GST_PAD_SINK,
 316     GST_PAD_ALWAYS,
 317     GST_STATIC_CAPS ("application/x-rtp, "
 318         "clock-rate = (int) [ 1, 2147483647 ]"
 319         /* "payload = (int) , "
 320          * "encoding-name = (string) "
 321          */ )
 322     );
 323
 324 static GstStaticPadTemplate gst_rtp_jitter_buffer_sink_rtcp_template =
 325 GST_STATIC_PAD_TEMPLATE ("sink_rtcp",
 326     GST_PAD_SINK,
 327     GST_PAD_REQUEST,
 328     GST_STATIC_CAPS ("application/x-rtcp")
 329     );
 330
 331 static GstStaticPadTemplate gst_rtp_jitter_buffer_src_template =
 332 GST_STATIC_PAD_TEMPLATE ("src",
 333     GST_PAD_SRC,
 334     GST_PAD_ALWAYS,
 335     GST_STATIC_CAPS ("application/x-rtp"
 336         /* "payload = (int) , "
 337          * "clock-rate = (int) , "
 338          * "encoding-name = (string) "
 339          */ )
 340     );
 341
 342 static guint gst_rtp_jitter_buffer_signals[LAST_SIGNAL] = { 0 };
 343
 344 #define gst_rtp_jitter_buffer_parent_class parent_class
 345 G_DEFINE_TYPE (GstRtpJitterBuffer, gst_rtp_jitter_buffer, GST_TYPE_ELEMENT);
 346
 347 /* object overrides */
 348 static void gst_rtp_jitter_buffer_set_property (GObject * object,
 349     guint prop_id, const GValue * value, GParamSpec * pspec);
 350 static void gst_rtp_jitter_buffer_get_property (GObject * object,
 351     guint prop_id, GValue * value, GParamSpec * pspec);
 352 static void gst_rtp_jitter_buffer_finalize (GObject * object);
 353
 354 /* element overrides */
 355 static GstStateChangeReturn gst_rtp_jitter_buffer_change_state (GstElement
 356     * element, GstStateChange transition);
 357 static GstPad *gst_rtp_jitter_buffer_request_new_pad (GstElement * element,
 358     GstPadTemplate * templ, const gchar * name, const GstCaps * filter);
 359 static void gst_rtp_jitter_buffer_release_pad (GstElement * element,
 360     GstPad * pad);
 361 static GstClock *gst_rtp_jitter_buffer_provide_clock (GstElement * element);
 362
 363 /* pad overrides */
 364 static GstCaps *gst_rtp_jitter_buffer_getcaps (GstPad * pad, GstCaps * filter);
 365 static GstIterator *gst_rtp_jitter_buffer_iterate_internal_links (GstPad * pad,
 366     GstObject * parent);
 367
 368 /* sinkpad overrides */
 369 static gboolean gst_rtp_jitter_buffer_sink_event (GstPad * pad,
 370     GstObject * parent, GstEvent * event);
 371 static GstFlowReturn gst_rtp_jitter_buffer_chain (GstPad * pad,
 372     GstObject * parent, GstBuffer * buffer);
 373
 374 static gboolean gst_rtp_jitter_buffer_sink_rtcp_event (GstPad * pad,
 375     GstObject * parent, GstEvent * event);
 376 static GstFlowReturn gst_rtp_jitter_buffer_chain_rtcp (GstPad * pad,
 377     GstObject * parent, GstBuffer * buffer);
 378
 379 static gboolean gst_rtp_jitter_buffer_sink_query (GstPad * pad,
 380     GstObject * parent, GstQuery * query);
 381
 382 /* srcpad overrides */
 383 static gboolean gst_rtp_jitter_buffer_src_event (GstPad * pad,
 384     GstObject * parent, GstEvent * event);
 385 static gboolean gst_rtp_jitter_buffer_src_activate_mode (GstPad * pad,
 386     GstObject * parent, GstPadMode mode, gboolean active);
 387 static void gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer);
 388 static gboolean gst_rtp_jitter_buffer_src_query (GstPad * pad,
 389     GstObject * parent, GstQuery * query);
 390
 391 static void
 392 gst_rtp_jitter_buffer_clear_pt_map (GstRtpJitterBuffer * jitterbuffer);
 393 static GstClockTime
 394 gst_rtp_jitter_buffer_set_active (GstRtpJitterBuffer * jitterbuffer,
 395     gboolean active, guint64 base_time);
 396 static void do_handle_sync (GstRtpJitterBuffer * jitterbuffer);
 397
 398 static void unschedule_current_timer (GstRtpJitterBuffer * jitterbuffer);
 399 static void remove_all_timers (GstRtpJitterBuffer * jitterbuffer);
 400
 401 static void wait_next_timeout (GstRtpJitterBuffer * jitterbuffer);
 402
 403 static GstStructure *gst_rtp_jitter_buffer_create_stats (GstRtpJitterBuffer *
 404     jitterbuffer);
 405
 406 static void
 407 gst_rtp_jitter_buffer_class_init (GstRtpJitterBufferClass * klass)
 408 {
 409   GObjectClass *gobject_class;
 410   GstElementClass *gstelement_class;
 411
 412   gobject_class = (GObjectClass *) klass;
 413   gstelement_class = (GstElementClass *) klass;
 414
 415   g_type_class_add_private (klass, sizeof (GstRtpJitterBufferPrivate));
 416
 417   gobject_class->finalize = gst_rtp_jitter_buffer_finalize;
 418
 419   gobject_class->set_property = gst_rtp_jitter_buffer_set_property;
 420   gobject_class->get_property = gst_rtp_jitter_buffer_get_property;
 421
 422   /**
 423    * GstRtpJitterBuffer:latency:
 424    *
 425    * The maximum latency of the jitterbuffer. Packets will be kept in the buffer
 426    * for at most this time.
 427    */
 428   g_object_class_install_property (gobject_class, PROP_LATENCY,
 429       g_param_spec_uint ("latency", "Buffer latency in ms",
 430           "Amount of ms to buffer", 0, G_MAXUINT, DEFAULT_LATENCY_MS,
 431           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 432   /**
 433    * GstRtpJitterBuffer:drop-on-latency:
 434    *
 435    * Drop oldest buffers when the queue is completely filled.
 436    */
 437   g_object_class_install_property (gobject_class, PROP_DROP_ON_LATENCY,
 438       g_param_spec_boolean ("drop-on-latency",
 439           "Drop buffers when maximum latency is reached",
 440           "Tells the jitterbuffer to never exceed the given latency in size",
 441           DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 442   /**
 443    * GstRtpJitterBuffer:ts-offset:
 444    *
 445    * Adjust GStreamer output buffer timestamps in the jitterbuffer with offset.
 446    * This is mainly used to ensure interstream synchronisation.
 447    */
 448   g_object_class_install_property (gobject_class, PROP_TS_OFFSET,
 449       g_param_spec_int64 ("ts-offset", "Timestamp Offset",
 450           "Adjust buffer timestamps with offset in nanoseconds", G_MININT64,
 451           G_MAXINT64, DEFAULT_TS_OFFSET,
 452           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 453
 454   /**
 455    * GstRtpJitterBuffer:do-lost:
 456    *
 457    * Send out a GstRTPPacketLost event downstream when a packet is considered
 458    * lost.
 459    */
 460   g_object_class_install_property (gobject_class, PROP_DO_LOST,
 461       g_param_spec_boolean ("do-lost", "Do Lost",
 462           "Send an event downstream when a packet is lost", DEFAULT_DO_LOST,
 463           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 464
 465   /**
 466    * GstRtpJitterBuffer:mode:
 467    *
 468    * Control the buffering and timestamping mode used by the jitterbuffer.
 469    */
 470   g_object_class_install_property (gobject_class, PROP_MODE,
 471       g_param_spec_enum ("mode", "Mode",
 472           "Control the buffering algorithm in use", RTP_TYPE_JITTER_BUFFER_MODE,
 473           DEFAULT_MODE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 474   /**
 475    * GstRtpJitterBuffer:percent:
 476    *
 477    * The percent of the jitterbuffer that is filled.
 478    */
 479   g_object_class_install_property (gobject_class, PROP_PERCENT,
 480       g_param_spec_int ("percent", "percent",
 481           "The buffer filled percent", 0, 100,
 482           0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 483   /**
 484    * GstRtpJitterBuffer:do-retransmission:
 485    *
 486    * Send out a GstRTPRetransmission event upstream when a packet is considered
 487    * late and should be retransmitted.
 488    *
 489    * Since: 1.2
 490    */
 491   g_object_class_install_property (gobject_class, PROP_DO_RETRANSMISSION,
 492       g_param_spec_boolean ("do-retransmission", "Do Retransmission",
 493           "Send retransmission events upstream when a packet is late",
 494           DEFAULT_DO_RETRANSMISSION,
 495           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 496
 497   /**
 498    * GstRtpJitterBuffer:rtx-delay:
 499    *
 500    * When a packet did not arrive at the expected time, wait this extra amount
 501    * of time before sending a retransmission event.
 502    *
 503    * When -1 is used, the max jitter will be used as extra delay.
 504    *
 505    * Since: 1.2
 506    */
 507   g_object_class_install_property (gobject_class, PROP_RTX_DELAY,
 508       g_param_spec_int ("rtx-delay", "RTX Delay",
 509           "Extra time in ms to wait before sending retransmission "
 510           "event (-1 automatic)", -1, G_MAXINT, DEFAULT_RTX_DELAY,
 511           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 512   /**
 513    * GstRtpJitterBuffer:rtx-delay-reorder:
 514    *
 515    * Assume that a retransmission event should be sent when we see
 516    * this much packet reordering.
 517    *
 518    * When -1 is used, the value will be estimated based on observed packet
 519    * reordering.
 520    *
 521    * Since: 1.2
 522    */
 523   g_object_class_install_property (gobject_class, PROP_RTX_DELAY_REORDER,
 524       g_param_spec_int ("rtx-delay-reorder", "RTX Delay Reorder",
 525           "Sending retransmission event when this much reordering (-1 automatic)",
 526           -1, G_MAXINT, DEFAULT_RTX_DELAY_REORDER,
 527           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 528   /**
 529    * GstRtpJitterBuffer::rtx-retry-timeout:
 530    *
 531    * When no packet has been received after sending a retransmission event
 532    * for this time, retry sending a retransmission event.
 533    *
 534    * When -1 is used, the value will be estimated based on observed round
 535    * trip time.
 536    *
 537    * Since: 1.2
 538    */
 539   g_object_class_install_property (gobject_class, PROP_RTX_RETRY_TIMEOUT,
 540       g_param_spec_int ("rtx-retry-timeout", "RTX Retry Timeout",
 541           "Retry sending a transmission event after this timeout in "
 542           "ms (-1 automatic)", -1, G_MAXINT, DEFAULT_RTX_RETRY_TIMEOUT,
 543           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 544   /**
 545    * GstRtpJitterBuffer:rtx-retry-period:
 546    *
 547    * The amount of time to try to get a retransmission.
 548    *
 549    * When -1 is used, the value will be estimated based on the jitterbuffer
 550    * latency and the observed round trip time.
 551    *
 552    * Since: 1.2
 553    */
 554   g_object_class_install_property (gobject_class, PROP_RTX_RETRY_PERIOD,
 555       g_param_spec_int ("rtx-retry-period", "RTX Retry Period",
 556           "Try to get a retransmission for this many ms "
 557           "(-1 automatic)", -1, G_MAXINT, DEFAULT_RTX_RETRY_PERIOD,
 558           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 559   /**
 560    * GstRtpJitterBuffer:stats:
 561    *
 562    * Various jitterbuffer statistics. This property returns a GstStructure
 563    * with name application/x-rtp-jitterbuffer-stats with the following fields:
 564    *
 565    *  "rtx-count"         G_TYPE_UINT64 The number of retransmissions requested
 566    *  "rtx-success-count" G_TYPE_UINT64 The number of successful retransmissions
 567    *  "rtx-per-packet"    G_TYPE_DOUBLE Average number of RTX per packet
 568    *  "rtx-rtt"           G_TYPE_UINT64 Average round trip time per RTX
 569    *
 570    * Since: 1.4
 571    */
 572   g_object_class_install_property (gobject_class, PROP_STATS,
 573       g_param_spec_boxed ("stats", "Statistics",
 574           "Various statistics", GST_TYPE_STRUCTURE,
 575           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 576
 577   /**
 578    * GstRtpJitterBuffer::request-pt-map:
 579    * @buffer: the object which received the signal
 580    * @pt: the pt
 581    *
 582    * Request the payload type as #GstCaps for @pt.
 583    */
 584   gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP] =
 585       g_signal_new ("request-pt-map", G_TYPE_FROM_CLASS (klass),
 586       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
 587           request_pt_map), NULL, NULL, g_cclosure_marshal_generic,
 588       GST_TYPE_CAPS, 1, G_TYPE_UINT);
 589   /**
 590    * GstRtpJitterBuffer::handle-sync:
 591    * @buffer: the object which received the signal
 592    * @struct: a GstStructure containing sync values.
 593    *
 594    * Be notified of new sync values.
 595    */
 596   gst_rtp_jitter_buffer_signals[SIGNAL_HANDLE_SYNC] =
 597       g_signal_new ("handle-sync", G_TYPE_FROM_CLASS (klass),
 598       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
 599           handle_sync), NULL, NULL, g_cclosure_marshal_VOID__BOXED,
 600       G_TYPE_NONE, 1, GST_TYPE_STRUCTURE | G_SIGNAL_TYPE_STATIC_SCOPE);
 601
 602   /**
 603    * GstRtpJitterBuffer::on-npt-stop:
 604    * @buffer: the object which received the signal
 605    *
 606    * Signal that the jitterbufer has pushed the RTP packet that corresponds to
 607    * the npt-stop position.
 608    */
 609   gst_rtp_jitter_buffer_signals[SIGNAL_ON_NPT_STOP] =
 610       g_signal_new ("on-npt-stop", G_TYPE_FROM_CLASS (klass),
 611       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
 612           on_npt_stop), NULL, NULL, g_cclosure_marshal_VOID__VOID,
 613       G_TYPE_NONE, 0, G_TYPE_NONE);
 614
 615   /**
 616    * GstRtpJitterBuffer::clear-pt-map:
 617    * @buffer: the object which received the signal
 618    *
 619    * Invalidate the clock-rate as obtained with the
 620    * #GstRtpJitterBuffer::request-pt-map signal.
 621    */
 622   gst_rtp_jitter_buffer_signals[SIGNAL_CLEAR_PT_MAP] =
 623       g_signal_new ("clear-pt-map", G_TYPE_FROM_CLASS (klass),
 624       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 625       G_STRUCT_OFFSET (GstRtpJitterBufferClass, clear_pt_map), NULL, NULL,
 626       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0, G_TYPE_NONE);
 627
 628   /**
 629    * GstRtpJitterBuffer::set-active:
 630    * @buffer: the object which received the signal
 631    *
 632    * Start pushing out packets with the given base time. This signal is only
 633    * useful in buffering mode.
 634    *
 635    * Returns: the time of the last pushed packet.
 636    */
 637   gst_rtp_jitter_buffer_signals[SIGNAL_SET_ACTIVE] =
 638       g_signal_new ("set-active", G_TYPE_FROM_CLASS (klass),
 639       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 640       G_STRUCT_OFFSET (GstRtpJitterBufferClass, set_active), NULL, NULL,
 641       g_cclosure_marshal_generic, G_TYPE_UINT64, 2, G_TYPE_BOOLEAN,
 642       G_TYPE_UINT64);
 643
 644   gstelement_class->change_state =
 645       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_change_state);
 646   gstelement_class->request_new_pad =
 647       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_request_new_pad);
 648   gstelement_class->release_pad =
 649       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_release_pad);
 650   gstelement_class->provide_clock =
 651       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_provide_clock);
 652
 653   gst_element_class_add_pad_template (gstelement_class,
 654       gst_static_pad_template_get (&gst_rtp_jitter_buffer_src_template));
 655   gst_element_class_add_pad_template (gstelement_class,
 656       gst_static_pad_template_get (&gst_rtp_jitter_buffer_sink_template));
 657   gst_element_class_add_pad_template (gstelement_class,
 658       gst_static_pad_template_get (&gst_rtp_jitter_buffer_sink_rtcp_template));
 659
 660   gst_element_class_set_static_metadata (gstelement_class,
 661       "RTP packet jitter-buffer", "Filter/Network/RTP",
 662       "A buffer that deals with network jitter and other transmission faults",
 663       "Philippe Kalaf <philippe.kalaf@collabora.co.uk>, "
 664       "Wim Taymans <wim.taymans@gmail.com>");
 665
 666   klass->clear_pt_map = GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_clear_pt_map);
 667   klass->set_active = GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_set_active);
 668
 669   GST_DEBUG_CATEGORY_INIT
 670       (rtpjitterbuffer_debug, "rtpjitterbuffer", 0, "RTP Jitter Buffer");
 671 }
 672
 673 static void
 674 gst_rtp_jitter_buffer_init (GstRtpJitterBuffer * jitterbuffer)
 675 {
 676   GstRtpJitterBufferPrivate *priv;
 677
 678   priv = GST_RTP_JITTER_BUFFER_GET_PRIVATE (jitterbuffer);
 679   jitterbuffer->priv = priv;
 680
 681   priv->latency_ms = DEFAULT_LATENCY_MS;
 682   priv->latency_ns = priv->latency_ms * GST_MSECOND;
 683   priv->drop_on_latency = DEFAULT_DROP_ON_LATENCY;
 684   priv->do_lost = DEFAULT_DO_LOST;
 685   priv->do_retransmission = DEFAULT_DO_RETRANSMISSION;
 686   priv->rtx_delay = DEFAULT_RTX_DELAY;
 687   priv->rtx_delay_reorder = DEFAULT_RTX_DELAY_REORDER;
 688   priv->rtx_retry_timeout = DEFAULT_RTX_RETRY_TIMEOUT;
 689   priv->rtx_retry_period = DEFAULT_RTX_RETRY_PERIOD;
 690
 691   priv->timers = g_array_new (FALSE, TRUE, sizeof (TimerData));
 692   priv->jbuf = rtp_jitter_buffer_new ();
 693   g_mutex_init (&priv->jbuf_lock);
 694   g_cond_init (&priv->jbuf_timer);
 695   g_cond_init (&priv->jbuf_event);
 696
 697   /* reset skew detection initialy */
 698   rtp_jitter_buffer_reset_skew (priv->jbuf);
 699   rtp_jitter_buffer_set_delay (priv->jbuf, priv->latency_ns);
 700   rtp_jitter_buffer_set_buffering (priv->jbuf, FALSE);
 701   priv->active = TRUE;
 702
 703   priv->srcpad =
 704       gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_src_template,
 705       "src");
 706
 707   gst_pad_set_activatemode_function (priv->srcpad,
 708       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_activate_mode));
 709   gst_pad_set_query_function (priv->srcpad,
 710       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_query));
 711   gst_pad_set_event_function (priv->srcpad,
 712       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_event));
 713
 714   priv->sinkpad =
 715       gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_sink_template,
 716       "sink");
 717
 718   gst_pad_set_chain_function (priv->sinkpad,
 719       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_chain));
 720   gst_pad_set_event_function (priv->sinkpad,
 721       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_sink_event));
 722   gst_pad_set_query_function (priv->sinkpad,
 723       GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_sink_query));
 724
 725   gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->srcpad);
 726   gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->sinkpad);
 727
 728   GST_OBJECT_FLAG_SET (jitterbuffer, GST_ELEMENT_FLAG_PROVIDE_CLOCK);
 729 }
 730
 731 #define ITEM_TYPE_BUFFER        0
 732 #define ITEM_TYPE_LOST          1
 733
 734 static RTPJitterBufferItem *
 735 alloc_item (gpointer data, guint type, GstClockTime dts, GstClockTime pts,
 736     guint seqnum, guint count, guint rtptime)
 737 {
 738   RTPJitterBufferItem *item;
 739
 740   item = g_slice_new (RTPJitterBufferItem);
 741   item->data = data;
 742   item->next = NULL;
 743   item->prev = NULL;
 744   item->type = type;
 745   item->dts = dts;
 746   item->pts = pts;
 747   item->seqnum = seqnum;
 748   item->count = count;
 749   item->rtptime = rtptime;
 750
 751   return item;
 752 }
 753
 754 static void
 755 free_item (RTPJitterBufferItem * item)
 756 {
 757   if (item->data)
 758     gst_mini_object_unref (item->data);
 759   g_slice_free (RTPJitterBufferItem, item);
 760 }
 761
 762 static void
 763 gst_rtp_jitter_buffer_finalize (GObject * object)
 764 {
 765   GstRtpJitterBuffer *jitterbuffer;
 766   GstRtpJitterBufferPrivate *priv;
 767
 768   jitterbuffer = GST_RTP_JITTER_BUFFER (object);
 769   priv = jitterbuffer->priv;
 770
 771   g_array_free (priv->timers, TRUE);
 772   g_mutex_clear (&priv->jbuf_lock);
 773   g_cond_clear (&priv->jbuf_timer);
 774   g_cond_clear (&priv->jbuf_event);
 775
 776   rtp_jitter_buffer_flush (priv->jbuf, (GFunc) free_item, NULL);
 777   g_object_unref (priv->jbuf);
 778
 779   G_OBJECT_CLASS (parent_class)->finalize (object);
 780 }
 781
 782 static GstIterator *
 783 gst_rtp_jitter_buffer_iterate_internal_links (GstPad * pad, GstObject * parent)
 784 {
 785   GstRtpJitterBuffer *jitterbuffer;
 786   GstPad *otherpad = NULL;
 787   GstIterator *it;
 788   GValue val = { 0, };
 789
 790   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
 791
 792   if (pad == jitterbuffer->priv->sinkpad) {
 793     otherpad = jitterbuffer->priv->srcpad;
 794   } else if (pad == jitterbuffer->priv->srcpad) {
 795     otherpad = jitterbuffer->priv->sinkpad;
 796   } else if (pad == jitterbuffer->priv->rtcpsinkpad) {
 797     otherpad = NULL;
 798   }
 799
 800   g_value_init (&val, GST_TYPE_PAD);
 801   g_value_set_object (&val, otherpad);
 802   it = gst_iterator_new_single (GST_TYPE_PAD, &val);
 803   g_value_unset (&val);
 804
 805   return it;
 806 }
 807
 808 static GstPad *
 809 create_rtcp_sink (GstRtpJitterBuffer * jitterbuffer)
 810 {
 811   GstRtpJitterBufferPrivate *priv;
 812
 813   priv = jitterbuffer->priv;
 814
 815   GST_DEBUG_OBJECT (jitterbuffer, "creating RTCP sink pad");
 816
 817   priv->rtcpsinkpad =
 818       gst_pad_new_from_static_template
 819       (&gst_rtp_jitter_buffer_sink_rtcp_template, "sink_rtcp");
 820   gst_pad_set_chain_function (priv->rtcpsinkpad,
 821       gst_rtp_jitter_buffer_chain_rtcp);
 822   gst_pad_set_event_function (priv->rtcpsinkpad,
 823       (GstPadEventFunction) gst_rtp_jitter_buffer_sink_rtcp_event);
 824   gst_pad_set_iterate_internal_links_function (priv->rtcpsinkpad,
 825       gst_rtp_jitter_buffer_iterate_internal_links);
 826   gst_pad_set_active (priv->rtcpsinkpad, TRUE);
 827   gst_element_add_pad (GST_ELEMENT_CAST (jitterbuffer), priv->rtcpsinkpad);
 828
 829   return priv->rtcpsinkpad;
 830 }
 831
 832 static void
 833 remove_rtcp_sink (GstRtpJitterBuffer * jitterbuffer)
 834 {
 835   GstRtpJitterBufferPrivate *priv;
 836
 837   priv = jitterbuffer->priv;
 838
 839   GST_DEBUG_OBJECT (jitterbuffer, "removing RTCP sink pad");
 840
 841   gst_pad_set_active (priv->rtcpsinkpad, FALSE);
 842
 843   gst_element_remove_pad (GST_ELEMENT_CAST (jitterbuffer), priv->rtcpsinkpad);
 844   priv->rtcpsinkpad = NULL;
 845 }
 846
 847 static GstPad *
 848 gst_rtp_jitter_buffer_request_new_pad (GstElement * element,
 849     GstPadTemplate * templ, const gchar * name, const GstCaps * filter)
 850 {
 851   GstRtpJitterBuffer *jitterbuffer;
 852   GstElementClass *klass;
 853   GstPad *result;
 854   GstRtpJitterBufferPrivate *priv;
 855
 856   g_return_val_if_fail (templ != NULL, NULL);
 857   g_return_val_if_fail (GST_IS_RTP_JITTER_BUFFER (element), NULL);
 858
 859   jitterbuffer = GST_RTP_JITTER_BUFFER (element);
 860   priv = jitterbuffer->priv;
 861   klass = GST_ELEMENT_GET_CLASS (element);
 862
 863   GST_DEBUG_OBJECT (element, "requesting pad %s", GST_STR_NULL (name));
 864
 865   /* figure out the template */
 866   if (templ == gst_element_class_get_pad_template (klass, "sink_rtcp")) {
 867     if (priv->rtcpsinkpad != NULL)
 868       goto exists;
 869
 870     result = create_rtcp_sink (jitterbuffer);
 871   } else
 872     goto wrong_template;
 873
 874   return result;
 875
 876   /* ERRORS */
 877 wrong_template:
 878   {
 879     g_warning ("rtpjitterbuffer: this is not our template");
 880     return NULL;
 881   }
 882 exists:
 883   {
 884     g_warning ("rtpjitterbuffer: pad already requested");
 885     return NULL;
 886   }
 887 }
 888
 889 static void
 890 gst_rtp_jitter_buffer_release_pad (GstElement * element, GstPad * pad)
 891 {
 892   GstRtpJitterBuffer *jitterbuffer;
 893   GstRtpJitterBufferPrivate *priv;
 894
 895   g_return_if_fail (GST_IS_RTP_JITTER_BUFFER (element));
 896   g_return_if_fail (GST_IS_PAD (pad));
 897
 898   jitterbuffer = GST_RTP_JITTER_BUFFER (element);
 899   priv = jitterbuffer->priv;
 900
 901   GST_DEBUG_OBJECT (element, "releasing pad %s:%s", GST_DEBUG_PAD_NAME (pad));
 902
 903   if (priv->rtcpsinkpad == pad) {
 904     remove_rtcp_sink (jitterbuffer);
 905   } else
 906     goto wrong_pad;
 907
 908   return;
 909
 910   /* ERRORS */
 911 wrong_pad:
 912   {
 913     g_warning ("gstjitterbuffer: asked to release an unknown pad");
 914     return;
 915   }
 916 }
 917
 918 static GstClock *
 919 gst_rtp_jitter_buffer_provide_clock (GstElement * element)
 920 {
 921   return gst_system_clock_obtain ();
 922 }
 923
 924 static void
 925 gst_rtp_jitter_buffer_clear_pt_map (GstRtpJitterBuffer * jitterbuffer)
 926 {
 927   GstRtpJitterBufferPrivate *priv;
 928
 929   priv = jitterbuffer->priv;
 930
 931   /* this will trigger a new pt-map request signal, FIXME, do something better. */
 932
 933   JBUF_LOCK (priv);
 934   priv->clock_rate = -1;
 935   /* do not clear current content, but refresh state for new arrival */
 936   GST_DEBUG_OBJECT (jitterbuffer, "reset jitterbuffer");
 937   rtp_jitter_buffer_reset_skew (priv->jbuf);
 938   priv->last_popped_seqnum = -1;
 939   priv->next_seqnum = -1;
 940   JBUF_UNLOCK (priv);
 941 }
 942
 943 static GstClockTime
 944 gst_rtp_jitter_buffer_set_active (GstRtpJitterBuffer * jbuf, gboolean active,
 945     guint64 offset)
 946 {
 947   GstRtpJitterBufferPrivate *priv;
 948   GstClockTime last_out;
 949   RTPJitterBufferItem *item;
 950
 951   priv = jbuf->priv;
 952
 953   JBUF_LOCK (priv);
 954   GST_DEBUG_OBJECT (jbuf, "setting active %d with offset %" GST_TIME_FORMAT,
 955       active, GST_TIME_ARGS (offset));
 956
 957   if (active != priv->active) {
 958     /* add the amount of time spent in paused to the output offset. All
 959      * outgoing buffers will have this offset applied to their timestamps in
 960      * order to make them arrive in time in the sink. */
 961     priv->out_offset = offset;
 962     GST_DEBUG_OBJECT (jbuf, "out offset %" GST_TIME_FORMAT,
 963         GST_TIME_ARGS (priv->out_offset));
 964     priv->active = active;
 965     JBUF_SIGNAL_EVENT (priv);
 966   }
 967   if (!active) {
 968     rtp_jitter_buffer_set_buffering (priv->jbuf, TRUE);
 969   }
 970   if ((item = rtp_jitter_buffer_peek (priv->jbuf))) {
 971     /* head buffer timestamp and offset gives our output time */
 972     last_out = item->dts + priv->ts_offset;
 973   } else {
 974     /* use last known time when the buffer is empty */
 975     last_out = priv->last_out_time;
 976   }
 977   JBUF_UNLOCK (priv);
 978
 979   return last_out;
 980 }
 981
 982 static GstCaps *
 983 gst_rtp_jitter_buffer_getcaps (GstPad * pad, GstCaps * filter)
 984 {
 985   GstRtpJitterBuffer *jitterbuffer;
 986   GstRtpJitterBufferPrivate *priv;
 987   GstPad *other;
 988   GstCaps *caps;
 989   GstCaps *templ;
 990
 991   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
 992   priv = jitterbuffer->priv;
 993
 994   other = (pad == priv->srcpad ? priv->sinkpad : priv->srcpad);
 995
 996   caps = gst_pad_peer_query_caps (other, filter);
 997
 998   templ = gst_pad_get_pad_template_caps (pad);
 999   if (caps == NULL) {
1000     GST_DEBUG_OBJECT (jitterbuffer, "use template");
1001     caps = templ;
1002   } else {
1003     GstCaps *intersect;
1004
1005     GST_DEBUG_OBJECT (jitterbuffer, "intersect with template");
1006
1007     intersect = gst_caps_intersect (caps, templ);
1008     gst_caps_unref (caps);
1009     gst_caps_unref (templ);
1010
1011     caps = intersect;
1012   }
1013   gst_object_unref (jitterbuffer);
1014
1015   return caps;
1016 }
1017
1018 /*
1019  * Must be called with JBUF_LOCK held
1020  */
1021
1022 static gboolean
1023 gst_jitter_buffer_sink_parse_caps (GstRtpJitterBuffer * jitterbuffer,
1024     GstCaps * caps)
1025 {
1026   GstRtpJitterBufferPrivate *priv;
1027   GstStructure *caps_struct;
1028   guint val;
1029   GstClockTime tval;
1030
1031   priv = jitterbuffer->priv;
1032
1033   /* first parse the caps */
1034   caps_struct = gst_caps_get_structure (caps, 0);
1035
1036   GST_DEBUG_OBJECT (jitterbuffer, "got caps");
1037
1038   /* we need a clock-rate to convert the rtp timestamps to GStreamer time and to
1039    * measure the amount of data in the buffer */
1040   if (!gst_structure_get_int (caps_struct, "clock-rate", &priv->clock_rate))
1041     goto error;
1042
1043   if (priv->clock_rate <= 0)
1044     goto wrong_rate;
1045
1046   GST_DEBUG_OBJECT (jitterbuffer, "got clock-rate %d", priv->clock_rate);
1047
1048   rtp_jitter_buffer_set_clock_rate (priv->jbuf, priv->clock_rate);
1049
1050   /* The clock base is the RTP timestamp corrsponding to the npt-start value. We
1051    * can use this to track the amount of time elapsed on the sender. */
1052   if (gst_structure_get_uint (caps_struct, "clock-base", &val))
1053     priv->clock_base = val;
1054   else
1055     priv->clock_base = -1;
1056
1057   priv->ext_timestamp = priv->clock_base;
1058
1059   GST_DEBUG_OBJECT (jitterbuffer, "got clock-base %" G_GINT64_FORMAT,
1060       priv->clock_base);
1061
1062   if (gst_structure_get_uint (caps_struct, "seqnum-base", &val)) {
1063     /* first expected seqnum, only update when we didn't have a previous base. */
1064     if (priv->next_in_seqnum == -1)
1065       priv->next_in_seqnum = val;
1066     if (priv->next_seqnum == -1)
1067       priv->next_seqnum = val;
1068   }
1069
1070   GST_DEBUG_OBJECT (jitterbuffer, "got seqnum-base %d", priv->next_in_seqnum);
1071
1072   /* the start and stop times. The seqnum-base corresponds to the start time. We
1073    * will keep track of the seqnums on the output and when we reach the one
1074    * corresponding to npt-stop, we emit the npt-stop-reached signal */
1075   if (gst_structure_get_clock_time (caps_struct, "npt-start", &tval))
1076     priv->npt_start = tval;
1077   else
1078     priv->npt_start = 0;
1079
1080   if (gst_structure_get_clock_time (caps_struct, "npt-stop", &tval))
1081     priv->npt_stop = tval;
1082   else
1083     priv->npt_stop = -1;
1084
1085   GST_DEBUG_OBJECT (jitterbuffer,
1086       "npt start/stop: %" GST_TIME_FORMAT "-%" GST_TIME_FORMAT,
1087       GST_TIME_ARGS (priv->npt_start), GST_TIME_ARGS (priv->npt_stop));
1088
1089   return TRUE;
1090
1091   /* ERRORS */
1092 error:
1093   {
1094     GST_DEBUG_OBJECT (jitterbuffer, "No clock-rate in caps!");
1095     return FALSE;
1096   }
1097 wrong_rate:
1098   {
1099     GST_DEBUG_OBJECT (jitterbuffer, "Invalid clock-rate %d", priv->clock_rate);
1100     return FALSE;
1101   }
1102 }
1103
1104 static void
1105 gst_rtp_jitter_buffer_flush_start (GstRtpJitterBuffer * jitterbuffer)
1106 {
1107   GstRtpJitterBufferPrivate *priv;
1108
1109   priv = jitterbuffer->priv;
1110
1111   JBUF_LOCK (priv);
1112   /* mark ourselves as flushing */
1113   priv->srcresult = GST_FLOW_FLUSHING;
1114   GST_DEBUG_OBJECT (jitterbuffer, "Disabling pop on queue");
1115   /* this unblocks any waiting pops on the src pad task */
1116   JBUF_SIGNAL_EVENT (priv);
1117   JBUF_UNLOCK (priv);
1118 }
1119
1120 static void
1121 gst_rtp_jitter_buffer_flush_stop (GstRtpJitterBuffer * jitterbuffer)
1122 {
1123   GstRtpJitterBufferPrivate *priv;
1124
1125   priv = jitterbuffer->priv;
1126
1127   JBUF_LOCK (priv);
1128   GST_DEBUG_OBJECT (jitterbuffer, "Enabling pop on queue");
1129   /* Mark as non flushing */
1130   priv->srcresult = GST_FLOW_OK;
1131   gst_segment_init (&priv->segment, GST_FORMAT_TIME);
1132   priv->last_popped_seqnum = -1;
1133   priv->last_out_time = -1;
1134   priv->next_seqnum = -1;
1135   priv->ips_rtptime = -1;
1136   priv->ips_dts = GST_CLOCK_TIME_NONE;
1137   priv->packet_spacing = 0;
1138   priv->next_in_seqnum = -1;
1139   priv->clock_rate = -1;
1140   priv->eos = FALSE;
1141   priv->estimated_eos = -1;
1142   priv->last_elapsed = 0;
1143   priv->ext_timestamp = -1;
1144   GST_DEBUG_OBJECT (jitterbuffer, "flush and reset jitterbuffer");
1145   rtp_jitter_buffer_flush (priv->jbuf, (GFunc) free_item, NULL);
1146   rtp_jitter_buffer_reset_skew (priv->jbuf);
1147   remove_all_timers (jitterbuffer);
1148   JBUF_UNLOCK (priv);
1149 }
1150
1151 static gboolean
1152 gst_rtp_jitter_buffer_src_activate_mode (GstPad * pad, GstObject * parent,
1153     GstPadMode mode, gboolean active)
1154 {
1155   gboolean result;
1156   GstRtpJitterBuffer *jitterbuffer = NULL;
1157
1158   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
1159
1160   switch (mode) {
1161     case GST_PAD_MODE_PUSH:
1162       if (active) {
1163         /* allow data processing */
1164         gst_rtp_jitter_buffer_flush_stop (jitterbuffer);
1165
1166         /* start pushing out buffers */
1167         GST_DEBUG_OBJECT (jitterbuffer, "Starting task on srcpad");
1168         result = gst_pad_start_task (jitterbuffer->priv->srcpad,
1169             (GstTaskFunction) gst_rtp_jitter_buffer_loop, jitterbuffer, NULL);
1170       } else {
1171         /* make sure all data processing stops ASAP */
1172         gst_rtp_jitter_buffer_flush_start (jitterbuffer);
1173
1174         /* NOTE this will hardlock if the state change is called from the src pad
1175          * task thread because we will _join() the thread. */
1176         GST_DEBUG_OBJECT (jitterbuffer, "Stopping task on srcpad");
1177         result = gst_pad_stop_task (pad);
1178       }
1179       break;
1180     default:
1181       result = FALSE;
1182       break;
1183   }
1184   return result;
1185 }
1186
1187 static GstStateChangeReturn
1188 gst_rtp_jitter_buffer_change_state (GstElement * element,
1189     GstStateChange transition)
1190 {
1191   GstRtpJitterBuffer *jitterbuffer;
1192   GstRtpJitterBufferPrivate *priv;
1193   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
1194
1195   jitterbuffer = GST_RTP_JITTER_BUFFER (element);
1196   priv = jitterbuffer->priv;
1197
1198   switch (transition) {
1199     case GST_STATE_CHANGE_NULL_TO_READY:
1200       break;
1201     case GST_STATE_CHANGE_READY_TO_PAUSED:
1202       JBUF_LOCK (priv);
1203       /* reset negotiated values */
1204       priv->clock_rate = -1;
1205       priv->clock_base = -1;
1206       priv->peer_latency = 0;
1207       priv->last_pt = -1;
1208       /* block until we go to PLAYING */
1209       priv->blocked = TRUE;
1210       priv->timer_running = TRUE;
1211       priv->timer_thread =
1212           g_thread_new ("timer", (GThreadFunc) wait_next_timeout, jitterbuffer);
1213       JBUF_UNLOCK (priv);
1214       break;
1215     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1216       JBUF_LOCK (priv);
1217       /* unblock to allow streaming in PLAYING */
1218       priv->blocked = FALSE;
1219       JBUF_SIGNAL_EVENT (priv);
1220       JBUF_SIGNAL_TIMER (priv);
1221       JBUF_UNLOCK (priv);
1222       break;
1223     default:
1224       break;
1225   }
1226
1227   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1228
1229   switch (transition) {
1230     case GST_STATE_CHANGE_READY_TO_PAUSED:
1231       /* we are a live element because we sync to the clock, which we can only
1232        * do in the PLAYING state */
1233       if (ret != GST_STATE_CHANGE_FAILURE)
1234         ret = GST_STATE_CHANGE_NO_PREROLL;
1235       break;
1236     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1237       JBUF_LOCK (priv);
1238       /* block to stop streaming when PAUSED */
1239       priv->blocked = TRUE;
1240       unschedule_current_timer (jitterbuffer);
1241       JBUF_UNLOCK (priv);
1242       if (ret != GST_STATE_CHANGE_FAILURE)
1243         ret = GST_STATE_CHANGE_NO_PREROLL;
1244       break;
1245     case GST_STATE_CHANGE_PAUSED_TO_READY:
1246       JBUF_LOCK (priv);
1247       gst_buffer_replace (&priv->last_sr, NULL);
1248       priv->timer_running = FALSE;
1249       unschedule_current_timer (jitterbuffer);
1250       JBUF_SIGNAL_TIMER (priv);
1251       JBUF_UNLOCK (priv);
1252       g_thread_join (priv->timer_thread);
1253       priv->timer_thread = NULL;
1254       break;
1255     case GST_STATE_CHANGE_READY_TO_NULL:
1256       break;
1257     default:
1258       break;
1259   }
1260
1261   return ret;
1262 }
1263
1264 static gboolean
1265 gst_rtp_jitter_buffer_src_event (GstPad * pad, GstObject * parent,
1266     GstEvent * event)
1267 {
1268   gboolean ret = TRUE;
1269   GstRtpJitterBuffer *jitterbuffer;
1270   GstRtpJitterBufferPrivate *priv;
1271
1272   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
1273   priv = jitterbuffer->priv;
1274
1275   GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
1276
1277   switch (GST_EVENT_TYPE (event)) {
1278     case GST_EVENT_LATENCY:
1279     {
1280       GstClockTime latency;
1281
1282       gst_event_parse_latency (event, &latency);
1283
1284       GST_DEBUG_OBJECT (jitterbuffer,
1285           "configuring latency of %" GST_TIME_FORMAT, GST_TIME_ARGS (latency));
1286
1287       JBUF_LOCK (priv);
1288       /* adjust the overall buffer delay to the total pipeline latency in
1289        * buffering mode because if downstream consumes too fast (because of
1290        * large latency or queues, we would start rebuffering again. */
1291       if (rtp_jitter_buffer_get_mode (priv->jbuf) ==
1292           RTP_JITTER_BUFFER_MODE_BUFFER) {
1293         rtp_jitter_buffer_set_delay (priv->jbuf, latency);
1294       }
1295       JBUF_UNLOCK (priv);
1296
1297       ret = gst_pad_push_event (priv->sinkpad, event);
1298       break;
1299     }
1300     default:
1301       ret = gst_pad_push_event (priv->sinkpad, event);
1302       break;
1303   }
1304
1305   return ret;
1306 }
1307
1308 static gboolean
1309 gst_rtp_jitter_buffer_sink_event (GstPad * pad, GstObject * parent,
1310     GstEvent * event)
1311 {
1312   gboolean ret = TRUE;
1313   GstRtpJitterBuffer *jitterbuffer;
1314   GstRtpJitterBufferPrivate *priv;
1315
1316   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
1317   priv = jitterbuffer->priv;
1318
1319   GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
1320
1321   switch (GST_EVENT_TYPE (event)) {
1322     case GST_EVENT_CAPS:
1323     {
1324       GstCaps *caps;
1325
1326       gst_event_parse_caps (event, &caps);
1327
1328       JBUF_LOCK (priv);
1329       ret = gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
1330       JBUF_UNLOCK (priv);
1331
1332       /* set same caps on srcpad on success */
1333       if (ret)
1334         ret = gst_pad_push_event (priv->srcpad, event);
1335       else
1336         gst_event_unref (event);
1337       break;
1338     }
1339     case GST_EVENT_SEGMENT:
1340     {
1341       gst_event_copy_segment (event, &priv->segment);
1342
1343       /* we need time for now */
1344       if (priv->segment.format != GST_FORMAT_TIME)
1345         goto newseg_wrong_format;
1346
1347       GST_DEBUG_OBJECT (jitterbuffer,
1348           "newsegment:  %" GST_SEGMENT_FORMAT, &priv->segment);
1349
1350       /* FIXME, push SEGMENT in the queue. Sorting order might be difficult. */
1351       ret = gst_pad_push_event (priv->srcpad, event);
1352       break;
1353     }
1354     case GST_EVENT_FLUSH_START:
1355       ret = gst_pad_push_event (priv->srcpad, event);
1356       gst_rtp_jitter_buffer_flush_start (jitterbuffer);
1357       /* wait for the loop to go into PAUSED */
1358       gst_pad_pause_task (priv->srcpad);
1359       break;
1360     case GST_EVENT_FLUSH_STOP:
1361       ret = gst_pad_push_event (priv->srcpad, event);
1362       ret =
1363           gst_rtp_jitter_buffer_src_activate_mode (priv->srcpad, parent,
1364           GST_PAD_MODE_PUSH, TRUE);
1365       break;
1366     case GST_EVENT_EOS:
1367     {
1368       /* push EOS in queue. We always push it at the head */
1369       JBUF_LOCK (priv);
1370       /* check for flushing, we need to discard the event and return FALSE when
1371        * we are flushing */
1372       ret = priv->srcresult == GST_FLOW_OK;
1373       if (ret && !priv->eos) {
1374         GST_INFO_OBJECT (jitterbuffer, "queuing EOS");
1375         priv->eos = TRUE;
1376         JBUF_SIGNAL_EVENT (priv);
1377       } else if (priv->eos) {
1378         GST_DEBUG_OBJECT (jitterbuffer, "dropping EOS, we are already EOS");
1379       } else {
1380         GST_DEBUG_OBJECT (jitterbuffer, "dropping EOS, reason %s",
1381             gst_flow_get_name (priv->srcresult));
1382       }
1383       JBUF_UNLOCK (priv);
1384       gst_event_unref (event);
1385       break;
1386     }
1387     default:
1388       ret = gst_pad_push_event (priv->srcpad, event);
1389       break;
1390   }
1391
1392 done:
1393
1394   return ret;
1395
1396   /* ERRORS */
1397 newseg_wrong_format:
1398   {
1399     GST_DEBUG_OBJECT (jitterbuffer, "received non TIME newsegment");
1400     ret = FALSE;
1401     gst_event_unref (event);
1402     goto done;
1403   }
1404 }
1405
1406 static gboolean
1407 gst_rtp_jitter_buffer_sink_rtcp_event (GstPad * pad, GstObject * parent,
1408     GstEvent * event)
1409 {
1410   gboolean ret = TRUE;
1411   GstRtpJitterBuffer *jitterbuffer;
1412
1413   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
1414
1415   GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
1416
1417   switch (GST_EVENT_TYPE (event)) {
1418     case GST_EVENT_FLUSH_START:
1419       gst_event_unref (event);
1420       break;
1421     case GST_EVENT_FLUSH_STOP:
1422       gst_event_unref (event);
1423       break;
1424     default:
1425       ret = gst_pad_event_default (pad, parent, event);
1426       break;
1427   }
1428
1429   return ret;
1430 }
1431
1432 /*
1433  * Must be called with JBUF_LOCK held, will release the LOCK when emiting the
1434  * signal. The function returns GST_FLOW_ERROR when a parsing error happened and
1435  * GST_FLOW_FLUSHING when the element is shutting down. On success
1436  * GST_FLOW_OK is returned.
1437  */
1438 static GstFlowReturn
1439 gst_rtp_jitter_buffer_get_clock_rate (GstRtpJitterBuffer * jitterbuffer,
1440     guint8 pt)
1441 {
1442   GValue ret = { 0 };
1443   GValue args[2] = { {0}, {0} };
1444   GstCaps *caps;
1445   gboolean res;
1446
1447   g_value_init (&args[0], GST_TYPE_ELEMENT);
1448   g_value_set_object (&args[0], jitterbuffer);
1449   g_value_init (&args[1], G_TYPE_UINT);
1450   g_value_set_uint (&args[1], pt);
1451
1452   g_value_init (&ret, GST_TYPE_CAPS);
1453   g_value_set_boxed (&ret, NULL);
1454
1455   JBUF_UNLOCK (jitterbuffer->priv);
1456   g_signal_emitv (args, gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP], 0,
1457       &ret);
1458   JBUF_LOCK_CHECK (jitterbuffer->priv, out_flushing);
1459
1460   g_value_unset (&args[0]);
1461   g_value_unset (&args[1]);
1462   caps = (GstCaps *) g_value_dup_boxed (&ret);
1463   g_value_unset (&ret);
1464   if (!caps)
1465     goto no_caps;
1466
1467   res = gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
1468   gst_caps_unref (caps);
1469
1470   if (G_UNLIKELY (!res))
1471     goto parse_failed;
1472
1473   return GST_FLOW_OK;
1474
1475   /* ERRORS */
1476 no_caps:
1477   {
1478     GST_DEBUG_OBJECT (jitterbuffer, "could not get caps");
1479     return GST_FLOW_ERROR;
1480   }
1481 out_flushing:
1482   {
1483     GST_DEBUG_OBJECT (jitterbuffer, "we are flushing");
1484     return GST_FLOW_FLUSHING;
1485   }
1486 parse_failed:
1487   {
1488     GST_DEBUG_OBJECT (jitterbuffer, "parse failed");
1489     return GST_FLOW_ERROR;
1490   }
1491 }
1492
1493 /* call with jbuf lock held */
1494 static void
1495 check_buffering_percent (GstRtpJitterBuffer * jitterbuffer, gint * percent)
1496 {
1497   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1498
1499   /* too short a stream, or too close to EOS will never really fill buffer */
1500   if (*percent != -1 && priv->npt_stop != -1 &&
1501       priv->npt_stop - priv->npt_start <=
1502       rtp_jitter_buffer_get_delay (priv->jbuf)) {
1503     GST_DEBUG_OBJECT (jitterbuffer, "short stream; faking full buffer");
1504     rtp_jitter_buffer_set_buffering (priv->jbuf, FALSE);
1505     *percent = 100;
1506   }
1507 }
1508
1509 static void
1510 post_buffering_percent (GstRtpJitterBuffer * jitterbuffer, gint percent)
1511 {
1512   GstMessage *message;
1513
1514   /* Post a buffering message */
1515   message = gst_message_new_buffering (GST_OBJECT_CAST (jitterbuffer), percent);
1516   gst_message_set_buffering_stats (message, GST_BUFFERING_LIVE, -1, -1, -1);
1517
1518   gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer), message);
1519 }
1520
1521 static GstClockTime
1522 apply_offset (GstRtpJitterBuffer * jitterbuffer, GstClockTime timestamp)
1523 {
1524   GstRtpJitterBufferPrivate *priv;
1525
1526   priv = jitterbuffer->priv;
1527
1528   if (timestamp == -1)
1529     return -1;
1530
1531   /* apply the timestamp offset, this is used for inter stream sync */
1532   timestamp += priv->ts_offset;
1533   /* add the offset, this is used when buffering */
1534   timestamp += priv->out_offset;
1535
1536   return timestamp;
1537 }
1538
1539 static TimerData *
1540 find_timer (GstRtpJitterBuffer * jitterbuffer, TimerType type, guint16 seqnum)
1541 {
1542   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1543   TimerData *timer = NULL;
1544   gint i, len;
1545
1546   len = priv->timers->len;
1547   for (i = 0; i < len; i++) {
1548     TimerData *test = &g_array_index (priv->timers, TimerData, i);
1549     if (test->seqnum == seqnum && test->type == type) {
1550       timer = test;
1551       break;
1552     }
1553   }
1554   return timer;
1555 }
1556
1557 static void
1558 unschedule_current_timer (GstRtpJitterBuffer * jitterbuffer)
1559 {
1560   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1561
1562   if (priv->clock_id) {
1563     GST_DEBUG_OBJECT (jitterbuffer, "unschedule current timer");
1564     gst_clock_id_unschedule (priv->clock_id);
1565     priv->clock_id = NULL;
1566   }
1567 }
1568
1569 static GstClockTime
1570 get_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer)
1571 {
1572   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1573   GstClockTime test_timeout;
1574
1575   if ((test_timeout = timer->timeout) == -1)
1576     return -1;
1577
1578   if (timer->type != TIMER_TYPE_EXPECTED) {
1579     /* add our latency and offset to get output times. */
1580     test_timeout = apply_offset (jitterbuffer, test_timeout);
1581     test_timeout += priv->latency_ns;
1582   }
1583   return test_timeout;
1584 }
1585
1586 static void
1587 recalculate_timer (GstRtpJitterBuffer * jitterbuffer, TimerData * timer)
1588 {
1589   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1590
1591   if (priv->clock_id) {
1592     GstClockTime timeout = get_timeout (jitterbuffer, timer);
1593
1594     GST_DEBUG ("%" GST_TIME_FORMAT " <> %" GST_TIME_FORMAT,
1595         GST_TIME_ARGS (timeout), GST_TIME_ARGS (priv->timer_timeout));
1596
1597     if (timeout == -1 || timeout < priv->timer_timeout)
1598       unschedule_current_timer (jitterbuffer);
1599   }
1600 }
1601
1602 static TimerData *
1603 add_timer (GstRtpJitterBuffer * jitterbuffer, TimerType type,
1604     guint16 seqnum, guint num, GstClockTime timeout, GstClockTime delay,
1605     GstClockTime duration)
1606 {
1607   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1608   TimerData *timer;
1609   gint len;
1610
1611   GST_DEBUG_OBJECT (jitterbuffer,
1612       "add timer for seqnum %d to %" GST_TIME_FORMAT ", delay %"
1613       GST_TIME_FORMAT, seqnum, GST_TIME_ARGS (timeout), GST_TIME_ARGS (delay));
1614
1615   len = priv->timers->len;
1616   g_array_set_size (priv->timers, len + 1);
1617   timer = &g_array_index (priv->timers, TimerData, len);
1618   timer->idx = len;
1619   timer->type = type;
1620   timer->seqnum = seqnum;
1621   timer->num = num;
1622   timer->timeout = timeout + delay;
1623   timer->duration = duration;
1624   if (type == TIMER_TYPE_EXPECTED) {
1625     timer->rtx_base = timeout;
1626     timer->rtx_delay = delay;
1627     timer->rtx_retry = 0;
1628   }
1629   timer->num_rtx_retry = 0;
1630   recalculate_timer (jitterbuffer, timer);
1631   JBUF_SIGNAL_TIMER (priv);
1632
1633   return timer;
1634 }
1635
1636 static void
1637 reschedule_timer (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
1638     guint16 seqnum, GstClockTime timeout, GstClockTime delay, gboolean reset)
1639 {
1640   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1641   gboolean seqchange, timechange;
1642   guint16 oldseq;
1643
1644   seqchange = timer->seqnum != seqnum;
1645   timechange = timer->timeout != timeout;
1646
1647   if (!seqchange && !timechange)
1648     return;
1649
1650   oldseq = timer->seqnum;
1651
1652   GST_DEBUG_OBJECT (jitterbuffer,
1653       "replace timer for seqnum %d->%d to %" GST_TIME_FORMAT,
1654       oldseq, seqnum, GST_TIME_ARGS (timeout + delay));
1655
1656   timer->timeout = timeout + delay;
1657   timer->seqnum = seqnum;
1658   if (reset) {
1659     timer->rtx_base = timeout;
1660     timer->rtx_delay = delay;
1661     timer->rtx_retry = 0;
1662   }
1663
1664   if (priv->clock_id) {
1665     /* we changed the seqnum and there is a timer currently waiting with this
1666      * seqnum, unschedule it */
1667     if (seqchange && priv->timer_seqnum == oldseq)
1668       unschedule_current_timer (jitterbuffer);
1669     /* we changed the time, check if it is earlier than what we are waiting
1670      * for and unschedule if so */
1671     else if (timechange)
1672       recalculate_timer (jitterbuffer, timer);
1673   }
1674 }
1675
1676 static TimerData *
1677 set_timer (GstRtpJitterBuffer * jitterbuffer, TimerType type,
1678     guint16 seqnum, GstClockTime timeout)
1679 {
1680   TimerData *timer;
1681
1682   /* find the seqnum timer */
1683   timer = find_timer (jitterbuffer, type, seqnum);
1684   if (timer == NULL) {
1685     timer = add_timer (jitterbuffer, type, seqnum, 0, timeout, 0, -1);
1686   } else {
1687     reschedule_timer (jitterbuffer, timer, seqnum, timeout, 0, FALSE);
1688   }
1689   return timer;
1690 }
1691
1692 static void
1693 remove_timer (GstRtpJitterBuffer * jitterbuffer, TimerData * timer)
1694 {
1695   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1696   guint idx;
1697
1698   if (priv->clock_id && priv->timer_seqnum == timer->seqnum)
1699     unschedule_current_timer (jitterbuffer);
1700
1701   idx = timer->idx;
1702   GST_DEBUG_OBJECT (jitterbuffer, "removed index %d", idx);
1703   g_array_remove_index_fast (priv->timers, idx);
1704   timer->idx = idx;
1705 }
1706
1707 static void
1708 remove_all_timers (GstRtpJitterBuffer * jitterbuffer)
1709 {
1710   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1711   GST_DEBUG_OBJECT (jitterbuffer, "removed all timers");
1712   g_array_set_size (priv->timers, 0);
1713   unschedule_current_timer (jitterbuffer);
1714 }
1715
1716 /* we just received a packet with seqnum and dts.
1717  *
1718  * First check for old seqnum that we are still expecting. If the gap with the
1719  * current seqnum is too big, unschedule the timeouts.
1720  *
1721  * If we have a valid packet spacing estimate we can set a timer for when we
1722  * should receive the next packet.
1723  * If we don't have a valid estimate, we remove any timer we might have
1724  * had for this packet.
1725  */
1726 static void
1727 update_timers (GstRtpJitterBuffer * jitterbuffer, guint16 seqnum,
1728     GstClockTime dts, gboolean do_next_seqnum)
1729 {
1730   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1731   TimerData *timer = NULL;
1732   gint i, len;
1733
1734   /* go through all timers and unschedule the ones with a large gap, also find
1735    * the timer for the seqnum */
1736   len = priv->timers->len;
1737   for (i = 0; i < len; i++) {
1738     TimerData *test = &g_array_index (priv->timers, TimerData, i);
1739     gint gap;
1740
1741     gap = gst_rtp_buffer_compare_seqnum (test->seqnum, seqnum);
1742
1743     GST_DEBUG_OBJECT (jitterbuffer, "%d, #%d<->#%d gap %d", i,
1744         test->seqnum, seqnum, gap);
1745
1746     if (gap == 0) {
1747       GST_DEBUG ("found timer for current seqnum");
1748       /* the timer for the current seqnum */
1749       timer = test;
1750     } else if (gap > priv->rtx_delay_reorder) {
1751       /* max gap, we exceeded the max reorder distance and we don't expect the
1752        * missing packet to be this reordered */
1753       if (test->num_rtx_retry == 0 && test->type == TIMER_TYPE_EXPECTED)
1754         reschedule_timer (jitterbuffer, test, test->seqnum, -1, 0, FALSE);
1755     }
1756   }
1757
1758   if (priv->packet_spacing > 0 && do_next_seqnum && priv->do_retransmission) {
1759     GstClockTime expected, delay;
1760
1761     /* calculate expected arrival time of the next seqnum */
1762     expected = dts + priv->packet_spacing;
1763     delay = priv->rtx_delay * GST_MSECOND;
1764
1765     /* and update/install timer for next seqnum */
1766     if (timer)
1767       reschedule_timer (jitterbuffer, timer, priv->next_in_seqnum, expected,
1768           delay, TRUE);
1769     else
1770       add_timer (jitterbuffer, TIMER_TYPE_EXPECTED, priv->next_in_seqnum, 0,
1771           expected, delay, priv->packet_spacing);
1772   } else if (timer && timer->type != TIMER_TYPE_DEADLINE) {
1773
1774     if (timer->num_rtx_retry > 0) {
1775       GstClockTime rtx_last;
1776
1777       /* we scheduled a retry for this packet and now we have it */
1778       priv->num_rtx_success++;
1779       /* all the previous retry attempts failed */
1780       priv->num_rtx_failed += timer->num_rtx_retry - 1;
1781       /* number of retries before receiving the packet */
1782       if (priv->avg_rtx_num == 0.0)
1783         priv->avg_rtx_num = timer->num_rtx_retry;
1784       else
1785         priv->avg_rtx_num = (timer->num_rtx_retry + 7 * priv->avg_rtx_num) / 8;
1786       /* calculate the delay between retransmission request and receiving this
1787        * packet, start with when we scheduled this timeout last */
1788       rtx_last = timer->rtx_last;
1789       if (dts > rtx_last) {
1790         GstClockTime delay;
1791         /* we have a valid delay if this packet arrived after we scheduled the
1792          * request */
1793         delay = dts - rtx_last;
1794         if (priv->avg_rtx_rtt == 0)
1795           priv->avg_rtx_rtt = delay;
1796         else
1797           priv->avg_rtx_rtt = (delay + 7 * priv->avg_rtx_rtt) / 8;
1798       }
1799       GST_LOG_OBJECT (jitterbuffer,
1800           "RTX success %" G_GUINT64_FORMAT ", failed %" G_GUINT64_FORMAT
1801           ", requests %" G_GUINT64_FORMAT ", dups %" G_GUINT64_FORMAT
1802           ", avg-num %g, avg-rtt %" G_GUINT64_FORMAT, priv->num_rtx_success,
1803           priv->num_rtx_failed, priv->num_rtx_requests, priv->num_duplicates,
1804           priv->avg_rtx_num, priv->avg_rtx_rtt);
1805     }
1806     /* if we had a timer, remove it, we don't know when to expect the next
1807      * packet. */
1808     remove_timer (jitterbuffer, timer);
1809     /* we signal the _loop function because this new packet could be the one
1810      * it was waiting for */
1811     JBUF_SIGNAL_EVENT (priv);
1812   }
1813 }
1814
1815 static void
1816 calculate_packet_spacing (GstRtpJitterBuffer * jitterbuffer, guint32 rtptime,
1817     GstClockTime dts)
1818 {
1819   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1820
1821   /* we need consecutive seqnums with a different
1822    * rtptime to estimate the packet spacing. */
1823   if (priv->ips_rtptime != rtptime) {
1824     /* rtptime changed, check dts diff */
1825     if (priv->ips_dts != -1 && dts != -1 && dts > priv->ips_dts) {
1826       priv->packet_spacing = dts - priv->ips_dts;
1827       GST_DEBUG_OBJECT (jitterbuffer,
1828           "new packet spacing %" GST_TIME_FORMAT,
1829           GST_TIME_ARGS (priv->packet_spacing));
1830     }
1831     priv->ips_rtptime = rtptime;
1832     priv->ips_dts = dts;
1833   }
1834 }
1835
1836 static void
1837 calculate_expected (GstRtpJitterBuffer * jitterbuffer, guint32 expected,
1838     guint16 seqnum, GstClockTime dts, gint gap)
1839 {
1840   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
1841   GstClockTime total_duration, duration, expected_dts;
1842   TimerType type;
1843
1844   GST_DEBUG_OBJECT (jitterbuffer,
1845       "dts %" GST_TIME_FORMAT ", last %" GST_TIME_FORMAT,
1846       GST_TIME_ARGS (dts), GST_TIME_ARGS (priv->last_in_dts));
1847
1848   /* the total duration spanned by the missing packets */
1849   if (dts >= priv->last_in_dts)
1850     total_duration = dts - priv->last_in_dts;
1851   else
1852     total_duration = 0;
1853
1854   /* interpolate between the current time and the last time based on
1855    * number of packets we are missing, this is the estimated duration
1856    * for the missing packet based on equidistant packet spacing. */
1857   duration = total_duration / (gap + 1);
1858
1859   GST_DEBUG_OBJECT (jitterbuffer, "duration %" GST_TIME_FORMAT,
1860       GST_TIME_ARGS (duration));
1861
1862   if (total_duration > priv->latency_ns) {
1863     GstClockTime gap_time;
1864     guint lost_packets;
1865
1866     gap_time = total_duration - priv->latency_ns;
1867
1868     if (duration > 0) {
1869       lost_packets = gap_time / duration;
1870       gap_time = lost_packets * duration;
1871     } else {
1872       lost_packets = gap;
1873     }
1874
1875     /* too many lost packets, some of the missing packets are already
1876      * too late and we can generate lost packet events for them. */
1877     GST_DEBUG_OBJECT (jitterbuffer, "too many lost packets %" GST_TIME_FORMAT
1878         " > %" GST_TIME_FORMAT ", consider %u lost",
1879         GST_TIME_ARGS (total_duration), GST_TIME_ARGS (priv->latency_ns),
1880         lost_packets);
1881
1882     /* this timer will fire immediately and the lost event will be pushed from
1883      * the timer thread */
1884     add_timer (jitterbuffer, TIMER_TYPE_LOST, expected, lost_packets,
1885         priv->last_in_dts + duration, 0, gap_time);
1886
1887     expected += lost_packets;
1888     priv->last_in_dts += gap_time;
1889   }
1890
1891   expected_dts = priv->last_in_dts + duration;
1892
1893   if (priv->do_retransmission) {
1894     TimerData *timer;
1895
1896     type = TIMER_TYPE_EXPECTED;
1897     /* if we had a timer for the first missing packet, update it. */
1898     if ((timer = find_timer (jitterbuffer, type, expected))) {
1899       GstClockTime timeout = timer->timeout;
1900
1901       timer->duration = duration;
1902       if (timeout > expected_dts) {
1903         GstClockTime delay = timeout - expected_dts - timer->rtx_retry;
1904         reschedule_timer (jitterbuffer, timer, timer->seqnum, expected_dts,
1905             delay, TRUE);
1906       }
1907       expected++;
1908       expected_dts += duration;
1909     }
1910   } else {
1911     type = TIMER_TYPE_LOST;
1912   }
1913
1914   while (expected < seqnum) {
1915     add_timer (jitterbuffer, type, expected, 0, expected_dts, 0, duration);
1916     expected_dts += duration;
1917     expected++;
1918   }
1919 }
1920
1921 static GstFlowReturn
1922 gst_rtp_jitter_buffer_chain (GstPad * pad, GstObject * parent,
1923     GstBuffer * buffer)
1924 {
1925   GstRtpJitterBuffer *jitterbuffer;
1926   GstRtpJitterBufferPrivate *priv;
1927   guint16 seqnum;
1928   guint32 expected, rtptime;
1929   GstFlowReturn ret = GST_FLOW_OK;
1930   GstClockTime dts, pts;
1931   guint64 latency_ts;
1932   gboolean tail;
1933   gint percent = -1;
1934   guint8 pt;
1935   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
1936   gboolean do_next_seqnum = FALSE;
1937   RTPJitterBufferItem *item;
1938
1939   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
1940
1941   priv = jitterbuffer->priv;
1942
1943   if (G_UNLIKELY (!gst_rtp_buffer_map (buffer, GST_MAP_READ, &rtp)))
1944     goto invalid_buffer;
1945
1946   pt = gst_rtp_buffer_get_payload_type (&rtp);
1947   seqnum = gst_rtp_buffer_get_seq (&rtp);
1948   rtptime = gst_rtp_buffer_get_timestamp (&rtp);
1949   gst_rtp_buffer_unmap (&rtp);
1950
1951   /* make sure we have PTS and DTS set */
1952   pts = GST_BUFFER_PTS (buffer);
1953   dts = GST_BUFFER_DTS (buffer);
1954   if (dts == -1)
1955     dts = pts;
1956   else if (pts == -1)
1957     pts = dts;
1958
1959   /* take the DTS of the buffer. This is the time when the packet was
1960    * received and is used to calculate jitter and clock skew. We will adjust
1961    * this DTS with the smoothed value after processing it in the
1962    * jitterbuffer and assign it as the PTS. */
1963   /* bring to running time */
1964   dts = gst_segment_to_running_time (&priv->segment, GST_FORMAT_TIME, dts);
1965
1966   GST_DEBUG_OBJECT (jitterbuffer,
1967       "Received packet #%d at time %" GST_TIME_FORMAT ", discont %d", seqnum,
1968       GST_TIME_ARGS (dts), GST_BUFFER_IS_DISCONT (buffer));
1969
1970   JBUF_LOCK_CHECK (priv, out_flushing);
1971
1972   if (G_UNLIKELY (priv->last_pt != pt)) {
1973     GstCaps *caps;
1974
1975     GST_DEBUG_OBJECT (jitterbuffer, "pt changed from %u to %u", priv->last_pt,
1976         pt);
1977
1978     priv->last_pt = pt;
1979     /* reset clock-rate so that we get a new one */
1980     priv->clock_rate = -1;
1981
1982     /* Try to get the clock-rate from the caps first if we can. If there are no
1983      * caps we must fire the signal to get the clock-rate. */
1984     if ((caps = gst_pad_get_current_caps (pad))) {
1985       gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
1986       gst_caps_unref (caps);
1987     }
1988   }
1989
1990   if (G_UNLIKELY (priv->clock_rate == -1)) {
1991     /* no clock rate given on the caps, try to get one with the signal */
1992     if (gst_rtp_jitter_buffer_get_clock_rate (jitterbuffer,
1993             pt) == GST_FLOW_FLUSHING)
1994       goto out_flushing;
1995
1996     if (G_UNLIKELY (priv->clock_rate == -1))
1997       goto no_clock_rate;
1998   }
1999
2000   /* don't accept more data on EOS */
2001   if (G_UNLIKELY (priv->eos))
2002     goto have_eos;
2003
2004   expected = priv->next_in_seqnum;
2005
2006   /* now check against our expected seqnum */
2007   if (G_LIKELY (expected != -1)) {
2008     gint gap;
2009
2010     /* now calculate gap */
2011     gap = gst_rtp_buffer_compare_seqnum (expected, seqnum);
2012
2013     GST_DEBUG_OBJECT (jitterbuffer, "expected #%d, got #%d, gap of %d",
2014         expected, seqnum, gap);
2015
2016     if (G_LIKELY (gap == 0)) {
2017       /* packet is expected */
2018       calculate_packet_spacing (jitterbuffer, rtptime, dts);
2019       do_next_seqnum = TRUE;
2020     } else {
2021       gboolean reset = FALSE;
2022
2023       if (gap < 0) {
2024         /* we received an old packet */
2025         if (G_UNLIKELY (gap < -RTP_MAX_MISORDER)) {
2026           /* too old packet, reset */
2027           GST_DEBUG_OBJECT (jitterbuffer, "reset: buffer too old %d < %d", gap,
2028               -RTP_MAX_MISORDER);
2029           reset = TRUE;
2030         } else {
2031           GST_DEBUG_OBJECT (jitterbuffer, "old packet received");
2032         }
2033       } else {
2034         /* new packet, we are missing some packets */
2035         if (G_UNLIKELY (gap > RTP_MAX_DROPOUT)) {
2036           /* packet too far in future, reset */
2037           GST_DEBUG_OBJECT (jitterbuffer, "reset: buffer too new %d > %d", gap,
2038               RTP_MAX_DROPOUT);
2039           reset = TRUE;
2040         } else {
2041           GST_DEBUG_OBJECT (jitterbuffer, "%d missing packets", gap);
2042           /* fill in the gap with EXPECTED timers */
2043           calculate_expected (jitterbuffer, expected, seqnum, dts, gap);
2044
2045           do_next_seqnum = TRUE;
2046         }
2047       }
2048       if (G_UNLIKELY (reset)) {
2049         GST_DEBUG_OBJECT (jitterbuffer, "flush and reset jitterbuffer");
2050         rtp_jitter_buffer_flush (priv->jbuf, (GFunc) free_item, NULL);
2051         rtp_jitter_buffer_reset_skew (priv->jbuf);
2052         remove_all_timers (jitterbuffer);
2053         priv->last_popped_seqnum = -1;
2054         priv->next_seqnum = seqnum;
2055         do_next_seqnum = TRUE;
2056       }
2057       /* reset spacing estimation when gap */
2058       priv->ips_rtptime = -1;
2059       priv->ips_dts = GST_CLOCK_TIME_NONE;
2060     }
2061   } else {
2062     GST_DEBUG_OBJECT (jitterbuffer, "First buffer #%d", seqnum);
2063     /* we don't know what the next_in_seqnum should be, wait for the last
2064      * possible moment to push this buffer, maybe we get an earlier seqnum
2065      * while we wait */
2066     set_timer (jitterbuffer, TIMER_TYPE_DEADLINE, seqnum, dts);
2067     do_next_seqnum = TRUE;
2068     /* take rtptime and dts to calculate packet spacing */
2069     priv->ips_rtptime = rtptime;
2070     priv->ips_dts = dts;
2071   }
2072   if (do_next_seqnum) {
2073     priv->last_in_seqnum = seqnum;
2074     priv->last_in_dts = dts;
2075     priv->next_in_seqnum = (seqnum + 1) & 0xffff;
2076   }
2077
2078   /* let's check if this buffer is too late, we can only accept packets with
2079    * bigger seqnum than the one we last pushed. */
2080   if (G_LIKELY (priv->last_popped_seqnum != -1)) {
2081     gint gap;
2082
2083     gap = gst_rtp_buffer_compare_seqnum (priv->last_popped_seqnum, seqnum);
2084
2085     /* priv->last_popped_seqnum >= seqnum, we're too late. */
2086     if (G_UNLIKELY (gap <= 0))
2087       goto too_late;
2088   }
2089
2090   /* let's drop oldest packet if the queue is already full and drop-on-latency
2091    * is set. We can only do this when there actually is a latency. When no
2092    * latency is set, we just pump it in the queue and let the other end push it
2093    * out as fast as possible. */
2094   if (priv->latency_ms && priv->drop_on_latency) {
2095     latency_ts =
2096         gst_util_uint64_scale_int (priv->latency_ms, priv->clock_rate, 1000);
2097
2098     if (G_UNLIKELY (rtp_jitter_buffer_get_ts_diff (priv->jbuf) >= latency_ts)) {
2099       RTPJitterBufferItem *old_item;
2100
2101       old_item = rtp_jitter_buffer_pop (priv->jbuf, &percent);
2102       GST_DEBUG_OBJECT (jitterbuffer, "Queue full, dropping old packet %p",
2103           old_item);
2104       priv->next_seqnum = (old_item->seqnum + 1) & 0xffff;
2105       free_item (old_item);
2106     }
2107   }
2108
2109   item = alloc_item (buffer, ITEM_TYPE_BUFFER, dts, pts, seqnum, 1, rtptime);
2110
2111   /* now insert the packet into the queue in sorted order. This function returns
2112    * FALSE if a packet with the same seqnum was already in the queue, meaning we
2113    * have a duplicate. */
2114   if (G_UNLIKELY (!rtp_jitter_buffer_insert (priv->jbuf, item,
2115               &tail, &percent)))
2116     goto duplicate;
2117
2118   /* update timers */
2119   update_timers (jitterbuffer, seqnum, dts, do_next_seqnum);
2120
2121   /* we had an unhandled SR, handle it now */
2122   if (priv->last_sr)
2123     do_handle_sync (jitterbuffer);
2124
2125   /* signal addition of new buffer when the _loop is waiting. */
2126   if (priv->active)
2127     JBUF_SIGNAL_EVENT (priv);
2128
2129   /* let's unschedule and unblock any waiting buffers. We only want to do this
2130    * when the tail buffer changed */
2131   if (G_UNLIKELY (priv->clock_id && tail)) {
2132     GST_DEBUG_OBJECT (jitterbuffer, "Unscheduling waiting new buffer");
2133     unschedule_current_timer (jitterbuffer);
2134   }
2135
2136   GST_DEBUG_OBJECT (jitterbuffer, "Pushed packet #%d, now %d packets, tail: %d",
2137       seqnum, rtp_jitter_buffer_num_packets (priv->jbuf), tail);
2138
2139   check_buffering_percent (jitterbuffer, &percent);
2140
2141 finished:
2142   JBUF_UNLOCK (priv);
2143
2144   if (percent != -1)
2145     post_buffering_percent (jitterbuffer, percent);
2146
2147   return ret;
2148
2149   /* ERRORS */
2150 invalid_buffer:
2151   {
2152     /* this is not fatal but should be filtered earlier */
2153     GST_ELEMENT_WARNING (jitterbuffer, STREAM, DECODE, (NULL),
2154         ("Received invalid RTP payload, dropping"));
2155     gst_buffer_unref (buffer);
2156     return GST_FLOW_OK;
2157   }
2158 no_clock_rate:
2159   {
2160     GST_WARNING_OBJECT (jitterbuffer,
2161         "No clock-rate in caps!, dropping buffer");
2162     gst_buffer_unref (buffer);
2163     goto finished;
2164   }
2165 out_flushing:
2166   {
2167     ret = priv->srcresult;
2168     GST_DEBUG_OBJECT (jitterbuffer, "flushing %s", gst_flow_get_name (ret));
2169     gst_buffer_unref (buffer);
2170     goto finished;
2171   }
2172 have_eos:
2173   {
2174     ret = GST_FLOW_EOS;
2175     GST_WARNING_OBJECT (jitterbuffer, "we are EOS, refusing buffer");
2176     gst_buffer_unref (buffer);
2177     goto finished;
2178   }
2179 too_late:
2180   {
2181     GST_WARNING_OBJECT (jitterbuffer, "Packet #%d too late as #%d was already"
2182         " popped, dropping", seqnum, priv->last_popped_seqnum);
2183     priv->num_late++;
2184     gst_buffer_unref (buffer);
2185     goto finished;
2186   }
2187 duplicate:
2188   {
2189     GST_WARNING_OBJECT (jitterbuffer, "Duplicate packet #%d detected, dropping",
2190         seqnum);
2191     priv->num_duplicates++;
2192     free_item (item);
2193     goto finished;
2194   }
2195 }
2196
2197 static GstClockTime
2198 compute_elapsed (GstRtpJitterBuffer * jitterbuffer, RTPJitterBufferItem * item)
2199 {
2200   guint64 ext_time, elapsed;
2201   guint32 rtp_time;
2202   GstRtpJitterBufferPrivate *priv;
2203
2204   priv = jitterbuffer->priv;
2205   rtp_time = item->rtptime;
2206
2207   GST_LOG_OBJECT (jitterbuffer, "rtp %" G_GUINT32_FORMAT ", ext %"
2208       G_GUINT64_FORMAT, rtp_time, priv->ext_timestamp);
2209
2210   if (rtp_time < priv->ext_timestamp) {
2211     ext_time = priv->ext_timestamp;
2212   } else {
2213     ext_time = gst_rtp_buffer_ext_timestamp (&priv->ext_timestamp, rtp_time);
2214   }
2215
2216   if (ext_time > priv->clock_base)
2217     elapsed = ext_time - priv->clock_base;
2218   else
2219     elapsed = 0;
2220
2221   elapsed = gst_util_uint64_scale_int (elapsed, GST_SECOND, priv->clock_rate);
2222   return elapsed;
2223 }
2224
2225 static void
2226 update_estimated_eos (GstRtpJitterBuffer * jitterbuffer,
2227     RTPJitterBufferItem * item)
2228 {
2229   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2230
2231   if (priv->npt_stop != -1 && priv->ext_timestamp != -1
2232       && priv->clock_base != -1 && priv->clock_rate > 0) {
2233     guint64 elapsed, estimated;
2234
2235     elapsed = compute_elapsed (jitterbuffer, item);
2236
2237     if (elapsed > priv->last_elapsed || !priv->last_elapsed) {
2238       guint64 left;
2239       GstClockTime out_time;
2240
2241       priv->last_elapsed = elapsed;
2242
2243       left = priv->npt_stop - priv->npt_start;
2244       GST_LOG_OBJECT (jitterbuffer, "left %" GST_TIME_FORMAT,
2245           GST_TIME_ARGS (left));
2246
2247       out_time = item->dts;
2248
2249       if (elapsed > 0)
2250         estimated = gst_util_uint64_scale (out_time, left, elapsed);
2251       else {
2252         /* if there is almost nothing left,
2253          * we may never advance enough to end up in the above case */
2254         if (left < GST_SECOND)
2255           estimated = GST_SECOND;
2256         else
2257           estimated = -1;
2258       }
2259
2260       GST_LOG_OBJECT (jitterbuffer, "elapsed %" GST_TIME_FORMAT ", estimated %"
2261           GST_TIME_FORMAT, GST_TIME_ARGS (elapsed), GST_TIME_ARGS (estimated));
2262
2263       if (estimated != -1 && priv->estimated_eos != estimated) {
2264         set_timer (jitterbuffer, TIMER_TYPE_EOS, -1, estimated);
2265         priv->estimated_eos = estimated;
2266       }
2267     }
2268   }
2269 }
2270
2271 /* take a buffer from the queue and push it */
2272 static GstFlowReturn
2273 pop_and_push_next (GstRtpJitterBuffer * jitterbuffer, guint16 seqnum)
2274 {
2275   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2276   GstFlowReturn result;
2277   RTPJitterBufferItem *item;
2278   GstBuffer *outbuf;
2279   GstEvent *outevent;
2280   GstClockTime dts, pts;
2281   gint percent = -1;
2282   gboolean is_buffer, do_push = TRUE;
2283
2284   /* when we get here we are ready to pop and push the buffer */
2285   item = rtp_jitter_buffer_pop (priv->jbuf, &percent);
2286
2287   is_buffer = GST_IS_BUFFER (item->data);
2288
2289   if (is_buffer) {
2290     check_buffering_percent (jitterbuffer, &percent);
2291
2292     /* we need to make writable to change the flags and timestamps */
2293     outbuf = gst_buffer_make_writable (item->data);
2294
2295     if (G_UNLIKELY (priv->discont)) {
2296       /* set DISCONT flag when we missed a packet. We pushed the buffer writable
2297        * into the jitterbuffer so we can modify now. */
2298       GST_DEBUG_OBJECT (jitterbuffer, "mark output buffer discont");
2299       GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_DISCONT);
2300       priv->discont = FALSE;
2301     }
2302     if (G_UNLIKELY (priv->ts_discont)) {
2303       GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_RESYNC);
2304       priv->ts_discont = FALSE;
2305     }
2306
2307     dts = gst_segment_to_position (&priv->segment, GST_FORMAT_TIME, item->dts);
2308     pts = gst_segment_to_position (&priv->segment, GST_FORMAT_TIME, item->pts);
2309
2310     /* apply timestamp with offset to buffer now */
2311     GST_BUFFER_DTS (outbuf) = apply_offset (jitterbuffer, dts);
2312     GST_BUFFER_PTS (outbuf) = apply_offset (jitterbuffer, pts);
2313
2314     /* update the elapsed time when we need to check against the npt stop time. */
2315     update_estimated_eos (jitterbuffer, item);
2316
2317     priv->last_out_time = GST_BUFFER_PTS (outbuf);
2318   } else {
2319     outevent = item->data;
2320     if (item->type == ITEM_TYPE_LOST) {
2321       priv->discont = TRUE;
2322       if (!priv->do_lost)
2323         do_push = FALSE;
2324     }
2325   }
2326
2327   /* now we are ready to push the buffer. Save the seqnum and release the lock
2328    * so the other end can push stuff in the queue again. */
2329   priv->last_popped_seqnum = seqnum;
2330   priv->next_seqnum = (seqnum + item->count) & 0xffff;
2331   JBUF_UNLOCK (priv);
2332
2333   item->data = NULL;
2334   free_item (item);
2335
2336   if (is_buffer) {
2337     /* push buffer */
2338     if (percent != -1)
2339       post_buffering_percent (jitterbuffer, percent);
2340
2341     GST_DEBUG_OBJECT (jitterbuffer,
2342         "Pushing buffer %d, dts %" GST_TIME_FORMAT ", pts %" GST_TIME_FORMAT,
2343         seqnum, GST_TIME_ARGS (GST_BUFFER_DTS (outbuf)),
2344         GST_TIME_ARGS (GST_BUFFER_PTS (outbuf)));
2345     result = gst_pad_push (priv->srcpad, outbuf);
2346   } else {
2347     GST_DEBUG_OBJECT (jitterbuffer, "Pushing event %d", seqnum);
2348
2349     if (do_push)
2350       gst_pad_push_event (priv->srcpad, outevent);
2351     else
2352       gst_event_unref (outevent);
2353
2354     result = GST_FLOW_OK;
2355   }
2356   JBUF_LOCK_CHECK (priv, out_flushing);
2357
2358   return result;
2359
2360   /* ERRORS */
2361 out_flushing:
2362   {
2363     return priv->srcresult;
2364   }
2365 }
2366
2367 #define GST_FLOW_WAIT GST_FLOW_CUSTOM_SUCCESS
2368
2369 /* Peek a buffer and compare the seqnum to the expected seqnum.
2370  * If all is fine, the buffer is pushed.
2371  * If something is wrong, we wait for some event
2372  */
2373 static GstFlowReturn
2374 handle_next_buffer (GstRtpJitterBuffer * jitterbuffer)
2375 {
2376   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2377   GstFlowReturn result = GST_FLOW_OK;
2378   RTPJitterBufferItem *item;
2379   guint16 seqnum;
2380   guint32 next_seqnum;
2381   gint gap;
2382
2383   /* only push buffers when PLAYING and active and not buffering */
2384   if (priv->blocked || !priv->active ||
2385       rtp_jitter_buffer_is_buffering (priv->jbuf))
2386     return GST_FLOW_WAIT;
2387
2388 again:
2389   /* peek a buffer, we're just looking at the sequence number.
2390    * If all is fine, we'll pop and push it. If the sequence number is wrong we
2391    * wait for a timeout or something to change.
2392    * The peeked buffer is valid for as long as we hold the jitterbuffer lock. */
2393   item = rtp_jitter_buffer_peek (priv->jbuf);
2394   if (item == NULL)
2395     goto wait;
2396
2397   /* get the seqnum and the next expected seqnum */
2398   seqnum = item->seqnum;
2399
2400   next_seqnum = priv->next_seqnum;
2401
2402   /* get the gap between this and the previous packet. If we don't know the
2403    * previous packet seqnum assume no gap. */
2404   if (G_UNLIKELY (next_seqnum == -1)) {
2405     GST_DEBUG_OBJECT (jitterbuffer, "First buffer #%d", seqnum);
2406     /* we don't know what the next_seqnum should be, the chain function should
2407      * have scheduled a DEADLINE timer that will increment next_seqnum when it
2408      * fires, so wait for that */
2409     result = GST_FLOW_WAIT;
2410   } else {
2411     /* else calculate GAP */
2412     gap = gst_rtp_buffer_compare_seqnum (next_seqnum, seqnum);
2413
2414     if (G_LIKELY (gap == 0)) {
2415       /* no missing packet, pop and push */
2416       result = pop_and_push_next (jitterbuffer, seqnum);
2417     } else if (G_UNLIKELY (gap < 0)) {
2418       RTPJitterBufferItem *item;
2419       /* if we have a packet that we already pushed or considered dropped, pop it
2420        * off and get the next packet */
2421       GST_DEBUG_OBJECT (jitterbuffer, "Old packet #%d, next #%d dropping",
2422           seqnum, next_seqnum);
2423       item = rtp_jitter_buffer_pop (priv->jbuf, NULL);
2424       free_item (item);
2425       goto again;
2426     } else {
2427       /* the chain function has scheduled timers to request retransmission or
2428        * when to consider the packet lost, wait for that */
2429       GST_DEBUG_OBJECT (jitterbuffer,
2430           "Sequence number GAP detected: expected %d instead of %d (%d missing)",
2431           next_seqnum, seqnum, gap);
2432       result = GST_FLOW_WAIT;
2433     }
2434   }
2435   return result;
2436
2437 wait:
2438   {
2439     GST_DEBUG_OBJECT (jitterbuffer, "no buffer, going to wait");
2440     if (priv->eos)
2441       result = GST_FLOW_EOS;
2442     else
2443       result = GST_FLOW_WAIT;
2444     return result;
2445   }
2446 }
2447
2448 /* the timeout for when we expected a packet expired */
2449 static gboolean
2450 do_expected_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
2451     GstClockTime now)
2452 {
2453   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2454   GstEvent *event;
2455   guint delay;
2456
2457   GST_DEBUG_OBJECT (jitterbuffer, "expected %d didn't arrive", timer->seqnum);
2458
2459   delay = timer->rtx_delay + timer->rtx_retry;
2460   event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
2461       gst_structure_new ("GstRTPRetransmissionRequest",
2462           "seqnum", G_TYPE_UINT, (guint) timer->seqnum,
2463           "running-time", G_TYPE_UINT64, timer->rtx_base,
2464           "delay", G_TYPE_UINT, GST_TIME_AS_MSECONDS (delay),
2465           "retry", G_TYPE_UINT, timer->num_rtx_retry,
2466           "frequency", G_TYPE_UINT, priv->rtx_retry_timeout,
2467           "period", G_TYPE_UINT, priv->rtx_retry_period,
2468           "deadline", G_TYPE_UINT, priv->latency_ms,
2469           "packet-spacing", G_TYPE_UINT64, priv->packet_spacing, NULL));
2470
2471   priv->num_rtx_requests++;
2472   timer->num_rtx_retry++;
2473   timer->rtx_last = now;
2474
2475   /* calculate the timeout for the next retransmission attempt */
2476   timer->rtx_retry += (priv->rtx_retry_timeout * GST_MSECOND);
2477   GST_DEBUG_OBJECT (jitterbuffer, "base %" GST_TIME_FORMAT ", delay %"
2478       GST_TIME_FORMAT ", retry %" GST_TIME_FORMAT,
2479       GST_TIME_ARGS (timer->rtx_base), GST_TIME_ARGS (timer->rtx_delay),
2480       GST_TIME_ARGS (timer->rtx_retry));
2481
2482   if (timer->rtx_retry + timer->rtx_delay >
2483       (priv->rtx_retry_period * GST_MSECOND)) {
2484     GST_DEBUG_OBJECT (jitterbuffer, "reschedule as LOST timer");
2485     /* too many retransmission request, we now convert the timer
2486      * to a lost timer, leave the num_rtx_retry as it is for stats */
2487     timer->type = TIMER_TYPE_LOST;
2488     timer->rtx_delay = 0;
2489     timer->rtx_retry = 0;
2490   }
2491   reschedule_timer (jitterbuffer, timer, timer->seqnum,
2492       timer->rtx_base + timer->rtx_retry, timer->rtx_delay, FALSE);
2493
2494   JBUF_UNLOCK (priv);
2495   gst_pad_push_event (priv->sinkpad, event);
2496   JBUF_LOCK (priv);
2497
2498   return FALSE;
2499 }
2500
2501 /* a packet is lost */
2502 static gboolean
2503 do_lost_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
2504     GstClockTime now)
2505 {
2506   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2507   GstClockTime duration, timestamp;
2508   guint seqnum, lost_packets, num_rtx_retry;
2509   gboolean late;
2510   GstEvent *event;
2511   RTPJitterBufferItem *item;
2512
2513   seqnum = timer->seqnum;
2514   timestamp = apply_offset (jitterbuffer, timer->timeout);
2515   duration = timer->duration;
2516   if (duration == GST_CLOCK_TIME_NONE && priv->packet_spacing > 0)
2517     duration = priv->packet_spacing;
2518   lost_packets = MAX (timer->num, 1);
2519   late = timer->num > 0;
2520   num_rtx_retry = timer->num_rtx_retry;
2521
2522   /* we had a gap and thus we lost some packets. Create an event for this.  */
2523   if (lost_packets > 1)
2524     GST_DEBUG_OBJECT (jitterbuffer, "Packets #%d -> #%d lost", seqnum,
2525         seqnum + lost_packets - 1);
2526   else
2527     GST_DEBUG_OBJECT (jitterbuffer, "Packet #%d lost", seqnum);
2528
2529   priv->num_late += lost_packets;
2530   priv->num_rtx_failed += num_rtx_retry;
2531
2532   /* create paket lost event */
2533   event = gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM,
2534       gst_structure_new ("GstRTPPacketLost",
2535           "seqnum", G_TYPE_UINT, (guint) seqnum,
2536           "timestamp", G_TYPE_UINT64, timestamp,
2537           "duration", G_TYPE_UINT64, duration,
2538           "late", G_TYPE_BOOLEAN, late,
2539           "retry", G_TYPE_UINT, num_rtx_retry, NULL));
2540
2541   item = alloc_item (event, ITEM_TYPE_LOST, -1, -1, seqnum, lost_packets, -1);
2542   rtp_jitter_buffer_insert (priv->jbuf, item, NULL, NULL);
2543
2544   /* remove timer now */
2545   remove_timer (jitterbuffer, timer);
2546   JBUF_SIGNAL_EVENT (priv);
2547
2548   return TRUE;
2549 }
2550
2551 static gboolean
2552 do_eos_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
2553     GstClockTime now)
2554 {
2555   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2556
2557   GST_INFO_OBJECT (jitterbuffer, "got the NPT timeout");
2558   remove_timer (jitterbuffer, timer);
2559   JBUF_SIGNAL_EVENT (priv);
2560
2561   return TRUE;
2562 }
2563
2564 static gboolean
2565 do_deadline_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
2566     GstClockTime now)
2567 {
2568   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2569
2570   GST_INFO_OBJECT (jitterbuffer, "got deadline timeout");
2571
2572   priv->next_seqnum = timer->seqnum;
2573   remove_timer (jitterbuffer, timer);
2574   JBUF_SIGNAL_EVENT (priv);
2575
2576   return TRUE;
2577 }
2578
2579 static gboolean
2580 do_timeout (GstRtpJitterBuffer * jitterbuffer, TimerData * timer,
2581     GstClockTime now)
2582 {
2583   gboolean removed = FALSE;
2584
2585   switch (timer->type) {
2586     case TIMER_TYPE_EXPECTED:
2587       removed = do_expected_timeout (jitterbuffer, timer, now);
2588       break;
2589     case TIMER_TYPE_LOST:
2590       removed = do_lost_timeout (jitterbuffer, timer, now);
2591       break;
2592     case TIMER_TYPE_DEADLINE:
2593       removed = do_deadline_timeout (jitterbuffer, timer, now);
2594       break;
2595     case TIMER_TYPE_EOS:
2596       removed = do_eos_timeout (jitterbuffer, timer, now);
2597       break;
2598   }
2599   return removed;
2600 }
2601
2602 /* called when we need to wait for the next timeout.
2603  *
2604  * We loop over the array of recorded timeouts and wait for the earliest one.
2605  * When it timed out, do the logic associated with the timer.
2606  *
2607  * If there are no timers, we wait on a gcond until something new happens.
2608  */
2609 static void
2610 wait_next_timeout (GstRtpJitterBuffer * jitterbuffer)
2611 {
2612   GstRtpJitterBufferPrivate *priv = jitterbuffer->priv;
2613   GstClockTime now = 0;
2614
2615   JBUF_LOCK (priv);
2616   while (priv->timer_running) {
2617     TimerData *timer = NULL;
2618     GstClockTime timer_timeout = -1;
2619     gint i, len;
2620
2621     GST_DEBUG_OBJECT (jitterbuffer, "now %" GST_TIME_FORMAT,
2622         GST_TIME_ARGS (now));
2623
2624     len = priv->timers->len;
2625     for (i = 0; i < len; i++) {
2626       TimerData *test = &g_array_index (priv->timers, TimerData, i);
2627       GstClockTime test_timeout = get_timeout (jitterbuffer, test);
2628       gboolean save_best = FALSE;
2629
2630       GST_DEBUG_OBJECT (jitterbuffer, "%d, %d, %d, %" GST_TIME_FORMAT,
2631           i, test->type, test->seqnum, GST_TIME_ARGS (test_timeout));
2632
2633       /* find the smallest timeout */
2634       if (timer == NULL) {
2635         save_best = TRUE;
2636       } else if (timer_timeout == -1) {
2637         /* we already have an immediate timeout, the new timer must be an
2638          * immediate timer with smaller seqnum to become the best */
2639         if (test_timeout == -1 && test->seqnum < timer->seqnum)
2640           save_best = TRUE;
2641       } else if (test_timeout == -1) {
2642         /* first immediate timer */
2643         save_best = TRUE;
2644       } else if (test_timeout < timer_timeout) {
2645         /* earlier timer */
2646         save_best = TRUE;
2647       } else if (test_timeout == timer_timeout && test->seqnum < timer->seqnum) {
2648         /* same timer, smaller seqnum */
2649         save_best = TRUE;
2650       }
2651       if (save_best) {
2652         GST_DEBUG_OBJECT (jitterbuffer, "new best %d", i);
2653         timer = test;
2654         timer_timeout = test_timeout;
2655       }
2656     }
2657     if (timer && !priv->blocked) {
2658       GstClock *clock;
2659       GstClockTime sync_time;
2660       GstClockID id;
2661       GstClockReturn ret;
2662       GstClockTimeDiff clock_jitter;
2663
2664       if (timer_timeout == -1 || timer_timeout <= now) {
2665         do_timeout (jitterbuffer, timer, now);
2666         /* check here, do_timeout could have released the lock */
2667         if (!priv->timer_running)
2668           break;
2669         continue;
2670       }
2671
2672       GST_OBJECT_LOCK (jitterbuffer);
2673       clock = GST_ELEMENT_CLOCK (jitterbuffer);
2674       if (!clock) {
2675         GST_OBJECT_UNLOCK (jitterbuffer);
2676         /* let's just push if there is no clock */
2677         GST_DEBUG_OBJECT (jitterbuffer, "No clock, timeout right away");
2678         now = timer_timeout;
2679         continue;
2680       }
2681
2682       /* prepare for sync against clock */
2683       sync_time = timer_timeout + GST_ELEMENT_CAST (jitterbuffer)->base_time;
2684       /* add latency of peer to get input time */
2685       sync_time += priv->peer_latency;
2686
2687       GST_DEBUG_OBJECT (jitterbuffer, "sync to timestamp %" GST_TIME_FORMAT
2688           " with sync time %" GST_TIME_FORMAT,
2689           GST_TIME_ARGS (timer_timeout), GST_TIME_ARGS (sync_time));
2690
2691       /* create an entry for the clock */
2692       id = priv->clock_id = gst_clock_new_single_shot_id (clock, sync_time);
2693       priv->timer_timeout = timer_timeout;
2694       priv->timer_seqnum = timer->seqnum;
2695       GST_OBJECT_UNLOCK (jitterbuffer);
2696
2697       /* release the lock so that the other end can push stuff or unlock */
2698       JBUF_UNLOCK (priv);
2699
2700       ret = gst_clock_id_wait (id, &clock_jitter);
2701
2702       JBUF_LOCK (priv);
2703       if (!priv->timer_running)
2704         break;
2705
2706       if (ret != GST_CLOCK_UNSCHEDULED) {
2707         now = timer_timeout + MAX (clock_jitter, 0);
2708         GST_DEBUG_OBJECT (jitterbuffer, "sync done, %d, #%d, %" G_GINT64_FORMAT,
2709             ret, priv->timer_seqnum, clock_jitter);
2710       } else {
2711         GST_DEBUG_OBJECT (jitterbuffer, "sync unscheduled");
2712       }
2713       /* and free the entry */
2714       gst_clock_id_unref (id);
2715       priv->clock_id = NULL;
2716     } else {
2717       /* no timers, wait for activity */
2718       JBUF_WAIT_TIMER (priv);
2719     }
2720   }
2721   JBUF_UNLOCK (priv);
2722
2723   GST_DEBUG_OBJECT (jitterbuffer, "we are stopping");
2724   return;
2725 }
2726
2727 /*
2728  * This funcion implements the main pushing loop on the source pad.
2729  *
2730  * It first tries to push as many buffers as possible. If there is a seqnum
2731  * mismatch, we wait for the next timeouts.
2732  */
2733 static void
2734 gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer)
2735 {
2736   GstRtpJitterBufferPrivate *priv;
2737   GstFlowReturn result;
2738
2739   priv = jitterbuffer->priv;
2740
2741   JBUF_LOCK_CHECK (priv, flushing);
2742   do {
2743     result = handle_next_buffer (jitterbuffer);
2744     if (G_LIKELY (result == GST_FLOW_WAIT)) {
2745       /* now wait for the next event */
2746       JBUF_WAIT_EVENT (priv, flushing);
2747       result = GST_FLOW_OK;
2748     }
2749   }
2750   while (result == GST_FLOW_OK);
2751   /* store result for upstream */
2752   priv->srcresult = result;
2753   JBUF_UNLOCK (priv);
2754
2755   /* if we get here we need to pause */
2756   goto pause;
2757
2758   /* ERRORS */
2759 flushing:
2760   {
2761     result = priv->srcresult;
2762     JBUF_UNLOCK (priv);
2763     goto pause;
2764   }
2765 pause:
2766   {
2767     const gchar *reason = gst_flow_get_name (result);
2768     GstEvent *event;
2769
2770     GST_DEBUG_OBJECT (jitterbuffer, "pausing task, reason %s", reason);
2771     gst_pad_pause_task (priv->srcpad);
2772     if (result == GST_FLOW_EOS) {
2773       event = gst_event_new_eos ();
2774       gst_pad_push_event (priv->srcpad, event);
2775     }
2776     return;
2777   }
2778 }
2779
2780 /* collect the info from the lastest RTCP packet and the jitterbuffer sync, do
2781  * some sanity checks and then emit the handle-sync signal with the parameters.
2782  * This function must be called with the LOCK */
2783 static void
2784 do_handle_sync (GstRtpJitterBuffer * jitterbuffer)
2785 {
2786   GstRtpJitterBufferPrivate *priv;
2787   guint64 base_rtptime, base_time;
2788   guint32 clock_rate;
2789   guint64 last_rtptime;
2790   guint64 clock_base;
2791   guint64 ext_rtptime, diff;
2792   gboolean drop = FALSE;
2793
2794   priv = jitterbuffer->priv;
2795
2796   /* get the last values from the jitterbuffer */
2797   rtp_jitter_buffer_get_sync (priv->jbuf, &base_rtptime, &base_time,
2798       &clock_rate, &last_rtptime);
2799
2800   clock_base = priv->clock_base;
2801   ext_rtptime = priv->ext_rtptime;
2802
2803   GST_DEBUG_OBJECT (jitterbuffer, "ext SR %" G_GUINT64_FORMAT ", base %"
2804       G_GUINT64_FORMAT ", clock-rate %" G_GUINT32_FORMAT
2805       ", clock-base %" G_GUINT64_FORMAT ", last-rtptime %" G_GUINT64_FORMAT,
2806       ext_rtptime, base_rtptime, clock_rate, clock_base, last_rtptime);
2807
2808   if (base_rtptime == -1 || clock_rate == -1 || base_time == -1) {
2809     GST_DEBUG_OBJECT (jitterbuffer, "dropping, no RTP values");
2810     drop = TRUE;
2811   } else {
2812     /* we can't accept anything that happened before we did the last resync */
2813     if (base_rtptime > ext_rtptime) {
2814       GST_DEBUG_OBJECT (jitterbuffer, "dropping, older than base time");
2815       drop = TRUE;
2816     } else {
2817       /* the SR RTP timestamp must be something close to what we last observed
2818        * in the jitterbuffer */
2819       if (ext_rtptime > last_rtptime) {
2820         /* check how far ahead it is to our RTP timestamps */
2821         diff = ext_rtptime - last_rtptime;
2822         /* if bigger than 1 second, we drop it */
2823         if (diff > clock_rate) {
2824           GST_DEBUG_OBJECT (jitterbuffer, "too far ahead");
2825           /* should drop this, but some RTSP servers end up with bogus
2826            * way too ahead RTCP packet when repeated PAUSE/PLAY,
2827            * so still trigger rptbin sync but invalidate RTCP data
2828            * (sync might use other methods) */
2829           ext_rtptime = -1;
2830         }
2831         GST_DEBUG_OBJECT (jitterbuffer, "ext last %" G_GUINT64_FORMAT ", diff %"
2832             G_GUINT64_FORMAT, last_rtptime, diff);
2833       }
2834     }
2835   }
2836
2837   if (!drop) {
2838     GstStructure *s;
2839
2840     s = gst_structure_new ("application/x-rtp-sync",
2841         "base-rtptime", G_TYPE_UINT64, base_rtptime,
2842         "base-time", G_TYPE_UINT64, base_time,
2843         "clock-rate", G_TYPE_UINT, clock_rate,
2844         "clock-base", G_TYPE_UINT64, clock_base,
2845         "sr-ext-rtptime", G_TYPE_UINT64, ext_rtptime,
2846         "sr-buffer", GST_TYPE_BUFFER, priv->last_sr, NULL);
2847
2848     GST_DEBUG_OBJECT (jitterbuffer, "signaling sync");
2849     gst_buffer_replace (&priv->last_sr, NULL);
2850     JBUF_UNLOCK (priv);
2851     g_signal_emit (jitterbuffer,
2852         gst_rtp_jitter_buffer_signals[SIGNAL_HANDLE_SYNC], 0, s);
2853     JBUF_LOCK (priv);
2854     gst_structure_free (s);
2855   } else {
2856     GST_DEBUG_OBJECT (jitterbuffer, "dropping RTCP packet");
2857   }
2858 }
2859
2860 static GstFlowReturn
2861 gst_rtp_jitter_buffer_chain_rtcp (GstPad * pad, GstObject * parent,
2862     GstBuffer * buffer)
2863 {
2864   GstRtpJitterBuffer *jitterbuffer;
2865   GstRtpJitterBufferPrivate *priv;
2866   GstFlowReturn ret = GST_FLOW_OK;
2867   guint32 ssrc;
2868   GstRTCPPacket packet;
2869   guint64 ext_rtptime;
2870   guint32 rtptime;
2871   GstRTCPBuffer rtcp = { NULL, };
2872
2873   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
2874
2875   if (G_UNLIKELY (!gst_rtcp_buffer_validate (buffer)))
2876     goto invalid_buffer;
2877
2878   priv = jitterbuffer->priv;
2879
2880   gst_rtcp_buffer_map (buffer, GST_MAP_READ, &rtcp);
2881
2882   if (!gst_rtcp_buffer_get_first_packet (&rtcp, &packet))
2883     goto empty_buffer;
2884
2885   /* first packet must be SR or RR or else the validate would have failed */
2886   switch (gst_rtcp_packet_get_type (&packet)) {
2887     case GST_RTCP_TYPE_SR:
2888       gst_rtcp_packet_sr_get_sender_info (&packet, &ssrc, NULL, &rtptime,
2889           NULL, NULL);
2890       break;
2891     default:
2892       goto ignore_buffer;
2893   }
2894   gst_rtcp_buffer_unmap (&rtcp);
2895
2896   GST_DEBUG_OBJECT (jitterbuffer, "received RTCP of SSRC %08x", ssrc);
2897
2898   JBUF_LOCK (priv);
2899   /* convert the RTP timestamp to our extended timestamp, using the same offset
2900    * we used in the jitterbuffer */
2901   ext_rtptime = priv->jbuf->ext_rtptime;
2902   ext_rtptime = gst_rtp_buffer_ext_timestamp (&ext_rtptime, rtptime);
2903
2904   priv->ext_rtptime = ext_rtptime;
2905   gst_buffer_replace (&priv->last_sr, buffer);
2906
2907   do_handle_sync (jitterbuffer);
2908   JBUF_UNLOCK (priv);
2909
2910 done:
2911   gst_buffer_unref (buffer);
2912
2913   return ret;
2914
2915 invalid_buffer:
2916   {
2917     /* this is not fatal but should be filtered earlier */
2918     GST_ELEMENT_WARNING (jitterbuffer, STREAM, DECODE, (NULL),
2919         ("Received invalid RTCP payload, dropping"));
2920     ret = GST_FLOW_OK;
2921     goto done;
2922   }
2923 empty_buffer:
2924   {
2925     /* this is not fatal but should be filtered earlier */
2926     GST_ELEMENT_WARNING (jitterbuffer, STREAM, DECODE, (NULL),
2927         ("Received empty RTCP payload, dropping"));
2928     gst_rtcp_buffer_unmap (&rtcp);
2929     ret = GST_FLOW_OK;
2930     goto done;
2931   }
2932 ignore_buffer:
2933   {
2934     GST_DEBUG_OBJECT (jitterbuffer, "ignoring RTCP packet");
2935     gst_rtcp_buffer_unmap (&rtcp);
2936     ret = GST_FLOW_OK;
2937     goto done;
2938   }
2939 }
2940
2941 static gboolean
2942 gst_rtp_jitter_buffer_sink_query (GstPad * pad, GstObject * parent,
2943     GstQuery * query)
2944 {
2945   gboolean res = FALSE;
2946
2947   switch (GST_QUERY_TYPE (query)) {
2948     case GST_QUERY_CAPS:
2949     {
2950       GstCaps *filter, *caps;
2951
2952       gst_query_parse_caps (query, &filter);
2953       caps = gst_rtp_jitter_buffer_getcaps (pad, filter);
2954       gst_query_set_caps_result (query, caps);
2955       gst_caps_unref (caps);
2956       res = TRUE;
2957       break;
2958     }
2959     default:
2960       if (GST_QUERY_IS_SERIALIZED (query)) {
2961         GST_WARNING_OBJECT (pad, "unhandled serialized query");
2962         res = FALSE;
2963       } else {
2964         res = gst_pad_query_default (pad, parent, query);
2965       }
2966       break;
2967   }
2968   return res;
2969 }
2970
2971 static gboolean
2972 gst_rtp_jitter_buffer_src_query (GstPad * pad, GstObject * parent,
2973     GstQuery * query)
2974 {
2975   GstRtpJitterBuffer *jitterbuffer;
2976   GstRtpJitterBufferPrivate *priv;
2977   gboolean res = FALSE;
2978
2979   jitterbuffer = GST_RTP_JITTER_BUFFER (parent);
2980   priv = jitterbuffer->priv;
2981
2982   switch (GST_QUERY_TYPE (query)) {
2983     case GST_QUERY_LATENCY:
2984     {
2985       /* We need to send the query upstream and add the returned latency to our
2986        * own */
2987       GstClockTime min_latency, max_latency;
2988       gboolean us_live;
2989       GstClockTime our_latency;
2990
2991       if ((res = gst_pad_peer_query (priv->sinkpad, query))) {
2992         gst_query_parse_latency (query, &us_live, &min_latency, &max_latency);
2993
2994         GST_DEBUG_OBJECT (jitterbuffer, "Peer latency: min %"
2995             GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
2996             GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
2997
2998         /* store this so that we can safely sync on the peer buffers. */
2999         JBUF_LOCK (priv);
3000         priv->peer_latency = min_latency;
3001         our_latency = priv->latency_ns;
3002         JBUF_UNLOCK (priv);
3003
3004         GST_DEBUG_OBJECT (jitterbuffer, "Our latency: %" GST_TIME_FORMAT,
3005             GST_TIME_ARGS (our_latency));
3006
3007         /* we add some latency but can buffer an infinite amount of time */
3008         min_latency += our_latency;
3009         max_latency = -1;
3010
3011         GST_DEBUG_OBJECT (jitterbuffer, "Calculated total latency : min %"
3012             GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
3013             GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
3014
3015         gst_query_set_latency (query, TRUE, min_latency, max_latency);
3016       }
3017       break;
3018     }
3019     case GST_QUERY_POSITION:
3020     {
3021       GstClockTime start, last_out;
3022       GstFormat fmt;
3023
3024       gst_query_parse_position (query, &fmt, NULL);
3025       if (fmt != GST_FORMAT_TIME) {
3026         res = gst_pad_query_default (pad, parent, query);
3027         break;
3028       }
3029
3030       JBUF_LOCK (priv);
3031       start = priv->npt_start;
3032       last_out = priv->last_out_time;
3033       JBUF_UNLOCK (priv);
3034
3035       GST_DEBUG_OBJECT (jitterbuffer, "npt start %" GST_TIME_FORMAT
3036           ", last out %" GST_TIME_FORMAT, GST_TIME_ARGS (start),
3037           GST_TIME_ARGS (last_out));
3038
3039       if (GST_CLOCK_TIME_IS_VALID (start) && GST_CLOCK_TIME_IS_VALID (last_out)) {
3040         /* bring 0-based outgoing time to stream time */
3041         gst_query_set_position (query, GST_FORMAT_TIME, start + last_out);
3042         res = TRUE;
3043       } else {
3044         res = gst_pad_query_default (pad, parent, query);
3045       }
3046       break;
3047     }
3048     case GST_QUERY_CAPS:
3049     {
3050       GstCaps *filter, *caps;
3051
3052       gst_query_parse_caps (query, &filter);
3053       caps = gst_rtp_jitter_buffer_getcaps (pad, filter);
3054       gst_query_set_caps_result (query, caps);
3055       gst_caps_unref (caps);
3056       res = TRUE;
3057       break;
3058     }
3059     default:
3060       res = gst_pad_query_default (pad, parent, query);
3061       break;
3062   }
3063
3064   return res;
3065 }
3066
3067 static void
3068 gst_rtp_jitter_buffer_set_property (GObject * object,
3069     guint prop_id, const GValue * value, GParamSpec * pspec)
3070 {
3071   GstRtpJitterBuffer *jitterbuffer;
3072   GstRtpJitterBufferPrivate *priv;
3073
3074   jitterbuffer = GST_RTP_JITTER_BUFFER (object);
3075   priv = jitterbuffer->priv;
3076
3077   switch (prop_id) {
3078     case PROP_LATENCY:
3079     {
3080       guint new_latency, old_latency;
3081
3082       new_latency = g_value_get_uint (value);
3083
3084       JBUF_LOCK (priv);
3085       old_latency = priv->latency_ms;
3086       priv->latency_ms = new_latency;
3087       priv->latency_ns = priv->latency_ms * GST_MSECOND;
3088       rtp_jitter_buffer_set_delay (priv->jbuf, priv->latency_ns);
3089       JBUF_UNLOCK (priv);
3090
3091       /* post message if latency changed, this will inform the parent pipeline
3092        * that a latency reconfiguration is possible/needed. */
3093       if (new_latency != old_latency) {
3094         GST_DEBUG_OBJECT (jitterbuffer, "latency changed to: %" GST_TIME_FORMAT,
3095             GST_TIME_ARGS (new_latency * GST_MSECOND));
3096
3097         gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer),
3098             gst_message_new_latency (GST_OBJECT_CAST (jitterbuffer)));
3099       }
3100       break;
3101     }
3102     case PROP_DROP_ON_LATENCY:
3103       JBUF_LOCK (priv);
3104       priv->drop_on_latency = g_value_get_boolean (value);
3105       JBUF_UNLOCK (priv);
3106       break;
3107     case PROP_TS_OFFSET:
3108       JBUF_LOCK (priv);
3109       priv->ts_offset = g_value_get_int64 (value);
3110       priv->ts_discont = TRUE;
3111       JBUF_UNLOCK (priv);
3112       break;
3113     case PROP_DO_LOST:
3114       JBUF_LOCK (priv);
3115       priv->do_lost = g_value_get_boolean (value);
3116       JBUF_UNLOCK (priv);
3117       break;
3118     case PROP_MODE:
3119       JBUF_LOCK (priv);
3120       rtp_jitter_buffer_set_mode (priv->jbuf, g_value_get_enum (value));
3121       JBUF_UNLOCK (priv);
3122       break;
3123     case PROP_DO_RETRANSMISSION:
3124       JBUF_LOCK (priv);
3125       priv->do_retransmission = g_value_get_boolean (value);
3126       JBUF_UNLOCK (priv);
3127       break;
3128     case PROP_RTX_DELAY:
3129       JBUF_LOCK (priv);
3130       priv->rtx_delay = g_value_get_int (value);
3131       JBUF_UNLOCK (priv);
3132       break;
3133     case PROP_RTX_DELAY_REORDER:
3134       JBUF_LOCK (priv);
3135       priv->rtx_delay_reorder = g_value_get_int (value);
3136       JBUF_UNLOCK (priv);
3137       break;
3138     case PROP_RTX_RETRY_TIMEOUT:
3139       JBUF_LOCK (priv);
3140       priv->rtx_retry_timeout = g_value_get_int (value);
3141       JBUF_UNLOCK (priv);
3142       break;
3143     case PROP_RTX_RETRY_PERIOD:
3144       JBUF_LOCK (priv);
3145       priv->rtx_retry_period = g_value_get_int (value);
3146       JBUF_UNLOCK (priv);
3147       break;
3148     default:
3149       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
3150       break;
3151   }
3152 }
3153
3154 static void
3155 gst_rtp_jitter_buffer_get_property (GObject * object,
3156     guint prop_id, GValue * value, GParamSpec * pspec)
3157 {
3158   GstRtpJitterBuffer *jitterbuffer;
3159   GstRtpJitterBufferPrivate *priv;
3160
3161   jitterbuffer = GST_RTP_JITTER_BUFFER (object);
3162   priv = jitterbuffer->priv;
3163
3164   switch (prop_id) {
3165     case PROP_LATENCY:
3166       JBUF_LOCK (priv);
3167       g_value_set_uint (value, priv->latency_ms);
3168       JBUF_UNLOCK (priv);
3169       break;
3170     case PROP_DROP_ON_LATENCY:
3171       JBUF_LOCK (priv);
3172       g_value_set_boolean (value, priv->drop_on_latency);
3173       JBUF_UNLOCK (priv);
3174       break;
3175     case PROP_TS_OFFSET:
3176       JBUF_LOCK (priv);
3177       g_value_set_int64 (value, priv->ts_offset);
3178       JBUF_UNLOCK (priv);
3179       break;
3180     case PROP_DO_LOST:
3181       JBUF_LOCK (priv);
3182       g_value_set_boolean (value, priv->do_lost);
3183       JBUF_UNLOCK (priv);
3184       break;
3185     case PROP_MODE:
3186       JBUF_LOCK (priv);
3187       g_value_set_enum (value, rtp_jitter_buffer_get_mode (priv->jbuf));
3188       JBUF_UNLOCK (priv);
3189       break;
3190     case PROP_PERCENT:
3191     {
3192       gint percent;
3193
3194       JBUF_LOCK (priv);
3195       if (priv->srcresult != GST_FLOW_OK)
3196         percent = 100;
3197       else
3198         percent = rtp_jitter_buffer_get_percent (priv->jbuf);
3199
3200       g_value_set_int (value, percent);
3201       JBUF_UNLOCK (priv);
3202       break;
3203     }
3204     case PROP_DO_RETRANSMISSION:
3205       JBUF_LOCK (priv);
3206       g_value_set_boolean (value, priv->do_retransmission);
3207       JBUF_UNLOCK (priv);
3208       break;
3209     case PROP_RTX_DELAY:
3210       JBUF_LOCK (priv);
3211       g_value_set_int (value, priv->rtx_delay);
3212       JBUF_UNLOCK (priv);
3213       break;
3214     case PROP_RTX_DELAY_REORDER:
3215       JBUF_LOCK (priv);
3216       g_value_set_int (value, priv->rtx_delay_reorder);
3217       JBUF_UNLOCK (priv);
3218       break;
3219     case PROP_RTX_RETRY_TIMEOUT:
3220       JBUF_LOCK (priv);
3221       g_value_set_int (value, priv->rtx_retry_timeout);
3222       JBUF_UNLOCK (priv);
3223       break;
3224     case PROP_RTX_RETRY_PERIOD:
3225       JBUF_LOCK (priv);
3226       g_value_set_int (value, priv->rtx_retry_period);
3227       JBUF_UNLOCK (priv);
3228       break;
3229     case PROP_STATS:
3230       g_value_take_boxed (value,
3231           gst_rtp_jitter_buffer_create_stats (jitterbuffer));
3232       break;
3233     default:
3234       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
3235       break;
3236   }
3237 }
3238
3239 static GstStructure *
3240 gst_rtp_jitter_buffer_create_stats (GstRtpJitterBuffer * jbuf)
3241 {
3242   GstStructure *s;
3243
3244   JBUF_LOCK (jbuf->priv);
3245   s = gst_structure_new ("application/x-rtp-jitterbuffer-stats",
3246       "rtx-count", G_TYPE_UINT64, jbuf->priv->num_rtx_requests,
3247       "rtx-success-count", G_TYPE_UINT64, jbuf->priv->num_rtx_success,
3248       "rtx-per-packet", G_TYPE_DOUBLE, jbuf->priv->avg_rtx_num,
3249       "rtx-rtt", G_TYPE_UINT64, jbuf->priv->avg_rtx_rtt, NULL);
3250   JBUF_UNLOCK (jbuf->priv);
3251
3252   return s;
3253 }