subprojects/gst-rtsp-server/examples/test-replay-server.c

   1 /* GStreamer
   2  * Copyright (C) 2019 Mathieu Duponchelle <mathieu@centricular.com>
   3  * Copyright (C) 2020 Seungha Yang <seungha@centricular.com>
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  18  * Boston, MA 02110-1301, USA.
  19  */
  20
  21
  22 #include <gst/gst.h>
  23
  24 #include <gst/rtsp-server/rtsp-server.h>
  25
  26 #include "test-replay-server.h"
  27
  28 GST_DEBUG_CATEGORY_STATIC (replay_server_debug);
  29 #define GST_CAT_DEFAULT (replay_server_debug)
  30
  31 static GstStaticCaps raw_video_caps = GST_STATIC_CAPS ("video/x-raw");
  32 static GstStaticCaps raw_audio_caps = GST_STATIC_CAPS ("audio/x-raw");
  33
  34 static GList
  35     * gst_rtsp_media_factory_replay_get_demuxers (GstRTSPMediaFactoryReplay *
  36     factory);
  37 static GList
  38     * gst_rtsp_media_factory_replay_get_payloaders (GstRTSPMediaFactoryReplay *
  39     factory);
  40 static GList
  41     * gst_rtsp_media_factory_replay_get_decoders (GstRTSPMediaFactoryReplay *
  42     factory);
  43
  44 typedef struct
  45 {
  46   GstPad *srcpad;
  47   gulong block_id;
  48 } GstReplayBinPad;
  49
  50 static void
  51 gst_replay_bin_pad_unblock_and_free (GstReplayBinPad * pad)
  52 {
  53   if (pad->srcpad && pad->block_id) {
  54     GST_DEBUG_OBJECT (pad->srcpad, "Unblock");
  55     gst_pad_remove_probe (pad->srcpad, pad->block_id);
  56     pad->block_id = 0;
  57   }
  58
  59   gst_clear_object (&pad->srcpad);
  60   g_free (pad);
  61 }
  62
  63 /* NOTE: this bin implementation is almost completely taken from rtsp-media-factory-uri
  64  * but this example doesn't use the GstRTSPMediaFactoryURI object so that
  65  * we can handle events and messages ourselves.
  66  * Specifically,
  67  * - Handle segment-done message for looping given source
  68  * - Drop all incoming seek event because client seek is not implemented
  69  *   and do initial segment seeking on no-more-pads signal
  70  */
  71 struct _GstReplayBin
  72 {
  73   GstBin parent;
  74
  75   gint64 num_loops;
  76
  77   GstCaps *raw_vcaps;
  78   GstCaps *raw_acaps;
  79
  80   guint pt;
  81
  82   /* without ref */
  83   GstElement *uridecodebin;
  84   GstElement *inner_bin;
  85
  86   /* holds ref */
  87   GstRTSPMediaFactoryReplay *factory;
  88
  89   GMutex lock;
  90
  91   GList *srcpads;
  92 };
  93
  94 static void gst_replay_bin_dispose (GObject * object);
  95 static void gst_replay_bin_finalize (GObject * object);
  96 static void gst_replay_bin_handle_message (GstBin * bin, GstMessage * message);
  97
  98 static gboolean autoplug_continue_cb (GstElement * dbin, GstPad * pad,
  99     GstCaps * caps, GstReplayBin * self);
 100 static void pad_added_cb (GstElement * dbin, GstPad * pad, GstReplayBin * self);
 101 static void no_more_pads_cb (GstElement * uribin, GstReplayBin * self);
 102
 103 #define gst_replay_bin_parent_class bin_parent_class
 104 G_DEFINE_TYPE (GstReplayBin, gst_replay_bin, GST_TYPE_BIN);
 105
 106 static void
 107 gst_replay_bin_class_init (GstReplayBinClass * klass)
 108 {
 109   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 110   GstBinClass *bin_class = GST_BIN_CLASS (klass);
 111
 112   gobject_class->dispose = gst_replay_bin_dispose;
 113   gobject_class->finalize = gst_replay_bin_finalize;
 114
 115   bin_class->handle_message = GST_DEBUG_FUNCPTR (gst_replay_bin_handle_message);
 116 }
 117
 118 static void
 119 gst_replay_bin_init (GstReplayBin * self)
 120 {
 121   self->raw_vcaps = gst_static_caps_get (&raw_video_caps);
 122   self->raw_acaps = gst_static_caps_get (&raw_audio_caps);
 123
 124   self->uridecodebin = gst_element_factory_make ("uridecodebin", NULL);
 125   if (!self->uridecodebin) {
 126     GST_ERROR_OBJECT (self, "uridecodebin is unavailable");
 127     return;
 128   }
 129
 130   /* our bin will dynamically expose payloaded pads */
 131   self->inner_bin = gst_bin_new ("dynpay0");
 132   gst_bin_add (GST_BIN_CAST (self), self->inner_bin);
 133   gst_bin_add (GST_BIN_CAST (self->inner_bin), self->uridecodebin);
 134
 135   g_signal_connect (self->uridecodebin, "autoplug-continue",
 136       G_CALLBACK (autoplug_continue_cb), self);
 137   g_signal_connect (self->uridecodebin, "pad-added",
 138       G_CALLBACK (pad_added_cb), self);
 139   g_signal_connect (self->uridecodebin, "no-more-pads",
 140       G_CALLBACK (no_more_pads_cb), self);
 141
 142   self->pt = 96;
 143
 144   g_mutex_init (&self->lock);
 145 }
 146
 147 static void
 148 gst_replay_bin_dispose (GObject * object)
 149 {
 150   GstReplayBin *self = GST_REPLAY_BIN (object);
 151
 152   GST_DEBUG_OBJECT (self, "dispose");
 153
 154   gst_clear_caps (&self->raw_vcaps);
 155   gst_clear_caps (&self->raw_acaps);
 156   gst_clear_object (&self->factory);
 157
 158   if (self->srcpads) {
 159     g_list_free_full (self->srcpads,
 160         (GDestroyNotify) gst_replay_bin_pad_unblock_and_free);
 161     self->srcpads = NULL;
 162   }
 163
 164   G_OBJECT_CLASS (bin_parent_class)->dispose (object);
 165 }
 166
 167 static void
 168 gst_replay_bin_finalize (GObject * object)
 169 {
 170   GstReplayBin *self = GST_REPLAY_BIN (object);
 171
 172   g_mutex_clear (&self->lock);
 173
 174   G_OBJECT_CLASS (bin_parent_class)->finalize (object);
 175 }
 176
 177 static gboolean
 178 send_eos_foreach_srcpad (GstElement * element, GstPad * pad, gpointer user_data)
 179 {
 180   GST_DEBUG_OBJECT (pad, "Sending EOS to downstream");
 181   gst_pad_push_event (pad, gst_event_new_eos ());
 182
 183   return TRUE;
 184 }
 185
 186 static void
 187 gst_replay_bin_do_segment_seek (GstElement * element, GstReplayBin * self)
 188 {
 189   gboolean ret;
 190
 191   ret = gst_element_seek (element, 1.0, GST_FORMAT_TIME,
 192       GST_SEEK_FLAG_ACCURATE | GST_SEEK_FLAG_SEGMENT,
 193       GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_NONE, -1);
 194
 195   if (!ret) {
 196     GST_WARNING_OBJECT (self, "segment seeking failed");
 197     gst_element_foreach_src_pad (element,
 198         (GstElementForeachPadFunc) send_eos_foreach_srcpad, NULL);
 199   }
 200 }
 201
 202 static void
 203 gst_replay_bin_handle_message (GstBin * bin, GstMessage * message)
 204 {
 205   GstReplayBin *self = GST_REPLAY_BIN (bin);
 206
 207   if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_SEGMENT_DONE) {
 208     gboolean next_loop = TRUE;
 209
 210     GST_DEBUG_OBJECT (self, "Have segment done message");
 211
 212     g_mutex_lock (&self->lock);
 213     if (self->num_loops != -1) {
 214       self->num_loops--;
 215
 216       if (self->num_loops < 1)
 217         next_loop = FALSE;
 218     }
 219
 220     if (next_loop) {
 221       /* Send seek event from non-streaming thread */
 222       gst_element_call_async (GST_ELEMENT_CAST (self->uridecodebin),
 223           (GstElementCallAsyncFunc) gst_replay_bin_do_segment_seek, self, NULL);
 224     } else {
 225       gst_element_foreach_src_pad (GST_ELEMENT_CAST (self->uridecodebin),
 226           (GstElementForeachPadFunc) send_eos_foreach_srcpad, NULL);
 227     }
 228
 229     g_mutex_unlock (&self->lock);
 230   }
 231
 232   GST_BIN_CLASS (bin_parent_class)->handle_message (bin, message);
 233 }
 234
 235 static GstElementFactory *
 236 find_payloader (GstReplayBin * self, GstCaps * caps)
 237 {
 238   GList *list;
 239   GstElementFactory *factory = NULL;
 240   gboolean autoplug_more = FALSE;
 241   GList *demuxers = NULL;
 242   GList *payloaders = NULL;
 243
 244   demuxers = gst_rtsp_media_factory_replay_get_demuxers (self->factory);
 245
 246   /* first find a demuxer that can link */
 247   list = gst_element_factory_list_filter (demuxers, caps, GST_PAD_SINK, FALSE);
 248
 249   if (list) {
 250     GstStructure *structure = gst_caps_get_structure (caps, 0);
 251     gboolean parsed = FALSE;
 252     gint mpegversion = 0;
 253
 254     if (!gst_structure_get_boolean (structure, "parsed", &parsed) &&
 255         gst_structure_has_name (structure, "audio/mpeg") &&
 256         gst_structure_get_int (structure, "mpegversion", &mpegversion) &&
 257         (mpegversion == 2 || mpegversion == 4)) {
 258       /* for AAC it's framed=true instead of parsed=true */
 259       gst_structure_get_boolean (structure, "framed", &parsed);
 260     }
 261
 262     /* Avoid plugging parsers in a loop. This is not 100% correct, as some
 263      * parsers don't set parsed=true in caps. We should do something like
 264      * decodebin does and track decode chains and elements plugged in those
 265      * chains...
 266      */
 267     if (parsed) {
 268       GList *walk;
 269       const gchar *klass;
 270
 271       for (walk = list; walk; walk = walk->next) {
 272         factory = GST_ELEMENT_FACTORY (walk->data);
 273         klass = gst_element_factory_get_metadata (factory,
 274             GST_ELEMENT_METADATA_KLASS);
 275         if (strstr (klass, "Parser"))
 276           /* caps have parsed=true, so skip this parser to avoid loops */
 277           continue;
 278
 279         autoplug_more = TRUE;
 280         break;
 281       }
 282     } else {
 283       /* caps don't have parsed=true set and we have a demuxer/parser */
 284       autoplug_more = TRUE;
 285     }
 286
 287     gst_plugin_feature_list_free (list);
 288   }
 289
 290   if (autoplug_more)
 291     /* we have a demuxer, try that one first */
 292     return NULL;
 293
 294   payloaders = gst_rtsp_media_factory_replay_get_payloaders (self->factory);
 295
 296   /* no demuxer try a depayloader */
 297   list = gst_element_factory_list_filter (payloaders,
 298       caps, GST_PAD_SINK, FALSE);
 299
 300   if (list == NULL) {
 301     GList *decoders =
 302         gst_rtsp_media_factory_replay_get_decoders (self->factory);
 303     /* no depayloader, try a decoder, we'll get to a payloader for a decoded
 304      * video or audio format, worst case. */
 305     list = gst_element_factory_list_filter (decoders,
 306         caps, GST_PAD_SINK, FALSE);
 307
 308     if (list != NULL) {
 309       /* we have a decoder, try that one first */
 310       gst_plugin_feature_list_free (list);
 311       return NULL;
 312     }
 313   }
 314
 315   if (list != NULL) {
 316     factory = GST_ELEMENT_FACTORY_CAST (list->data);
 317     g_object_ref (factory);
 318     gst_plugin_feature_list_free (list);
 319   }
 320
 321   return factory;
 322 }
 323
 324 static gboolean
 325 autoplug_continue_cb (GstElement * dbin, GstPad * pad, GstCaps * caps,
 326     GstReplayBin * self)
 327 {
 328   GstElementFactory *factory;
 329
 330   GST_DEBUG_OBJECT (self, "found pad %s:%s of caps %" GST_PTR_FORMAT,
 331       GST_DEBUG_PAD_NAME (pad), caps);
 332
 333   if (!(factory = find_payloader (self, caps)))
 334     goto no_factory;
 335
 336   /* we found a payloader, stop autoplugging so we can plug the
 337    * payloader. */
 338   GST_DEBUG_OBJECT (self, "found factory %s",
 339       gst_plugin_feature_get_name (GST_PLUGIN_FEATURE (factory)));
 340   gst_object_unref (factory);
 341
 342   return FALSE;
 343
 344 no_factory:
 345   {
 346     /* no payloader, continue autoplugging */
 347     GST_DEBUG_OBJECT (self, "no payloader found for caps %" GST_PTR_FORMAT,
 348         caps);
 349     return TRUE;
 350   }
 351 }
 352
 353 static GstPadProbeReturn
 354 replay_bin_sink_probe (GstPad * pad, GstPadProbeInfo * info,
 355     GstReplayBin * self)
 356 {
 357   GstPadProbeReturn ret = GST_PAD_PROBE_OK;
 358
 359   if (GST_IS_EVENT (GST_PAD_PROBE_INFO_DATA (info))) {
 360     GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);
 361
 362     switch (GST_EVENT_TYPE (event)) {
 363       case GST_EVENT_SEEK:
 364         /* Ideally this shouldn't happen because we are responding
 365          * seeking query with non-seekable */
 366         GST_DEBUG_OBJECT (pad, "Drop seek event");
 367         ret = GST_PAD_PROBE_DROP;
 368         break;
 369       default:
 370         break;
 371     }
 372   } else if (GST_IS_QUERY (GST_PAD_PROBE_INFO_DATA (info))) {
 373     GstQuery *query = GST_PAD_PROBE_INFO_QUERY (info);
 374
 375     switch (GST_QUERY_TYPE (query)) {
 376       case GST_QUERY_SEEKING:
 377       {
 378         /* FIXME: client seek is not implemented */
 379         gst_query_set_seeking (query, GST_FORMAT_TIME, FALSE, 0,
 380             GST_CLOCK_TIME_NONE);
 381         ret = GST_PAD_PROBE_HANDLED;
 382         break;
 383       }
 384       case GST_QUERY_SEGMENT:
 385         /* client seeking is not considered in here */
 386         gst_query_set_segment (query,
 387             1.0, GST_FORMAT_TIME, 0, GST_CLOCK_TIME_NONE);
 388         ret = GST_PAD_PROBE_HANDLED;
 389         break;
 390       default:
 391         break;
 392     }
 393   }
 394
 395   return ret;
 396 }
 397
 398 static GstPadProbeReturn
 399 replay_bin_src_block (GstPad * pad, GstPadProbeInfo * info, GstReplayBin * self)
 400 {
 401   GST_DEBUG_OBJECT (pad, "Block pad");
 402
 403   return GST_PAD_PROBE_OK;
 404 }
 405
 406 static void
 407 pad_added_cb (GstElement * dbin, GstPad * pad, GstReplayBin * self)
 408 {
 409   GstElementFactory *factory;
 410   GstElement *payloader;
 411   GstCaps *caps;
 412   GstPad *sinkpad, *srcpad, *ghostpad;
 413   GstPad *dpad = pad;
 414   GstElement *convert;
 415   gchar *padname, *payloader_name;
 416   GstElement *inner_bin = self->inner_bin;
 417   GstReplayBinPad *bin_pad;
 418
 419   GST_DEBUG_OBJECT (self, "added pad %s:%s", GST_DEBUG_PAD_NAME (pad));
 420
 421   /* ref to make refcounting easier later */
 422   gst_object_ref (pad);
 423   padname = gst_pad_get_name (pad);
 424
 425   /* get pad caps first, then call get_caps, then fail */
 426   if ((caps = gst_pad_get_current_caps (pad)) == NULL)
 427     if ((caps = gst_pad_query_caps (pad, NULL)) == NULL)
 428       goto no_caps;
 429
 430   /* check for raw caps */
 431   if (gst_caps_can_intersect (caps, self->raw_vcaps)) {
 432     /* we have raw video caps, insert converter */
 433     convert = gst_element_factory_make ("videoconvert", NULL);
 434   } else if (gst_caps_can_intersect (caps, self->raw_acaps)) {
 435     /* we have raw audio caps, insert converter */
 436     convert = gst_element_factory_make ("audioconvert", NULL);
 437   } else {
 438     convert = NULL;
 439   }
 440
 441   if (convert) {
 442     gst_bin_add (GST_BIN_CAST (inner_bin), convert);
 443     gst_element_sync_state_with_parent (convert);
 444
 445     sinkpad = gst_element_get_static_pad (convert, "sink");
 446     gst_pad_link (pad, sinkpad);
 447     gst_object_unref (sinkpad);
 448
 449     /* unref old pad, we reffed before */
 450     gst_object_unref (pad);
 451
 452     /* continue with new pad and caps */
 453     pad = gst_element_get_static_pad (convert, "src");
 454     if ((caps = gst_pad_get_current_caps (pad)) == NULL)
 455       if ((caps = gst_pad_query_caps (pad, NULL)) == NULL)
 456         goto no_caps;
 457   }
 458
 459   if (!(factory = find_payloader (self, caps)))
 460     goto no_factory;
 461
 462   gst_caps_unref (caps);
 463
 464   /* we have a payloader now */
 465   GST_DEBUG_OBJECT (self, "found payloader factory %s",
 466       gst_plugin_feature_get_name (GST_PLUGIN_FEATURE (factory)));
 467
 468   payloader_name = g_strdup_printf ("pay_%s", padname);
 469   payloader = gst_element_factory_create (factory, payloader_name);
 470   g_free (payloader_name);
 471   if (payloader == NULL)
 472     goto no_payloader;
 473
 474   g_object_set (payloader, "pt", self->pt, NULL);
 475   self->pt++;
 476
 477   if (g_object_class_find_property (G_OBJECT_GET_CLASS (payloader),
 478           "buffer-list"))
 479     g_object_set (payloader, "buffer-list", TRUE, NULL);
 480
 481   /* add the payloader to the pipeline */
 482   gst_bin_add (GST_BIN_CAST (inner_bin), payloader);
 483   gst_element_sync_state_with_parent (payloader);
 484
 485   /* link the pad to the sinkpad of the payloader */
 486   sinkpad = gst_element_get_static_pad (payloader, "sink");
 487   gst_pad_link (pad, sinkpad);
 488   gst_object_unref (pad);
 489
 490   /* Add pad probe to handle events */
 491   gst_pad_add_probe (sinkpad,
 492       GST_PAD_PROBE_TYPE_EVENT_UPSTREAM | GST_PAD_PROBE_TYPE_QUERY_UPSTREAM,
 493       (GstPadProbeCallback) replay_bin_sink_probe, self, NULL);
 494   gst_object_unref (sinkpad);
 495
 496   /* block data for initial segment seeking */
 497   bin_pad = g_new0 (GstReplayBinPad, 1);
 498
 499   /* Move ownership of pad to this struct */
 500   bin_pad->srcpad = gst_object_ref (dpad);
 501   bin_pad->block_id =
 502       gst_pad_add_probe (dpad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM,
 503       (GstPadProbeCallback) replay_bin_src_block, self, NULL);
 504   g_mutex_lock (&self->lock);
 505   self->srcpads = g_list_append (self->srcpads, bin_pad);
 506   g_mutex_unlock (&self->lock);
 507
 508   /* now expose the srcpad of the payloader as a ghostpad with the same name
 509    * as the uridecodebin pad name. */
 510   srcpad = gst_element_get_static_pad (payloader, "src");
 511   ghostpad = gst_ghost_pad_new (padname, srcpad);
 512   gst_object_unref (srcpad);
 513   g_free (padname);
 514
 515   gst_pad_set_active (ghostpad, TRUE);
 516   gst_element_add_pad (inner_bin, ghostpad);
 517
 518   return;
 519
 520   /* ERRORS */
 521 no_caps:
 522   {
 523     GST_WARNING ("could not get caps from pad");
 524     g_free (padname);
 525     gst_object_unref (pad);
 526     return;
 527   }
 528 no_factory:
 529   {
 530     GST_DEBUG ("no payloader found");
 531     g_free (padname);
 532     gst_caps_unref (caps);
 533     gst_object_unref (pad);
 534     return;
 535   }
 536 no_payloader:
 537   {
 538     GST_ERROR ("could not create payloader from factory");
 539     g_free (padname);
 540     gst_caps_unref (caps);
 541     gst_object_unref (pad);
 542     return;
 543   }
 544 }
 545
 546 static void
 547 gst_replay_bin_do_initial_segment_seek (GstElement * element,
 548     GstReplayBin * self)
 549 {
 550   gboolean ret;
 551   GstQuery *query;
 552   gboolean seekable;
 553
 554   query = gst_query_new_seeking (GST_FORMAT_TIME);
 555   ret = gst_element_query (element, query);
 556
 557   if (!ret) {
 558     GST_WARNING_OBJECT (self, "Cannot query seeking");
 559     gst_query_unref (query);
 560     goto done;
 561   }
 562
 563   gst_query_parse_seeking (query, NULL, &seekable, NULL, NULL);
 564   gst_query_unref (query);
 565
 566   if (!seekable) {
 567     GST_WARNING_OBJECT (self, "Source is not seekable");
 568     ret = FALSE;
 569     goto done;
 570   }
 571
 572   ret = gst_element_seek (element, 1.0, GST_FORMAT_TIME,
 573       GST_SEEK_FLAG_ACCURATE | GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_FLUSH,
 574       GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_NONE, -1);
 575
 576   if (!ret)
 577     GST_WARNING_OBJECT (self, "segment seeking failed");
 578
 579 done:
 580   /* Unblock all pads then */
 581   g_mutex_lock (&self->lock);
 582   if (self->srcpads) {
 583     g_list_free_full (self->srcpads,
 584         (GDestroyNotify) gst_replay_bin_pad_unblock_and_free);
 585     self->srcpads = NULL;
 586   }
 587   g_mutex_unlock (&self->lock);
 588
 589   if (!ret) {
 590     GST_WARNING_OBJECT (self, "Sending eos to all pads");
 591     gst_element_foreach_src_pad (element,
 592         (GstElementForeachPadFunc) send_eos_foreach_srcpad, NULL);
 593   }
 594 }
 595
 596 static void
 597 no_more_pads_cb (GstElement * uribin, GstReplayBin * self)
 598 {
 599   GST_DEBUG_OBJECT (self, "no-more-pads");
 600   gst_element_no_more_pads (GST_ELEMENT_CAST (self->inner_bin));
 601
 602   /* Flush seeking from streaming thread might not be good idea.
 603    * Do this from another (non-streaming) thread */
 604   gst_element_call_async (GST_ELEMENT_CAST (self->uridecodebin),
 605       (GstElementCallAsyncFunc) gst_replay_bin_do_initial_segment_seek,
 606       self, NULL);
 607 }
 608
 609 static GstElement *
 610 gst_replay_bin_new (const gchar * uri, gint64 num_loops,
 611     GstRTSPMediaFactoryReplay * factory, const gchar * name)
 612 {
 613   GstReplayBin *self;
 614
 615   g_return_val_if_fail (uri != NULL, NULL);
 616   g_return_val_if_fail (GST_IS_RTSP_MEDIA_FACTORY (factory), NULL);
 617
 618   if (!name)
 619     name = "GstRelayBin";
 620
 621   self = GST_REPLAY_BIN (g_object_new (GST_TYPE_REPLAY_BIN,
 622           "name", name, NULL));
 623
 624   if (!self->uridecodebin) {
 625     gst_object_unref (self);
 626     return NULL;
 627   }
 628
 629   g_object_set (self->uridecodebin, "uri", uri, NULL);
 630   self->factory = g_object_ref (factory);
 631   self->num_loops = num_loops;
 632
 633   return GST_ELEMENT_CAST (self);
 634 }
 635
 636 struct _GstRTSPMediaFactoryReplay
 637 {
 638   GstRTSPMediaFactory parent;
 639
 640   gchar *uri;
 641
 642   GList *demuxers;
 643   GList *payloaders;
 644   GList *decoders;
 645
 646   gint64 num_loops;
 647 };
 648
 649 enum
 650 {
 651   PROP_0,
 652   PROP_URI,
 653   PROP_NUM_LOOPS,
 654 };
 655
 656 #define DEFAULT_NUM_LOOPS (-1)
 657
 658 static void gst_rtsp_media_factory_replay_get_property (GObject * object,
 659     guint propid, GValue * value, GParamSpec * pspec);
 660 static void gst_rtsp_media_factory_replay_set_property (GObject * object,
 661     guint propid, const GValue * value, GParamSpec * pspec);
 662 static void gst_rtsp_media_factory_replay_finalize (GObject * object);
 663
 664 static GstElement
 665     * gst_rtsp_media_factory_replay_create_element (GstRTSPMediaFactory *
 666     factory, const GstRTSPUrl * url);
 667
 668 typedef struct
 669 {
 670   GList *demux;
 671   GList *payload;
 672   GList *decode;
 673 } FilterData;
 674
 675 static gboolean
 676 payloader_filter (GstPluginFeature * feature, FilterData * self);
 677
 678 #define gst_rtsp_media_factory_replay_parent_class parent_class
 679 G_DEFINE_TYPE (GstRTSPMediaFactoryReplay,
 680     gst_rtsp_media_factory_replay, GST_TYPE_RTSP_MEDIA_FACTORY);
 681
 682 static void
 683 gst_rtsp_media_factory_replay_class_init (GstRTSPMediaFactoryReplayClass
 684     * klass)
 685 {
 686   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 687   GstRTSPMediaFactoryClass *mf_class = GST_RTSP_MEDIA_FACTORY_CLASS (klass);
 688
 689   gobject_class->get_property = gst_rtsp_media_factory_replay_get_property;
 690   gobject_class->set_property = gst_rtsp_media_factory_replay_set_property;
 691   gobject_class->finalize = gst_rtsp_media_factory_replay_finalize;
 692
 693   g_object_class_install_property (gobject_class, PROP_URI,
 694       g_param_spec_string ("uri", "URI",
 695           "The URI of the resource to stream", NULL,
 696           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 697
 698   g_object_class_install_property (gobject_class, PROP_NUM_LOOPS,
 699       g_param_spec_int64 ("num-loops", "Num Loops",
 700           "The number of loops (-1 = infinite)", -1, G_MAXINT64,
 701           DEFAULT_NUM_LOOPS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 702
 703   mf_class->create_element =
 704       GST_DEBUG_FUNCPTR (gst_rtsp_media_factory_replay_create_element);
 705 }
 706
 707 static void
 708 gst_rtsp_media_factory_replay_init (GstRTSPMediaFactoryReplay * self)
 709 {
 710   FilterData data = { NULL, };
 711
 712   /* get the feature list using the filter */
 713   gst_registry_feature_filter (gst_registry_get (), (GstPluginFeatureFilter)
 714       payloader_filter, FALSE, &data);
 715
 716   /* sort */
 717   self->demuxers =
 718       g_list_sort (data.demux, gst_plugin_feature_rank_compare_func);
 719   self->payloaders =
 720       g_list_sort (data.payload, gst_plugin_feature_rank_compare_func);
 721   self->decoders =
 722       g_list_sort (data.decode, gst_plugin_feature_rank_compare_func);
 723
 724   self->num_loops = DEFAULT_NUM_LOOPS;
 725 }
 726
 727 static void
 728 gst_rtsp_media_factory_replay_get_property (GObject * object, guint propid,
 729     GValue * value, GParamSpec * pspec)
 730 {
 731   GstRTSPMediaFactoryReplay *self = GST_RTSP_MEDIA_FACTORY_REPLAY (object);
 732
 733   switch (propid) {
 734     case PROP_URI:
 735       g_value_take_string (value, self->uri);
 736       break;
 737     case PROP_NUM_LOOPS:
 738       g_value_set_int64 (value, self->num_loops);
 739       break;
 740     default:
 741       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 742   }
 743 }
 744
 745 static void
 746 gst_rtsp_media_factory_replay_set_property (GObject * object, guint propid,
 747     const GValue * value, GParamSpec * pspec)
 748 {
 749   GstRTSPMediaFactoryReplay *self = GST_RTSP_MEDIA_FACTORY_REPLAY (object);
 750
 751   switch (propid) {
 752     case PROP_URI:
 753       g_free (self->uri);
 754       self->uri = g_value_dup_string (value);
 755       break;
 756     case PROP_NUM_LOOPS:
 757       self->num_loops = g_value_get_int64 (value);
 758       break;
 759     default:
 760       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 761   }
 762 }
 763
 764 static void
 765 gst_rtsp_media_factory_replay_finalize (GObject * object)
 766 {
 767   GstRTSPMediaFactoryReplay *self = GST_RTSP_MEDIA_FACTORY_REPLAY (object);
 768
 769   g_free (self->uri);
 770
 771   gst_plugin_feature_list_free (self->demuxers);
 772   gst_plugin_feature_list_free (self->payloaders);
 773   gst_plugin_feature_list_free (self->decoders);
 774
 775   G_OBJECT_CLASS (parent_class)->finalize (object);
 776 }
 777
 778 static GstElement *
 779 gst_rtsp_media_factory_replay_create_element (GstRTSPMediaFactory * factory,
 780     const GstRTSPUrl * url)
 781 {
 782   GstRTSPMediaFactoryReplay *self = GST_RTSP_MEDIA_FACTORY_REPLAY (factory);
 783
 784   return gst_replay_bin_new (self->uri, self->num_loops, self,
 785       "GstRTSPMediaFactoryReplay");
 786 }
 787
 788 static gboolean
 789 payloader_filter (GstPluginFeature * feature, FilterData * data)
 790 {
 791   const gchar *klass;
 792   GstElementFactory *fact;
 793   GList **list = NULL;
 794
 795   /* we only care about element factories */
 796   if (G_UNLIKELY (!GST_IS_ELEMENT_FACTORY (feature)))
 797     return FALSE;
 798
 799   if (gst_plugin_feature_get_rank (feature) < GST_RANK_MARGINAL)
 800     return FALSE;
 801
 802   fact = GST_ELEMENT_FACTORY_CAST (feature);
 803
 804   klass = gst_element_factory_get_metadata (fact, GST_ELEMENT_METADATA_KLASS);
 805
 806   if (strstr (klass, "Decoder"))
 807     list = &data->decode;
 808   else if (strstr (klass, "Demux"))
 809     list = &data->demux;
 810   else if (strstr (klass, "Parser") && strstr (klass, "Codec"))
 811     list = &data->demux;
 812   else if (strstr (klass, "Payloader") && strstr (klass, "RTP"))
 813     list = &data->payload;
 814
 815   if (list) {
 816     GST_LOG ("adding %s", GST_OBJECT_NAME (fact));
 817     *list = g_list_prepend (*list, gst_object_ref (fact));
 818   }
 819
 820   return FALSE;
 821 }
 822
 823 static GList *
 824 gst_rtsp_media_factory_replay_get_demuxers (GstRTSPMediaFactoryReplay * factory)
 825 {
 826   return factory->demuxers;
 827 }
 828
 829 static GList *
 830 gst_rtsp_media_factory_replay_get_payloaders (GstRTSPMediaFactoryReplay *
 831     factory)
 832 {
 833   return factory->payloaders;
 834 }
 835
 836 static GList *
 837 gst_rtsp_media_factory_replay_get_decoders (GstRTSPMediaFactoryReplay * factory)
 838 {
 839   return factory->decoders;
 840 }
 841
 842 static GstRTSPMediaFactory *
 843 gst_rtsp_media_factory_replay_new (const gchar * uri, gint64 num_loops)
 844 {
 845   GstRTSPMediaFactory *factory;
 846
 847   factory =
 848       GST_RTSP_MEDIA_FACTORY (g_object_new
 849       (GST_TYPE_RTSP_MEDIA_FACTORY_REPLAY, "uri", uri, "num-loops", num_loops,
 850           NULL));
 851
 852   return factory;
 853 }
 854
 855 int
 856 main (int argc, char *argv[])
 857 {
 858   GMainLoop *loop;
 859   GstRTSPServer *server;
 860   GstRTSPMountPoints *mounts;
 861   GstRTSPMediaFactory *factory;
 862   GOptionContext *optctx;
 863   GError *error = NULL;
 864   gchar *service;
 865   gchar *uri = NULL;
 866   gint64 num_loops = -1;
 867   GOptionEntry options[] = {
 868     {"num-loops", 0, 0, G_OPTION_ARG_INT64, &num_loops,
 869         "The number of loops (default = -1, infinite)", NULL},
 870     {NULL}
 871   };
 872
 873   optctx = g_option_context_new ("RTSP Replay Server");
 874   g_option_context_add_main_entries (optctx, options, NULL);
 875   g_option_context_add_group (optctx, gst_init_get_option_group ());
 876   if (!g_option_context_parse (optctx, &argc, &argv, &error)) {
 877     g_printerr ("Error parsing options: %s\n", error->message);
 878     g_option_context_free (optctx);
 879     g_clear_error (&error);
 880     return -1;
 881   }
 882   if (argc < 2) {
 883     g_print ("%s\n", g_option_context_get_help (optctx, TRUE, NULL));
 884     return 1;
 885   }
 886
 887   g_option_context_free (optctx);
 888
 889   /* check if URI is valid, otherwise convert filename to URI if it's a file */
 890   if (gst_uri_is_valid (argv[1])) {
 891     uri = g_strdup (argv[1]);
 892   } else if (g_file_test (argv[1], G_FILE_TEST_EXISTS)) {
 893     uri = gst_filename_to_uri (argv[1], NULL);
 894   } else {
 895     g_printerr ("Unrecognised command line argument '%s'.\n"
 896         "Please pass an URI or file as argument!\n", argv[1]);
 897     return -1;
 898   }
 899
 900   if (num_loops < -1 || num_loops == 0) {
 901     g_printerr ("num-loop should be non-zero or -1");
 902     return -1;
 903   }
 904
 905   GST_DEBUG_CATEGORY_INIT (replay_server_debug, "replay-server", 0,
 906       "RTSP replay server");
 907
 908   if (num_loops != -1)
 909     g_print ("Run loop %" G_GINT64_FORMAT " times\n", num_loops);
 910
 911   loop = g_main_loop_new (NULL, FALSE);
 912
 913   server = gst_rtsp_server_new ();
 914
 915   mounts = gst_rtsp_server_get_mount_points (server);
 916   factory = gst_rtsp_media_factory_replay_new (uri, num_loops);
 917   g_free (uri);
 918
 919   gst_rtsp_mount_points_add_factory (mounts, "/test", factory);
 920
 921   g_object_unref (mounts);
 922
 923   gst_rtsp_server_attach (server, NULL);
 924
 925   service = gst_rtsp_server_get_service (server);
 926   g_print ("stream ready at rtsp://127.0.0.1:%s/test\n", service);
 927   g_free (service);
 928   g_main_loop_run (loop);
 929
 930   return 0;
 931 }