tests/check/elements/rtpssrcdemux.c

   1 /* GStreamer
   2  *
   3  * Copyright (C) 2018 Collabora Ltd.
   4  *               Author: Nicolas Dufresne <nicolas.dufresne@collabora.com>
   5  * Copyright (C) 2019 Pexip
   6  *               Author: Havard Graff <havard@pexip.com>
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Library General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Library General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Library General Public
  19  * License along with this library; if not, write to the
  20  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  21  * Boston, MA 02110-1301, USA.
  22  */
  23 #ifdef HAVE_CONFIG_H
  24 #include "config.h"
  25 #endif
  26
  27 #include <gst/rtp/gstrtpbuffer.h>
  28 #include <gst/rtp/gstrtcpbuffer.h>
  29
  30 #include <gst/check/gstcheck.h>
  31 #include <gst/check/gstharness.h>
  32
  33 #ifdef HAVE_VALGRIND
  34 # include <valgrind/valgrind.h>
  35 #else
  36 # define RUNNING_ON_VALGRIND 0
  37 #endif
  38
  39 #define TEST_BUF_CLOCK_RATE 8000
  40 #define TEST_BUF_PT 0
  41 #define TEST_BUF_SSRC 0x01BADBAD
  42 #define TEST_BUF_MS  20
  43 #define TEST_BUF_DURATION (TEST_BUF_MS * GST_MSECOND)
  44 #define TEST_BUF_SIZE (64000 * TEST_BUF_MS / 1000)
  45 #define TEST_RTP_TS_DURATION (TEST_BUF_CLOCK_RATE * TEST_BUF_MS / 1000)
  46
  47 static GstCaps *
  48 generate_caps (void)
  49 {
  50   return gst_caps_new_simple ("application/x-rtp",
  51       "media", G_TYPE_STRING, "audio",
  52       "clock-rate", G_TYPE_INT, TEST_BUF_CLOCK_RATE, NULL);
  53 }
  54
  55 static GstBuffer *
  56 create_buffer (guint seq_num, guint32 ssrc)
  57 {
  58   GstBuffer *buf;
  59   guint8 *payload;
  60   guint i;
  61   GstClockTime dts = seq_num * TEST_BUF_DURATION;
  62   guint32 rtp_ts = seq_num * TEST_RTP_TS_DURATION;
  63   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
  64
  65   buf = gst_rtp_buffer_new_allocate (TEST_BUF_SIZE, 0, 0);
  66   GST_BUFFER_DTS (buf) = dts;
  67
  68   gst_rtp_buffer_map (buf, GST_MAP_READWRITE, &rtp);
  69   gst_rtp_buffer_set_payload_type (&rtp, TEST_BUF_PT);
  70   gst_rtp_buffer_set_seq (&rtp, seq_num);
  71   gst_rtp_buffer_set_timestamp (&rtp, rtp_ts);
  72   gst_rtp_buffer_set_ssrc (&rtp, ssrc);
  73
  74   payload = gst_rtp_buffer_get_payload (&rtp);
  75   for (i = 0; i < TEST_BUF_SIZE; i++)
  76     payload[i] = 0xff;
  77
  78   gst_rtp_buffer_unmap (&rtp);
  79
  80   return buf;
  81 }
  82
  83
  84 typedef struct
  85 {
  86   GstHarness *rtp_sink;
  87   GstHarness *rtcp_sink;
  88   GstHarness *rtp_src;
  89   GstHarness *rtcp_src;
  90 } TestContext;
  91
  92 static void
  93 rtpssrcdemux_pad_added (G_GNUC_UNUSED GstElement * demux, GstPad * src_pad,
  94     TestContext * ctx)
  95 {
  96   GstHarness *h;
  97
  98   h = gst_harness_new_with_element (ctx->rtp_sink->element, NULL,
  99       GST_PAD_NAME (src_pad));
 100
 101   /* FIXME We should also check that pads have current caps, but this is not
 102    * currently the case as both pads are created when the first pad receive a
 103    * buffer. If the other pad is not linked, you'll get a pad without caps.
 104    * Changing this implies not having both pads on 'on-new-ssrc' which would
 105    * break rtpbin assumption. */
 106
 107   if (g_str_has_prefix (GST_PAD_NAME (src_pad), "src_")) {
 108     g_assert (ctx->rtp_src == NULL);
 109     ctx->rtp_src = h;
 110   } else if (g_str_has_prefix (GST_PAD_NAME (src_pad), "rtcp_src_")) {
 111     g_assert (ctx->rtcp_src == NULL);
 112     ctx->rtcp_src = h;
 113   } else {
 114     g_assert_not_reached ();
 115   }
 116 }
 117
 118 GST_START_TEST (test_event_forwarding)
 119 {
 120   TestContext ctx = { NULL, NULL, NULL, NULL };
 121   GstHarness *h;
 122   GstEvent *event;
 123   GstCaps *caps;
 124   GstStructure *s;
 125   guint ssrc;
 126
 127   ctx.rtp_sink = h = gst_harness_new_with_padnames ("rtpssrcdemux", "sink",
 128       NULL);
 129   g_signal_connect (h->element, "pad_added",
 130       G_CALLBACK (rtpssrcdemux_pad_added), &ctx);
 131
 132   ctx.rtcp_sink = gst_harness_new_with_element (h->element, "rtcp_sink", NULL);
 133
 134   gst_harness_set_src_caps (h, generate_caps ());
 135   gst_harness_push (h, create_buffer (0, TEST_BUF_SSRC));
 136
 137   g_assert (ctx.rtp_src);
 138   g_assert (ctx.rtcp_src);
 139
 140   gst_harness_push_event (h, gst_event_new_eos ());
 141
 142   /* We expect stream-start/caps/segment/eos */
 143   g_assert_cmpint (gst_harness_events_in_queue (ctx.rtp_src), ==, 4);
 144
 145   event = gst_harness_pull_event (ctx.rtp_src);
 146   g_assert_cmpint (event->type, ==, GST_EVENT_STREAM_START);
 147   gst_event_unref (event);
 148
 149   event = gst_harness_pull_event (ctx.rtp_src);
 150   g_assert_cmpint (event->type, ==, GST_EVENT_CAPS);
 151   gst_event_parse_caps (event, &caps);
 152   s = gst_caps_get_structure (caps, 0);
 153   g_assert (gst_structure_has_field (s, "ssrc"));
 154   g_assert (gst_structure_get_uint (s, "ssrc", &ssrc));
 155   g_assert_cmpuint (ssrc, ==, TEST_BUF_SSRC);
 156   gst_event_unref (event);
 157
 158   event = gst_harness_pull_event (ctx.rtp_src);
 159   g_assert_cmpint (event->type, ==, GST_EVENT_SEGMENT);
 160   gst_event_unref (event);
 161
 162   event = gst_harness_pull_event (ctx.rtp_src);
 163   g_assert_cmpint (event->type, ==, GST_EVENT_EOS);
 164   gst_event_unref (event);
 165
 166   /* We pushed on the RTP pad, no events should have reached the RTCP pad */
 167   g_assert_cmpint (gst_harness_events_in_queue (ctx.rtcp_src), ==, 0);
 168
 169   /* push EOS on the rtcp sink pad, to make sure it EOS properly, the harness
 170    * will create the missing stream-start */
 171   gst_harness_push_event (ctx.rtcp_sink, gst_event_new_eos ());
 172
 173   g_assert_cmpint (gst_harness_events_in_queue (ctx.rtp_src), ==, 0);
 174   g_assert_cmpint (gst_harness_events_in_queue (ctx.rtcp_src), ==, 1);
 175
 176   event = gst_harness_pull_event (ctx.rtcp_src);
 177   g_assert_cmpint (event->type, ==, GST_EVENT_EOS);
 178   gst_event_unref (event);
 179
 180   gst_harness_teardown (ctx.rtp_src);
 181   gst_harness_teardown (ctx.rtcp_src);
 182   gst_harness_teardown (ctx.rtcp_sink);
 183   gst_harness_teardown (ctx.rtp_sink);
 184 }
 185
 186 GST_END_TEST;
 187
 188 typedef struct
 189 {
 190   gint ready;
 191   GMutex mutex;
 192   GCond cond;
 193 } LockTestContext;
 194
 195 static void
 196 new_ssrc_pad_cb (G_GNUC_UNUSED GstElement * element, G_GNUC_UNUSED guint ssrc,
 197     G_GNUC_UNUSED GstPad * pad, LockTestContext * ctx)
 198 {
 199   g_message ("Signalling ready");
 200   g_atomic_int_set (&ctx->ready, 1);
 201
 202   g_message ("Waiting no more ready");
 203   while (g_atomic_int_get (&ctx->ready))
 204     g_usleep (G_USEC_PER_SEC / 100);
 205
 206   g_mutex_lock (&ctx->mutex);
 207   g_mutex_unlock (&ctx->mutex);
 208 }
 209
 210 static gpointer
 211 push_buffer_func (gpointer user_data)
 212 {
 213   GstHarness *h = user_data;
 214   gst_harness_push (h, create_buffer (0, 0xdeadbeef));
 215   return NULL;
 216 }
 217
 218 GST_START_TEST (test_oob_event_locking)
 219 {
 220   GstHarness *h = gst_harness_new_with_padnames ("rtpssrcdemux", "sink", NULL);
 221   LockTestContext ctx;
 222   GThread *thread;
 223
 224   memset (&ctx, 0, sizeof (LockTestContext));
 225   g_mutex_init (&ctx.mutex);
 226   g_cond_init (&ctx.cond);
 227
 228   gst_harness_set_src_caps_str (h, "application/x-rtp");
 229   g_signal_connect (h->element,
 230       "new-ssrc-pad", G_CALLBACK (new_ssrc_pad_cb), &ctx);
 231
 232   thread = g_thread_new ("streaming-thread", push_buffer_func, h);
 233
 234   g_mutex_lock (&ctx.mutex);
 235
 236   g_message ("Waiting for ready");
 237   while (!g_atomic_int_get (&ctx.ready))
 238     g_usleep (G_USEC_PER_SEC / 100);
 239   g_message ("Signal no more ready");
 240   g_atomic_int_set (&ctx.ready, 0);
 241
 242   gst_harness_push_event (h,
 243       gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM_OOB, NULL));
 244
 245   g_mutex_unlock (&ctx.mutex);
 246
 247   g_thread_join (thread);
 248   g_mutex_clear (&ctx.mutex);
 249   g_cond_clear (&ctx.cond);
 250   gst_harness_teardown (h);
 251 }
 252
 253 GST_END_TEST;
 254
 255
 256 static void
 257 new_ssrc_pad_found (GstElement * element, G_GNUC_UNUSED guint ssrc,
 258     GstPad * pad, GSList ** src_h)
 259 {
 260   GstHarness *h = gst_harness_new_with_element (element, NULL, NULL);
 261   gst_harness_add_element_src_pad (h, pad);
 262   *src_h = g_slist_prepend (*src_h, h);
 263 }
 264
 265 GST_START_TEST (test_rtpssrcdemux_max_streams)
 266 {
 267   GstHarness *h = gst_harness_new_with_padnames ("rtpssrcdemux", "sink", NULL);
 268   GSList *src_h = NULL;
 269   gint i;
 270
 271   g_object_set (h->element, "max-streams", 64, NULL);
 272   gst_harness_set_src_caps_str (h, "application/x-rtp");
 273   g_signal_connect (h->element,
 274       "new-ssrc-pad", (GCallback) new_ssrc_pad_found, &src_h);
 275   gst_harness_play (h);
 276
 277   for (i = 0; i < 128; ++i) {
 278     fail_unless_equals_int (GST_FLOW_OK,
 279         gst_harness_push (h, create_buffer (0, i)));
 280   }
 281
 282   fail_unless_equals_int (g_slist_length (src_h), 64);
 283   g_slist_free_full (src_h, (GDestroyNotify) gst_harness_teardown);
 284   gst_harness_teardown (h);
 285 }
 286
 287 GST_END_TEST;
 288
 289 static void
 290 new_rtcp_ssrc_pad_found (GstElement * element, guint ssrc,
 291     G_GNUC_UNUSED GstPad * rtp_pad, GSList ** src_h)
 292 {
 293   GstHarness *h;
 294   gchar *name;
 295
 296   name = g_strdup_printf ("rtcp_src_%u", ssrc);
 297   h = gst_harness_new_with_element (element, NULL, name);
 298   g_free (name);
 299   *src_h = g_slist_prepend (*src_h, h);
 300 }
 301
 302 GST_START_TEST (test_rtpssrcdemux_rtcp_app)
 303 {
 304   GstHarness *h =
 305       gst_harness_new_with_padnames ("rtpssrcdemux", "rtcp_sink", NULL);
 306   GSList *src_h = NULL;
 307   guint8 rtcp_app_pkt[] = { 0x81, 0xcc, 0x00, 0x05, 0x00, 0x00, 0x5d, 0xaf,
 308     0x20, 0x20, 0x20, 0x20, 0x21, 0x02, 0x00, 0x0a,
 309     0x00, 0x00, 0x5d, 0xaf, 0x00, 0x00, 0x16, 0x03
 310   };
 311
 312   gst_harness_set_src_caps_str (h, "application/x-rtcp");
 313   g_signal_connect (h->element,
 314       "new-ssrc-pad", (GCallback) new_rtcp_ssrc_pad_found, &src_h);
 315   gst_harness_play (h);
 316
 317   fail_unless_equals_int (GST_FLOW_OK,
 318       gst_harness_push (h, gst_buffer_new_wrapped_full (0, rtcp_app_pkt,
 319               sizeof rtcp_app_pkt, 0, sizeof rtcp_app_pkt, NULL, NULL)));
 320
 321   fail_unless_equals_int (g_slist_length (src_h), 1);
 322   g_slist_free_full (src_h, (GDestroyNotify) gst_harness_teardown);
 323   gst_harness_teardown (h);
 324 }
 325
 326 GST_END_TEST;
 327
 328 GST_START_TEST (test_rtpssrcdemux_invalid_rtp)
 329 {
 330   GstHarness *h = gst_harness_new_with_padnames ("rtpssrcdemux", "sink", NULL);
 331   guint8 bad_pkt[] = {
 332     0x01, 0x02, 0x03
 333   };
 334
 335   gst_harness_set_src_caps_str (h, "application/x-rtp");
 336   gst_harness_play (h);
 337
 338   fail_unless_equals_int (GST_FLOW_OK,
 339       gst_harness_push (h, gst_buffer_new_wrapped_full (0, bad_pkt,
 340               sizeof bad_pkt, 0, sizeof bad_pkt, NULL, NULL)));
 341
 342   gst_harness_teardown (h);
 343 }
 344
 345 GST_END_TEST;
 346
 347 GST_START_TEST (test_rtpssrcdemux_invalid_rtcp)
 348 {
 349   GstHarness *h =
 350       gst_harness_new_with_padnames ("rtpssrcdemux", "rtcp_sink", NULL);
 351   guint8 bad_pkt[] = {
 352     0x01, 0x02, 0x03
 353   };
 354
 355   gst_harness_set_src_caps_str (h, "application/x-rtcp");
 356   gst_harness_play (h);
 357
 358   fail_unless_equals_int (GST_FLOW_OK,
 359       gst_harness_push (h, gst_buffer_new_wrapped_full (0, bad_pkt,
 360               sizeof bad_pkt, 0, sizeof bad_pkt, NULL, NULL)));
 361
 362   gst_harness_teardown (h);
 363 }
 364
 365 GST_END_TEST;
 366
 367 static GstBuffer *
 368 generate_rtcp_sr_buffer (guint ssrc)
 369 {
 370   GstBuffer *buf;
 371   GstRTCPBuffer rtcp = GST_RTCP_BUFFER_INIT;
 372   GstRTCPPacket packet;
 373
 374   buf = gst_rtcp_buffer_new (1000);
 375   fail_unless (gst_rtcp_buffer_map (buf, GST_MAP_READWRITE, &rtcp));
 376   fail_unless (gst_rtcp_buffer_add_packet (&rtcp, GST_RTCP_TYPE_SR, &packet));
 377   gst_rtcp_packet_sr_set_sender_info (&packet, ssrc, 0, 0, 1, 1);
 378   gst_rtcp_buffer_unmap (&rtcp);
 379   return buf;
 380 }
 381
 382 typedef struct
 383 {
 384   GstHarness *rtp_h;
 385   GstHarness *rtcp_h;
 386 } SimulCtx;
 387
 388 static void
 389 _simul_ctx_new_ssrc_pad_cb (GstElement * element, guint ssrc,
 390     GstPad * rtp_pad, SimulCtx * ctx)
 391 {
 392   GstPad *rtcp_pad;
 393   gchar *name;
 394
 395   gst_harness_add_element_src_pad (ctx->rtp_h, rtp_pad);
 396
 397   name = g_strdup_printf ("rtcp_src_%u", ssrc);
 398   rtcp_pad = gst_element_get_static_pad (element, name);
 399   gst_harness_add_element_src_pad (ctx->rtcp_h, rtcp_pad);
 400   gst_object_unref (rtcp_pad);
 401   g_free (name);
 402 }
 403
 404 static gpointer
 405 _simul_ctx_push_rtp_buffers (gpointer user_data)
 406 {
 407   SimulCtx *ctx = user_data;
 408
 409   gst_harness_set_src_caps_str (ctx->rtp_h, "application/x-rtp");
 410   gst_harness_push (ctx->rtp_h, create_buffer (0, 1111));
 411   return NULL;
 412 }
 413
 414 static gpointer
 415 _simul_ctx_push_rtcp_buffers (gpointer user_data)
 416 {
 417   SimulCtx *ctx = user_data;
 418
 419   g_usleep (10);
 420   gst_harness_set_src_caps_str (ctx->rtcp_h, "application/x-rtcp");
 421   gst_harness_push (ctx->rtcp_h, generate_rtcp_sr_buffer (1111));
 422   return NULL;
 423 }
 424
 425 GST_START_TEST (test_rtp_and_rtcp_arrives_simultaneously)
 426 {
 427   guint r;
 428   guint repeats = 1000;
 429   if (RUNNING_ON_VALGRIND)
 430     repeats = 2;
 431
 432   for (r = 0; r < repeats; r++) {
 433     SimulCtx ctx;
 434     GThread *t0, *t1;
 435
 436     ctx.rtp_h = gst_harness_new_with_padnames ("rtpssrcdemux", "sink", NULL);
 437     ctx.rtcp_h =
 438         gst_harness_new_with_element (ctx.rtp_h->element, "rtcp_sink", NULL);
 439
 440     g_signal_connect (ctx.rtp_h->element,
 441         "new-ssrc-pad", (GCallback) _simul_ctx_new_ssrc_pad_cb, &ctx);
 442
 443     t0 = g_thread_new ("push rtp", _simul_ctx_push_rtp_buffers, &ctx);
 444     t1 = g_thread_new ("push rtcp", _simul_ctx_push_rtcp_buffers, &ctx);
 445
 446     g_thread_join (t0);
 447     g_thread_join (t1);
 448
 449     gst_harness_teardown (ctx.rtp_h);
 450     gst_harness_teardown (ctx.rtcp_h);
 451   }
 452 }
 453
 454 GST_END_TEST;
 455
 456 static Suite *
 457 rtpssrcdemux_suite (void)
 458 {
 459   Suite *s = suite_create ("rtpssrcdemux");
 460   TCase *tc_chain = tcase_create ("general");
 461
 462   suite_add_tcase (s, tc_chain);
 463   tcase_add_test (tc_chain, test_event_forwarding);
 464   tcase_add_test (tc_chain, test_oob_event_locking);
 465   tcase_add_test (tc_chain, test_rtpssrcdemux_max_streams);
 466   tcase_add_test (tc_chain, test_rtpssrcdemux_rtcp_app);
 467   tcase_add_test (tc_chain, test_rtpssrcdemux_invalid_rtp);
 468   tcase_add_test (tc_chain, test_rtpssrcdemux_invalid_rtcp);
 469   tcase_add_test (tc_chain, test_rtp_and_rtcp_arrives_simultaneously);
 470
 471   return s;
 472 }
 473
 474 GST_CHECK_MAIN (rtpssrcdemux);