tests/check/elements/rtpjitterbuffer.c

   1 /* GStreamer
   2  *
   3  * Copyright (C) 2009 Nokia Corporation and its subsidary(-ies)
   4  *               contact: <stefan.kost@nokia.com>
   5  * Copyright (C) 2012 Cisco Systems, Inc
   6  *               Authors: Kelley Rogers <kelro@cisco.com>
   7  *               Havard Graff <hgraff@cisco.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 #include <gst/check/gstcheck.h>
  26 #include <gst/check/gsttestclock.h>
  27
  28 #include <gst/rtp/gstrtpbuffer.h>
  29
  30 /* For ease of programming we use globals to keep refs for our floating
  31  * src and sink pads we create; otherwise we always have to do get_pad,
  32  * get_peer, and then remove references in every test function */
  33 static GstPad *mysrcpad, *mysinkpad;
  34 /* we also have a list of src buffers */
  35 static GList *inbuffers = NULL;
  36 static gint num_dropped = 0;
  37
  38 #define RTP_CAPS_STRING    \
  39     "application/x-rtp, "               \
  40     "media = (string)audio, "           \
  41     "payload = (int) 0, "               \
  42     "clock-rate = (int) 8000, "         \
  43     "encoding-name = (string)PCMU"
  44
  45 #define RTP_FRAME_SIZE 20
  46
  47 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  48     GST_PAD_SINK,
  49     GST_PAD_ALWAYS,
  50     GST_STATIC_CAPS ("application/x-rtp")
  51     );
  52 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  53     GST_PAD_SRC,
  54     GST_PAD_ALWAYS,
  55     GST_STATIC_CAPS ("application/x-rtp, "
  56         "clock-rate = (int) [ 1, 2147483647 ]")
  57     );
  58
  59 static void
  60 buffer_dropped (gpointer data, GstMiniObject * obj)
  61 {
  62   GST_DEBUG ("dropping buffer %p", obj);
  63   num_dropped++;
  64 }
  65
  66 static GstElement *
  67 setup_jitterbuffer (gint num_buffers)
  68 {
  69   GstElement *jitterbuffer;
  70   GstClock *clock;
  71   GstBuffer *buffer;
  72   GstCaps *caps;
  73   /* a 20 sample audio block (2,5 ms) generated with
  74    * gst-launch audiotestsrc wave=silence blocksize=40 num-buffers=3 !
  75    *    "audio/x-raw,channels=1,rate=8000" ! mulawenc ! rtppcmupay !
  76    *     fakesink dump=1
  77    */
  78   guint8 in[] = {               /* first 4 bytes are rtp-header, next 4 bytes are timestamp */
  79     0x80, 0x80, 0x1c, 0x24, 0x46, 0xcd, 0xb7, 0x11, 0x3c, 0x3a, 0x7c, 0x5b,
  80     0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
  81     0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff
  82   };
  83   GstClockTime ts = G_GUINT64_CONSTANT (0);
  84   GstClockTime tso = gst_util_uint64_scale (RTP_FRAME_SIZE, GST_SECOND, 8000);
  85   /*guint latency = GST_TIME_AS_MSECONDS (num_buffers * tso); */
  86   gint i;
  87
  88   GST_DEBUG ("setup_jitterbuffer");
  89   jitterbuffer = gst_check_setup_element ("rtpjitterbuffer");
  90   /* we need a clock here */
  91   clock = gst_system_clock_obtain ();
  92   gst_element_set_clock (jitterbuffer, clock);
  93   gst_object_unref (clock);
  94   /* setup latency */
  95   /* latency would be 7 for 3 buffers here, default is 200
  96      g_object_set (G_OBJECT (jitterbuffer), "latency", latency, NULL);
  97      GST_INFO_OBJECT (jitterbuffer, "set latency to %u ms", latency);
  98    */
  99
 100   mysrcpad = gst_check_setup_src_pad (jitterbuffer, &srctemplate);
 101   mysinkpad = gst_check_setup_sink_pad (jitterbuffer, &sinktemplate);
 102   gst_pad_set_active (mysrcpad, TRUE);
 103   gst_pad_set_active (mysinkpad, TRUE);
 104
 105   /* create n buffers */
 106   caps = gst_caps_from_string (RTP_CAPS_STRING);
 107   gst_check_setup_events (mysrcpad, jitterbuffer, caps, GST_FORMAT_TIME);
 108   gst_caps_unref (caps);
 109
 110   for (i = 0; i < num_buffers; i++) {
 111     buffer = gst_buffer_new_and_alloc (sizeof (in));
 112     gst_buffer_fill (buffer, 0, in, sizeof (in));
 113     GST_BUFFER_DTS (buffer) = ts;
 114     GST_BUFFER_PTS (buffer) = ts;
 115     GST_BUFFER_DURATION (buffer) = tso;
 116     gst_mini_object_weak_ref (GST_MINI_OBJECT (buffer), buffer_dropped, NULL);
 117     GST_DEBUG ("created buffer: %p", buffer);
 118
 119     if (!i)
 120       GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT);
 121
 122     inbuffers = g_list_append (inbuffers, buffer);
 123
 124     /* hackish way to update the rtp header */
 125     in[1] = 0x00;
 126     in[3]++;                    /* seqnumber */
 127     in[7] += RTP_FRAME_SIZE;    /* inc. timestamp with framesize */
 128     ts += tso;
 129   }
 130   num_dropped = 0;
 131
 132   return jitterbuffer;
 133 }
 134
 135 static GstStateChangeReturn
 136 start_jitterbuffer (GstElement * jitterbuffer)
 137 {
 138   GstStateChangeReturn ret;
 139   GstClockTime now;
 140   GstClock *clock;
 141
 142   clock = gst_element_get_clock (jitterbuffer);
 143   now = gst_clock_get_time (clock);
 144   gst_object_unref (clock);
 145
 146   gst_element_set_base_time (jitterbuffer, now);
 147   ret = gst_element_set_state (jitterbuffer, GST_STATE_PLAYING);
 148
 149   return ret;
 150 }
 151
 152 static void
 153 cleanup_jitterbuffer (GstElement * jitterbuffer)
 154 {
 155   GST_DEBUG ("cleanup_jitterbuffer");
 156
 157   g_list_foreach (buffers, (GFunc) gst_mini_object_unref, NULL);
 158   g_list_free (buffers);
 159   buffers = NULL;
 160
 161   g_list_free (inbuffers);
 162   inbuffers = NULL;
 163
 164   gst_pad_set_active (mysrcpad, FALSE);
 165   gst_pad_set_active (mysinkpad, FALSE);
 166   gst_check_teardown_src_pad (jitterbuffer);
 167   gst_check_teardown_sink_pad (jitterbuffer);
 168   gst_check_teardown_element (jitterbuffer);
 169 }
 170
 171 static void
 172 check_jitterbuffer_results (GstElement * jitterbuffer, gint num_buffers)
 173 {
 174   GstBuffer *buffer;
 175   GList *node;
 176   GstClockTime ts = G_GUINT64_CONSTANT (0);
 177   GstClockTime tso = gst_util_uint64_scale (RTP_FRAME_SIZE, GST_SECOND, 8000);
 178   GstMapInfo map;
 179   guint16 prev_sn = 0, cur_sn;
 180   guint32 prev_ts = 0, cur_ts;
 181
 182   /* sleep for twice the latency */
 183   g_usleep (400 * 1000);
 184
 185   GST_INFO ("of %d buffer %d/%d received/dropped", num_buffers,
 186       g_list_length (buffers), num_dropped);
 187   /* if this fails, not all buffers have been processed */
 188   fail_unless_equals_int ((g_list_length (buffers) + num_dropped), num_buffers);
 189
 190   /* check the buffer list */
 191   fail_unless_equals_int (g_list_length (buffers), num_buffers);
 192   for (node = buffers; node; node = g_list_next (node)) {
 193     fail_if ((buffer = (GstBuffer *) node->data) == NULL);
 194     fail_if (GST_BUFFER_PTS (buffer) != ts);
 195     fail_if (GST_BUFFER_DTS (buffer) != ts);
 196     gst_buffer_map (buffer, &map, GST_MAP_READ);
 197     cur_sn = ((guint16) map.data[2] << 8) | map.data[3];
 198     cur_ts = ((guint32) map.data[4] << 24) | ((guint32) map.data[5] << 16) |
 199         ((guint32) map.data[6] << 8) | map.data[7];
 200     gst_buffer_unmap (buffer, &map);
 201
 202     if (node != buffers) {
 203       fail_unless (cur_sn > prev_sn);
 204       fail_unless (cur_ts > prev_ts);
 205
 206       prev_sn = cur_sn;
 207       prev_ts = cur_ts;
 208     }
 209     ts += tso;
 210   }
 211 }
 212
 213 GST_START_TEST (test_push_forward_seq)
 214 {
 215   GstElement *jitterbuffer;
 216   const guint num_buffers = 3;
 217   GstBuffer *buffer;
 218   GList *node;
 219
 220   jitterbuffer = setup_jitterbuffer (num_buffers);
 221   fail_unless (start_jitterbuffer (jitterbuffer)
 222       == GST_STATE_CHANGE_SUCCESS, "could not set to playing");
 223
 224   /* push buffers: 0,1,2, */
 225   for (node = inbuffers; node; node = g_list_next (node)) {
 226     buffer = (GstBuffer *) node->data;
 227     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 228   }
 229
 230   /* check the buffer list */
 231   check_jitterbuffer_results (jitterbuffer, num_buffers);
 232
 233   /* cleanup */
 234   cleanup_jitterbuffer (jitterbuffer);
 235 }
 236
 237 GST_END_TEST;
 238
 239 GST_START_TEST (test_push_backward_seq)
 240 {
 241   GstElement *jitterbuffer;
 242   const guint num_buffers = 4;
 243   GstBuffer *buffer;
 244   GList *node;
 245
 246   jitterbuffer = setup_jitterbuffer (num_buffers);
 247   fail_unless (start_jitterbuffer (jitterbuffer)
 248       == GST_STATE_CHANGE_SUCCESS, "could not set to playing");
 249
 250   /* push buffers: 0,3,2,1 */
 251   buffer = (GstBuffer *) inbuffers->data;
 252   fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 253   for (node = g_list_last (inbuffers); node != inbuffers;
 254       node = g_list_previous (node)) {
 255     buffer = (GstBuffer *) node->data;
 256     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 257   }
 258
 259   /* check the buffer list */
 260   check_jitterbuffer_results (jitterbuffer, num_buffers);
 261
 262   /* cleanup */
 263   cleanup_jitterbuffer (jitterbuffer);
 264 }
 265
 266 GST_END_TEST;
 267
 268 GST_START_TEST (test_push_unordered)
 269 {
 270   GstElement *jitterbuffer;
 271   const guint num_buffers = 4;
 272   GstBuffer *buffer;
 273
 274   jitterbuffer = setup_jitterbuffer (num_buffers);
 275   fail_unless (start_jitterbuffer (jitterbuffer)
 276       == GST_STATE_CHANGE_SUCCESS, "could not set to playing");
 277
 278   /* push buffers; 0,2,1,3 */
 279   buffer = (GstBuffer *) inbuffers->data;
 280   fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 281   buffer = g_list_nth_data (inbuffers, 2);
 282   fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 283   buffer = g_list_nth_data (inbuffers, 1);
 284   fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 285   buffer = g_list_nth_data (inbuffers, 3);
 286   fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 287
 288   /* check the buffer list */
 289   check_jitterbuffer_results (jitterbuffer, num_buffers);
 290
 291   /* cleanup */
 292   cleanup_jitterbuffer (jitterbuffer);
 293 }
 294
 295 GST_END_TEST;
 296
 297 GST_START_TEST (test_basetime)
 298 {
 299   GstElement *jitterbuffer;
 300   const guint num_buffers = 3;
 301   GstBuffer *buffer;
 302   GList *node;
 303   GstClockTime tso = gst_util_uint64_scale (RTP_FRAME_SIZE, GST_SECOND, 8000);
 304
 305   jitterbuffer = setup_jitterbuffer (num_buffers);
 306   fail_unless (start_jitterbuffer (jitterbuffer)
 307       == GST_STATE_CHANGE_SUCCESS, "could not set to playing");
 308
 309   /* push buffers: 2,1,0 */
 310   for (node = g_list_last (inbuffers); node; node = g_list_previous (node)) {
 311     buffer = (GstBuffer *) node->data;
 312     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 313   }
 314
 315   /* sleep for twice the latency */
 316   g_usleep (400 * 1000);
 317
 318   /* if this fails, not all buffers have been processed */
 319   fail_unless_equals_int ((g_list_length (buffers) + num_dropped), num_buffers);
 320
 321   buffer = (GstBuffer *) buffers->data;
 322   fail_unless (GST_BUFFER_DTS (buffer) != (num_buffers * tso));
 323   fail_unless (GST_BUFFER_PTS (buffer) != (num_buffers * tso));
 324
 325   /* cleanup */
 326   cleanup_jitterbuffer (jitterbuffer);
 327 }
 328
 329 GST_END_TEST;
 330
 331 static const guint payload_size = 160;
 332 static const guint clock_rate = 8000;
 333 static const guint pcmu_payload_type = 0;
 334 static const guint test_ssrc = 0x01BADBAD;
 335
 336 typedef struct
 337 {
 338   GstElement *jitter_buffer;
 339   GstPad *test_sink_pad, *test_src_pad;
 340   GstClock *clock;
 341   GAsyncQueue *buf_queue;
 342   GAsyncQueue *sink_event_queue;
 343   GAsyncQueue *src_event_queue;
 344   gint lost_event_count;
 345   gint rtx_event_count;
 346 } TestData;
 347
 348 static GstCaps *
 349 generate_caps (void)
 350 {
 351   return gst_caps_new_simple ("application/x-rtp",
 352       "media", G_TYPE_STRING, "audio",
 353       "clock-rate", G_TYPE_INT, clock_rate,
 354       "encoding-name", G_TYPE_STRING, "PCMU",
 355       "payload", G_TYPE_INT, pcmu_payload_type,
 356       "ssrc", G_TYPE_UINT, test_ssrc, NULL);
 357 }
 358
 359 static GstBuffer *
 360 generate_test_buffer (GstClockTime gst_ts,
 361     gboolean marker_bit, guint seq_num, guint32 rtp_ts)
 362 {
 363   GstBuffer *buf;
 364   guint8 *payload;
 365   guint i;
 366   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 367
 368   buf = gst_rtp_buffer_new_allocate (payload_size, 0, 0);
 369   GST_BUFFER_DTS (buf) = gst_ts;
 370   GST_BUFFER_PTS (buf) = gst_ts;
 371
 372   gst_rtp_buffer_map (buf, GST_MAP_READWRITE, &rtp);
 373   gst_rtp_buffer_set_payload_type (&rtp, pcmu_payload_type);
 374   gst_rtp_buffer_set_marker (&rtp, marker_bit);
 375   gst_rtp_buffer_set_seq (&rtp, seq_num);
 376   gst_rtp_buffer_set_timestamp (&rtp, rtp_ts);
 377   gst_rtp_buffer_set_ssrc (&rtp, test_ssrc);
 378
 379   payload = gst_rtp_buffer_get_payload (&rtp);
 380   for (i = 0; i < payload_size; i++)
 381     payload[i] = 0xff;
 382
 383   gst_rtp_buffer_unmap (&rtp);
 384
 385   return buf;
 386 }
 387
 388 static GstFlowReturn
 389 test_sink_pad_chain_cb (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 390 {
 391   TestData *data = gst_pad_get_element_private (pad);
 392   g_async_queue_push (data->buf_queue, buffer);
 393   return GST_FLOW_OK;
 394 }
 395
 396 static gboolean
 397 test_sink_pad_event_cb (GstPad * pad, GstObject * parent, GstEvent * event)
 398 {
 399   TestData *data = gst_pad_get_element_private (pad);
 400   const GstStructure *structure = gst_event_get_structure (event);
 401
 402   GST_DEBUG ("got event %" GST_PTR_FORMAT, event);
 403
 404   if (strcmp (gst_structure_get_name (structure), "GstRTPPacketLost") == 0) {
 405     data->lost_event_count++;
 406     GST_DEBUG ("lost event count %d", data->lost_event_count);
 407   }
 408
 409   g_async_queue_push (data->sink_event_queue, event);
 410   return TRUE;
 411 }
 412
 413 static gboolean
 414 test_src_pad_event_cb (GstPad * pad, GstObject * parent, GstEvent * event)
 415 {
 416   TestData *data = gst_pad_get_element_private (pad);
 417   const GstStructure *structure = gst_event_get_structure (event);
 418
 419   GST_DEBUG ("got event %" GST_PTR_FORMAT, event);
 420
 421   if (structure
 422       && strcmp (gst_structure_get_name (structure),
 423           "GstRTPRetransmissionRequest") == 0) {
 424     data->rtx_event_count++;
 425     GST_DEBUG ("rtx event count %d", data->rtx_event_count);
 426   }
 427
 428   g_async_queue_push (data->src_event_queue, event);
 429   return TRUE;
 430 }
 431
 432 static void
 433 setup_testharness (TestData * data)
 434 {
 435   GstPad *jb_sink_pad, *jb_src_pad;
 436   GstSegment seg;
 437   GstMiniObject *obj;
 438
 439   /* create the testclock */
 440   data->clock = gst_test_clock_new ();
 441   g_assert (data->clock);
 442   gst_test_clock_set_time (GST_TEST_CLOCK (data->clock), 0);
 443
 444   /* rig up the jitter buffer */
 445   data->jitter_buffer = gst_element_factory_make ("rtpjitterbuffer", NULL);
 446   g_assert (data->jitter_buffer);
 447   gst_element_set_clock (data->jitter_buffer, data->clock);
 448   g_object_set (data->jitter_buffer, "do-lost", TRUE, NULL);
 449   g_assert_cmpint (gst_element_set_state (data->jitter_buffer,
 450           GST_STATE_PLAYING), !=, GST_STATE_CHANGE_FAILURE);
 451
 452   /* set up the buf and event queues */
 453   data->buf_queue =
 454       g_async_queue_new_full ((GDestroyNotify) gst_mini_object_unref);
 455   data->sink_event_queue =
 456       g_async_queue_new_full ((GDestroyNotify) gst_mini_object_unref);
 457   data->src_event_queue =
 458       g_async_queue_new_full ((GDestroyNotify) gst_mini_object_unref);
 459
 460   data->lost_event_count = 0;
 461   data->rtx_event_count = 0;
 462
 463   /* link in the test source-pad */
 464   data->test_src_pad = gst_pad_new ("src", GST_PAD_SRC);
 465   gst_pad_set_element_private (data->test_src_pad, data);
 466   gst_pad_set_event_function (data->test_src_pad, test_src_pad_event_cb);
 467   jb_sink_pad = gst_element_get_static_pad (data->jitter_buffer, "sink");
 468   g_assert_cmpint (gst_pad_link (data->test_src_pad, jb_sink_pad), ==,
 469       GST_PAD_LINK_OK);
 470   gst_object_unref (jb_sink_pad);
 471
 472   /* link in the test sink-pad */
 473   data->test_sink_pad = gst_pad_new ("sink", GST_PAD_SINK);
 474   gst_pad_set_element_private (data->test_sink_pad, data);
 475   gst_pad_set_caps (data->test_sink_pad, generate_caps ());
 476   gst_pad_set_chain_function (data->test_sink_pad, test_sink_pad_chain_cb);
 477   gst_pad_set_event_function (data->test_sink_pad, test_sink_pad_event_cb);
 478   jb_src_pad = gst_element_get_static_pad (data->jitter_buffer, "src");
 479   g_assert_cmpint (gst_pad_link (jb_src_pad, data->test_sink_pad), ==,
 480       GST_PAD_LINK_OK);
 481   gst_object_unref (jb_src_pad);
 482
 483   g_assert (gst_pad_set_active (data->test_src_pad, TRUE));
 484   g_assert (gst_pad_set_active (data->test_sink_pad, TRUE));
 485
 486   gst_segment_init (&seg, GST_FORMAT_TIME);
 487
 488   gst_pad_push_event (data->test_src_pad,
 489       gst_event_new_stream_start ("stream0"));
 490   gst_pad_set_caps (data->test_src_pad, generate_caps ());
 491   gst_pad_push_event (data->test_src_pad, gst_event_new_segment (&seg));
 492
 493   while ((obj = g_async_queue_try_pop (data->sink_event_queue)))
 494     gst_mini_object_unref (obj);
 495 }
 496
 497 static void
 498 destroy_testharness (TestData * data)
 499 {
 500   /* clean up */
 501   g_assert_cmpint (gst_element_set_state (data->jitter_buffer, GST_STATE_NULL),
 502       ==, GST_STATE_CHANGE_SUCCESS);
 503   gst_object_unref (data->jitter_buffer);
 504   data->jitter_buffer = NULL;
 505
 506   gst_object_unref (data->test_src_pad);
 507   data->test_src_pad = NULL;
 508
 509   gst_object_unref (data->test_sink_pad);
 510   data->test_sink_pad = NULL;
 511
 512   gst_object_unref (data->clock);
 513   data->clock = NULL;
 514
 515   g_async_queue_unref (data->buf_queue);
 516   data->buf_queue = NULL;
 517
 518   g_async_queue_unref (data->sink_event_queue);
 519   data->sink_event_queue = NULL;
 520   g_async_queue_unref (data->src_event_queue);
 521   data->src_event_queue = NULL;
 522
 523   data->lost_event_count = 0;
 524 }
 525
 526 static void
 527 verify_lost_event (GstEvent * event, guint32 expected_seqnum,
 528     GstClockTime expected_timestamp, GstClockTime expected_duration,
 529     gboolean expected_late)
 530 {
 531   const GstStructure *s = gst_event_get_structure (event);
 532   const GValue *value;
 533   guint32 seqnum;
 534   GstClockTime timestamp;
 535   GstClockTime duration;
 536   gboolean late;
 537
 538   g_assert (gst_structure_get_uint (s, "seqnum", &seqnum));
 539
 540   value = gst_structure_get_value (s, "timestamp");
 541   g_assert (value && G_VALUE_HOLDS_UINT64 (value));
 542   timestamp = g_value_get_uint64 (value);
 543
 544   value = gst_structure_get_value (s, "duration");
 545   g_assert (value && G_VALUE_HOLDS_UINT64 (value));
 546   duration = g_value_get_uint64 (value);
 547
 548   g_assert (gst_structure_get_boolean (s, "late", &late));
 549
 550   g_assert_cmpint (seqnum, ==, expected_seqnum);
 551   g_assert_cmpint (timestamp, ==, expected_timestamp);
 552   g_assert_cmpint (duration, ==, expected_duration);
 553   g_assert (late == expected_late);
 554 }
 555
 556 static void
 557 verify_rtx_event (GstEvent * event, guint32 expected_seqnum,
 558     GstClockTime expected_timestamp, guint expected_delay,
 559     GstClockTime expected_spacing)
 560 {
 561   const GstStructure *s = gst_event_get_structure (event);
 562   const GValue *value;
 563   guint32 seqnum;
 564   GstClockTime timestamp, spacing;
 565   guint delay;
 566
 567   g_assert (gst_structure_get_uint (s, "seqnum", &seqnum));
 568
 569   value = gst_structure_get_value (s, "running-time");
 570   g_assert (value && G_VALUE_HOLDS_UINT64 (value));
 571   timestamp = g_value_get_uint64 (value);
 572
 573   value = gst_structure_get_value (s, "delay");
 574   g_assert (value && G_VALUE_HOLDS_UINT (value));
 575   delay = g_value_get_uint (value);
 576
 577   value = gst_structure_get_value (s, "packet-spacing");
 578   g_assert (value && G_VALUE_HOLDS_UINT64 (value));
 579   spacing = g_value_get_uint64 (value);
 580
 581   g_assert_cmpint (seqnum, ==, expected_seqnum);
 582   g_assert_cmpint (timestamp, ==, expected_timestamp);
 583   g_assert_cmpint (delay, ==, expected_delay);
 584   g_assert_cmpint (spacing, ==, expected_spacing);
 585 }
 586
 587 GST_START_TEST (test_only_one_lost_event_on_large_gaps)
 588 {
 589   TestData data;
 590   GstClockID id, test_id;
 591   GstBuffer *in_buf, *out_buf;
 592   GstEvent *out_event;
 593   gint jb_latency_ms = 200;
 594   guint buffer_size_ms = (payload_size * 1000) / clock_rate;
 595   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 596
 597   setup_testharness (&data);
 598
 599   g_object_set (data.jitter_buffer, "latency", jb_latency_ms, NULL);
 600
 601   /* push the first buffer in */
 602   in_buf = generate_test_buffer (0 * GST_MSECOND, TRUE, 0, 0);
 603   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 0);
 604   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 605
 606   /* wait for the first buffer to be synced to timestamp + latency */
 607   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 608
 609   /* increase the time to timestamp + latency and release the wait */
 610   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock),
 611       jb_latency_ms * GST_MSECOND);
 612   g_assert (gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock))
 613       == id);
 614
 615   /* check for the buffer coming out that was pushed in */
 616   out_buf = g_async_queue_pop (data.buf_queue);
 617   g_assert (out_buf != NULL);
 618   g_assert_cmpint (GST_BUFFER_DTS (out_buf), ==, 0);
 619   g_assert_cmpint (GST_BUFFER_PTS (out_buf), ==, 0);
 620
 621   /* move time ahead 10 seconds */
 622   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 10 * GST_SECOND);
 623
 624   /* wait a bit */
 625   g_usleep (G_USEC_PER_SEC / 10);
 626
 627   /* check that no buffers have been pushed out and no pending waits */
 628   g_assert_cmpint (g_async_queue_length (data.buf_queue), ==, 0);
 629   g_assert (gst_test_clock_peek_next_pending_id (GST_TEST_CLOCK (data.clock),
 630           &id) == FALSE);
 631
 632   /* a buffer now arrives perfectly on time */
 633   in_buf = generate_test_buffer (10 * GST_SECOND, FALSE, 500, 500 * 160);
 634   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 10 * GST_SECOND);
 635   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 636
 637   /* release the wait */
 638   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 639   gst_test_clock_advance_time (GST_TEST_CLOCK (data.clock), GST_MSECOND * 20);
 640   test_id = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
 641   g_assert (id == test_id);
 642
 643   /* we should now receive a packet-lost-event for buffers 1 through 489 */
 644   out_event = g_async_queue_pop (data.sink_event_queue);
 645   g_assert (out_event != NULL);
 646   g_assert_cmpint (data.lost_event_count, ==, 1);
 647   verify_lost_event (out_event, 1, 1 * GST_MSECOND * 20, GST_MSECOND * 20 * 490,
 648       TRUE);
 649
 650   /* churn through sync_times until the new buffer gets pushed out */
 651   while (g_async_queue_length (data.buf_queue) < 1) {
 652     if (gst_test_clock_peek_next_pending_id (GST_TEST_CLOCK (data.clock), &id)) {
 653       GstClockTime t = gst_clock_id_get_time (id);
 654       if (t > gst_clock_get_time (data.clock)) {
 655         gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), t);
 656       }
 657       gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
 658     }
 659   }
 660
 661   out_buf = g_async_queue_pop (data.buf_queue);
 662   g_assert (out_buf != NULL);
 663   g_assert (GST_BUFFER_FLAG_IS_SET (out_buf, GST_BUFFER_FLAG_DISCONT));
 664   gst_rtp_buffer_map (out_buf, GST_MAP_READ, &rtp);
 665   g_assert_cmpint (gst_rtp_buffer_get_seq (&rtp), ==, 500);
 666   gst_rtp_buffer_unmap (&rtp);
 667   g_assert_cmpint (GST_BUFFER_DTS (out_buf), ==, (10 * GST_SECOND));
 668   g_assert_cmpint (GST_BUFFER_PTS (out_buf), ==, (10 * GST_SECOND));
 669
 670   /* we get as many lost events as the the number of buffers the jitterbuffer
 671    * is able to wait for (+ the one we already got) */
 672   g_assert_cmpint (data.lost_event_count, ==, jb_latency_ms / buffer_size_ms);
 673
 674   destroy_testharness (&data);
 675 }
 676
 677 GST_END_TEST;
 678
 679 GST_START_TEST (test_two_lost_one_arrives_in_time)
 680 {
 681   TestData data;
 682   GstClockID id;
 683   GstBuffer *in_buf, *out_buf;
 684   GstEvent *out_event;
 685   gint jb_latency_ms = 100;
 686   GstClockTime buffer_time, now;
 687   gint b;
 688   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 689
 690   setup_testharness (&data);
 691
 692   g_object_set (data.jitter_buffer, "latency", jb_latency_ms, NULL);
 693
 694   /* push the first buffer in */
 695   in_buf = generate_test_buffer (0 * GST_MSECOND, TRUE, 0, 0);
 696   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 0);
 697   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 698   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 699   now = jb_latency_ms * GST_MSECOND;
 700   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), now);
 701   g_assert (gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock))
 702       == id);
 703   out_buf = g_async_queue_pop (data.buf_queue);
 704   g_assert (out_buf != NULL);
 705
 706   /* push some buffers arriving in perfect time! */
 707   for (b = 1; b < 3; b++) {
 708     buffer_time = b * GST_MSECOND * 20;
 709     in_buf = generate_test_buffer (buffer_time, TRUE, b, b * 160);
 710     gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), now + buffer_time);
 711     g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 712
 713     /* check for the buffer coming out that was pushed in */
 714     out_buf = g_async_queue_pop (data.buf_queue);
 715     g_assert (out_buf != NULL);
 716     g_assert_cmpint (GST_BUFFER_DTS (out_buf), ==, buffer_time);
 717     g_assert_cmpint (GST_BUFFER_PTS (out_buf), ==, buffer_time);
 718   }
 719
 720   /* hop over 2 packets and make another one (gap of 2) */
 721   b = 5;
 722   buffer_time = b * GST_MSECOND * 20;
 723   in_buf = generate_test_buffer (buffer_time, TRUE, b, b * 160);
 724   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 725
 726   /* verify that the jitterbuffer now wait for the latest moment it can push */
 727   /* the first lost buffer (buffer 3) out on (buffer-timestamp (60) + latency (10) = 70) */
 728   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 729   g_assert_cmpint (gst_clock_id_get_time (id), ==,
 730       (3 * GST_MSECOND * 20) + (jb_latency_ms * GST_MSECOND));
 731
 732   /* let the time expire... */
 733   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock),
 734       gst_clock_id_get_time (id));
 735   g_assert (gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock))
 736       == id);
 737
 738   /* we should now receive a packet-lost-event for buffer 3 */
 739   out_event = g_async_queue_pop (data.sink_event_queue);
 740   g_assert (out_event != NULL);
 741   g_assert_cmpint (data.lost_event_count, ==, 1);
 742   verify_lost_event (out_event, 3, 3 * GST_MSECOND * 20, GST_MSECOND * 20,
 743       FALSE);
 744
 745   /* buffer 4 now arrives just in time (time is 70, buffer 4 expires at 90) */
 746   b = 4;
 747   buffer_time = b * GST_MSECOND * 20;
 748   in_buf = generate_test_buffer (buffer_time, TRUE, b, b * 160);
 749   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 750
 751   /* verify that buffer 4 made it through! */
 752   out_buf = g_async_queue_pop (data.buf_queue);
 753   g_assert (out_buf != NULL);
 754   g_assert (GST_BUFFER_FLAG_IS_SET (out_buf, GST_BUFFER_FLAG_DISCONT));
 755   gst_rtp_buffer_map (out_buf, GST_MAP_READ, &rtp);
 756   g_assert_cmpint (gst_rtp_buffer_get_seq (&rtp), ==, 4);
 757   gst_rtp_buffer_unmap (&rtp);
 758
 759   /* and see that buffer 5 now arrives in a normal fashion */
 760   out_buf = g_async_queue_pop (data.buf_queue);
 761   g_assert (out_buf != NULL);
 762   g_assert (!GST_BUFFER_FLAG_IS_SET (out_buf, GST_BUFFER_FLAG_DISCONT));
 763   gst_rtp_buffer_map (out_buf, GST_MAP_READ, &rtp);
 764   g_assert_cmpint (gst_rtp_buffer_get_seq (&rtp), ==, 5);
 765   gst_rtp_buffer_unmap (&rtp);
 766
 767   /* should still have only seen 1 packet lost event */
 768   g_assert_cmpint (data.lost_event_count, ==, 1);
 769
 770   destroy_testharness (&data);
 771 }
 772
 773 GST_END_TEST;
 774
 775 GST_START_TEST (test_late_packets_still_makes_lost_events)
 776 {
 777   TestData data;
 778   GstClockID id;
 779   GstBuffer *in_buf, *out_buf;
 780   GstEvent *out_event;
 781   gint jb_latency_ms = 10;
 782   GstClockTime buffer_time;
 783   gint b;
 784   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 785
 786   setup_testharness (&data);
 787
 788   g_object_set (data.jitter_buffer, "latency", jb_latency_ms, NULL);
 789
 790   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 10 * GST_SECOND);
 791
 792   /* push the first buffer in */
 793   in_buf = generate_test_buffer (0 * GST_MSECOND, TRUE, 0, 0);
 794   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 795
 796   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 797   g_assert (gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock))
 798       == id);
 799   out_buf = g_async_queue_pop (data.buf_queue);
 800   g_assert (out_buf != NULL);
 801
 802   /* push some buffers in! */
 803   for (b = 1; b < 3; b++) {
 804     buffer_time = b * GST_MSECOND * 20;
 805     in_buf = generate_test_buffer (buffer_time, TRUE, b, b * 160);
 806     g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 807
 808     /* check for the buffer coming out that was pushed in */
 809     out_buf = g_async_queue_pop (data.buf_queue);
 810     g_assert (out_buf != NULL);
 811     g_assert_cmpint (GST_BUFFER_DTS (out_buf), ==, buffer_time);
 812     g_assert_cmpint (GST_BUFFER_PTS (out_buf), ==, buffer_time);
 813   }
 814
 815   /* hop over 2 packets and make another one (gap of 2) */
 816   b = 5;
 817   buffer_time = b * GST_MSECOND * 20;
 818   in_buf = generate_test_buffer (buffer_time, TRUE, b, b * 160);
 819   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 820
 821   /* we should now receive a packet-lost-event for buffer 3 and 4 */
 822   out_event = g_async_queue_pop (data.sink_event_queue);
 823   g_assert (out_event != NULL);
 824   g_assert_cmpint (data.lost_event_count, ==, 1);
 825   verify_lost_event (out_event, 3, 3 * GST_MSECOND * 20, GST_MSECOND * 20 * 2,
 826       TRUE);
 827
 828   /* verify that buffer 5 made it through! */
 829   out_buf = g_async_queue_pop (data.buf_queue);
 830   g_assert (out_buf != NULL);
 831   g_assert (GST_BUFFER_FLAG_IS_SET (out_buf, GST_BUFFER_FLAG_DISCONT));
 832   gst_rtp_buffer_map (out_buf, GST_MAP_READ, &rtp);
 833   g_assert_cmpint (gst_rtp_buffer_get_seq (&rtp), ==, 5);
 834   gst_rtp_buffer_unmap (&rtp);
 835
 836   /* should still have only seen 1 packet lost event */
 837   g_assert_cmpint (data.lost_event_count, ==, 1);
 838
 839   destroy_testharness (&data);
 840 }
 841
 842 GST_END_TEST;
 843
 844 GST_START_TEST (test_all_packets_are_timestamped_zero)
 845 {
 846   TestData data;
 847   GstClockID id;
 848   GstBuffer *in_buf, *out_buf;
 849   GstEvent *out_event;
 850   gint jb_latency_ms = 10;
 851   gint b;
 852   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 853
 854   setup_testharness (&data);
 855
 856   g_object_set (data.jitter_buffer, "latency", jb_latency_ms, NULL);
 857
 858   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 10 * GST_SECOND);
 859
 860   /* push the first buffer in */
 861   in_buf = generate_test_buffer (0 * GST_MSECOND, TRUE, 0, 0);
 862   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 863
 864   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 865   g_assert (gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock))
 866       == id);
 867   out_buf = g_async_queue_pop (data.buf_queue);
 868   g_assert (out_buf != NULL);
 869
 870   /* push some buffers in! */
 871   for (b = 1; b < 3; b++) {
 872     in_buf = generate_test_buffer (0, TRUE, b, 0);
 873     g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 874
 875     /* check for the buffer coming out that was pushed in */
 876     out_buf = g_async_queue_pop (data.buf_queue);
 877     g_assert (out_buf != NULL);
 878     g_assert_cmpint (GST_BUFFER_DTS (out_buf), ==, 0);
 879     g_assert_cmpint (GST_BUFFER_PTS (out_buf), ==, 0);
 880   }
 881
 882   /* hop over 2 packets and make another one (gap of 2) */
 883   b = 5;
 884   in_buf = generate_test_buffer (0, TRUE, b, 0);
 885   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 886
 887   /* we should now receive a packet-lost-event for buffer 3 and 4 */
 888   out_event = g_async_queue_pop (data.sink_event_queue);
 889   g_assert (out_event != NULL);
 890   verify_lost_event (out_event, 3, 0, 0, FALSE);
 891
 892   out_event = g_async_queue_pop (data.sink_event_queue);
 893   g_assert (out_event != NULL);
 894   verify_lost_event (out_event, 4, 0, 0, FALSE);
 895
 896   g_assert_cmpint (data.lost_event_count, ==, 2);
 897
 898   /* verify that buffer 5 made it through! */
 899   out_buf = g_async_queue_pop (data.buf_queue);
 900   g_assert (out_buf != NULL);
 901   g_assert (GST_BUFFER_FLAG_IS_SET (out_buf, GST_BUFFER_FLAG_DISCONT));
 902   gst_rtp_buffer_map (out_buf, GST_MAP_READ, &rtp);
 903   g_assert_cmpint (gst_rtp_buffer_get_seq (&rtp), ==, 5);
 904   gst_rtp_buffer_unmap (&rtp);
 905
 906   /* should still have only seen 2 packet lost events */
 907   g_assert_cmpint (data.lost_event_count, ==, 2);
 908
 909   destroy_testharness (&data);
 910 }
 911
 912 GST_END_TEST;
 913
 914 GST_START_TEST (test_rtx_expected_next)
 915 {
 916   TestData data;
 917   GstClockID id, tid;
 918   GstBuffer *in_buf, *out_buf;
 919   GstEvent *out_event;
 920   gint jb_latency_ms = 200;
 921
 922   setup_testharness (&data);
 923   g_object_set (data.jitter_buffer, "do-retransmission", TRUE, NULL);
 924   g_object_set (data.jitter_buffer, "latency", jb_latency_ms, NULL);
 925   g_object_set (data.jitter_buffer, "rtx-retry-period", 120, NULL);
 926
 927   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 0);
 928
 929   /* push the first buffer in */
 930   in_buf = generate_test_buffer (0 * GST_MSECOND, TRUE, 0, 0);
 931   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 932
 933   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 20 * GST_MSECOND);
 934
 935   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 936
 937   /* put second buffer, the jitterbuffer should now know that the packet spacing
 938    * is 20ms and should ask for retransmission of seqnum 2 in 20ms */
 939   in_buf = generate_test_buffer (20 * GST_MSECOND, TRUE, 1, 160);
 940   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
 941
 942   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 943   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 60 * GST_MSECOND);
 944   g_assert (gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock))
 945       == id);
 946
 947   out_event = g_async_queue_pop (data.src_event_queue);
 948   g_assert (out_event != NULL);
 949   verify_rtx_event (out_event, 2, 40 * GST_MSECOND, 20, 20 * GST_MSECOND);
 950
 951   /* now we wait for the next timeout */
 952   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 953   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 100 * GST_MSECOND);
 954   tid = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
 955   g_assert (id == tid);
 956
 957   out_event = g_async_queue_pop (data.src_event_queue);
 958   g_assert (out_event != NULL);
 959   verify_rtx_event (out_event, 2, 40 * GST_MSECOND, 60, 20 * GST_MSECOND);
 960
 961   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 962   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 140 * GST_MSECOND);
 963   tid = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
 964   g_assert (id == tid);
 965
 966   out_event = g_async_queue_pop (data.src_event_queue);
 967   g_assert (out_event != NULL);
 968   verify_rtx_event (out_event, 2, 40 * GST_MSECOND, 100, 20 * GST_MSECOND);
 969
 970   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 971   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 200 * GST_MSECOND);
 972   tid = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
 973   g_assert (id == tid);
 974
 975   out_buf = g_async_queue_pop (data.buf_queue);
 976   g_assert (out_buf != NULL);
 977
 978
 979   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
 980   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 260 * GST_MSECOND);
 981   g_assert (gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock))
 982       == id);
 983
 984   /* we should now receive a packet-lost-event for buffer 2 */
 985   out_event = g_async_queue_pop (data.sink_event_queue);
 986   g_assert (out_event != NULL);
 987   verify_lost_event (out_event, 2, 40 * GST_MSECOND, 20 * GST_MSECOND, FALSE);
 988
 989   destroy_testharness (&data);
 990 }
 991
 992 GST_END_TEST;
 993
 994 GST_START_TEST (test_rtx_two_missing)
 995 {
 996   TestData data;
 997   GstClockID id, tid;
 998   GstBuffer *in_buf, *out_buf;
 999   GstEvent *out_event;
1000   gint jb_latency_ms = 200;
1001   gint i;
1002   GstStructure *rtx_stats;
1003   const GValue *rtx_stat;
1004   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
1005
1006   setup_testharness (&data);
1007   g_object_set (data.jitter_buffer, "do-retransmission", TRUE, NULL);
1008   g_object_set (data.jitter_buffer, "latency", jb_latency_ms, NULL);
1009   g_object_set (data.jitter_buffer, "rtx-retry-period", 120, NULL);
1010
1011   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 0);
1012
1013   /* push the first buffer in */
1014   in_buf = generate_test_buffer (0 * GST_MSECOND, TRUE, 0, 0);
1015   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
1016
1017   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 20 * GST_MSECOND);
1018
1019   /* put second buffer, the jitterbuffer should now know that the packet spacing
1020    * is 20ms and should ask for retransmission of seqnum 2 at 60ms */
1021   in_buf = generate_test_buffer (20 * GST_MSECOND, TRUE, 1, 160);
1022   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
1023
1024   /* push buffer 4, 2 and 3 are missing now, we should get retransmission events
1025    * for 3 at 100ms*/
1026   in_buf = generate_test_buffer (80 * GST_MSECOND, TRUE, 4, 4 * 160);
1027   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
1028
1029   /* wait for first retransmission request */
1030   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 60 * GST_MSECOND);
1031   do {
1032     gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
1033   } while (id !=
1034       gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock)));
1035
1036   /* we should have 2 events now, one for 2 and another for 3 */
1037   out_event = g_async_queue_pop (data.src_event_queue);
1038   g_assert (out_event != NULL);
1039   verify_rtx_event (out_event, 2, 40 * GST_MSECOND, 20, 20 * GST_MSECOND);
1040   out_event = g_async_queue_pop (data.src_event_queue);
1041   g_assert (out_event != NULL);
1042   verify_rtx_event (out_event, 3, 60 * GST_MSECOND, 0, 20 * GST_MSECOND);
1043
1044   /* now we wait for the next timeout */
1045   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
1046   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 100 * GST_MSECOND);
1047   tid = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
1048   g_assert (id == tid);
1049
1050   /* we should have 2 events now, one for 2 and another for 3 */
1051   out_event = g_async_queue_pop (data.src_event_queue);
1052   g_assert (out_event != NULL);
1053   verify_rtx_event (out_event, 2, 40 * GST_MSECOND, 60, 20 * GST_MSECOND);
1054   out_event = g_async_queue_pop (data.src_event_queue);
1055   g_assert (out_event != NULL);
1056   verify_rtx_event (out_event, 3, 60 * GST_MSECOND, 40, 20 * GST_MSECOND);
1057
1058   /* make buffer 3 */
1059   in_buf = generate_test_buffer (60 * GST_MSECOND, TRUE, 3, 3 * 160);
1060   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
1061
1062   /* make more buffers */
1063   for (i = 5; i < 15; i++) {
1064     in_buf = generate_test_buffer (i * 20 * GST_MSECOND, TRUE, i, i * 160);
1065     g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
1066   }
1067
1068   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
1069   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 140 * GST_MSECOND);
1070   tid = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
1071   g_assert (id == tid);
1072
1073   /* now we only get requests for 2 */
1074   out_event = g_async_queue_pop (data.src_event_queue);
1075   g_assert (out_event != NULL);
1076   verify_rtx_event (out_event, 2, 40 * GST_MSECOND, 100, 20 * GST_MSECOND);
1077
1078   /* this is when buffer 0 deadline expires */
1079   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
1080   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 200 * GST_MSECOND);
1081   tid = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
1082   g_assert (id == tid);
1083
1084   for (i = 0; i < 2; i++) {
1085     GST_DEBUG ("popping %d", i);
1086     out_buf = g_async_queue_pop (data.buf_queue);
1087     g_assert (out_buf != NULL);
1088     gst_rtp_buffer_map (out_buf, GST_MAP_READ, &rtp);
1089     g_assert_cmpint (gst_rtp_buffer_get_seq (&rtp), ==, i);
1090     gst_rtp_buffer_unmap (&rtp);
1091   }
1092
1093   /* this is when 2 is lost */
1094   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
1095   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 240 * GST_MSECOND);
1096   tid = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
1097   g_assert (id == tid);
1098
1099   /* we should now receive a packet-lost-event for buffer 2 */
1100   out_event = g_async_queue_pop (data.sink_event_queue);
1101   g_assert (out_event != NULL);
1102   verify_lost_event (out_event, 2, 40 * GST_MSECOND, 20 * GST_MSECOND, FALSE);
1103
1104   /* verify that buffers made it through! */
1105   for (i = 3; i < 15; i++) {
1106     GST_DEBUG ("popping %d", i);
1107     out_buf = g_async_queue_pop (data.buf_queue);
1108     g_assert (out_buf != NULL);
1109     gst_rtp_buffer_map (out_buf, GST_MAP_READ, &rtp);
1110     g_assert_cmpint (gst_rtp_buffer_get_seq (&rtp), ==, i);
1111     gst_rtp_buffer_unmap (&rtp);
1112   }
1113   /* should still have only seen 1 packet lost events */
1114   g_assert_cmpint (data.lost_event_count, ==, 1);
1115
1116   g_object_get (data.jitter_buffer, "stats", &rtx_stats, NULL);
1117
1118   rtx_stat = gst_structure_get_value (rtx_stats, "rtx-count");
1119   g_assert_cmpuint (g_value_get_uint64 (rtx_stat), ==, 5);
1120
1121   rtx_stat = gst_structure_get_value (rtx_stats, "rtx-success-count");
1122   g_assert_cmpuint (g_value_get_uint64 (rtx_stat), ==, 1);
1123
1124   rtx_stat = gst_structure_get_value (rtx_stats, "rtx-rtt");
1125   g_assert_cmpuint (g_value_get_uint64 (rtx_stat), ==, 0);
1126
1127   destroy_testharness (&data);
1128 }
1129
1130 GST_END_TEST;
1131
1132 GST_START_TEST (test_rtx_packet_delay)
1133 {
1134   TestData data;
1135   GstClockID id, tid;
1136   GstBuffer *in_buf, *out_buf;
1137   GstEvent *out_event;
1138   gint jb_latency_ms = 200;
1139   gint i;
1140   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
1141
1142   setup_testharness (&data);
1143   g_object_set (data.jitter_buffer, "do-retransmission", TRUE, NULL);
1144   g_object_set (data.jitter_buffer, "latency", jb_latency_ms, NULL);
1145   g_object_set (data.jitter_buffer, "rtx-retry-period", 120, NULL);
1146
1147   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 0);
1148
1149   /* push the first buffer in */
1150   in_buf = generate_test_buffer (0 * GST_MSECOND, TRUE, 0, 0);
1151   GST_BUFFER_FLAG_SET (in_buf, GST_BUFFER_FLAG_DISCONT);
1152   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
1153
1154   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 20 * GST_MSECOND);
1155
1156   /* put second buffer, the jitterbuffer should now know that the packet spacing
1157    * is 20ms and should ask for retransmission of seqnum 2 at 60ms */
1158   in_buf = generate_test_buffer (20 * GST_MSECOND, TRUE, 1, 160);
1159   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
1160
1161   /* push buffer 8, 2 -> 7 are missing now. note that the rtp time is the same
1162    * as packet 1 because it was part of a fragmented payload. This means that
1163    * the estimate for 2 could be refined now to 20ms. also packet 2, 3 and 4 are
1164    * exceeding the max allowed reorder distance and should request a
1165    * retransmission right away */
1166   in_buf = generate_test_buffer (20 * GST_MSECOND, TRUE, 8, 8 * 160);
1167   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
1168
1169   /* we should now receive retransmission requests for 2 -> 5 */
1170   out_event = g_async_queue_pop (data.src_event_queue);
1171   g_assert (out_event != NULL);
1172   verify_rtx_event (out_event, 2, 20 * GST_MSECOND, 40, 20 * GST_MSECOND);
1173
1174   for (i = 3; i < 5; i++) {
1175     GST_DEBUG ("popping %d", i);
1176     out_event = g_async_queue_pop (data.src_event_queue);
1177     g_assert (out_event != NULL);
1178     verify_rtx_event (out_event, i, 20 * GST_MSECOND, 0, 20 * GST_MSECOND);
1179   }
1180   g_assert_cmpint (data.rtx_event_count, ==, 3);
1181
1182   /* push 9, this should immediately request retransmission of 5 */
1183   in_buf = generate_test_buffer (20 * GST_MSECOND, TRUE, 9, 9 * 160);
1184   g_assert_cmpint (gst_pad_push (data.test_src_pad, in_buf), ==, GST_FLOW_OK);
1185
1186   /* we should now receive retransmission requests for 5 */
1187   out_event = g_async_queue_pop (data.src_event_queue);
1188   g_assert (out_event != NULL);
1189   verify_rtx_event (out_event, 5, 20 * GST_MSECOND, 0, 20 * GST_MSECOND);
1190
1191   /* wait for timeout for rtx 6 -> 7 */
1192   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
1193   tid = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
1194   g_assert (id == tid);
1195
1196   for (i = 6; i < 8; i++) {
1197     GST_DEBUG ("popping %d", i);
1198     out_event = g_async_queue_pop (data.src_event_queue);
1199     g_assert (out_event != NULL);
1200     verify_rtx_event (out_event, i, 20 * GST_MSECOND, 0, 20 * GST_MSECOND);
1201   }
1202
1203   /* churn through sync_times until the new buffer gets pushed out */
1204   while (g_async_queue_length (data.buf_queue) < 1) {
1205     if (gst_test_clock_peek_next_pending_id (GST_TEST_CLOCK (data.clock), &id)) {
1206       GstClockTime t = gst_clock_id_get_time (id);
1207       if (t > gst_clock_get_time (data.clock)) {
1208         gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), t);
1209       }
1210       gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
1211     }
1212   }
1213
1214   /* verify that buffer 0 and 1 made it through! */
1215   for (i = 0; i < 2; i++) {
1216     out_buf = g_async_queue_pop (data.buf_queue);
1217     g_assert (out_buf != NULL);
1218     if (i == 0)
1219       g_assert (GST_BUFFER_FLAG_IS_SET (out_buf, GST_BUFFER_FLAG_DISCONT));
1220     gst_rtp_buffer_map (out_buf, GST_MAP_READ, &rtp);
1221     g_assert_cmpint (gst_rtp_buffer_get_seq (&rtp), ==, i);
1222     gst_rtp_buffer_unmap (&rtp);
1223   }
1224
1225   /* churn through sync_times until the next buffer gets pushed out */
1226   while (g_async_queue_length (data.buf_queue) < 1) {
1227     if (gst_test_clock_peek_next_pending_id (GST_TEST_CLOCK (data.clock), &id)) {
1228       GstClockTime t = gst_clock_id_get_time (id);
1229       if (t >= 240 * GST_MSECOND)
1230         break;
1231       if (t > gst_clock_get_time (data.clock)) {
1232         gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), t);
1233       }
1234       gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
1235     }
1236   }
1237
1238   for (i = 2; i < 8; i++) {
1239     GST_DEBUG ("popping lost event %d", i);
1240     out_event = g_async_queue_pop (data.sink_event_queue);
1241     g_assert (out_event != NULL);
1242     verify_lost_event (out_event, i, 20 * GST_MSECOND, 0, FALSE);
1243   }
1244
1245   /* verify that buffer 8 made it through! */
1246   for (i = 8; i < 10; i++) {
1247     GST_DEBUG ("popping buffer %d", i);
1248     out_buf = g_async_queue_pop (data.buf_queue);
1249     g_assert (out_buf != NULL);
1250     if (i == 8)
1251       g_assert (GST_BUFFER_FLAG_IS_SET (out_buf, GST_BUFFER_FLAG_DISCONT));
1252     gst_rtp_buffer_map (out_buf, GST_MAP_READ, &rtp);
1253     g_assert_cmpint (gst_rtp_buffer_get_seq (&rtp), ==, i);
1254     gst_rtp_buffer_unmap (&rtp);
1255   }
1256
1257   GST_DEBUG ("waiting for 240ms");
1258   gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
1259   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 240 * GST_MSECOND);
1260   tid = gst_test_clock_process_next_clock_id (GST_TEST_CLOCK (data.clock));
1261   g_assert (id == tid);
1262
1263   GST_DEBUG ("popping lost event 10");
1264   out_event = g_async_queue_pop (data.sink_event_queue);
1265   g_assert (out_event != NULL);
1266   verify_lost_event (out_event, 10, 40 * GST_MSECOND, 20 * GST_MSECOND, FALSE);
1267
1268   /* should have seen 6 packet lost events */
1269   g_assert_cmpint (data.lost_event_count, ==, 7);
1270   g_assert_cmpint (data.rtx_event_count, ==, 26);
1271
1272   destroy_testharness (&data);
1273 }
1274
1275 GST_END_TEST;
1276
1277 static Suite *
1278 rtpjitterbuffer_suite (void)
1279 {
1280   Suite *s = suite_create ("rtpjitterbuffer");
1281   TCase *tc_chain = tcase_create ("general");
1282
1283   suite_add_tcase (s, tc_chain);
1284   tcase_add_test (tc_chain, test_push_forward_seq);
1285   tcase_add_test (tc_chain, test_push_backward_seq);
1286   tcase_add_test (tc_chain, test_push_unordered);
1287   tcase_add_test (tc_chain, test_basetime);
1288   tcase_add_test (tc_chain, test_only_one_lost_event_on_large_gaps);
1289   tcase_add_test (tc_chain, test_two_lost_one_arrives_in_time);
1290   tcase_add_test (tc_chain, test_late_packets_still_makes_lost_events);
1291   tcase_add_test (tc_chain, test_all_packets_are_timestamped_zero);
1292   tcase_add_test (tc_chain, test_rtx_expected_next);
1293   tcase_add_test (tc_chain, test_rtx_two_missing);
1294   tcase_add_test (tc_chain, test_rtx_packet_delay);
1295
1296   return s;
1297 }
1298
1299 GST_CHECK_MAIN (rtpjitterbuffer);