2 * Farsight Voice+Video library
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>
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.
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.
19 * You should have received a copy of the GNU Library General Public
20 * License along with this library; if not, write to the
21 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
22 * Boston, MA 02111-1307, USA.
27 * SECTION:element-gstrtpjitterbuffer
28 * @short_description: buffer, reorder and remove duplicate RTP packets to
29 * compensate for network oddities.
33 * This element reorders and removes duplicate RTP packets as they are received
34 * from a network source. It will also wait for missing packets up to a
35 * configurable time limit using the ::latency property. Packets arriving too
36 * late are considered to be lost packets.
39 * This element acts as a live element and so adds ::latency to the pipeline.
42 * The element needs the clock-rate of the RTP payload in order to estimate the
43 * delay. This information is obtained either from the caps on the sink pad or,
44 * when no caps are present, from the ::request-pt-map signal. To clear the
45 * previous pt-map use the ::clear-pt-map signal.
48 * This element will automatically be used inside gstrtpbin.
50 * <title>Example pipelines</title>
53 * gst-launch rtspsrc location=rtsp://192.168.1.133:8554/mpeg1or2AudioVideoTest ! gstrtpjitterbuffer ! rtpmpvdepay ! mpeg2dec ! xvimagesink
55 * Connect to a streaming server and decode the MPEG video. The jitterbuffer is
56 * inserted into the pipeline to smooth out network jitter and to reorder the
57 * out-of-order RTP packets.
61 * Last reviewed on 2007-05-28 (0.10.5)
70 #include <gst/rtp/gstrtpbuffer.h>
72 #include "gstrtpbin-marshal.h"
74 #include "gstrtpjitterbuffer.h"
75 #include "rtpjitterbuffer.h"
77 GST_DEBUG_CATEGORY (rtpjitterbuffer_debug);
78 #define GST_CAT_DEFAULT (rtpjitterbuffer_debug)
80 /* low and high threshold tell the queue when to start and stop buffering */
81 #define LOW_THRESHOLD 0.2
82 #define HIGH_THRESHOLD 0.8
84 /* elementfactory information */
85 static const GstElementDetails gst_rtp_jitter_buffer_details =
86 GST_ELEMENT_DETAILS ("RTP packet jitter-buffer",
88 "A buffer that deals with network jitter and other transmission faults",
89 "Philippe Kalaf <philippe.kalaf@collabora.co.uk>, "
90 "Wim Taymans <wim.taymans@gmail.com>");
92 /* RTPJitterBuffer signals and args */
95 SIGNAL_REQUEST_PT_MAP,
100 #define DEFAULT_LATENCY_MS 200
101 #define DEFAULT_DROP_ON_LATENCY FALSE
102 #define DEFAULT_TS_OFFSET 0
108 PROP_DROP_ON_LATENCY,
112 #define JBUF_LOCK(priv) (g_mutex_lock ((priv)->jbuf_lock))
114 #define JBUF_LOCK_CHECK(priv,label) G_STMT_START { \
116 if (priv->srcresult != GST_FLOW_OK) \
120 #define JBUF_UNLOCK(priv) (g_mutex_unlock ((priv)->jbuf_lock))
121 #define JBUF_WAIT(priv) (g_cond_wait ((priv)->jbuf_cond, (priv)->jbuf_lock))
123 #define JBUF_WAIT_CHECK(priv,label) G_STMT_START { \
125 if (priv->srcresult != GST_FLOW_OK) \
129 #define JBUF_SIGNAL(priv) (g_cond_signal ((priv)->jbuf_cond))
131 struct _GstRtpJitterBufferPrivate
133 GstPad *sinkpad, *srcpad;
135 RTPJitterBuffer *jbuf;
142 gboolean drop_on_latency;
145 /* the last seqnum we pushed out */
146 guint32 last_popped_seqnum;
147 /* the next expected seqnum */
153 /* clock rate and rtp timestamp offset */
156 gint64 prev_ts_offset;
158 /* when we are shutting down */
159 GstFlowReturn srcresult;
165 guint32 waiting_seqnum;
166 /* the latency of the upstream peer, we have to take this into account when
167 * synchronizing the buffers. */
168 GstClockTime peer_latency;
170 /* some accounting */
172 guint64 num_duplicates;
175 #define GST_RTP_JITTER_BUFFER_GET_PRIVATE(o) \
176 (G_TYPE_INSTANCE_GET_PRIVATE ((o), GST_TYPE_RTP_JITTER_BUFFER, \
177 GstRtpJitterBufferPrivate))
179 static GstStaticPadTemplate gst_rtp_jitter_buffer_sink_template =
180 GST_STATIC_PAD_TEMPLATE ("sink",
183 GST_STATIC_CAPS ("application/x-rtp, "
184 "clock-rate = (int) [ 1, 2147483647 ]"
185 /* "payload = (int) , "
186 * "encoding-name = (string) "
190 static GstStaticPadTemplate gst_rtp_jitter_buffer_src_template =
191 GST_STATIC_PAD_TEMPLATE ("src",
194 GST_STATIC_CAPS ("application/x-rtp"
195 /* "payload = (int) , "
196 * "clock-rate = (int) , "
197 * "encoding-name = (string) "
201 static guint gst_rtp_jitter_buffer_signals[LAST_SIGNAL] = { 0 };
203 GST_BOILERPLATE (GstRtpJitterBuffer, gst_rtp_jitter_buffer, GstElement,
206 /* object overrides */
207 static void gst_rtp_jitter_buffer_set_property (GObject * object,
208 guint prop_id, const GValue * value, GParamSpec * pspec);
209 static void gst_rtp_jitter_buffer_get_property (GObject * object,
210 guint prop_id, GValue * value, GParamSpec * pspec);
211 static void gst_rtp_jitter_buffer_finalize (GObject * object);
213 /* element overrides */
214 static GstStateChangeReturn gst_rtp_jitter_buffer_change_state (GstElement
215 * element, GstStateChange transition);
218 static GstCaps *gst_rtp_jitter_buffer_getcaps (GstPad * pad);
220 /* sinkpad overrides */
221 static gboolean gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps);
222 static gboolean gst_rtp_jitter_buffer_sink_event (GstPad * pad,
224 static GstFlowReturn gst_rtp_jitter_buffer_chain (GstPad * pad,
227 /* srcpad overrides */
229 gst_rtp_jitter_buffer_src_activate_push (GstPad * pad, gboolean active);
230 static void gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer);
231 static gboolean gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query);
234 gst_rtp_jitter_buffer_clear_pt_map (GstRtpJitterBuffer * jitterbuffer);
237 gst_rtp_jitter_buffer_base_init (gpointer klass)
239 GstElementClass *element_class = GST_ELEMENT_CLASS (klass);
241 gst_element_class_add_pad_template (element_class,
242 gst_static_pad_template_get (&gst_rtp_jitter_buffer_src_template));
243 gst_element_class_add_pad_template (element_class,
244 gst_static_pad_template_get (&gst_rtp_jitter_buffer_sink_template));
245 gst_element_class_set_details (element_class, &gst_rtp_jitter_buffer_details);
249 gst_rtp_jitter_buffer_class_init (GstRtpJitterBufferClass * klass)
251 GObjectClass *gobject_class;
252 GstElementClass *gstelement_class;
254 gobject_class = (GObjectClass *) klass;
255 gstelement_class = (GstElementClass *) klass;
257 g_type_class_add_private (klass, sizeof (GstRtpJitterBufferPrivate));
259 gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_finalize);
261 gobject_class->set_property = gst_rtp_jitter_buffer_set_property;
262 gobject_class->get_property = gst_rtp_jitter_buffer_get_property;
265 * GstRtpJitterBuffer::latency:
267 * The maximum latency of the jitterbuffer. Packets will be kept in the buffer
268 * for at most this time.
270 g_object_class_install_property (gobject_class, PROP_LATENCY,
271 g_param_spec_uint ("latency", "Buffer latency in ms",
272 "Amount of ms to buffer", 0, G_MAXUINT, DEFAULT_LATENCY_MS,
275 * GstRtpJitterBuffer::drop-on-latency:
277 * Drop oldest buffers when the queue is completely filled.
279 g_object_class_install_property (gobject_class, PROP_DROP_ON_LATENCY,
280 g_param_spec_boolean ("drop-on-latency",
281 "Drop buffers when maximum latency is reached",
282 "Tells the jitterbuffer to never exceed the given latency in size",
283 DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE));
285 * GstRtpJitterBuffer::ts-offset:
287 * Adjust RTP timestamps in the jitterbuffer with offset.
289 g_object_class_install_property (gobject_class, PROP_TS_OFFSET,
290 g_param_spec_int64 ("ts-offset",
292 "Adjust buffer RTP timestamps with offset in nanoseconds", G_MININT64,
293 G_MAXINT64, DEFAULT_TS_OFFSET, G_PARAM_READWRITE));
295 * GstRtpJitterBuffer::request-pt-map:
296 * @buffer: the object which received the signal
299 * Request the payload type as #GstCaps for @pt.
301 gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP] =
302 g_signal_new ("request-pt-map", G_TYPE_FROM_CLASS (klass),
303 G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
304 request_pt_map), NULL, NULL, gst_rtp_bin_marshal_BOXED__UINT,
305 GST_TYPE_CAPS, 1, G_TYPE_UINT);
307 * GstRtpJitterBuffer::clear-pt-map:
308 * @buffer: the object which received the signal
310 * Invalidate the clock-rate as obtained with the ::request-pt-map signal.
312 gst_rtp_jitter_buffer_signals[SIGNAL_CLEAR_PT_MAP] =
313 g_signal_new ("clear-pt-map", G_TYPE_FROM_CLASS (klass),
314 G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpJitterBufferClass,
315 clear_pt_map), NULL, NULL, g_cclosure_marshal_VOID__VOID,
316 G_TYPE_NONE, 0, G_TYPE_NONE);
318 gstelement_class->change_state = gst_rtp_jitter_buffer_change_state;
320 klass->clear_pt_map = GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_clear_pt_map);
322 GST_DEBUG_CATEGORY_INIT
323 (rtpjitterbuffer_debug, "gstrtpjitterbuffer", 0, "RTP Jitter Buffer");
327 gst_rtp_jitter_buffer_init (GstRtpJitterBuffer * jitterbuffer,
328 GstRtpJitterBufferClass * klass)
330 GstRtpJitterBufferPrivate *priv;
332 priv = GST_RTP_JITTER_BUFFER_GET_PRIVATE (jitterbuffer);
333 jitterbuffer->priv = priv;
335 priv->latency_ms = DEFAULT_LATENCY_MS;
336 priv->drop_on_latency = DEFAULT_DROP_ON_LATENCY;
338 priv->jbuf = rtp_jitter_buffer_new ();
339 priv->jbuf_lock = g_mutex_new ();
340 priv->jbuf_cond = g_cond_new ();
342 priv->waiting_seqnum = -1;
345 gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_src_template,
348 gst_pad_set_activatepush_function (priv->srcpad,
349 GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_activate_push));
350 gst_pad_set_query_function (priv->srcpad,
351 GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_query));
352 gst_pad_set_getcaps_function (priv->srcpad,
353 GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_getcaps));
356 gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_sink_template,
359 gst_pad_set_chain_function (priv->sinkpad,
360 GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_chain));
361 gst_pad_set_event_function (priv->sinkpad,
362 GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_sink_event));
363 gst_pad_set_setcaps_function (priv->sinkpad,
364 GST_DEBUG_FUNCPTR (gst_jitter_buffer_sink_setcaps));
365 gst_pad_set_getcaps_function (priv->sinkpad,
366 GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_getcaps));
368 gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->srcpad);
369 gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->sinkpad);
373 gst_rtp_jitter_buffer_finalize (GObject * object)
375 GstRtpJitterBuffer *jitterbuffer;
377 jitterbuffer = GST_RTP_JITTER_BUFFER (object);
379 g_mutex_free (jitterbuffer->priv->jbuf_lock);
380 g_cond_free (jitterbuffer->priv->jbuf_cond);
382 g_object_unref (jitterbuffer->priv->jbuf);
384 G_OBJECT_CLASS (parent_class)->finalize (object);
388 gst_rtp_jitter_buffer_clear_pt_map (GstRtpJitterBuffer * jitterbuffer)
390 GstRtpJitterBufferPrivate *priv;
392 priv = jitterbuffer->priv;
394 /* this will trigger a new pt-map request signal, FIXME, do something better. */
395 priv->clock_rate = -1;
399 gst_rtp_jitter_buffer_getcaps (GstPad * pad)
401 GstRtpJitterBuffer *jitterbuffer;
402 GstRtpJitterBufferPrivate *priv;
405 const GstCaps *templ;
407 jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
408 priv = jitterbuffer->priv;
410 other = (pad == priv->srcpad ? priv->sinkpad : priv->srcpad);
412 caps = gst_pad_peer_get_caps (other);
414 templ = gst_pad_get_pad_template_caps (pad);
416 GST_DEBUG_OBJECT (jitterbuffer, "copy template");
417 caps = gst_caps_copy (templ);
421 GST_DEBUG_OBJECT (jitterbuffer, "intersect with template");
423 intersect = gst_caps_intersect (caps, templ);
424 gst_caps_unref (caps);
428 gst_object_unref (jitterbuffer);
434 gst_jitter_buffer_sink_parse_caps (GstRtpJitterBuffer * jitterbuffer,
437 GstRtpJitterBufferPrivate *priv;
438 GstStructure *caps_struct;
441 priv = jitterbuffer->priv;
443 /* first parse the caps */
444 caps_struct = gst_caps_get_structure (caps, 0);
446 GST_DEBUG_OBJECT (jitterbuffer, "got caps");
448 /* we need a clock-rate to convert the rtp timestamps to GStreamer time and to
449 * measure the amount of data in the buffer */
450 if (!gst_structure_get_int (caps_struct, "clock-rate", &priv->clock_rate))
453 if (priv->clock_rate <= 0)
456 rtp_jitter_buffer_set_clock_rate (priv->jbuf, priv->clock_rate);
458 GST_DEBUG_OBJECT (jitterbuffer, "got clock-rate %d", priv->clock_rate);
460 /* gah, clock-base is uint. If we don't have a base, we will use the first
461 * buffer timestamp as the base time. This will screw up sync but it's better
463 if (gst_structure_get_uint (caps_struct, "clock-base", &val))
464 priv->clock_base = val;
466 priv->clock_base = -1;
468 GST_DEBUG_OBJECT (jitterbuffer, "got clock-base %" G_GINT64_FORMAT,
471 /* first expected seqnum */
472 if (gst_structure_get_uint (caps_struct, "seqnum-base", &val))
473 priv->next_seqnum = val;
475 priv->next_seqnum = -1;
477 GST_DEBUG_OBJECT (jitterbuffer, "got seqnum-base %d", priv->next_seqnum);
484 GST_DEBUG_OBJECT (jitterbuffer, "No clock-rate in caps!");
489 GST_DEBUG_OBJECT (jitterbuffer, "Invalid clock-rate %d", priv->clock_rate);
495 gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps)
497 GstRtpJitterBuffer *jitterbuffer;
498 GstRtpJitterBufferPrivate *priv;
501 jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
502 priv = jitterbuffer->priv;
504 res = gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
506 /* set same caps on srcpad on success */
508 gst_pad_set_caps (priv->srcpad, caps);
510 gst_object_unref (jitterbuffer);
516 gst_rtp_jitter_buffer_flush_start (GstRtpJitterBuffer * jitterbuffer)
518 GstRtpJitterBufferPrivate *priv;
520 priv = jitterbuffer->priv;
523 /* mark ourselves as flushing */
524 priv->srcresult = GST_FLOW_WRONG_STATE;
525 GST_DEBUG_OBJECT (jitterbuffer, "Disabling pop on queue");
526 /* this unblocks any waiting pops on the src pad task */
528 /* unlock clock, we just unschedule, the entry will be released by the
529 * locking streaming thread. */
531 gst_clock_id_unschedule (priv->clock_id);
536 gst_rtp_jitter_buffer_flush_stop (GstRtpJitterBuffer * jitterbuffer)
538 GstRtpJitterBufferPrivate *priv;
540 priv = jitterbuffer->priv;
543 GST_DEBUG_OBJECT (jitterbuffer, "Enabling pop on queue");
544 /* Mark as non flushing */
545 priv->srcresult = GST_FLOW_OK;
546 gst_segment_init (&priv->segment, GST_FORMAT_TIME);
547 priv->last_popped_seqnum = -1;
548 priv->next_seqnum = -1;
549 priv->clock_rate = -1;
551 rtp_jitter_buffer_flush (priv->jbuf);
552 rtp_jitter_buffer_reset_skew (priv->jbuf);
557 gst_rtp_jitter_buffer_src_activate_push (GstPad * pad, gboolean active)
559 gboolean result = TRUE;
560 GstRtpJitterBuffer *jitterbuffer = NULL;
562 jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
565 /* allow data processing */
566 gst_rtp_jitter_buffer_flush_stop (jitterbuffer);
568 /* start pushing out buffers */
569 GST_DEBUG_OBJECT (jitterbuffer, "Starting task on srcpad");
570 gst_pad_start_task (jitterbuffer->priv->srcpad,
571 (GstTaskFunction) gst_rtp_jitter_buffer_loop, jitterbuffer);
573 /* make sure all data processing stops ASAP */
574 gst_rtp_jitter_buffer_flush_start (jitterbuffer);
576 /* NOTE this will hardlock if the state change is called from the src pad
577 * task thread because we will _join() the thread. */
578 GST_DEBUG_OBJECT (jitterbuffer, "Stopping task on srcpad");
579 result = gst_pad_stop_task (pad);
582 gst_object_unref (jitterbuffer);
587 static GstStateChangeReturn
588 gst_rtp_jitter_buffer_change_state (GstElement * element,
589 GstStateChange transition)
591 GstRtpJitterBuffer *jitterbuffer;
592 GstRtpJitterBufferPrivate *priv;
593 GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
595 jitterbuffer = GST_RTP_JITTER_BUFFER (element);
596 priv = jitterbuffer->priv;
598 switch (transition) {
599 case GST_STATE_CHANGE_NULL_TO_READY:
601 case GST_STATE_CHANGE_READY_TO_PAUSED:
603 /* reset negotiated values */
604 priv->clock_rate = -1;
605 priv->clock_base = -1;
606 priv->peer_latency = 0;
607 /* block until we go to PLAYING */
608 priv->blocked = TRUE;
609 /* reset skew detection initialy */
610 rtp_jitter_buffer_reset_skew (priv->jbuf);
613 case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
615 /* unblock to allow streaming in PLAYING */
616 priv->blocked = FALSE;
624 ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
626 switch (transition) {
627 case GST_STATE_CHANGE_READY_TO_PAUSED:
628 /* we are a live element because we sync to the clock, which we can only
629 * do in the PLAYING state */
630 if (ret != GST_STATE_CHANGE_FAILURE)
631 ret = GST_STATE_CHANGE_NO_PREROLL;
633 case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
635 /* block to stop streaming when PAUSED */
636 priv->blocked = TRUE;
638 if (ret != GST_STATE_CHANGE_FAILURE)
639 ret = GST_STATE_CHANGE_NO_PREROLL;
641 case GST_STATE_CHANGE_PAUSED_TO_READY:
643 case GST_STATE_CHANGE_READY_TO_NULL:
653 * Performs comparison 'b - a' with check for overflows.
656 priv_compare_rtp_seq_lt (guint16 a, guint16 b)
658 /* check if diff more than half of the 16bit range */
659 if (abs (b - a) > (1 << 15)) {
660 /* one of a/b has wrapped */
668 gst_rtp_jitter_buffer_sink_event (GstPad * pad, GstEvent * event)
671 GstRtpJitterBuffer *jitterbuffer;
672 GstRtpJitterBufferPrivate *priv;
674 jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
675 priv = jitterbuffer->priv;
677 GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));
679 switch (GST_EVENT_TYPE (event)) {
680 case GST_EVENT_NEWSEGMENT:
684 gint64 start, stop, time;
687 gst_event_parse_new_segment_full (event, &update, &rate, &arate, &format,
688 &start, &stop, &time);
690 /* we need time for now */
691 if (format != GST_FORMAT_TIME)
692 goto newseg_wrong_format;
694 GST_DEBUG_OBJECT (jitterbuffer,
695 "newsegment: update %d, rate %g, arate %g, start %" GST_TIME_FORMAT
696 ", stop %" GST_TIME_FORMAT ", time %" GST_TIME_FORMAT,
697 update, rate, arate, GST_TIME_ARGS (start), GST_TIME_ARGS (stop),
698 GST_TIME_ARGS (time));
700 /* now configure the values, we need these to time the release of the
701 * buffers on the srcpad. */
702 gst_segment_set_newsegment_full (&priv->segment, update,
703 rate, arate, format, start, stop, time);
705 /* FIXME, push SEGMENT in the queue. Sorting order might be difficult. */
706 ret = gst_pad_push_event (priv->srcpad, event);
709 case GST_EVENT_FLUSH_START:
710 gst_rtp_jitter_buffer_flush_start (jitterbuffer);
711 ret = gst_pad_push_event (priv->srcpad, event);
713 case GST_EVENT_FLUSH_STOP:
714 ret = gst_pad_push_event (priv->srcpad, event);
715 ret = gst_rtp_jitter_buffer_src_activate_push (priv->srcpad, TRUE);
719 /* push EOS in queue. We always push it at the head */
721 /* check for flushing, we need to discard the event and return FALSE when
723 ret = priv->srcresult == GST_FLOW_OK;
724 if (ret && !priv->eos) {
725 GST_DEBUG_OBJECT (jitterbuffer, "queuing EOS");
728 } else if (priv->eos) {
729 GST_DEBUG_OBJECT (jitterbuffer, "dropping EOS, we are already EOS");
731 GST_DEBUG_OBJECT (jitterbuffer, "dropping EOS, reason %s",
732 gst_flow_get_name (priv->srcresult));
735 gst_event_unref (event);
739 ret = gst_pad_push_event (priv->srcpad, event);
744 gst_object_unref (jitterbuffer);
751 GST_DEBUG_OBJECT (jitterbuffer, "received non TIME newsegment");
758 gst_rtp_jitter_buffer_get_clock_rate (GstRtpJitterBuffer * jitterbuffer,
762 GValue args[2] = { {0}, {0} };
766 g_value_init (&args[0], GST_TYPE_ELEMENT);
767 g_value_set_object (&args[0], jitterbuffer);
768 g_value_init (&args[1], G_TYPE_UINT);
769 g_value_set_uint (&args[1], pt);
771 g_value_init (&ret, GST_TYPE_CAPS);
772 g_value_set_boxed (&ret, NULL);
774 g_signal_emitv (args, gst_rtp_jitter_buffer_signals[SIGNAL_REQUEST_PT_MAP], 0,
777 caps = (GstCaps *) g_value_get_boxed (&ret);
781 res = gst_jitter_buffer_sink_parse_caps (jitterbuffer, caps);
788 GST_DEBUG_OBJECT (jitterbuffer, "could not get caps");
794 gst_rtp_jitter_buffer_chain (GstPad * pad, GstBuffer * buffer)
796 GstRtpJitterBuffer *jitterbuffer;
797 GstRtpJitterBufferPrivate *priv;
799 GstFlowReturn ret = GST_FLOW_OK;
800 GstClockTime timestamp;
804 jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
806 if (!gst_rtp_buffer_validate (buffer))
809 priv = jitterbuffer->priv;
811 if (priv->clock_rate == -1) {
814 /* no clock rate given on the caps, try to get one with the signal */
815 pt = gst_rtp_buffer_get_payload_type (buffer);
817 gst_rtp_jitter_buffer_get_clock_rate (jitterbuffer, pt);
818 if (priv->clock_rate == -1)
821 rtp_jitter_buffer_set_clock_rate (priv->jbuf, priv->clock_rate);
824 /* take the timestamp of the buffer. This is the time when the packet was
825 * received and is used to calculate jitter and clock skew. We will adjust
826 * this timestamp with the smoothed value after processing it in the
828 timestamp = GST_BUFFER_TIMESTAMP (buffer);
829 /* bring to running time */
830 timestamp = gst_segment_to_running_time (&priv->segment, GST_FORMAT_TIME,
833 seqnum = gst_rtp_buffer_get_seq (buffer);
834 GST_DEBUG_OBJECT (jitterbuffer,
835 "Received packet #%d at time %" GST_TIME_FORMAT, seqnum,
836 GST_TIME_ARGS (timestamp));
838 JBUF_LOCK_CHECK (priv, out_flushing);
839 /* don't accept more data on EOS */
843 /* let's check if this buffer is too late, we cannot accept packets with
844 * bigger seqnum than the one we already pushed. */
845 if (priv->last_popped_seqnum != -1) {
846 if (priv_compare_rtp_seq_lt (priv->last_popped_seqnum, seqnum) < 0)
850 /* let's drop oldest packet if the queue is already full and drop-on-latency
851 * is set. We can only do this when there actually is a latency. When no
852 * latency is set, we just pump it in the queue and let the other end push it
853 * out as fast as possible. */
854 if (priv->latency_ms && priv->drop_on_latency) {
857 gst_util_uint64_scale_int (priv->latency_ms, priv->clock_rate, 1000);
859 if (rtp_jitter_buffer_get_ts_diff (priv->jbuf) >= latency_ts) {
862 GST_DEBUG_OBJECT (jitterbuffer, "Queue full, dropping old packet #%d",
865 old_buf = rtp_jitter_buffer_pop (priv->jbuf);
866 gst_buffer_unref (old_buf);
870 /* now insert the packet into the queue in sorted order. This function returns
871 * FALSE if a packet with the same seqnum was already in the queue, meaning we
872 * have a duplicate. */
873 if (!rtp_jitter_buffer_insert (priv->jbuf, buffer, timestamp, &tail))
876 /* signal addition of new buffer when the _loop is waiting. */
880 /* let's unschedule and unblock any waiting buffers. We only want to do this
881 * when the tail buffer changed */
882 if (priv->clock_id && tail) {
883 GST_DEBUG_OBJECT (jitterbuffer,
884 "Unscheduling waiting buffer, new tail buffer");
885 gst_clock_id_unschedule (priv->clock_id);
888 GST_DEBUG_OBJECT (jitterbuffer, "Pushed packet #%d, now %d packets",
889 seqnum, rtp_jitter_buffer_num_packets (priv->jbuf));
894 gst_object_unref (jitterbuffer);
901 /* this is fatal and should be filtered earlier */
902 GST_ELEMENT_ERROR (jitterbuffer, STREAM, DECODE, (NULL),
903 ("Received invalid RTP payload"));
904 gst_buffer_unref (buffer);
905 gst_object_unref (jitterbuffer);
906 return GST_FLOW_ERROR;
910 GST_WARNING_OBJECT (jitterbuffer, "No clock-rate in caps!");
911 gst_buffer_unref (buffer);
912 gst_object_unref (jitterbuffer);
913 return GST_FLOW_NOT_NEGOTIATED;
917 ret = priv->srcresult;
918 GST_DEBUG_OBJECT (jitterbuffer, "flushing %s", gst_flow_get_name (ret));
919 gst_buffer_unref (buffer);
924 ret = GST_FLOW_UNEXPECTED;
925 GST_WARNING_OBJECT (jitterbuffer, "we are EOS, refusing buffer");
926 gst_buffer_unref (buffer);
931 GST_WARNING_OBJECT (jitterbuffer, "Packet #%d too late as #%d was already"
932 " popped, dropping", seqnum, priv->last_popped_seqnum);
934 gst_buffer_unref (buffer);
939 GST_WARNING_OBJECT (jitterbuffer, "Duplicate packet #%d detected, dropping",
941 priv->num_duplicates++;
942 gst_buffer_unref (buffer);
948 apply_offset (GstRtpJitterBuffer * jitterbuffer, GstClockTime timestamp)
950 GstRtpJitterBufferPrivate *priv;
952 priv = jitterbuffer->priv;
957 /* apply the timestamp offset */
958 timestamp += priv->ts_offset;
964 * This funcion will push out buffers on the source pad.
966 * For each pushed buffer, the seqnum is recorded, if the next buffer B has a
967 * different seqnum (missing packets before B), this function will wait for the
968 * missing packet to arrive up to the timestamp of buffer B.
971 gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer)
973 GstRtpJitterBufferPrivate *priv;
974 GstBuffer *outbuf = NULL;
975 GstFlowReturn result;
977 GstClockTime timestamp, out_time;
979 priv = jitterbuffer->priv;
981 JBUF_LOCK_CHECK (priv, flushing);
983 GST_DEBUG_OBJECT (jitterbuffer, "Peeking item");
986 /* always wait if we are blocked */
987 if (!priv->blocked) {
988 /* if we have a packet, we can grab it */
989 if (rtp_jitter_buffer_num_packets (priv->jbuf) > 0)
991 /* no packets but we are EOS, do eos logic */
995 /* wait for packets or flushing now */
996 priv->waiting = TRUE;
997 JBUF_WAIT_CHECK (priv, flushing);
998 priv->waiting = FALSE;
1001 /* peek a buffer, we're just looking at the timestamp and the sequence number.
1002 * If all is fine, we'll pop and push it. If the sequence number is wrong we
1003 * wait on the timestamp. In the chain function we will unlock the wait when a
1004 * new buffer is available. The peeked buffer is valid for as long as we hold
1005 * the jitterbuffer lock. */
1006 outbuf = rtp_jitter_buffer_peek (priv->jbuf);
1007 seqnum = gst_rtp_buffer_get_seq (outbuf);
1009 /* get the timestamp, this is already corrected for clock skew by the
1011 timestamp = GST_BUFFER_TIMESTAMP (outbuf);
1013 GST_DEBUG_OBJECT (jitterbuffer,
1014 "Peeked buffer #%d, timestamp %" GST_TIME_FORMAT ", now %d left",
1015 seqnum, GST_TIME_ARGS (timestamp),
1016 rtp_jitter_buffer_num_packets (priv->jbuf));
1018 /* apply our timestamp offset to the incomming buffer, this will be our output
1020 out_time = apply_offset (jitterbuffer, timestamp);
1022 /* If we don't know what the next seqnum should be (== -1) we have to wait
1023 * because it might be possible that we are not receiving this buffer in-order,
1024 * a buffer with a lower seqnum could arrive later and we want to push that
1025 * earlier buffer before this buffer then.
1026 * If we know the expected seqnum, we can compare it to the current seqnum to
1027 * determine if we have missing a packet. If we have a missing packet (which
1028 * must be before this packet) we can wait for it until the deadline for this
1029 * packet expires. */
1030 if ((priv->next_seqnum == -1 || priv->next_seqnum != seqnum)
1031 && out_time != -1) {
1033 GstClockTime sync_time;
1037 if (priv->next_seqnum != -1) {
1038 /* we expected next_seqnum but received something else, that's a gap */
1039 GST_WARNING_OBJECT (jitterbuffer,
1040 "Sequence number GAP detected: expected %d instead of %d",
1041 priv->next_seqnum, seqnum);
1043 /* we don't know what the next_seqnum should be, wait for the last
1044 * possible moment to push this buffer, maybe we get an earlier seqnum
1046 GST_DEBUG_OBJECT (jitterbuffer, "First buffer %d, do sync", seqnum);
1049 GST_OBJECT_LOCK (jitterbuffer);
1050 clock = GST_ELEMENT_CLOCK (jitterbuffer);
1052 GST_OBJECT_UNLOCK (jitterbuffer);
1053 /* let's just push if there is no clock */
1057 GST_DEBUG_OBJECT (jitterbuffer, "sync to timestamp %" GST_TIME_FORMAT,
1058 GST_TIME_ARGS (out_time));
1060 /* prepare for sync against clock */
1061 sync_time = out_time + GST_ELEMENT_CAST (jitterbuffer)->base_time;
1062 /* add latency, this includes our own latency and the peer latency. */
1063 sync_time += (priv->latency_ms * GST_MSECOND);
1064 sync_time += priv->peer_latency;
1066 /* create an entry for the clock */
1067 id = priv->clock_id = gst_clock_new_single_shot_id (clock, sync_time);
1068 priv->waiting_seqnum = seqnum;
1069 GST_OBJECT_UNLOCK (jitterbuffer);
1071 /* release the lock so that the other end can push stuff or unlock */
1074 ret = gst_clock_id_wait (id, NULL);
1077 /* and free the entry */
1078 gst_clock_id_unref (id);
1079 priv->clock_id = NULL;
1080 priv->waiting_seqnum = -1;
1082 /* at this point, the clock could have been unlocked by a timeout, a new
1083 * tail element was added to the queue or because we are shutting down. Check
1084 * for shutdown first. */
1085 if (priv->srcresult != GST_FLOW_OK)
1088 /* if we got unscheduled and we are not flushing, it's because a new tail
1089 * element became available in the queue. Grab it and try to push or sync. */
1090 if (ret == GST_CLOCK_UNSCHEDULED) {
1091 GST_DEBUG_OBJECT (jitterbuffer,
1092 "Wait got unscheduled, will retry to push with new buffer");
1095 /* Get new timestamp, latency might have changed */
1096 out_time = apply_offset (jitterbuffer, timestamp);
1099 /* when we get here we are ready to pop and push the buffer */
1100 outbuf = rtp_jitter_buffer_pop (priv->jbuf);
1102 /* check if we are pushing something unexpected */
1103 if (priv->next_seqnum != -1 && priv->next_seqnum != seqnum) {
1106 /* calc number of missing packets, careful for wraparounds */
1107 dropped = priv_compare_rtp_seq_lt (priv->next_seqnum, seqnum);
1109 GST_DEBUG_OBJECT (jitterbuffer,
1110 "Pushing DISCONT after dropping %d (%d to %d)", dropped,
1111 priv->next_seqnum, seqnum);
1114 priv->num_late += dropped;
1116 /* set DISCONT flag when we missed a packet. */
1117 outbuf = gst_buffer_make_metadata_writable (outbuf);
1118 GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_DISCONT);
1121 /* apply timestamp with offset to buffer now */
1122 GST_BUFFER_TIMESTAMP (outbuf) = out_time;
1124 /* now we are ready to push the buffer. Save the seqnum and release the lock
1125 * so the other end can push stuff in the queue again. */
1126 priv->last_popped_seqnum = seqnum;
1127 priv->next_seqnum = (seqnum + 1) & 0xffff;
1131 GST_DEBUG_OBJECT (jitterbuffer,
1132 "Pushing buffer %d, timestamp %" GST_TIME_FORMAT, seqnum,
1133 GST_TIME_ARGS (out_time));
1134 result = gst_pad_push (priv->srcpad, outbuf);
1135 if (result != GST_FLOW_OK)
1143 /* store result, we are flushing now */
1144 GST_DEBUG_OBJECT (jitterbuffer, "We are EOS, pushing EOS downstream");
1145 priv->srcresult = GST_FLOW_UNEXPECTED;
1146 gst_pad_pause_task (priv->srcpad);
1147 gst_pad_push_event (priv->srcpad, gst_event_new_eos ());
1153 GST_DEBUG_OBJECT (jitterbuffer, "we are flushing");
1154 gst_pad_pause_task (priv->srcpad);
1156 gst_buffer_unref (outbuf);
1162 const gchar *reason = gst_flow_get_name (result);
1164 GST_DEBUG_OBJECT (jitterbuffer, "pausing task, reason %s", reason);
1168 priv->srcresult = result;
1169 /* we don't post errors or anything because upstream will do that for us
1170 * when we pass the return value upstream. */
1171 gst_pad_pause_task (priv->srcpad);
1178 gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query)
1180 GstRtpJitterBuffer *jitterbuffer;
1181 GstRtpJitterBufferPrivate *priv;
1182 gboolean res = FALSE;
1184 jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
1185 priv = jitterbuffer->priv;
1187 switch (GST_QUERY_TYPE (query)) {
1188 case GST_QUERY_LATENCY:
1190 /* We need to send the query upstream and add the returned latency to our
1192 GstClockTime min_latency, max_latency;
1194 GstClockTime our_latency;
1196 if ((res = gst_pad_peer_query (priv->sinkpad, query))) {
1197 gst_query_parse_latency (query, &us_live, &min_latency, &max_latency);
1199 GST_DEBUG_OBJECT (jitterbuffer, "Peer latency: min %"
1200 GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
1201 GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
1203 /* store this so that we can safely sync on the peer buffers. */
1205 priv->peer_latency = min_latency;
1206 our_latency = ((guint64) priv->latency_ms) * GST_MSECOND;
1209 GST_DEBUG_OBJECT (jitterbuffer, "Our latency: %" GST_TIME_FORMAT,
1210 GST_TIME_ARGS (our_latency));
1212 min_latency += our_latency;
1213 /* max_latency can be -1, meaning there is no upper limit for the
1215 if (max_latency != -1)
1216 max_latency += our_latency;
1218 GST_DEBUG_OBJECT (jitterbuffer, "Calculated total latency : min %"
1219 GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
1220 GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
1222 gst_query_set_latency (query, TRUE, min_latency, max_latency);
1227 res = gst_pad_query_default (pad, query);
1234 gst_rtp_jitter_buffer_set_property (GObject * object,
1235 guint prop_id, const GValue * value, GParamSpec * pspec)
1237 GstRtpJitterBuffer *jitterbuffer;
1238 GstRtpJitterBufferPrivate *priv;
1240 jitterbuffer = GST_RTP_JITTER_BUFFER (object);
1241 priv = jitterbuffer->priv;
1246 guint new_latency, old_latency;
1248 new_latency = g_value_get_uint (value);
1251 old_latency = priv->latency_ms;
1252 priv->latency_ms = new_latency;
1255 /* post message if latency changed, this will inform the parent pipeline
1256 * that a latency reconfiguration is possible/needed. */
1257 if (new_latency != old_latency) {
1258 GST_DEBUG_OBJECT (jitterbuffer, "latency changed to: %" GST_TIME_FORMAT,
1259 GST_TIME_ARGS (new_latency * GST_MSECOND));
1261 gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer),
1262 gst_message_new_latency (GST_OBJECT_CAST (jitterbuffer)));
1266 case PROP_DROP_ON_LATENCY:
1267 priv->drop_on_latency = g_value_get_boolean (value);
1269 case PROP_TS_OFFSET:
1271 priv->ts_offset = g_value_get_int64 (value);
1275 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1281 gst_rtp_jitter_buffer_get_property (GObject * object,
1282 guint prop_id, GValue * value, GParamSpec * pspec)
1284 GstRtpJitterBuffer *jitterbuffer;
1285 GstRtpJitterBufferPrivate *priv;
1287 jitterbuffer = GST_RTP_JITTER_BUFFER (object);
1288 priv = jitterbuffer->priv;
1293 g_value_set_uint (value, priv->latency_ms);
1296 case PROP_DROP_ON_LATENCY:
1297 g_value_set_boolean (value, priv->drop_on_latency);
1299 case PROP_TS_OFFSET:
1301 g_value_set_int64 (value, priv->ts_offset);
1305 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);