1 /* GStreamer
2 * Copyright (C) <2015> Jan Schmidt <jan@centricular.com>
3 * Copyright (C) <2007> Wim Taymans <wim.taymans@gmail.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 * SECTION:element-urisourcebin
23 * @title: urisourcebin
24 *
25 * urisourcebin is an element for accessing URIs in a uniform manner.
26 *
27 * It handles selecting a URI source element and potentially download
28 * buffering for network sources. It produces one or more source pads,
29 * depending on the input source, for feeding to decoding chains or decodebin.
30 *
31 * The main configuration is via the #GstURISourceBin:uri property.
32 *
33 * <emphasis>urisourcebin is still experimental API and a technology preview.
34 * Its behaviour and exposed API is subject to change.</emphasis>
35 */
36
37 /* FIXME 0.11: suppress warnings for deprecated API such as GValueArray
38 * with newer GLib versions (>= 2.31.0) */
39 #define GLIB_DISABLE_DEPRECATION_WARNINGS
40
41 #ifdef HAVE_CONFIG_H
42 # include "config.h"
43 #endif
44
45 #include <string.h>
46
47 #include <gst/gst.h>
48 #include <gst/gst-i18n-plugin.h>
49 #include <gst/pbutils/missing-plugins.h>
50
51 #include "gstplay-enum.h"
52 #include "gstrawcaps.h"
53 #include "gstplayback.h"
54 #include "gstplaybackutils.h"
55
56 #define GST_TYPE_URI_SOURCE_BIN \
57 (gst_uri_source_bin_get_type())
58 #define GST_URI_SOURCE_BIN(obj) \
59 (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_URI_SOURCE_BIN,GstURISourceBin))
60 #define GST_URI_SOURCE_BIN_CLASS(klass) \
61 (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_URI_SOURCE_BIN,GstURISourceBinClass))
62 #define GST_IS_URI_SOURCE_BIN(obj) \
63 (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_URI_SOURCE_BIN))
64 #define GST_IS_URI_SOURCE_BIN_CLASS(klass) \
65 (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_URI_SOURCE_BIN))
66 #define GST_URI_SOURCE_BIN_CAST(obj) ((GstURISourceBin *) (obj))
67
68 typedef struct _GstURISourceBin GstURISourceBin;
69 typedef struct _GstURISourceBinClass GstURISourceBinClass;
70 typedef struct _ChildSrcPadInfo ChildSrcPadInfo;
71 typedef struct _OutputSlotInfo OutputSlotInfo;
72
73 #define GST_URI_SOURCE_BIN_LOCK(urisrc) (g_mutex_lock(&((GstURISourceBin*)(urisrc))->lock))
74 #define GST_URI_SOURCE_BIN_UNLOCK(urisrc) (g_mutex_unlock(&((GstURISourceBin*)(urisrc))->lock))
75
76 #define BUFFERING_LOCK(ubin) G_STMT_START { \
77 GST_LOG_OBJECT (ubin, \
78 "buffering locking from thread %p", \
79 g_thread_self ()); \
80 g_mutex_lock (&GST_URI_SOURCE_BIN_CAST(ubin)->buffering_lock); \
81 GST_LOG_OBJECT (ubin, \
82 "buffering lock from thread %p", \
83 g_thread_self ()); \
84 } G_STMT_END
85
86 #define BUFFERING_UNLOCK(ubin) G_STMT_START { \
87 GST_LOG_OBJECT (ubin, \
88 "buffering unlocking from thread %p", \
89 g_thread_self ()); \
90 g_mutex_unlock (&GST_URI_SOURCE_BIN_CAST(ubin)->buffering_lock); \
91 } G_STMT_END
92
93 /* Track a source pad from a child that
94 * is linked or needs linking to an output
95 * slot */
96 struct _ChildSrcPadInfo
97 {
98 guint blocking_probe_id;
99 guint event_probe_id;
100 GstPad *demux_src_pad;
101 GstCaps *cur_caps; /* holds ref */
102
103 /* Configured output slot, if any */
104 OutputSlotInfo *output_slot;
105 };
106
107 struct _OutputSlotInfo
108 {
109 ChildSrcPadInfo *linked_info; /* demux source pad info feeding this slot, if any */
110 GstElement *queue; /* queue2 or downloadbuffer */
111 GstPad *sinkpad; /* Sink pad of the queue eleemnt */
112 GstPad *srcpad; /* Output ghost pad */
113 gboolean is_eos; /* Did EOS get fed into the buffering element */
114
115 gulong bitrate_changed_id; /* queue bitrate changed notification */
116 };
117
118 /**
119 * GstURISourceBin
120 *
121 * urisourcebin element struct
122 */
123 struct _GstURISourceBin
124 {
125 GstBin parent_instance;
126
127 GMutex lock; /* lock for constructing */
128
129 GMutex factories_lock;
130 guint32 factories_cookie;
131 GList *factories; /* factories we can use for selecting elements */
132
133 gchar *uri;
134 guint64 connection_speed;
135
136 gboolean is_stream;
137 gboolean is_adaptive;
138 gboolean need_queue;
139 guint64 buffer_duration; /* When buffering, buffer duration (ns) */
140 guint buffer_size; /* When buffering, buffer size (bytes) */
141 gboolean download;
142 gboolean use_buffering;
143 gdouble low_watermark;
144 gdouble high_watermark;
145
146 GstElement *source;
147 GList *typefinds; /* list of typefind element */
148
149 GstElement *demuxer; /* Adaptive demuxer if any */
150 GSList *out_slots;
151
152 guint numpads;
153
154 /* for dynamic sources */
155 guint src_np_sig_id; /* new-pad signal id */
156
157 guint64 ring_buffer_max_size; /* 0 means disabled */
158
159 GList *pending_pads; /* Pads we have blocked pending assignment
160 to an output source pad */
161 GList *inactive_output_pads; /* output pads that were unghosted */
162
163 GList *buffering_status; /* element currently buffering messages */
164 gint last_buffering_pct; /* Avoid sending buffering over and over */
165 GMutex buffering_lock;
166 GMutex buffering_post_lock;
167 };
168
169 struct _GstURISourceBinClass
170 {
171 GstBinClass parent_class;
172
173 /* emitted when all data has been drained out
174 * FIXME : What do we need this for ?? */
175 void (*drained) (GstElement * element);
176 /* emitted when all data has been fed into buffering slots (i.e the
177 * actual sources are done) */
178 void (*about_to_finish) (GstElement * element);
179 };
180
181 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src_%u",
182 GST_PAD_SRC,
183 GST_PAD_SOMETIMES,
184 GST_STATIC_CAPS_ANY);
185
186 static GstStaticCaps default_raw_caps = GST_STATIC_CAPS (DEFAULT_RAW_CAPS);
187
188 GST_DEBUG_CATEGORY_STATIC (gst_uri_source_bin_debug);
189 #define GST_CAT_DEFAULT gst_uri_source_bin_debug
190
191 /* signals */
192 enum
193 {
194 SIGNAL_DRAINED,
195 SIGNAL_ABOUT_TO_FINISH,
196 SIGNAL_SOURCE_SETUP,
197 LAST_SIGNAL
198 };
199
200 /* properties */
201 #define DEFAULT_PROP_URI NULL
202 #define DEFAULT_PROP_SOURCE NULL
203 #define DEFAULT_CONNECTION_SPEED 0
204 #define DEFAULT_BUFFER_DURATION -1
205 #define DEFAULT_BUFFER_SIZE -1
206 #define DEFAULT_DOWNLOAD FALSE
207 #define DEFAULT_USE_BUFFERING TRUE
208 #define DEFAULT_RING_BUFFER_MAX_SIZE 0
209 #define DEFAULT_LOW_WATERMARK 0.01
210 #define DEFAULT_HIGH_WATERMARK 0.99
211
212 #define ACTUAL_DEFAULT_BUFFER_SIZE 10 * 1024 * 1024 /* The value used for byte limits when buffer-size == -1 */
213 #define ACTUAL_DEFAULT_BUFFER_DURATION 5 * GST_SECOND /* The value used for time limits when buffer-duration == -1 */
214
215 #define GET_BUFFER_SIZE(u) ((u)->buffer_size == -1 ? ACTUAL_DEFAULT_BUFFER_SIZE : (u)->buffer_size)
216 #define GET_BUFFER_DURATION(u) ((u)->buffer_duration == -1 ? ACTUAL_DEFAULT_BUFFER_DURATION : (u)->buffer_duration)
217
218 #define DEFAULT_CAPS (gst_static_caps_get (&default_raw_caps))
219 enum
220 {
221 PROP_0,
222 PROP_URI,
223 PROP_SOURCE,
224 PROP_CONNECTION_SPEED,
225 PROP_BUFFER_SIZE,
226 PROP_BUFFER_DURATION,
227 PROP_DOWNLOAD,
228 PROP_USE_BUFFERING,
229 PROP_RING_BUFFER_MAX_SIZE,
230 PROP_LOW_WATERMARK,
231 PROP_HIGH_WATERMARK,
232 PROP_STATISTICS,
233 };
234
235 #define CUSTOM_EOS_QUARK _custom_eos_quark_get ()
236 #define CUSTOM_EOS_QUARK_DATA "custom-eos"
237 static GQuark
_custom_eos_quark_get(void)238 _custom_eos_quark_get (void)
239 {
240 static gsize g_quark;
241
242 if (g_once_init_enter (&g_quark)) {
243 gsize quark =
244 (gsize) g_quark_from_static_string ("urisourcebin-custom-eos");
245 g_once_init_leave (&g_quark, quark);
246 }
247 return g_quark;
248 }
249
250 static void post_missing_plugin_error (GstElement * urisrc,
251 const gchar * element_name);
252
253 static guint gst_uri_source_bin_signals[LAST_SIGNAL] = { 0 };
254
255 GType gst_uri_source_bin_get_type (void);
256 #define gst_uri_source_bin_parent_class parent_class
257 G_DEFINE_TYPE (GstURISourceBin, gst_uri_source_bin, GST_TYPE_BIN);
258
259 static void gst_uri_source_bin_set_property (GObject * object, guint prop_id,
260 const GValue * value, GParamSpec * pspec);
261 static void gst_uri_source_bin_get_property (GObject * object, guint prop_id,
262 GValue * value, GParamSpec * pspec);
263 static void gst_uri_source_bin_finalize (GObject * obj);
264
265 static void handle_message (GstBin * bin, GstMessage * msg);
266
267 static gboolean gst_uri_source_bin_query (GstElement * element,
268 GstQuery * query);
269 static GstStateChangeReturn gst_uri_source_bin_change_state (GstElement *
270 element, GstStateChange transition);
271
272 static void remove_demuxer (GstURISourceBin * bin);
273 static void expose_output_pad (GstURISourceBin * urisrc, GstPad * pad);
274 static OutputSlotInfo *get_output_slot (GstURISourceBin * urisrc,
275 gboolean do_download, gboolean is_adaptive, GstCaps * caps);
276 static void free_output_slot (OutputSlotInfo * slot, GstURISourceBin * urisrc);
277 static void free_output_slot_async (GstURISourceBin * urisrc,
278 OutputSlotInfo * slot);
279 static GstPad *create_output_pad (GstURISourceBin * urisrc, GstPad * pad);
280 static void remove_buffering_msgs (GstURISourceBin * bin, GstObject * src);
281
282 static void update_queue_values (GstURISourceBin * urisrc);
283 static GstStructure *get_queue_statistics (GstURISourceBin * urisrc);
284
285 static void
gst_uri_source_bin_class_init(GstURISourceBinClass * klass)286 gst_uri_source_bin_class_init (GstURISourceBinClass * klass)
287 {
288 GObjectClass *gobject_class;
289 GstElementClass *gstelement_class;
290 GstBinClass *gstbin_class;
291
292 gobject_class = G_OBJECT_CLASS (klass);
293 gstelement_class = GST_ELEMENT_CLASS (klass);
294 gstbin_class = GST_BIN_CLASS (klass);
295
296 gobject_class->set_property = gst_uri_source_bin_set_property;
297 gobject_class->get_property = gst_uri_source_bin_get_property;
298 gobject_class->finalize = gst_uri_source_bin_finalize;
299
300 g_object_class_install_property (gobject_class, PROP_URI,
301 g_param_spec_string ("uri", "URI", "URI to decode",
302 DEFAULT_PROP_URI, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
303
304 g_object_class_install_property (gobject_class, PROP_SOURCE,
305 g_param_spec_object ("source", "Source", "Source object used",
306 GST_TYPE_ELEMENT, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
307
308 g_object_class_install_property (gobject_class, PROP_CONNECTION_SPEED,
309 g_param_spec_uint64 ("connection-speed", "Connection Speed",
310 "Network connection speed in kbps (0 = unknown)",
311 0, G_MAXUINT64 / 1000, DEFAULT_CONNECTION_SPEED,
312 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
313
314 g_object_class_install_property (gobject_class, PROP_BUFFER_SIZE,
315 g_param_spec_int ("buffer-size", "Buffer size (bytes)",
316 "Buffer size when buffering streams (-1 default value)",
317 -1, G_MAXINT, DEFAULT_BUFFER_SIZE,
318 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
319 g_object_class_install_property (gobject_class, PROP_BUFFER_DURATION,
320 g_param_spec_int64 ("buffer-duration", "Buffer duration (ns)",
321 "Buffer duration when buffering streams (-1 default value)",
322 -1, G_MAXINT64, DEFAULT_BUFFER_DURATION,
323 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
324
325 /**
326 * GstURISourceBin::download:
327 *
328 * For certain media type, enable download buffering.
329 */
330 g_object_class_install_property (gobject_class, PROP_DOWNLOAD,
331 g_param_spec_boolean ("download", "Download",
332 "Attempt download buffering when buffering network streams",
333 DEFAULT_DOWNLOAD, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
334
335 /**
336 * GstURISourceBin::use-buffering:
337 *
338 * Perform buffering using a queue2 element, and emit BUFFERING
339 * messages based on low-/high-percent thresholds of streaming data,
340 * such as adaptive-demuxer streams.
341 *
342 * When download buffering is activated and used for the current media
343 * type, this property does nothing.
344 *
345 */
346 g_object_class_install_property (gobject_class, PROP_USE_BUFFERING,
347 g_param_spec_boolean ("use-buffering", "Use Buffering",
348 "Perform buffering on demuxed/parsed media",
349 DEFAULT_USE_BUFFERING, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
350
351 /**
352 * GstURISourceBin::ring-buffer-max-size
353 *
354 * The maximum size of the ring buffer in kilobytes. If set to 0, the ring
355 * buffer is disabled. Default is 0.
356 *
357 */
358 g_object_class_install_property (gobject_class, PROP_RING_BUFFER_MAX_SIZE,
359 g_param_spec_uint64 ("ring-buffer-max-size",
360 "Max. ring buffer size (bytes)",
361 "Max. amount of data in the ring buffer (bytes, 0 = ring buffer disabled)",
362 0, G_MAXUINT, DEFAULT_RING_BUFFER_MAX_SIZE,
363 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
364
365 /**
366 * GstURISourceBin::low-watermark
367 *
368 * Proportion of the queue size (either in bytes or time) for buffering
369 * to restart when crossed from above. Only used if use-buffering is TRUE.
370 */
371 g_object_class_install_property (gobject_class, PROP_LOW_WATERMARK,
372 g_param_spec_double ("low-watermark", "Low watermark",
373 "Low threshold for buffering to start. Only used if use-buffering is True",
374 0.0, 1.0, DEFAULT_LOW_WATERMARK,
375 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
376
377 /**
378 * GstURISourceBin::high-watermark
379 *
380 * Proportion of the queue size (either in bytes or time) to complete
381 * buffering. Only used if use-buffering is TRUE.
382 */
383 g_object_class_install_property (gobject_class, PROP_HIGH_WATERMARK,
384 g_param_spec_double ("high-watermark", "High watermark",
385 "High threshold for buffering to finish. Only used if use-buffering is True",
386 0.0, 1.0, DEFAULT_HIGH_WATERMARK,
387 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
388
389 /**
390 * GstURISourceBin::statistics
391 *
392 * A GStructure containing the following values based on the values from
393 * all the queue's contained in this urisourcebin.
394 *
395 * "minimum-byte-level" G_TYPE_UINT Minimum of the current byte levels
396 * "maximum-byte-level" G_TYPE_UINT Maximum of the current byte levels
397 * "average-byte-level" G_TYPE_UINT Average of the current byte levels
398 * "minimum-time-level" G_TYPE_UINT64 Minimum of the current time levels
399 * "maximum-time-level" G_TYPE_UINT64 Maximum of the current time levels
400 * "average-time-level" G_TYPE_UINT64 Average of the current time levels
401 */
402 g_object_class_install_property (gobject_class, PROP_STATISTICS,
403 g_param_spec_boxed ("statistics", "Queue Statistics",
404 "A set of statistics over all the queue-like elements contained in "
405 "this element", GST_TYPE_STRUCTURE,
406 G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
407
408 /**
409 * GstURISourceBin::drained:
410 *
411 * This signal is emitted when the data for the current uri is played.
412 */
413 gst_uri_source_bin_signals[SIGNAL_DRAINED] =
414 g_signal_new ("drained", G_TYPE_FROM_CLASS (klass),
415 G_SIGNAL_RUN_LAST,
416 G_STRUCT_OFFSET (GstURISourceBinClass, drained), NULL, NULL,
417 g_cclosure_marshal_generic, G_TYPE_NONE, 0, G_TYPE_NONE);
418
419 /**
420 * GstURISourceBin::about-to-finish:
421 *
422 * This signal is emitted when the data for the current uri is played.
423 */
424 gst_uri_source_bin_signals[SIGNAL_ABOUT_TO_FINISH] =
425 g_signal_new ("about-to-finish", G_TYPE_FROM_CLASS (klass),
426 G_SIGNAL_RUN_LAST,
427 G_STRUCT_OFFSET (GstURISourceBinClass, about_to_finish), NULL, NULL,
428 g_cclosure_marshal_generic, G_TYPE_NONE, 0, G_TYPE_NONE);
429
430 /**
431 * GstURISourceBin::source-setup:
432 * @bin: the urisourcebin.
433 * @source: source element
434 *
435 * This signal is emitted after the source element has been created, so
436 * it can be configured by setting additional properties (e.g. set a
437 * proxy server for an http source, or set the device and read speed for
438 * an audio cd source). This is functionally equivalent to connecting to
439 * the notify::source signal, but more convenient.
440 *
441 * Since: 1.6.1
442 */
443 gst_uri_source_bin_signals[SIGNAL_SOURCE_SETUP] =
444 g_signal_new ("source-setup", G_TYPE_FROM_CLASS (klass),
445 G_SIGNAL_RUN_LAST, 0, NULL, NULL,
446 g_cclosure_marshal_generic, G_TYPE_NONE, 1, GST_TYPE_ELEMENT);
447
448 gst_element_class_add_pad_template (gstelement_class,
449 gst_static_pad_template_get (&srctemplate));
450 gst_element_class_set_static_metadata (gstelement_class,
451 "URI reader", "Generic/Bin/Source",
452 "Download and buffer a URI as needed",
453 "Jan Schmidt <jan@centricular.com>");
454
455 gstelement_class->query = GST_DEBUG_FUNCPTR (gst_uri_source_bin_query);
456 gstelement_class->change_state =
457 GST_DEBUG_FUNCPTR (gst_uri_source_bin_change_state);
458
459 gstbin_class->handle_message = GST_DEBUG_FUNCPTR (handle_message);
460 }
461
462 static void
gst_uri_source_bin_init(GstURISourceBin * urisrc)463 gst_uri_source_bin_init (GstURISourceBin * urisrc)
464 {
465 /* first filter out the interesting element factories */
466 g_mutex_init (&urisrc->factories_lock);
467
468 g_mutex_init (&urisrc->lock);
469
470 g_mutex_init (&urisrc->buffering_lock);
471 g_mutex_init (&urisrc->buffering_post_lock);
472
473 urisrc->uri = g_strdup (DEFAULT_PROP_URI);
474 urisrc->connection_speed = DEFAULT_CONNECTION_SPEED;
475
476 urisrc->buffer_duration = DEFAULT_BUFFER_DURATION;
477 urisrc->buffer_size = DEFAULT_BUFFER_SIZE;
478 urisrc->download = DEFAULT_DOWNLOAD;
479 urisrc->use_buffering = DEFAULT_USE_BUFFERING;
480 urisrc->ring_buffer_max_size = DEFAULT_RING_BUFFER_MAX_SIZE;
481 urisrc->last_buffering_pct = -1;
482 urisrc->low_watermark = DEFAULT_LOW_WATERMARK;
483 urisrc->high_watermark = DEFAULT_HIGH_WATERMARK;
484
485 GST_OBJECT_FLAG_SET (urisrc,
486 GST_ELEMENT_FLAG_SOURCE | GST_BIN_FLAG_STREAMS_AWARE);
487 gst_bin_set_suppressed_flags (GST_BIN (urisrc),
488 GST_ELEMENT_FLAG_SOURCE | GST_ELEMENT_FLAG_SINK);
489 }
490
491 static void
gst_uri_source_bin_finalize(GObject * obj)492 gst_uri_source_bin_finalize (GObject * obj)
493 {
494 GstURISourceBin *urisrc = GST_URI_SOURCE_BIN (obj);
495
496 remove_demuxer (urisrc);
497 g_mutex_clear (&urisrc->lock);
498 g_mutex_clear (&urisrc->factories_lock);
499 g_mutex_clear (&urisrc->buffering_lock);
500 g_mutex_clear (&urisrc->buffering_post_lock);
501 g_free (urisrc->uri);
502 if (urisrc->factories)
503 gst_plugin_feature_list_free (urisrc->factories);
504
505 G_OBJECT_CLASS (parent_class)->finalize (obj);
506 }
507
508 static void
gst_uri_source_bin_set_property(GObject * object,guint prop_id,const GValue * value,GParamSpec * pspec)509 gst_uri_source_bin_set_property (GObject * object, guint prop_id,
510 const GValue * value, GParamSpec * pspec)
511 {
512 GstURISourceBin *urisrc = GST_URI_SOURCE_BIN (object);
513
514 switch (prop_id) {
515 case PROP_URI:
516 GST_OBJECT_LOCK (urisrc);
517 g_free (urisrc->uri);
518 urisrc->uri = g_value_dup_string (value);
519 GST_OBJECT_UNLOCK (urisrc);
520 break;
521 case PROP_CONNECTION_SPEED:
522 GST_OBJECT_LOCK (urisrc);
523 urisrc->connection_speed = g_value_get_uint64 (value) * 1000;
524 GST_OBJECT_UNLOCK (urisrc);
525 break;
526 case PROP_BUFFER_SIZE:
527 urisrc->buffer_size = g_value_get_int (value);
528 update_queue_values (urisrc);
529 break;
530 case PROP_BUFFER_DURATION:
531 urisrc->buffer_duration = g_value_get_int64 (value);
532 update_queue_values (urisrc);
533 break;
534 case PROP_DOWNLOAD:
535 urisrc->download = g_value_get_boolean (value);
536 break;
537 case PROP_USE_BUFFERING:
538 urisrc->use_buffering = g_value_get_boolean (value);
539 break;
540 case PROP_RING_BUFFER_MAX_SIZE:
541 urisrc->ring_buffer_max_size = g_value_get_uint64 (value);
542 break;
543 case PROP_LOW_WATERMARK:
544 urisrc->low_watermark = g_value_get_double (value);
545 update_queue_values (urisrc);
546 break;
547 case PROP_HIGH_WATERMARK:
548 urisrc->high_watermark = g_value_get_double (value);
549 update_queue_values (urisrc);
550 break;
551 default:
552 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
553 break;
554 }
555 }
556
557 static void
gst_uri_source_bin_get_property(GObject * object,guint prop_id,GValue * value,GParamSpec * pspec)558 gst_uri_source_bin_get_property (GObject * object, guint prop_id,
559 GValue * value, GParamSpec * pspec)
560 {
561 GstURISourceBin *urisrc = GST_URI_SOURCE_BIN (object);
562
563 switch (prop_id) {
564 case PROP_URI:
565 GST_OBJECT_LOCK (urisrc);
566 g_value_set_string (value, urisrc->uri);
567 GST_OBJECT_UNLOCK (urisrc);
568 break;
569 case PROP_SOURCE:
570 GST_OBJECT_LOCK (urisrc);
571 g_value_set_object (value, urisrc->source);
572 GST_OBJECT_UNLOCK (urisrc);
573 break;
574 case PROP_CONNECTION_SPEED:
575 GST_OBJECT_LOCK (urisrc);
576 g_value_set_uint64 (value, urisrc->connection_speed / 1000);
577 GST_OBJECT_UNLOCK (urisrc);
578 break;
579 case PROP_BUFFER_SIZE:
580 GST_OBJECT_LOCK (urisrc);
581 g_value_set_int (value, urisrc->buffer_size);
582 GST_OBJECT_UNLOCK (urisrc);
583 break;
584 case PROP_BUFFER_DURATION:
585 GST_OBJECT_LOCK (urisrc);
586 g_value_set_int64 (value, urisrc->buffer_duration);
587 GST_OBJECT_UNLOCK (urisrc);
588 break;
589 case PROP_DOWNLOAD:
590 g_value_set_boolean (value, urisrc->download);
591 break;
592 case PROP_USE_BUFFERING:
593 g_value_set_boolean (value, urisrc->use_buffering);
594 break;
595 case PROP_RING_BUFFER_MAX_SIZE:
596 g_value_set_uint64 (value, urisrc->ring_buffer_max_size);
597 break;
598 case PROP_LOW_WATERMARK:
599 g_value_set_double (value, urisrc->low_watermark);
600 break;
601 case PROP_HIGH_WATERMARK:
602 g_value_set_double (value, urisrc->high_watermark);
603 break;
604 case PROP_STATISTICS:
605 g_value_take_boxed (value, get_queue_statistics (urisrc));
606 break;
607 default:
608 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
609 break;
610 }
611 }
612
613 static gboolean
copy_sticky_events(GstPad * pad,GstEvent ** event,gpointer user_data)614 copy_sticky_events (GstPad * pad, GstEvent ** event, gpointer user_data)
615 {
616 GstPad *gpad = GST_PAD_CAST (user_data);
617
618 GST_DEBUG_OBJECT (gpad, "store sticky event %" GST_PTR_FORMAT, *event);
619 gst_pad_store_sticky_event (gpad, *event);
620
621 return TRUE;
622 }
623
624 static GstPadProbeReturn
625 pending_pad_blocked (GstPad * pad, GstPadProbeInfo * info, gpointer user_data);
626
627 static GstPadProbeReturn
628 demux_pad_events (GstPad * pad, GstPadProbeInfo * info, gpointer user_data);
629
630 static void
free_child_src_pad_info(ChildSrcPadInfo * info)631 free_child_src_pad_info (ChildSrcPadInfo * info)
632 {
633 if (info->cur_caps)
634 gst_caps_unref (info->cur_caps);
635 g_free (info);
636 }
637
638 /* Called by the signal handlers when a demuxer has produced a new stream */
639 static void
new_demuxer_pad_added_cb(GstElement * element,GstPad * pad,GstURISourceBin * urisrc)640 new_demuxer_pad_added_cb (GstElement * element, GstPad * pad,
641 GstURISourceBin * urisrc)
642 {
643 ChildSrcPadInfo *info;
644
645 info = g_new0 (ChildSrcPadInfo, 1);
646 info->demux_src_pad = pad;
647 info->cur_caps = gst_pad_get_current_caps (pad);
648 if (info->cur_caps == NULL)
649 info->cur_caps = gst_pad_query_caps (pad, NULL);
650
651 g_object_set_data_full (G_OBJECT (pad), "urisourcebin.srcpadinfo",
652 info, (GDestroyNotify) free_child_src_pad_info);
653
654 GST_DEBUG_OBJECT (element, "new demuxer pad, name: <%s>. "
655 "Added as pending pad with caps %" GST_PTR_FORMAT,
656 GST_PAD_NAME (pad), info->cur_caps);
657
658 GST_URI_SOURCE_BIN_LOCK (urisrc);
659 urisrc->pending_pads = g_list_prepend (urisrc->pending_pads, pad);
660 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
661
662 /* Block the pad. On the first data on that pad if it hasn't
663 * been linked to an output slot, we'll create one */
664 info->blocking_probe_id =
665 gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM,
666 pending_pad_blocked, urisrc, NULL);
667 info->event_probe_id =
668 gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM |
669 GST_PAD_PROBE_TYPE_EVENT_FLUSH, demux_pad_events, urisrc, NULL);
670 }
671
672 static GstPadProbeReturn
pending_pad_blocked(GstPad * pad,GstPadProbeInfo * info,gpointer user_data)673 pending_pad_blocked (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
674 {
675 ChildSrcPadInfo *child_info;
676 OutputSlotInfo *slot;
677 GstURISourceBin *urisrc = GST_URI_SOURCE_BIN (user_data);
678 GstCaps *caps;
679
680 if (!(child_info =
681 g_object_get_data (G_OBJECT (pad), "urisourcebin.srcpadinfo")))
682 goto done;
683
684 GST_LOG_OBJECT (urisrc, "Removing pad %" GST_PTR_FORMAT " from pending list",
685 pad);
686
687 GST_URI_SOURCE_BIN_LOCK (urisrc);
688
689 /* Once blocked, this pad is no longer pending, one way or another */
690 urisrc->pending_pads = g_list_remove (urisrc->pending_pads, pad);
691
692 /* If already linked to a slot, nothing more to do */
693 if (child_info->output_slot) {
694 GST_LOG_OBJECT (urisrc, "Pad %" GST_PTR_FORMAT " is linked to queue %"
695 GST_PTR_FORMAT " on slot %p", pad, child_info->output_slot->queue,
696 child_info->output_slot);
697 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
698 goto done;
699 }
700
701 caps = gst_pad_get_current_caps (pad);
702 if (caps == NULL)
703 caps = gst_pad_query_caps (pad, NULL);
704
705 slot = get_output_slot (urisrc, FALSE, TRUE, caps);
706
707 gst_caps_unref (caps);
708
709 if (slot == NULL) {
710 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
711 goto done;
712 }
713
714 GST_LOG_OBJECT (urisrc, "Pad %" GST_PTR_FORMAT " linked to slot %p", pad,
715 slot);
716
717 child_info->output_slot = slot;
718 slot->linked_info = child_info;
719 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
720
721 gst_pad_link (pad, slot->sinkpad);
722
723 expose_output_pad (urisrc, slot->srcpad);
724
725 done:
726 return GST_PAD_PROBE_REMOVE;
727 }
728
729 /* Called with LOCK held */
730 /* Looks for a suitable pending pad to connect onto this
731 * finishing output slot that's about to EOS */
732 static gboolean
link_pending_pad_to_output(GstURISourceBin * urisrc,OutputSlotInfo * slot)733 link_pending_pad_to_output (GstURISourceBin * urisrc, OutputSlotInfo * slot)
734 {
735 GList *cur;
736 ChildSrcPadInfo *in_info = slot->linked_info;
737 ChildSrcPadInfo *out_info = NULL;
738 gboolean res = FALSE;
739 GstCaps *cur_caps;
740
741 /* Look for a suitable pending pad */
742 cur_caps = gst_pad_get_current_caps (slot->sinkpad);
743
744 GST_DEBUG_OBJECT (urisrc,
745 "Looking for a pending pad with caps %" GST_PTR_FORMAT, cur_caps);
746
747 for (cur = urisrc->pending_pads; cur != NULL; cur = g_list_next (cur)) {
748 GstPad *pending = (GstPad *) (cur->data);
749 ChildSrcPadInfo *cur_info = NULL;
750 if ((cur_info =
751 g_object_get_data (G_OBJECT (pending),
752 "urisourcebin.srcpadinfo"))) {
753 /* Don't re-link to the same pad in case of EOS while still pending */
754 if (in_info == cur_info)
755 continue;
756 if (cur_caps == NULL || gst_caps_is_equal (cur_caps, cur_info->cur_caps)) {
757 GST_DEBUG_OBJECT (urisrc, "Found suitable pending pad %" GST_PTR_FORMAT
758 " with caps %" GST_PTR_FORMAT " to link to this output slot",
759 cur_info->demux_src_pad, cur_info->cur_caps);
760 out_info = cur_info;
761 break;
762 }
763 }
764 }
765
766 if (cur_caps)
767 gst_caps_unref (cur_caps);
768
769 if (out_info) {
770 /* Block any upstream stuff while we switch out the pad */
771 guint block_id =
772 gst_pad_add_probe (slot->sinkpad, GST_PAD_PROBE_TYPE_BLOCK_UPSTREAM,
773 NULL, NULL, NULL);
774 GST_DEBUG_OBJECT (urisrc, "Linking pending pad %" GST_PTR_FORMAT
775 " to existing output slot %p", out_info->demux_src_pad, slot);
776
777 if (in_info) {
778 gst_pad_unlink (in_info->demux_src_pad, slot->sinkpad);
779 in_info->output_slot = NULL;
780 slot->linked_info = NULL;
781 }
782
783 if (gst_pad_link (out_info->demux_src_pad,
784 slot->sinkpad) == GST_PAD_LINK_OK) {
785 out_info->output_slot = slot;
786 slot->linked_info = out_info;
787
788 BUFFERING_LOCK (urisrc);
789 /* A re-linked slot is no longer EOS */
790 slot->is_eos = FALSE;
791 BUFFERING_UNLOCK (urisrc);
792 res = TRUE;
793 slot->is_eos = FALSE;
794 urisrc->pending_pads =
795 g_list_remove (urisrc->pending_pads, out_info->demux_src_pad);
796 } else {
797 GST_ERROR_OBJECT (urisrc,
798 "Failed to link new demuxer pad to the output slot we tried");
799 }
800 gst_pad_remove_probe (slot->sinkpad, block_id);
801 }
802
803 return res;
804 }
805
806 /* Called with lock held */
807 static gboolean
all_slots_are_eos(GstURISourceBin * urisrc)808 all_slots_are_eos (GstURISourceBin * urisrc)
809 {
810 GSList *tmp;
811
812 for (tmp = urisrc->out_slots; tmp; tmp = tmp->next) {
813 OutputSlotInfo *slot = (OutputSlotInfo *) tmp->data;
814 if (slot->is_eos == FALSE)
815 return FALSE;
816 }
817 return TRUE;
818 }
819
820 static GstPadProbeReturn
demux_pad_events(GstPad * pad,GstPadProbeInfo * info,gpointer user_data)821 demux_pad_events (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
822 {
823 GstURISourceBin *urisrc = GST_URI_SOURCE_BIN (user_data);
824 ChildSrcPadInfo *child_info;
825 GstPadProbeReturn ret = GST_PAD_PROBE_OK;
826 GstEvent *ev = GST_PAD_PROBE_INFO_EVENT (info);
827
828 if (!(child_info =
829 g_object_get_data (G_OBJECT (pad), "urisourcebin.srcpadinfo")))
830 goto done;
831
832 GST_URI_SOURCE_BIN_LOCK (urisrc);
833 /* If not linked to a slot, nothing more to do */
834 if (child_info->output_slot == NULL) {
835 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
836 goto done;
837 }
838
839 switch (GST_EVENT_TYPE (ev)) {
840 case GST_EVENT_EOS:
841 {
842 gboolean all_streams_eos;
843
844 GST_LOG_OBJECT (urisrc, "EOS on pad %" GST_PTR_FORMAT, pad);
845
846 if ((urisrc->pending_pads &&
847 link_pending_pad_to_output (urisrc, child_info->output_slot))) {
848 /* Found a new source pad to give this slot data - no need to send EOS */
849 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
850 ret = GST_PAD_PROBE_DROP;
851 goto done;
852 }
853
854 BUFFERING_LOCK (urisrc);
855 /* Mark that we fed an EOS to this slot */
856 child_info->output_slot->is_eos = TRUE;
857 all_streams_eos = all_slots_are_eos (urisrc);
858 BUFFERING_UNLOCK (urisrc);
859
860 /* EOS means this element is no longer buffering */
861 remove_buffering_msgs (urisrc,
862 GST_OBJECT_CAST (child_info->output_slot->queue));
863
864 /* Mark this custom EOS */
865 gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (ev), CUSTOM_EOS_QUARK,
866 (gchar *) CUSTOM_EOS_QUARK_DATA, NULL);
867 if (all_streams_eos) {
868 GST_DEBUG_OBJECT (urisrc, "POSTING ABOUT TO FINISH");
869 g_signal_emit (urisrc,
870 gst_uri_source_bin_signals[SIGNAL_ABOUT_TO_FINISH], 0, NULL);
871 }
872 }
873 break;
874 case GST_EVENT_CAPS:
875 {
876 GstCaps *caps;
877 gst_event_parse_caps (ev, &caps);
878 gst_caps_replace (&child_info->cur_caps, caps);
879 }
880 break;
881 case GST_EVENT_STREAM_START:
882 case GST_EVENT_FLUSH_STOP:
883 BUFFERING_LOCK (urisrc);
884 child_info->output_slot->is_eos = FALSE;
885 BUFFERING_UNLOCK (urisrc);
886 break;
887 default:
888 break;
889 }
890
891 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
892
893 done:
894 return ret;
895 }
896
897 static GstPadProbeReturn
pre_queue_event_probe(GstPad * pad,GstPadProbeInfo * info,gpointer user_data)898 pre_queue_event_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
899 {
900 GstURISourceBin *urisrc = GST_URI_SOURCE_BIN (user_data);
901 GstPadProbeReturn ret = GST_PAD_PROBE_OK;
902 GstEvent *ev = GST_PAD_PROBE_INFO_EVENT (info);
903
904 switch (GST_EVENT_TYPE (ev)) {
905 case GST_EVENT_EOS:
906 {
907 GST_LOG_OBJECT (urisrc, "EOS on pad %" GST_PTR_FORMAT, pad);
908 GST_DEBUG_OBJECT (urisrc, "POSTING ABOUT TO FINISH");
909 g_signal_emit (urisrc,
910 gst_uri_source_bin_signals[SIGNAL_ABOUT_TO_FINISH], 0, NULL);
911 }
912 break;
913 default:
914 break;
915 }
916 return ret;
917 }
918
919 static GstStructure *
get_queue_statistics(GstURISourceBin * urisrc)920 get_queue_statistics (GstURISourceBin * urisrc)
921 {
922 GstStructure *ret = NULL;
923 guint min_byte_level = 0, max_byte_level = 0;
924 guint64 min_time_level = 0, max_time_level = 0;
925 gdouble avg_byte_level = 0., avg_time_level = 0.;
926 guint i = 0;
927 GSList *cur;
928
929 GST_URI_SOURCE_BIN_LOCK (urisrc);
930
931 for (cur = urisrc->out_slots; cur != NULL; cur = g_slist_next (cur)) {
932 OutputSlotInfo *slot = (OutputSlotInfo *) (cur->data);
933 guint byte_limit = 0;
934 guint64 time_limit = 0;
935
936 g_object_get (slot->queue, "current-level-bytes", &byte_limit,
937 "current-level-time", &time_limit, NULL);
938
939 if (byte_limit < min_byte_level)
940 min_byte_level = byte_limit;
941 if (byte_limit > max_byte_level)
942 max_byte_level = byte_limit;
943 avg_byte_level = (avg_byte_level * i + byte_limit) / (gdouble) (i + 1);
944
945 if (time_limit < min_time_level)
946 min_time_level = time_limit;
947 if (time_limit > max_time_level)
948 max_time_level = time_limit;
949 avg_time_level = (avg_time_level * i + time_limit) / (gdouble) (i + 1);
950
951 i++;
952 }
953 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
954
955 ret = gst_structure_new ("application/x-urisourcebin-stats",
956 "minimum-byte-level", G_TYPE_UINT, (guint) min_byte_level,
957 "maximum-byte-level", G_TYPE_UINT, (guint) max_byte_level,
958 "average-byte-level", G_TYPE_UINT, (guint) avg_byte_level,
959 "minimum-time-level", G_TYPE_UINT64, (guint64) min_time_level,
960 "maximum-time-level", G_TYPE_UINT64, (guint64) max_time_level,
961 "average-time-level", G_TYPE_UINT64, (guint64) avg_time_level, NULL);
962
963 return ret;
964 }
965
966 static void
update_queue_values(GstURISourceBin * urisrc)967 update_queue_values (GstURISourceBin * urisrc)
968 {
969 gint64 duration;
970 guint buffer_size;
971 gdouble low_watermark, high_watermark;
972 guint64 cumulative_bitrate = 0;
973 GSList *cur;
974
975 GST_URI_SOURCE_BIN_LOCK (urisrc);
976 duration = GET_BUFFER_DURATION (urisrc);
977 buffer_size = GET_BUFFER_SIZE (urisrc);
978 low_watermark = urisrc->low_watermark;
979 high_watermark = urisrc->high_watermark;
980
981 for (cur = urisrc->out_slots; cur != NULL; cur = g_slist_next (cur)) {
982 OutputSlotInfo *slot = (OutputSlotInfo *) (cur->data);
983 guint64 bitrate = 0;
984
985 if (g_object_class_find_property (G_OBJECT_GET_CLASS (slot->queue),
986 "bitrate")) {
987 g_object_get (G_OBJECT (slot->queue), "bitrate", &bitrate, NULL);
988 }
989
990 if (bitrate > 0)
991 cumulative_bitrate += bitrate;
992 else {
993 GST_TRACE_OBJECT (urisrc, "Unknown bitrate detected from %" GST_PTR_FORMAT
994 ", resetting all bitrates", slot->queue);
995 cumulative_bitrate = 0;
996 break;
997 }
998 }
999
1000 GST_DEBUG_OBJECT (urisrc, "recalculating queue limits with cumulative "
1001 "bitrate %" G_GUINT64_FORMAT ", buffer size %u, buffer duration %"
1002 G_GINT64_FORMAT, cumulative_bitrate, buffer_size, duration);
1003
1004 for (cur = urisrc->out_slots; cur != NULL; cur = g_slist_next (cur)) {
1005 OutputSlotInfo *slot = (OutputSlotInfo *) (cur->data);
1006 guint byte_limit;
1007
1008 if (cumulative_bitrate > 0
1009 && g_object_class_find_property (G_OBJECT_GET_CLASS (slot->queue),
1010 "bitrate")) {
1011 guint64 bitrate;
1012 g_object_get (G_OBJECT (slot->queue), "bitrate", &bitrate, NULL);
1013 byte_limit =
1014 gst_util_uint64_scale (buffer_size, bitrate, cumulative_bitrate);
1015 } else {
1016 /* if not all queue's have valid bitrates, use the buffer-size as the
1017 * limit */
1018 byte_limit = buffer_size;
1019 }
1020
1021 GST_DEBUG_OBJECT (urisrc,
1022 "calculated new limits for queue-like element %" GST_PTR_FORMAT
1023 ", bytes:%u, time:%" G_GUINT64_FORMAT
1024 ", low-watermark:%f, high-watermark:%f",
1025 slot->queue, byte_limit, (guint64) duration, low_watermark,
1026 high_watermark);
1027 g_object_set (G_OBJECT (slot->queue), "max-size-bytes", byte_limit,
1028 "max-size-time", (guint64) duration, "low-watermark", low_watermark,
1029 "high-watermark", high_watermark, NULL);
1030 }
1031 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1032 }
1033
1034 static void
on_queue_bitrate_changed(GstElement * queue,GParamSpec * pspec,gpointer user_data)1035 on_queue_bitrate_changed (GstElement * queue, GParamSpec * pspec,
1036 gpointer user_data)
1037 {
1038 GstURISourceBin *urisrc = GST_URI_SOURCE_BIN (user_data);
1039
1040 gst_element_call_async (GST_ELEMENT (urisrc),
1041 (GstElementCallAsyncFunc) update_queue_values, NULL, NULL);
1042 }
1043
1044 /* Called with lock held */
1045 static OutputSlotInfo *
get_output_slot(GstURISourceBin * urisrc,gboolean do_download,gboolean is_adaptive,GstCaps * caps)1046 get_output_slot (GstURISourceBin * urisrc, gboolean do_download,
1047 gboolean is_adaptive, GstCaps * caps)
1048 {
1049 OutputSlotInfo *slot;
1050 GstPad *srcpad;
1051 GstElement *queue;
1052 const gchar *elem_name;
1053
1054 /* If we have caps, iterate the existing slots and look for an
1055 * unlinked one that can be used */
1056 if (caps && gst_caps_is_fixed (caps)) {
1057 GSList *cur;
1058 GstCaps *cur_caps;
1059
1060 for (cur = urisrc->out_slots; cur != NULL; cur = g_slist_next (cur)) {
1061 slot = (OutputSlotInfo *) (cur->data);
1062 if (slot->linked_info == NULL) {
1063 cur_caps = gst_pad_get_current_caps (slot->sinkpad);
1064 if (cur_caps == NULL || gst_caps_is_equal (caps, cur_caps)) {
1065 GST_LOG_OBJECT (urisrc, "Found existing slot %p to link to", slot);
1066 gst_caps_unref (cur_caps);
1067 slot->is_eos = FALSE;
1068 return slot;
1069 }
1070 gst_caps_unref (cur_caps);
1071 }
1072 }
1073 }
1074
1075 /* Otherwise create the new slot */
1076 if (do_download)
1077 elem_name = "downloadbuffer";
1078 else
1079 elem_name = "queue2";
1080
1081 queue = gst_element_factory_make (elem_name, NULL);
1082 if (!queue)
1083 goto no_buffer_element;
1084
1085 slot = g_new0 (OutputSlotInfo, 1);
1086 slot->queue = queue;
1087
1088 /* Set the slot onto the queue (needed in buffering msg handling) */
1089 g_object_set_data (G_OBJECT (queue), "urisourcebin.slotinfo", slot);
1090
1091 slot->bitrate_changed_id =
1092 g_signal_connect (G_OBJECT (queue), "notify::bitrate",
1093 (GCallback) on_queue_bitrate_changed, urisrc);
1094
1095 if (do_download) {
1096 gchar *temp_template, *filename;
1097 const gchar *tmp_dir, *prgname;
1098
1099 tmp_dir = g_get_user_cache_dir ();
1100 prgname = g_get_prgname ();
1101 if (prgname == NULL)
1102 prgname = "GStreamer";
1103
1104 filename = g_strdup_printf ("%s-XXXXXX", prgname);
1105
1106 /* build our filename */
1107 temp_template = g_build_filename (tmp_dir, filename, NULL);
1108
1109 GST_DEBUG_OBJECT (urisrc, "enable download buffering in %s (%s, %s, %s)",
1110 temp_template, tmp_dir, prgname, filename);
1111
1112 /* configure progressive download for selected media types */
1113 g_object_set (queue, "temp-template", temp_template, NULL);
1114
1115 g_free (filename);
1116 g_free (temp_template);
1117 } else {
1118 if (is_adaptive) {
1119 GST_LOG_OBJECT (urisrc, "Adding queue for adaptive streaming stream");
1120 g_object_set (queue, "use-buffering", urisrc->use_buffering,
1121 "use-tags-bitrate", TRUE, "use-rate-estimate", FALSE, NULL);
1122 } else {
1123 GST_LOG_OBJECT (urisrc, "Adding queue for buffering");
1124 g_object_set (queue, "use-buffering", urisrc->use_buffering, NULL);
1125 }
1126 g_object_set (queue, "ring-buffer-max-size",
1127 urisrc->ring_buffer_max_size, NULL);
1128 /* Disable max-size-buffers - queue based on data rate to the default time limit */
1129 g_object_set (queue, "max-size-buffers", 0, NULL);
1130
1131 /* Don't start buffering until the queue is empty (< 1%).
1132 * Start playback when the queue is 60% full, leaving a bit more room
1133 * for upstream to push more without getting bursty */
1134 g_object_set (queue, "low-percent", 1, "high-percent", 60, NULL);
1135
1136 g_object_set (queue, "low-watermark", urisrc->low_watermark,
1137 "high-watermark", urisrc->high_watermark, NULL);
1138 }
1139
1140 /* set the necessary limits on the queue-like elements */
1141 g_object_set (queue, "max-size-bytes", GET_BUFFER_SIZE (urisrc),
1142 "max-size-time", (guint64) GET_BUFFER_DURATION (urisrc), NULL);
1143 #if 0
1144 /* Disabled because this makes initial startup slower for radio streams */
1145 else {
1146 /* Buffer 4 seconds by default - some extra headroom over the
1147 * core default, because we trigger playback sooner */
1148 //g_object_set (queue, "max-size-time", 4 * GST_SECOND, NULL);
1149 }
1150 #endif
1151
1152 /* save queue pointer so we can remove it later */
1153 urisrc->out_slots = g_slist_prepend (urisrc->out_slots, slot);
1154
1155 gst_bin_add (GST_BIN_CAST (urisrc), queue);
1156 gst_element_sync_state_with_parent (queue);
1157
1158 slot->sinkpad = gst_element_get_static_pad (queue, "sink");
1159
1160 /* get the new raw srcpad */
1161 srcpad = gst_element_get_static_pad (queue, "src");
1162 g_object_set_data (G_OBJECT (srcpad), "urisourcebin.slotinfo", slot);
1163
1164 slot->srcpad = create_output_pad (urisrc, srcpad);
1165
1166 gst_object_unref (srcpad);
1167
1168 return slot;
1169
1170 no_buffer_element:
1171 {
1172 post_missing_plugin_error (GST_ELEMENT_CAST (urisrc), elem_name);
1173 return NULL;
1174 }
1175 }
1176
1177 static GstPadProbeReturn
source_pad_event_probe(GstPad * pad,GstPadProbeInfo * info,gpointer user_data)1178 source_pad_event_probe (GstPad * pad, GstPadProbeInfo * info,
1179 gpointer user_data)
1180 {
1181 GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);
1182 GstURISourceBin *urisrc = user_data;
1183
1184 GST_LOG_OBJECT (pad, "%s, urisrc %p", GST_EVENT_TYPE_NAME (event), event);
1185
1186 if (GST_EVENT_TYPE (event) == GST_EVENT_EOS &&
1187 gst_mini_object_get_qdata (GST_MINI_OBJECT_CAST (event),
1188 CUSTOM_EOS_QUARK)) {
1189 OutputSlotInfo *slot;
1190 GST_DEBUG_OBJECT (pad, "we received EOS");
1191
1192 /* remove custom-eos */
1193 gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (event), CUSTOM_EOS_QUARK,
1194 NULL, NULL);
1195
1196 GST_URI_SOURCE_BIN_LOCK (urisrc);
1197
1198 slot = g_object_get_data (G_OBJECT (pad), "urisourcebin.slotinfo");
1199
1200 if (slot) {
1201 GstEvent *eos;
1202 guint32 seqnum;
1203
1204 if (slot->linked_info) {
1205 if (slot->is_eos) {
1206 /* linked_info is old input which is stil linked without removal */
1207 GST_DEBUG_OBJECT (pad, "push actual EOS");
1208 seqnum = gst_event_get_seqnum (event);
1209 eos = gst_event_new_eos ();
1210 gst_event_set_seqnum (eos, seqnum);
1211 gst_pad_push_event (slot->srcpad, eos);
1212 } else {
1213 /* Do not clear output slot yet. A new input was
1214 * connected. We should just drop this EOS */
1215 }
1216 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1217 return GST_PAD_PROBE_DROP;
1218 }
1219
1220 seqnum = gst_event_get_seqnum (event);
1221 eos = gst_event_new_eos ();
1222 gst_event_set_seqnum (eos, seqnum);
1223 gst_pad_push_event (slot->srcpad, eos);
1224 free_output_slot_async (urisrc, slot);
1225 }
1226
1227 /* FIXME: Only emit drained if all output pads are done and there's no
1228 * pending pads */
1229 g_signal_emit (urisrc, gst_uri_source_bin_signals[SIGNAL_DRAINED], 0, NULL);
1230
1231 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1232 return GST_PAD_PROBE_DROP;
1233 }
1234 /* never drop events */
1235 return GST_PAD_PROBE_OK;
1236 }
1237
1238 /* called when we found a raw pad to expose. We set up a
1239 * padprobe to detect EOS before exposing the pad.
1240 * Called with LOCK held. */
1241 static GstPad *
create_output_pad(GstURISourceBin * urisrc,GstPad * pad)1242 create_output_pad (GstURISourceBin * urisrc, GstPad * pad)
1243 {
1244 GstPad *newpad;
1245 GstPadTemplate *pad_tmpl;
1246 gchar *padname;
1247
1248 gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM,
1249 source_pad_event_probe, urisrc, NULL);
1250
1251 pad_tmpl = gst_static_pad_template_get (&srctemplate);
1252
1253 padname = g_strdup_printf ("src_%u", urisrc->numpads);
1254 urisrc->numpads++;
1255
1256 newpad = gst_ghost_pad_new_from_template (padname, pad, pad_tmpl);
1257 gst_object_unref (pad_tmpl);
1258 g_free (padname);
1259
1260 GST_DEBUG_OBJECT (urisrc, "Created output pad %s:%s for pad %s:%s",
1261 GST_DEBUG_PAD_NAME (newpad), GST_DEBUG_PAD_NAME (pad));
1262
1263 return newpad;
1264 }
1265
1266 static void
expose_output_pad(GstURISourceBin * urisrc,GstPad * pad)1267 expose_output_pad (GstURISourceBin * urisrc, GstPad * pad)
1268 {
1269 GstPad *target;
1270
1271 if (gst_object_has_as_parent (GST_OBJECT (pad), GST_OBJECT (urisrc)))
1272 return; /* Pad is already exposed */
1273
1274 target = gst_ghost_pad_get_target (GST_GHOST_PAD (pad));
1275
1276 gst_pad_sticky_events_foreach (target, copy_sticky_events, pad);
1277 gst_object_unref (target);
1278
1279 gst_pad_set_active (pad, TRUE);
1280 gst_element_add_pad (GST_ELEMENT_CAST (urisrc), pad);
1281 }
1282
1283 static void
pad_removed_cb(GstElement * element,GstPad * pad,GstURISourceBin * urisrc)1284 pad_removed_cb (GstElement * element, GstPad * pad, GstURISourceBin * urisrc)
1285 {
1286 ChildSrcPadInfo *info;
1287
1288 GST_DEBUG_OBJECT (element, "pad removed name: <%s:%s>",
1289 GST_DEBUG_PAD_NAME (pad));
1290
1291 /* we only care about srcpads */
1292 if (!GST_PAD_IS_SRC (pad))
1293 return;
1294
1295 if (!(info = g_object_get_data (G_OBJECT (pad), "urisourcebin.srcpadinfo")))
1296 goto no_info;
1297
1298 GST_URI_SOURCE_BIN_LOCK (urisrc);
1299 /* Make sure this isn't in the pending pads list */
1300 urisrc->pending_pads = g_list_remove (urisrc->pending_pads, pad);
1301
1302 /* Send EOS to the output slot if the demuxer didn't already */
1303 if (info->output_slot) {
1304 GstStructure *s;
1305 GstEvent *event;
1306 OutputSlotInfo *slot;
1307
1308 slot = info->output_slot;
1309
1310 if (!slot->is_eos && urisrc->pending_pads &&
1311 link_pending_pad_to_output (urisrc, slot)) {
1312 /* Found a new source pad to give this slot data - no need to send EOS */
1313 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1314 return;
1315 }
1316
1317 BUFFERING_LOCK (urisrc);
1318 /* Unlink this pad from its output slot and send a fake EOS event
1319 * to drain the queue */
1320 slot->is_eos = TRUE;
1321 BUFFERING_UNLOCK (urisrc);
1322
1323 remove_buffering_msgs (urisrc, GST_OBJECT_CAST (slot->queue));
1324
1325 slot->linked_info = NULL;
1326
1327 info->output_slot = NULL;
1328
1329 GST_LOG_OBJECT (element,
1330 "Pad %" GST_PTR_FORMAT " was removed without EOS. Sending.", pad);
1331
1332 event = gst_event_new_eos ();
1333 s = gst_event_writable_structure (event);
1334 gst_structure_set (s, "urisourcebin-custom-eos", G_TYPE_BOOLEAN, TRUE,
1335 NULL);
1336 gst_pad_send_event (slot->sinkpad, event);
1337 } else {
1338 GST_LOG_OBJECT (urisrc, "Removed pad has no output slot");
1339 }
1340 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1341
1342 return;
1343
1344 /* ERRORS */
1345 no_info:
1346 {
1347 GST_WARNING_OBJECT (element, "no info found for pad");
1348 return;
1349 }
1350 }
1351
1352 /* helper function to lookup stuff in lists */
1353 static gboolean
array_has_value(const gchar * values[],const gchar * value)1354 array_has_value (const gchar * values[], const gchar * value)
1355 {
1356 gint i;
1357
1358 for (i = 0; values[i]; i++) {
1359 if (g_str_has_prefix (value, values[i]))
1360 return TRUE;
1361 }
1362 return FALSE;
1363 }
1364
1365 static gboolean
array_has_uri_value(const gchar * values[],const gchar * value)1366 array_has_uri_value (const gchar * values[], const gchar * value)
1367 {
1368 gint i;
1369
1370 for (i = 0; values[i]; i++) {
1371 if (!g_ascii_strncasecmp (value, values[i], strlen (values[i])))
1372 return TRUE;
1373 }
1374 return FALSE;
1375 }
1376
1377 /* list of URIs that we consider to be streams and that need buffering.
1378 * We have no mechanism yet to figure this out with a query. */
1379 static const gchar *stream_uris[] = { "http://", "https://", "mms://",
1380 "mmsh://", "mmsu://", "mmst://", "fd://", "myth://", "ssh://",
1381 "ftp://", "sftp://",
1382 NULL
1383 };
1384
1385 /* list of URIs that need a queue because they are pretty bursty */
1386 static const gchar *queue_uris[] = { "cdda://", NULL };
1387
1388 /* blacklisted URIs, we know they will always fail. */
1389 static const gchar *blacklisted_uris[] = { NULL };
1390
1391 /* media types that use adaptive streaming */
1392 static const gchar *adaptive_media[] = {
1393 "application/x-hls", "application/vnd.ms-sstr+xml",
1394 "application/dash+xml", NULL
1395 };
1396
1397 #define IS_STREAM_URI(uri) (array_has_uri_value (stream_uris, uri))
1398 #define IS_QUEUE_URI(uri) (array_has_uri_value (queue_uris, uri))
1399 #define IS_BLACKLISTED_URI(uri) (array_has_uri_value (blacklisted_uris, uri))
1400 #define IS_ADAPTIVE_MEDIA(media) (array_has_value (adaptive_media, media))
1401
1402 /*
1403 * Generate and configure a source element.
1404 */
1405 static GstElement *
gen_source_element(GstURISourceBin * urisrc)1406 gen_source_element (GstURISourceBin * urisrc)
1407 {
1408 GObjectClass *source_class;
1409 GstElement *source;
1410 GParamSpec *pspec;
1411 GstQuery *query;
1412 GstSchedulingFlags flags;
1413 GError *err = NULL;
1414
1415 if (!urisrc->uri)
1416 goto no_uri;
1417
1418 GST_LOG_OBJECT (urisrc, "finding source for %s", urisrc->uri);
1419
1420 if (!gst_uri_is_valid (urisrc->uri))
1421 goto invalid_uri;
1422
1423 if (IS_BLACKLISTED_URI (urisrc->uri))
1424 goto uri_blacklisted;
1425
1426 source = gst_element_make_from_uri (GST_URI_SRC, urisrc->uri, "source", &err);
1427 if (!source)
1428 goto no_source;
1429
1430 GST_LOG_OBJECT (urisrc, "found source type %s", G_OBJECT_TYPE_NAME (source));
1431
1432 query = gst_query_new_scheduling ();
1433 if (gst_element_query (source, query)) {
1434 gst_query_parse_scheduling (query, &flags, NULL, NULL, NULL);
1435 urisrc->is_stream = flags & GST_SCHEDULING_FLAG_BANDWIDTH_LIMITED;
1436 } else
1437 urisrc->is_stream = IS_STREAM_URI (urisrc->uri);
1438 gst_query_unref (query);
1439
1440 GST_LOG_OBJECT (urisrc, "source is stream: %d", urisrc->is_stream);
1441
1442 urisrc->need_queue = IS_QUEUE_URI (urisrc->uri);
1443 GST_LOG_OBJECT (urisrc, "source needs queue: %d", urisrc->need_queue);
1444
1445 source_class = G_OBJECT_GET_CLASS (source);
1446
1447 pspec = g_object_class_find_property (source_class, "connection-speed");
1448 if (pspec != NULL) {
1449 guint64 speed = urisrc->connection_speed / 1000;
1450 gboolean wrong_type = FALSE;
1451
1452 if (G_PARAM_SPEC_TYPE (pspec) == G_TYPE_PARAM_UINT) {
1453 GParamSpecUInt *pspecuint = G_PARAM_SPEC_UINT (pspec);
1454
1455 speed = CLAMP (speed, pspecuint->minimum, pspecuint->maximum);
1456 } else if (G_PARAM_SPEC_TYPE (pspec) == G_TYPE_PARAM_INT) {
1457 GParamSpecInt *pspecint = G_PARAM_SPEC_INT (pspec);
1458
1459 speed = CLAMP (speed, pspecint->minimum, pspecint->maximum);
1460 } else if (G_PARAM_SPEC_TYPE (pspec) == G_TYPE_PARAM_UINT64) {
1461 GParamSpecUInt64 *pspecuint = G_PARAM_SPEC_UINT64 (pspec);
1462
1463 speed = CLAMP (speed, pspecuint->minimum, pspecuint->maximum);
1464 } else if (G_PARAM_SPEC_TYPE (pspec) == G_TYPE_PARAM_INT64) {
1465 GParamSpecInt64 *pspecint = G_PARAM_SPEC_INT64 (pspec);
1466
1467 speed = CLAMP (speed, pspecint->minimum, pspecint->maximum);
1468 } else {
1469 GST_WARNING_OBJECT (urisrc,
1470 "The connection speed property %" G_GUINT64_FORMAT
1471 " of type %s is not useful. Not setting it", speed,
1472 g_type_name (G_PARAM_SPEC_TYPE (pspec)));
1473 wrong_type = TRUE;
1474 }
1475
1476 if (!wrong_type) {
1477 g_object_set (source, "connection-speed", speed, NULL);
1478
1479 GST_DEBUG_OBJECT (urisrc,
1480 "setting connection-speed=%" G_GUINT64_FORMAT " to source element",
1481 speed);
1482 }
1483 }
1484
1485 return source;
1486
1487 /* ERRORS */
1488 no_uri:
1489 {
1490 GST_ELEMENT_ERROR (urisrc, RESOURCE, NOT_FOUND,
1491 (_("No URI specified to play from.")), (NULL));
1492 return NULL;
1493 }
1494 invalid_uri:
1495 {
1496 GST_ELEMENT_ERROR (urisrc, RESOURCE, NOT_FOUND,
1497 (_("Invalid URI \"%s\"."), urisrc->uri), (NULL));
1498 g_clear_error (&err);
1499 return NULL;
1500 }
1501 uri_blacklisted:
1502 {
1503 GST_ELEMENT_ERROR (urisrc, RESOURCE, FAILED,
1504 (_("This stream type cannot be played yet.")), (NULL));
1505 return NULL;
1506 }
1507 no_source:
1508 {
1509 /* whoops, could not create the source element, dig a little deeper to
1510 * figure out what might be wrong. */
1511 if (err != NULL && err->code == GST_URI_ERROR_UNSUPPORTED_PROTOCOL) {
1512 gchar *prot;
1513
1514 prot = gst_uri_get_protocol (urisrc->uri);
1515 if (prot == NULL)
1516 goto invalid_uri;
1517
1518 gst_element_post_message (GST_ELEMENT_CAST (urisrc),
1519 gst_missing_uri_source_message_new (GST_ELEMENT (urisrc), prot));
1520
1521 GST_ELEMENT_ERROR (urisrc, CORE, MISSING_PLUGIN,
1522 (_("No URI handler implemented for \"%s\"."), prot), (NULL));
1523
1524 g_free (prot);
1525 } else {
1526 GST_ELEMENT_ERROR (urisrc, RESOURCE, NOT_FOUND,
1527 ("%s", (err) ? err->message : "URI was not accepted by any element"),
1528 ("No element accepted URI '%s'", urisrc->uri));
1529 }
1530
1531 g_clear_error (&err);
1532 return NULL;
1533 }
1534 }
1535
1536 static gboolean
is_all_raw_caps(GstCaps * caps,GstCaps * rawcaps,gboolean * all_raw)1537 is_all_raw_caps (GstCaps * caps, GstCaps * rawcaps, gboolean * all_raw)
1538 {
1539 GstCaps *intersection;
1540 gint capssize;
1541 gboolean res = FALSE;
1542
1543 if (caps == NULL)
1544 return FALSE;
1545
1546 capssize = gst_caps_get_size (caps);
1547 /* no caps, skip and move to the next pad */
1548 if (capssize == 0 || gst_caps_is_empty (caps) || gst_caps_is_any (caps))
1549 goto done;
1550
1551 intersection = gst_caps_intersect (caps, rawcaps);
1552 *all_raw = !gst_caps_is_empty (intersection)
1553 && (gst_caps_get_size (intersection) == capssize);
1554 gst_caps_unref (intersection);
1555
1556 res = TRUE;
1557
1558 done:
1559 return res;
1560 }
1561
1562 /**
1563 * has_all_raw_caps:
1564 * @pad: a #GstPad
1565 * @all_raw: pointer to hold the result
1566 *
1567 * check if the caps of the pad are all raw. The caps are all raw if
1568 * all of its structures contain audio/x-raw or video/x-raw.
1569 *
1570 * Returns: %FALSE @pad has no caps. Else TRUE and @all_raw set t the result.
1571 */
1572 static gboolean
has_all_raw_caps(GstPad * pad,GstCaps * rawcaps,gboolean * all_raw)1573 has_all_raw_caps (GstPad * pad, GstCaps * rawcaps, gboolean * all_raw)
1574 {
1575 GstCaps *caps;
1576 gboolean res = FALSE;
1577
1578 caps = gst_pad_query_caps (pad, NULL);
1579
1580 GST_DEBUG_OBJECT (pad, "have caps %" GST_PTR_FORMAT, caps);
1581
1582 res = is_all_raw_caps (caps, rawcaps, all_raw);
1583
1584 gst_caps_unref (caps);
1585 return res;
1586 }
1587
1588 static void
post_missing_plugin_error(GstElement * urisrc,const gchar * element_name)1589 post_missing_plugin_error (GstElement * urisrc, const gchar * element_name)
1590 {
1591 GstMessage *msg;
1592
1593 msg = gst_missing_element_message_new (urisrc, element_name);
1594 gst_element_post_message (urisrc, msg);
1595
1596 GST_ELEMENT_ERROR (urisrc, CORE, MISSING_PLUGIN,
1597 (_("Missing element '%s' - check your GStreamer installation."),
1598 element_name), (NULL));
1599 }
1600
1601 /**
1602 * analyse_source:
1603 * @urisrc: a #GstURISourceBin
1604 * @is_raw: are all pads raw data
1605 * @have_out: does the source have output
1606 * @is_dynamic: is this a dynamic source
1607 * @use_queue: put a queue before raw output pads
1608 *
1609 * Check the source of @urisrc and collect information about it.
1610 *
1611 * @is_raw will be set to TRUE if the source only produces raw pads. When this
1612 * function returns, all of the raw pad of the source will be added
1613 * to @urisrc
1614 *
1615 * @have_out: will be set to TRUE if the source has output pads.
1616 *
1617 * @is_dynamic: TRUE if the element will create (more) pads dynamically later
1618 * on.
1619 *
1620 * Returns: FALSE if a fatal error occured while scanning.
1621 */
1622 static gboolean
analyse_source(GstURISourceBin * urisrc,gboolean * is_raw,gboolean * have_out,gboolean * is_dynamic,gboolean use_queue)1623 analyse_source (GstURISourceBin * urisrc, gboolean * is_raw,
1624 gboolean * have_out, gboolean * is_dynamic, gboolean use_queue)
1625 {
1626 GstIterator *pads_iter;
1627 gboolean done = FALSE;
1628 gboolean res = TRUE;
1629 GstPad *pad;
1630 GValue item = { 0, };
1631 GstCaps *rawcaps = DEFAULT_CAPS;
1632
1633 *have_out = FALSE;
1634 *is_raw = FALSE;
1635 *is_dynamic = FALSE;
1636
1637 pads_iter = gst_element_iterate_src_pads (urisrc->source);
1638 while (!done) {
1639 switch (gst_iterator_next (pads_iter, &item)) {
1640 case GST_ITERATOR_ERROR:
1641 res = FALSE;
1642 /* FALLTROUGH */
1643 case GST_ITERATOR_DONE:
1644 done = TRUE;
1645 break;
1646 case GST_ITERATOR_RESYNC:
1647 /* reset results and resync */
1648 *have_out = FALSE;
1649 *is_raw = FALSE;
1650 *is_dynamic = FALSE;
1651 gst_iterator_resync (pads_iter);
1652 break;
1653 case GST_ITERATOR_OK:
1654 pad = g_value_dup_object (&item);
1655 /* we now officially have an ouput pad */
1656 *have_out = TRUE;
1657
1658 /* if FALSE, this pad has no caps and we continue with the next pad. */
1659 if (!has_all_raw_caps (pad, rawcaps, is_raw)) {
1660 gst_object_unref (pad);
1661 g_value_reset (&item);
1662 break;
1663 }
1664
1665 /* caps on source pad are all raw, we can add the pad */
1666 if (*is_raw) {
1667 GST_URI_SOURCE_BIN_LOCK (urisrc);
1668 if (use_queue) {
1669 OutputSlotInfo *slot = get_output_slot (urisrc, FALSE, FALSE, NULL);
1670 if (!slot)
1671 goto no_slot;
1672
1673 gst_pad_link (pad, slot->sinkpad);
1674
1675 /* get the new raw srcpad */
1676 gst_object_unref (pad);
1677 pad = slot->srcpad;
1678 } else {
1679 GstPad *tmppad = create_output_pad (urisrc, pad);
1680 gst_object_unref (pad);
1681
1682 pad = tmppad;
1683 }
1684 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1685 expose_output_pad (urisrc, pad);
1686 } else {
1687 gst_object_unref (pad);
1688 }
1689 g_value_reset (&item);
1690 break;
1691 }
1692 }
1693 g_value_unset (&item);
1694 gst_iterator_free (pads_iter);
1695 gst_caps_unref (rawcaps);
1696
1697 if (!*have_out) {
1698 GstElementClass *elemclass;
1699 GList *walk;
1700
1701 /* element has no output pads, check for padtemplates that list SOMETIMES
1702 * pads. */
1703 elemclass = GST_ELEMENT_GET_CLASS (urisrc->source);
1704
1705 walk = gst_element_class_get_pad_template_list (elemclass);
1706 while (walk != NULL) {
1707 GstPadTemplate *templ;
1708
1709 templ = (GstPadTemplate *) walk->data;
1710 if (GST_PAD_TEMPLATE_DIRECTION (templ) == GST_PAD_SRC) {
1711 if (GST_PAD_TEMPLATE_PRESENCE (templ) == GST_PAD_SOMETIMES)
1712 *is_dynamic = TRUE;
1713 break;
1714 }
1715 walk = g_list_next (walk);
1716 }
1717 }
1718
1719 return res;
1720 no_slot:
1721 {
1722 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1723 gst_object_unref (pad);
1724 g_value_unset (&item);
1725 gst_iterator_free (pads_iter);
1726 gst_caps_unref (rawcaps);
1727
1728 return FALSE;
1729 }
1730 }
1731
1732 /* Remove any adaptive demuxer element */
1733 static void
remove_demuxer(GstURISourceBin * bin)1734 remove_demuxer (GstURISourceBin * bin)
1735 {
1736 if (bin->demuxer) {
1737 GST_DEBUG_OBJECT (bin, "removing old demuxer element");
1738 gst_element_set_state (bin->demuxer, GST_STATE_NULL);
1739 gst_bin_remove (GST_BIN_CAST (bin), bin->demuxer);
1740 bin->demuxer = NULL;
1741 }
1742 }
1743
1744 /* make a demuxer and connect to all the signals */
1745 static GstElement *
make_demuxer(GstURISourceBin * urisrc,GstCaps * caps)1746 make_demuxer (GstURISourceBin * urisrc, GstCaps * caps)
1747 {
1748 GList *factories, *eligible, *cur;
1749 GstElement *demuxer = NULL;
1750 GParamSpec *pspec;
1751
1752 GST_LOG_OBJECT (urisrc, "making new adaptive demuxer");
1753
1754 /* now create the demuxer element */
1755
1756 /* FIXME: Fire a signal to get the demuxer? */
1757 factories = gst_element_factory_list_get_elements
1758 (GST_ELEMENT_FACTORY_TYPE_DEMUXER, GST_RANK_MARGINAL);
1759 eligible =
1760 gst_element_factory_list_filter (factories, caps, GST_PAD_SINK,
1761 gst_caps_is_fixed (caps));
1762 gst_plugin_feature_list_free (factories);
1763
1764 if (eligible == NULL)
1765 goto no_demuxer;
1766
1767 eligible = g_list_sort (eligible, gst_plugin_feature_rank_compare_func);
1768
1769 for (cur = eligible; cur != NULL; cur = g_list_next (cur)) {
1770 GstElementFactory *factory = (GstElementFactory *) (cur->data);
1771 const gchar *klass =
1772 gst_element_factory_get_metadata (factory, GST_ELEMENT_METADATA_KLASS);
1773
1774 /* Can't be a demuxer unless it has Demux in the klass name */
1775 if (!strstr (klass, "Demux") || !strstr (klass, "Adaptive"))
1776 continue;
1777
1778 demuxer = gst_element_factory_create (factory, NULL);
1779 break;
1780 }
1781 gst_plugin_feature_list_free (eligible);
1782
1783 if (!demuxer)
1784 goto no_demuxer;
1785
1786 GST_DEBUG_OBJECT (urisrc, "Created adaptive demuxer %" GST_PTR_FORMAT,
1787 demuxer);
1788
1789 /* set up callbacks to create the links between
1790 * demuxer streams and output */
1791 g_signal_connect (demuxer,
1792 "pad-added", G_CALLBACK (new_demuxer_pad_added_cb), urisrc);
1793 g_signal_connect (demuxer,
1794 "pad-removed", G_CALLBACK (pad_removed_cb), urisrc);
1795
1796 /* Propagate connection-speed property */
1797 pspec = g_object_class_find_property (G_OBJECT_GET_CLASS (demuxer),
1798 "connection-speed");
1799 if (pspec != NULL)
1800 g_object_set (demuxer,
1801 "connection-speed", urisrc->connection_speed / 1000, NULL);
1802
1803 return demuxer;
1804
1805 /* ERRORS */
1806 no_demuxer:
1807 {
1808 /* FIXME: Fire the right error */
1809 GST_ELEMENT_ERROR (urisrc, CORE, MISSING_PLUGIN, (NULL),
1810 ("No demuxer element, check your installation"));
1811 return NULL;
1812 }
1813 }
1814
1815 static void
handle_new_pad(GstURISourceBin * urisrc,GstPad * srcpad,GstCaps * caps)1816 handle_new_pad (GstURISourceBin * urisrc, GstPad * srcpad, GstCaps * caps)
1817 {
1818 gboolean is_raw;
1819 GstStructure *s;
1820 const gchar *media_type;
1821 gboolean do_download = FALSE;
1822
1823 GST_URI_SOURCE_BIN_LOCK (urisrc);
1824
1825 /* if this is a pad with all raw caps, we can expose it */
1826 if (is_all_raw_caps (caps, DEFAULT_CAPS, &is_raw) && is_raw) {
1827 GstPad *pad;
1828
1829 GST_DEBUG_OBJECT (urisrc, "Found pad with raw caps %" GST_PTR_FORMAT
1830 ", exposing", caps);
1831 pad = create_output_pad (urisrc, srcpad);
1832 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1833
1834 expose_output_pad (urisrc, pad);
1835 return;
1836 }
1837 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1838
1839 s = gst_caps_get_structure (caps, 0);
1840 media_type = gst_structure_get_name (s);
1841
1842 urisrc->is_adaptive = IS_ADAPTIVE_MEDIA (media_type);
1843
1844 if (urisrc->is_adaptive) {
1845 GstPad *sinkpad;
1846 GstPadLinkReturn link_res;
1847
1848 urisrc->demuxer = make_demuxer (urisrc, caps);
1849 if (!urisrc->demuxer)
1850 goto no_demuxer;
1851 gst_bin_add (GST_BIN_CAST (urisrc), urisrc->demuxer);
1852
1853 sinkpad = gst_element_get_static_pad (urisrc->demuxer, "sink");
1854 if (sinkpad == NULL)
1855 goto no_demuxer_sink;
1856
1857 link_res = gst_pad_link (srcpad, sinkpad);
1858
1859 gst_object_unref (sinkpad);
1860 if (link_res != GST_PAD_LINK_OK)
1861 goto could_not_link;
1862
1863 gst_element_sync_state_with_parent (urisrc->demuxer);
1864 } else if (!urisrc->is_stream) {
1865 GstPad *pad;
1866 /* We don't need slot here, expose immediately */
1867 GST_URI_SOURCE_BIN_LOCK (urisrc);
1868 pad = create_output_pad (urisrc, srcpad);
1869 expose_output_pad (urisrc, pad);
1870 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1871 } else {
1872 OutputSlotInfo *slot;
1873
1874 /* only enable download buffering if the upstream duration is known */
1875 if (urisrc->download) {
1876 GstQuery *query = gst_query_new_duration (GST_FORMAT_BYTES);
1877 if (gst_pad_query (srcpad, query)) {
1878 gint64 dur;
1879 gst_query_parse_duration (query, NULL, &dur);
1880 do_download = (dur != -1);
1881 }
1882 gst_query_unref (query);
1883 }
1884
1885 GST_DEBUG_OBJECT (urisrc, "check media-type %s, do_download:%d", media_type,
1886 do_download);
1887
1888 GST_URI_SOURCE_BIN_LOCK (urisrc);
1889 slot = get_output_slot (urisrc, do_download, FALSE, NULL);
1890
1891 if (slot == NULL || gst_pad_link (srcpad, slot->sinkpad) != GST_PAD_LINK_OK)
1892 goto could_not_link;
1893
1894 gst_pad_add_probe (srcpad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM,
1895 pre_queue_event_probe, urisrc, NULL);
1896
1897 expose_output_pad (urisrc, slot->srcpad);
1898 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1899 }
1900
1901 return;
1902
1903 /* ERRORS */
1904 no_demuxer:
1905 {
1906 /* error was posted */
1907 return;
1908 }
1909 no_demuxer_sink:
1910 {
1911 GST_ELEMENT_ERROR (urisrc, CORE, NEGOTIATION,
1912 (NULL), ("Adaptive demuxer element has no 'sink' pad"));
1913 return;
1914 }
1915 could_not_link:
1916 {
1917 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
1918 GST_ELEMENT_ERROR (urisrc, CORE, NEGOTIATION,
1919 (NULL), ("Can't link typefind to adaptive demuxer element"));
1920 return;
1921 }
1922 }
1923
1924 /* signaled when we have a stream and we need to configure the download
1925 * buffering or regular buffering */
1926 static void
type_found(GstElement * typefind,guint probability,GstCaps * caps,GstURISourceBin * urisrc)1927 type_found (GstElement * typefind, guint probability,
1928 GstCaps * caps, GstURISourceBin * urisrc)
1929 {
1930 GstPad *srcpad = gst_element_get_static_pad (typefind, "src");
1931
1932 GST_DEBUG_OBJECT (urisrc, "typefind found caps %" GST_PTR_FORMAT
1933 " on pad %" GST_PTR_FORMAT, caps, srcpad);
1934 handle_new_pad (urisrc, srcpad, caps);
1935
1936 gst_object_unref (GST_OBJECT (srcpad));
1937 }
1938
1939 /* setup typefind for any source. This will first plug a typefind element to the
1940 * source. After we find the type, we decide to whether to plug an adaptive
1941 * demuxer, or just link through queue2 (if needed) and expose the data */
1942 static gboolean
setup_typefind(GstURISourceBin * urisrc,GstPad * srcpad)1943 setup_typefind (GstURISourceBin * urisrc, GstPad * srcpad)
1944 {
1945 GstElement *typefind;
1946
1947 /* now create the typefind element */
1948 typefind = gst_element_factory_make ("typefind", NULL);
1949 if (!typefind)
1950 goto no_typefind;
1951
1952 /* Make sure the bin doesn't set the typefind running yet */
1953 gst_element_set_locked_state (typefind, TRUE);
1954
1955 gst_bin_add (GST_BIN_CAST (urisrc), typefind);
1956
1957 if (!srcpad) {
1958 if (!gst_element_link_pads (urisrc->source, NULL, typefind, "sink"))
1959 goto could_not_link;
1960 } else {
1961 GstPad *sinkpad = gst_element_get_static_pad (typefind, "sink");
1962 GstPadLinkReturn ret;
1963
1964 ret = gst_pad_link (srcpad, sinkpad);
1965 gst_object_unref (sinkpad);
1966 if (ret != GST_PAD_LINK_OK)
1967 goto could_not_link;
1968 }
1969
1970 urisrc->typefinds = g_list_append (urisrc->typefinds, typefind);
1971
1972 /* connect a signal to find out when the typefind element found
1973 * a type */
1974 g_signal_connect (typefind, "have-type", G_CALLBACK (type_found), urisrc);
1975
1976 /* Now it can start */
1977 gst_element_set_locked_state (typefind, FALSE);
1978 gst_element_sync_state_with_parent (typefind);
1979
1980 return TRUE;
1981
1982 /* ERRORS */
1983 no_typefind:
1984 {
1985 post_missing_plugin_error (GST_ELEMENT_CAST (urisrc), "typefind");
1986 GST_ELEMENT_ERROR (urisrc, CORE, MISSING_PLUGIN, (NULL),
1987 ("No typefind element, check your installation"));
1988 return FALSE;
1989 }
1990 could_not_link:
1991 {
1992 GST_ELEMENT_ERROR (urisrc, CORE, NEGOTIATION,
1993 (NULL), ("Can't link source to typefind element"));
1994 gst_bin_remove (GST_BIN_CAST (urisrc), typefind);
1995 return FALSE;
1996 }
1997 }
1998
1999 static void
free_output_slot(OutputSlotInfo * slot,GstURISourceBin * urisrc)2000 free_output_slot (OutputSlotInfo * slot, GstURISourceBin * urisrc)
2001 {
2002 GST_DEBUG_OBJECT (urisrc, "removing old queue element and freeing slot %p",
2003 slot);
2004 if (slot->bitrate_changed_id > 0)
2005 g_signal_handler_disconnect (slot->queue, slot->bitrate_changed_id);
2006 slot->bitrate_changed_id = 0;
2007
2008 gst_element_set_locked_state (slot->queue, TRUE);
2009 gst_element_set_state (slot->queue, GST_STATE_NULL);
2010 gst_bin_remove (GST_BIN_CAST (urisrc), slot->queue);
2011
2012 gst_object_unref (slot->sinkpad);
2013
2014 remove_buffering_msgs (urisrc, GST_OBJECT_CAST (slot->queue));
2015
2016 /* deactivate and remove the srcpad */
2017 gst_pad_set_active (slot->srcpad, FALSE);
2018 gst_element_remove_pad (GST_ELEMENT_CAST (urisrc), slot->srcpad);
2019
2020 g_free (slot);
2021 }
2022
2023 static void
call_free_output_slot(GstURISourceBin * urisrc,OutputSlotInfo * slot)2024 call_free_output_slot (GstURISourceBin * urisrc, OutputSlotInfo * slot)
2025 {
2026 GST_LOG_OBJECT (urisrc, "free output slot in thread pool");
2027 free_output_slot (slot, urisrc);
2028 }
2029
2030 /* must be called with GST_URI_SOURCE_BIN_LOCK */
2031 static void
free_output_slot_async(GstURISourceBin * urisrc,OutputSlotInfo * slot)2032 free_output_slot_async (GstURISourceBin * urisrc, OutputSlotInfo * slot)
2033 {
2034 GST_LOG_OBJECT (urisrc, "pushing output slot on thread pool to free");
2035 urisrc->out_slots = g_slist_remove (urisrc->out_slots, slot);
2036 gst_element_call_async (GST_ELEMENT_CAST (urisrc),
2037 (GstElementCallAsyncFunc) call_free_output_slot, slot, NULL);
2038 }
2039
2040 /* remove source and all related elements */
2041 static void
remove_source(GstURISourceBin * urisrc)2042 remove_source (GstURISourceBin * urisrc)
2043 {
2044 GstElement *source = urisrc->source;
2045
2046 if (source) {
2047 GST_DEBUG_OBJECT (urisrc, "removing old src element");
2048 gst_element_set_state (source, GST_STATE_NULL);
2049
2050 if (urisrc->src_np_sig_id) {
2051 g_signal_handler_disconnect (source, urisrc->src_np_sig_id);
2052 urisrc->src_np_sig_id = 0;
2053 }
2054 gst_bin_remove (GST_BIN_CAST (urisrc), source);
2055 urisrc->source = NULL;
2056 }
2057 if (urisrc->typefinds) {
2058 GList *iter, *next;
2059 GST_DEBUG_OBJECT (urisrc, "removing old typefind element");
2060 for (iter = urisrc->typefinds; iter; iter = next) {
2061 GstElement *typefind = iter->data;
2062
2063 next = g_list_next (iter);
2064
2065 gst_element_set_state (typefind, GST_STATE_NULL);
2066 gst_bin_remove (GST_BIN_CAST (urisrc), typefind);
2067 }
2068 g_list_free (urisrc->typefinds);
2069 urisrc->typefinds = NULL;
2070 }
2071
2072 GST_URI_SOURCE_BIN_LOCK (urisrc);
2073 g_slist_foreach (urisrc->out_slots, (GFunc) free_output_slot, urisrc);
2074 g_slist_free (urisrc->out_slots);
2075 urisrc->out_slots = NULL;
2076 GST_URI_SOURCE_BIN_UNLOCK (urisrc);
2077
2078 if (urisrc->demuxer) {
2079 GST_DEBUG_OBJECT (urisrc, "removing old adaptive demux element");
2080 gst_element_set_state (urisrc->demuxer, GST_STATE_NULL);
2081 gst_bin_remove (GST_BIN_CAST (urisrc), urisrc->demuxer);
2082 urisrc->demuxer = NULL;
2083 }
2084 }
2085
2086 /* is called when a dynamic source element created a new pad. */
2087 static void
source_new_pad(GstElement * element,GstPad * pad,GstURISourceBin * urisrc)2088 source_new_pad (GstElement * element, GstPad * pad, GstURISourceBin * urisrc)
2089 {
2090 GstCaps *caps;
2091
2092 GST_DEBUG_OBJECT (urisrc, "Found new pad %s.%s in source element %s",
2093 GST_DEBUG_PAD_NAME (pad), GST_ELEMENT_NAME (element));
2094 caps = gst_pad_get_current_caps (pad);
2095 if (caps == NULL)
2096 setup_typefind (urisrc, pad);
2097 else {
2098 handle_new_pad (urisrc, pad, caps);
2099 gst_caps_unref (caps);
2100 }
2101 }
2102
2103 static gboolean
is_live_source(GstElement * source)2104 is_live_source (GstElement * source)
2105 {
2106 GObjectClass *source_class = NULL;
2107 gboolean is_live = FALSE;
2108 GParamSpec *pspec;
2109
2110 source_class = G_OBJECT_GET_CLASS (source);
2111 pspec = g_object_class_find_property (source_class, "is-live");
2112 if (!pspec || G_PARAM_SPEC_VALUE_TYPE (pspec) != G_TYPE_BOOLEAN)
2113 return FALSE;
2114
2115 g_object_get (G_OBJECT (source), "is-live", &is_live, NULL);
2116
2117 return is_live;
2118 }
2119
2120 /* construct and run the source and demuxer elements until we found
2121 * all the streams or until a preroll queue has been filled.
2122 */
2123 static gboolean
setup_source(GstURISourceBin * urisrc)2124 setup_source (GstURISourceBin * urisrc)
2125 {
2126 gboolean is_raw, have_out, is_dynamic;
2127
2128 GST_DEBUG_OBJECT (urisrc, "setup source");
2129
2130 /* delete old src */
2131 remove_source (urisrc);
2132
2133 /* create and configure an element that can handle the uri */
2134 if (!(urisrc->source = gen_source_element (urisrc)))
2135 goto no_source;
2136
2137 /* state will be merged later - if file is not found, error will be
2138 * handled by the application right after. */
2139 gst_bin_add (GST_BIN_CAST (urisrc), urisrc->source);
2140
2141 /* notify of the new source used */
2142 g_object_notify (G_OBJECT (urisrc), "source");
2143
2144 g_signal_emit (urisrc, gst_uri_source_bin_signals[SIGNAL_SOURCE_SETUP],
2145 0, urisrc->source);
2146
2147 if (is_live_source (urisrc->source))
2148 urisrc->is_stream = FALSE;
2149
2150 /* remove the old demuxer now, if any */
2151 remove_demuxer (urisrc);
2152
2153 /* see if the source element emits raw audio/video all by itself,
2154 * if so, we can create streams for the pads and be done with it.
2155 * Also check that is has source pads, if not, we assume it will
2156 * do everything itself. */
2157 if (!analyse_source (urisrc, &is_raw, &have_out, &is_dynamic,
2158 urisrc->need_queue && urisrc->use_buffering))
2159 goto invalid_source;
2160
2161 if (is_raw) {
2162 GST_DEBUG_OBJECT (urisrc, "Source provides all raw data");
2163 /* source provides raw data, we added the pads and we can now signal a
2164 * no_more pads because we are done. */
2165 gst_element_no_more_pads (GST_ELEMENT_CAST (urisrc));
2166 return TRUE;
2167 }
2168 if (!have_out && !is_dynamic) {
2169 GST_DEBUG_OBJECT (urisrc, "Source has no output pads");
2170 return TRUE;
2171 }
2172 if (is_dynamic) {
2173 GST_DEBUG_OBJECT (urisrc, "Source has dynamic output pads");
2174 /* connect a handler for the new-pad signal */
2175 urisrc->src_np_sig_id =
2176 g_signal_connect (urisrc->source, "pad-added",
2177 G_CALLBACK (source_new_pad), urisrc);
2178 } else {
2179 if (urisrc->is_stream) {
2180 GST_DEBUG_OBJECT (urisrc, "Setting up streaming");
2181 /* do the stream things here */
2182 if (!setup_typefind (urisrc, NULL))
2183 goto streaming_failed;
2184 } else {
2185 GstIterator *pads_iter;
2186 gboolean done = FALSE;
2187 pads_iter = gst_element_iterate_src_pads (urisrc->source);
2188 while (!done) {
2189 GValue item = { 0, };
2190 GstPad *pad;
2191
2192 switch (gst_iterator_next (pads_iter, &item)) {
2193 case GST_ITERATOR_ERROR:
2194 GST_WARNING_OBJECT (urisrc,
2195 "Error iterating pads on source element");
2196 /* FALLTROUGH */
2197 case GST_ITERATOR_DONE:
2198 done = TRUE;
2199 break;
2200 case GST_ITERATOR_RESYNC:
2201 /* reset results and resync */
2202 gst_iterator_resync (pads_iter);
2203 break;
2204 case GST_ITERATOR_OK:
2205 pad = g_value_get_object (&item);
2206 if (!setup_typefind (urisrc, pad)) {
2207 gst_iterator_free (pads_iter);
2208 goto streaming_failed;
2209 }
2210 g_value_reset (&item);
2211 break;
2212 }
2213 }
2214 gst_iterator_free (pads_iter);
2215 }
2216 }
2217 return TRUE;
2218
2219 /* ERRORS */
2220 no_source:
2221 {
2222 /* error message was already posted */
2223 return FALSE;
2224 }
2225 invalid_source:
2226 {
2227 GST_ELEMENT_ERROR (urisrc, CORE, FAILED,
2228 (_("Source element is invalid.")), (NULL));
2229 return FALSE;
2230 }
2231 streaming_failed:
2232 {
2233 /* message was posted */
2234 return FALSE;
2235 }
2236 }
2237
2238 static void
value_list_append_structure_list(GValue * list_val,GstStructure ** first,GList * structure_list)2239 value_list_append_structure_list (GValue * list_val, GstStructure ** first,
2240 GList * structure_list)
2241 {
2242 GList *l;
2243
2244 for (l = structure_list; l != NULL; l = l->next) {
2245 GValue val = { 0, };
2246
2247 if (*first == NULL)
2248 *first = gst_structure_copy ((GstStructure *) l->data);
2249
2250 g_value_init (&val, GST_TYPE_STRUCTURE);
2251 g_value_take_boxed (&val, gst_structure_copy ((GstStructure *) l->data));
2252 gst_value_list_append_value (list_val, &val);
2253 g_value_unset (&val);
2254 }
2255 }
2256
2257 /* if it's a redirect message with multiple redirect locations we might
2258 * want to pick a different 'best' location depending on the required
2259 * bitrates and the connection speed */
2260 static GstMessage *
handle_redirect_message(GstURISourceBin * urisrc,GstMessage * msg)2261 handle_redirect_message (GstURISourceBin * urisrc, GstMessage * msg)
2262 {
2263 const GValue *locations_list, *location_val;
2264 GstMessage *new_msg;
2265 GstStructure *new_structure = NULL;
2266 GList *l_good = NULL, *l_neutral = NULL, *l_bad = NULL;
2267 GValue new_list = { 0, };
2268 guint size, i;
2269 const GstStructure *structure;
2270
2271 GST_DEBUG_OBJECT (urisrc, "redirect message: %" GST_PTR_FORMAT, msg);
2272 GST_DEBUG_OBJECT (urisrc, "connection speed: %" G_GUINT64_FORMAT,
2273 urisrc->connection_speed);
2274
2275 structure = gst_message_get_structure (msg);
2276 if (urisrc->connection_speed == 0 || structure == NULL)
2277 return msg;
2278
2279 locations_list = gst_structure_get_value (structure, "locations");
2280 if (locations_list == NULL)
2281 return msg;
2282
2283 size = gst_value_list_get_size (locations_list);
2284 if (size < 2)
2285 return msg;
2286
2287 /* maintain existing order as much as possible, just sort references
2288 * with too high a bitrate to the end (the assumption being that if
2289 * bitrates are given they are given for all interesting streams and
2290 * that the you-need-at-least-version-xyz redirect has the same bitrate
2291 * as the lowest referenced redirect alternative) */
2292 for (i = 0; i < size; ++i) {
2293 const GstStructure *s;
2294 gint bitrate = 0;
2295
2296 location_val = gst_value_list_get_value (locations_list, i);
2297 s = (const GstStructure *) g_value_get_boxed (location_val);
2298 if (!gst_structure_get_int (s, "minimum-bitrate", &bitrate) || bitrate <= 0) {
2299 GST_DEBUG_OBJECT (urisrc, "no bitrate: %" GST_PTR_FORMAT, s);
2300 l_neutral = g_list_append (l_neutral, (gpointer) s);
2301 } else if (bitrate > urisrc->connection_speed) {
2302 GST_DEBUG_OBJECT (urisrc, "bitrate too high: %" GST_PTR_FORMAT, s);
2303 l_bad = g_list_append (l_bad, (gpointer) s);
2304 } else if (bitrate <= urisrc->connection_speed) {
2305 GST_DEBUG_OBJECT (urisrc, "bitrate OK: %" GST_PTR_FORMAT, s);
2306 l_good = g_list_append (l_good, (gpointer) s);
2307 }
2308 }
2309
2310 g_value_init (&new_list, GST_TYPE_LIST);
2311 value_list_append_structure_list (&new_list, &new_structure, l_good);
2312 value_list_append_structure_list (&new_list, &new_structure, l_neutral);
2313 value_list_append_structure_list (&new_list, &new_structure, l_bad);
2314 gst_structure_take_value (new_structure, "locations", &new_list);
2315
2316 g_list_free (l_good);
2317 g_list_free (l_neutral);
2318 g_list_free (l_bad);
2319
2320 new_msg = gst_message_new_element (msg->src, new_structure);
2321 gst_message_unref (msg);
2322
2323 GST_DEBUG_OBJECT (urisrc, "new redirect message: %" GST_PTR_FORMAT, new_msg);
2324 return new_msg;
2325 }
2326
2327 static void
handle_buffering_message(GstURISourceBin * urisrc,GstMessage * msg)2328 handle_buffering_message (GstURISourceBin * urisrc, GstMessage * msg)
2329 {
2330 gint perc, msg_perc;
2331 gint smaller_perc = 100;
2332 GstMessage *smaller = NULL;
2333 GList *found = NULL;
2334 GList *iter;
2335 OutputSlotInfo *slot;
2336
2337 /* buffering messages must be aggregated as there might be multiple
2338 * multiqueue in the pipeline and their independent buffering messages
2339 * will confuse the application
2340 *
2341 * urisourcebin keeps a list of messages received from elements that are
2342 * buffering.
2343 * Rules are:
2344 * 0) Ignore buffering from elements that are draining (is_eos == TRUE)
2345 * 1) Always post the smaller buffering %
2346 * 2) If an element posts a 100% buffering message, remove it from the list
2347 * 3) When there are no more messages on the list, post 100% message
2348 * 4) When an element posts a new buffering message, update the one
2349 * on the list to this new value
2350 */
2351 gst_message_parse_buffering (msg, &msg_perc);
2352 GST_LOG_OBJECT (urisrc, "Got buffering msg from %" GST_PTR_FORMAT
2353 " with %d%%", GST_MESSAGE_SRC (msg), msg_perc);
2354
2355 slot = g_object_get_data (G_OBJECT (GST_MESSAGE_SRC (msg)),
2356 "urisourcebin.slotinfo");
2357
2358 BUFFERING_LOCK (urisrc);
2359 if (slot && slot->is_eos) {
2360 /* Ignore buffering messages from queues we marked as EOS,
2361 * we already removed those from the list of buffering
2362 * objects */
2363 BUFFERING_UNLOCK (urisrc);
2364 gst_message_replace (&msg, NULL);
2365 return;
2366 }
2367
2368
2369 g_mutex_lock (&urisrc->buffering_post_lock);
2370
2371 /*
2372 * Single loop for 2 things:
2373 * 1) Look for a message with the same source
2374 * 1.1) If the received message is 100%, remove it from the list
2375 * 2) Find the minimum buffering from the list from elements that aren't EOS
2376 */
2377 for (iter = urisrc->buffering_status; iter;) {
2378 GstMessage *bufstats = iter->data;
2379 gboolean is_eos = FALSE;
2380
2381 slot = g_object_get_data (G_OBJECT (GST_MESSAGE_SRC (bufstats)),
2382 "urisourcebin.slotinfo");
2383 if (slot)
2384 is_eos = slot->is_eos;
2385
2386 if (GST_MESSAGE_SRC (bufstats) == GST_MESSAGE_SRC (msg)) {
2387 found = iter;
2388 if (msg_perc < 100) {
2389 gst_message_unref (iter->data);
2390 bufstats = iter->data = gst_message_ref (msg);
2391 } else {
2392 GList *current = iter;
2393
2394 /* remove the element here and avoid confusing the loop */
2395 iter = g_list_next (iter);
2396
2397 gst_message_unref (current->data);
2398 urisrc->buffering_status =
2399 g_list_delete_link (urisrc->buffering_status, current);
2400
2401 continue;
2402 }
2403 }
2404
2405 /* only update minimum stat for non-EOS slots */
2406 if (!is_eos) {
2407 gst_message_parse_buffering (bufstats, &perc);
2408 if (perc < smaller_perc) {
2409 smaller_perc = perc;
2410 smaller = bufstats;
2411 }
2412 } else {
2413 GST_LOG_OBJECT (urisrc, "Ignoring buffering from EOS element");
2414 }
2415 iter = g_list_next (iter);
2416 }
2417
2418 if (found == NULL && msg_perc < 100) {
2419 if (msg_perc < smaller_perc) {
2420 smaller_perc = msg_perc;
2421 smaller = msg;
2422 }
2423 urisrc->buffering_status =
2424 g_list_prepend (urisrc->buffering_status, gst_message_ref (msg));
2425 }
2426
2427 if (smaller_perc == urisrc->last_buffering_pct) {
2428 /* Don't repeat our last buffering status */
2429 gst_message_replace (&msg, NULL);
2430 } else {
2431 urisrc->last_buffering_pct = smaller_perc;
2432
2433 /* now compute the buffering message that should be posted */
2434 if (smaller_perc == 100) {
2435 g_assert (urisrc->buffering_status == NULL);
2436 /* we are posting the original received msg */
2437 } else {
2438 gst_message_replace (&msg, smaller);
2439 }
2440 }
2441 BUFFERING_UNLOCK (urisrc);
2442
2443 if (msg) {
2444 GST_LOG_OBJECT (urisrc, "Sending buffering msg from %" GST_PTR_FORMAT
2445 " with %d%%", GST_MESSAGE_SRC (msg), smaller_perc);
2446 GST_BIN_CLASS (parent_class)->handle_message (GST_BIN (urisrc), msg);
2447 } else {
2448 GST_LOG_OBJECT (urisrc, "Dropped buffering msg as a repeat of %d%%",
2449 smaller_perc);
2450 }
2451 g_mutex_unlock (&urisrc->buffering_post_lock);
2452 }
2453
2454 /* Remove any buffering message from the given source */
2455 static void
remove_buffering_msgs(GstURISourceBin * urisrc,GstObject * src)2456 remove_buffering_msgs (GstURISourceBin * urisrc, GstObject * src)
2457 {
2458 GList *iter;
2459 gboolean removed = FALSE, post;
2460
2461 BUFFERING_LOCK (urisrc);
2462 g_mutex_lock (&urisrc->buffering_post_lock);
2463
2464 GST_DEBUG_OBJECT (urisrc, "Removing %" GST_PTR_FORMAT
2465 " buffering messages", src);
2466
2467 for (iter = urisrc->buffering_status; iter;) {
2468 GstMessage *bufstats = iter->data;
2469 if (GST_MESSAGE_SRC (bufstats) == src) {
2470 gst_message_unref (bufstats);
2471 urisrc->buffering_status =
2472 g_list_delete_link (urisrc->buffering_status, iter);
2473 removed = TRUE;
2474 break;
2475 }
2476 iter = g_list_next (iter);
2477 }
2478
2479 post = (removed && urisrc->buffering_status == NULL);
2480 BUFFERING_UNLOCK (urisrc);
2481
2482 if (post) {
2483 GST_DEBUG_OBJECT (urisrc, "Last buffering element done - posting 100%%");
2484
2485 /* removed the last buffering element, post 100% */
2486 gst_element_post_message (GST_ELEMENT_CAST (urisrc),
2487 gst_message_new_buffering (GST_OBJECT_CAST (urisrc), 100));
2488 }
2489
2490 g_mutex_unlock (&urisrc->buffering_post_lock);
2491 }
2492
2493 static void
handle_message(GstBin * bin,GstMessage * msg)2494 handle_message (GstBin * bin, GstMessage * msg)
2495 {
2496 GstURISourceBin *urisrc = GST_URI_SOURCE_BIN (bin);
2497
2498 switch (GST_MESSAGE_TYPE (msg)) {
2499 case GST_MESSAGE_ELEMENT:{
2500 if (gst_message_has_name (msg, "redirect")) {
2501 /* sort redirect messages based on the connection speed. This simplifies
2502 * the user of this element as it can in most cases just pick the first item
2503 * of the sorted list as a good redirection candidate. It can of course
2504 * choose something else from the list if it has a better way. */
2505 msg = handle_redirect_message (urisrc, msg);
2506 }
2507 break;
2508 }
2509 case GST_MESSAGE_BUFFERING:
2510 handle_buffering_message (urisrc, msg);
2511 msg = NULL;
2512 break;
2513 default:
2514 break;
2515 }
2516
2517 if (msg)
2518 GST_BIN_CLASS (parent_class)->handle_message (bin, msg);
2519 }
2520
2521 /* generic struct passed to all query fold methods
2522 * FIXME, move to core.
2523 */
2524 typedef struct
2525 {
2526 GstQuery *query;
2527 gint64 min;
2528 gint64 max;
2529 gboolean seekable;
2530 gboolean live;
2531 } QueryFold;
2532
2533 typedef void (*QueryInitFunction) (GstURISourceBin * urisrc, QueryFold * fold);
2534 typedef void (*QueryDoneFunction) (GstURISourceBin * urisrc, QueryFold * fold);
2535
2536 /* for duration/position we collect all durations/positions and take
2537 * the MAX of all valid results */
2538 static void
uri_source_query_init(GstURISourceBin * urisrc,QueryFold * fold)2539 uri_source_query_init (GstURISourceBin * urisrc, QueryFold * fold)
2540 {
2541 fold->min = 0;
2542 fold->max = -1;
2543 fold->seekable = TRUE;
2544 fold->live = 0;
2545 }
2546
2547 static gboolean
uri_source_query_duration_fold(const GValue * item,GValue * ret,QueryFold * fold)2548 uri_source_query_duration_fold (const GValue * item, GValue * ret,
2549 QueryFold * fold)
2550 {
2551 GstPad *pad = g_value_get_object (item);
2552
2553 if (gst_pad_query (pad, fold->query)) {
2554 gint64 duration;
2555
2556 g_value_set_boolean (ret, TRUE);
2557
2558 gst_query_parse_duration (fold->query, NULL, &duration);
2559
2560 GST_DEBUG_OBJECT (item, "got duration %" G_GINT64_FORMAT, duration);
2561
2562 if (duration > fold->max)
2563 fold->max = duration;
2564 }
2565 return TRUE;
2566 }
2567
2568 static void
uri_source_query_duration_done(GstURISourceBin * urisrc,QueryFold * fold)2569 uri_source_query_duration_done (GstURISourceBin * urisrc, QueryFold * fold)
2570 {
2571 GstFormat format;
2572
2573 gst_query_parse_duration (fold->query, &format, NULL);
2574 /* store max in query result */
2575 gst_query_set_duration (fold->query, format, fold->max);
2576
2577 GST_DEBUG ("max duration %" G_GINT64_FORMAT, fold->max);
2578 }
2579
2580 static gboolean
uri_source_query_position_fold(const GValue * item,GValue * ret,QueryFold * fold)2581 uri_source_query_position_fold (const GValue * item, GValue * ret,
2582 QueryFold * fold)
2583 {
2584 GstPad *pad = g_value_get_object (item);
2585
2586 if (gst_pad_query (pad, fold->query)) {
2587 gint64 position;
2588
2589 g_value_set_boolean (ret, TRUE);
2590
2591 gst_query_parse_position (fold->query, NULL, &position);
2592
2593 GST_DEBUG_OBJECT (item, "got position %" G_GINT64_FORMAT, position);
2594
2595 if (position > fold->max)
2596 fold->max = position;
2597 }
2598
2599 return TRUE;
2600 }
2601
2602 static void
uri_source_query_position_done(GstURISourceBin * urisrc,QueryFold * fold)2603 uri_source_query_position_done (GstURISourceBin * urisrc, QueryFold * fold)
2604 {
2605 GstFormat format;
2606
2607 gst_query_parse_position (fold->query, &format, NULL);
2608 /* store max in query result */
2609 gst_query_set_position (fold->query, format, fold->max);
2610
2611 GST_DEBUG_OBJECT (urisrc, "max position %" G_GINT64_FORMAT, fold->max);
2612 }
2613
2614 static gboolean
uri_source_query_latency_fold(const GValue * item,GValue * ret,QueryFold * fold)2615 uri_source_query_latency_fold (const GValue * item, GValue * ret,
2616 QueryFold * fold)
2617 {
2618 GstPad *pad = g_value_get_object (item);
2619
2620 if (gst_pad_query (pad, fold->query)) {
2621 GstClockTime min, max;
2622 gboolean live;
2623
2624 gst_query_parse_latency (fold->query, &live, &min, &max);
2625
2626 GST_DEBUG_OBJECT (pad,
2627 "got latency min %" GST_TIME_FORMAT ", max %" GST_TIME_FORMAT
2628 ", live %d", GST_TIME_ARGS (min), GST_TIME_ARGS (max), live);
2629
2630 if (live) {
2631 /* for the combined latency we collect the MAX of all min latencies and
2632 * the MIN of all max latencies */
2633 if (min > fold->min)
2634 fold->min = min;
2635 if (fold->max == -1)
2636 fold->max = max;
2637 else if (max < fold->max)
2638 fold->max = max;
2639
2640 fold->live = TRUE;
2641 }
2642 } else {
2643 GST_LOG_OBJECT (pad, "latency query failed");
2644 g_value_set_boolean (ret, FALSE);
2645 }
2646
2647 return TRUE;
2648 }
2649
2650 static void
uri_source_query_latency_done(GstURISourceBin * urisrc,QueryFold * fold)2651 uri_source_query_latency_done (GstURISourceBin * urisrc, QueryFold * fold)
2652 {
2653 /* store max in query result */
2654 gst_query_set_latency (fold->query, fold->live, fold->min, fold->max);
2655
2656 GST_DEBUG_OBJECT (urisrc,
2657 "latency min %" GST_TIME_FORMAT ", max %" GST_TIME_FORMAT
2658 ", live %d", GST_TIME_ARGS (fold->min), GST_TIME_ARGS (fold->max),
2659 fold->live);
2660 }
2661
2662 /* we are seekable if all srcpads are seekable */
2663 static gboolean
uri_source_query_seeking_fold(const GValue * item,GValue * ret,QueryFold * fold)2664 uri_source_query_seeking_fold (const GValue * item, GValue * ret,
2665 QueryFold * fold)
2666 {
2667 GstPad *pad = g_value_get_object (item);
2668
2669 if (gst_pad_query (pad, fold->query)) {
2670 gboolean seekable;
2671
2672 g_value_set_boolean (ret, TRUE);
2673 gst_query_parse_seeking (fold->query, NULL, &seekable, NULL, NULL);
2674
2675 GST_DEBUG_OBJECT (item, "got seekable %d", seekable);
2676
2677 if (fold->seekable)
2678 fold->seekable = seekable;
2679 }
2680
2681 return TRUE;
2682 }
2683
2684 static void
uri_source_query_seeking_done(GstURISourceBin * urisrc,QueryFold * fold)2685 uri_source_query_seeking_done (GstURISourceBin * urisrc, QueryFold * fold)
2686 {
2687 GstFormat format;
2688
2689 gst_query_parse_seeking (fold->query, &format, NULL, NULL, NULL);
2690 gst_query_set_seeking (fold->query, format, fold->seekable, 0, -1);
2691
2692 GST_DEBUG_OBJECT (urisrc, "seekable %d", fold->seekable);
2693 }
2694
2695 /* generic fold, return first valid result */
2696 static gboolean
uri_source_query_generic_fold(const GValue * item,GValue * ret,QueryFold * fold)2697 uri_source_query_generic_fold (const GValue * item, GValue * ret,
2698 QueryFold * fold)
2699 {
2700 GstPad *pad = g_value_get_object (item);
2701 gboolean res;
2702
2703 if ((res = gst_pad_query (pad, fold->query))) {
2704 g_value_set_boolean (ret, TRUE);
2705 GST_DEBUG_OBJECT (item, "answered query %p", fold->query);
2706 }
2707
2708 /* and stop as soon as we have a valid result */
2709 return !res;
2710 }
2711
2712 /* we're a bin, the default query handler iterates sink elements, which we don't
2713 * have normally. We should just query all source pads.
2714 */
2715 static gboolean
gst_uri_source_bin_query(GstElement * element,GstQuery * query)2716 gst_uri_source_bin_query (GstElement * element, GstQuery * query)
2717 {
2718 GstURISourceBin *urisrc;
2719 gboolean res = FALSE;
2720 GstIterator *iter;
2721 GstIteratorFoldFunction fold_func;
2722 QueryInitFunction fold_init = NULL;
2723 QueryDoneFunction fold_done = NULL;
2724 QueryFold fold_data;
2725 GValue ret = { 0 };
2726 gboolean default_ret = FALSE;
2727
2728 urisrc = GST_URI_SOURCE_BIN (element);
2729
2730 switch (GST_QUERY_TYPE (query)) {
2731 case GST_QUERY_DURATION:
2732 /* iterate and collect durations */
2733 fold_func = (GstIteratorFoldFunction) uri_source_query_duration_fold;
2734 fold_init = uri_source_query_init;
2735 fold_done = uri_source_query_duration_done;
2736 break;
2737 case GST_QUERY_POSITION:
2738 /* iterate and collect durations */
2739 fold_func = (GstIteratorFoldFunction) uri_source_query_position_fold;
2740 fold_init = uri_source_query_init;
2741 fold_done = uri_source_query_position_done;
2742 break;
2743 case GST_QUERY_LATENCY:
2744 /* iterate and collect durations */
2745 fold_func = (GstIteratorFoldFunction) uri_source_query_latency_fold;
2746 fold_init = uri_source_query_init;
2747 fold_done = uri_source_query_latency_done;
2748 default_ret = TRUE;
2749 break;
2750 case GST_QUERY_SEEKING:
2751 /* iterate and collect durations */
2752 fold_func = (GstIteratorFoldFunction) uri_source_query_seeking_fold;
2753 fold_init = uri_source_query_init;
2754 fold_done = uri_source_query_seeking_done;
2755 break;
2756 default:
2757 fold_func = (GstIteratorFoldFunction) uri_source_query_generic_fold;
2758 break;
2759 }
2760
2761 fold_data.query = query;
2762
2763 g_value_init (&ret, G_TYPE_BOOLEAN);
2764 g_value_set_boolean (&ret, default_ret);
2765
2766 iter = gst_element_iterate_src_pads (element);
2767 GST_DEBUG_OBJECT (element, "Sending query %p (type %d) to src pads",
2768 query, GST_QUERY_TYPE (query));
2769
2770 if (fold_init)
2771 fold_init (urisrc, &fold_data);
2772
2773 while (TRUE) {
2774 GstIteratorResult ires;
2775
2776 ires = gst_iterator_fold (iter, fold_func, &ret, &fold_data);
2777
2778 switch (ires) {
2779 case GST_ITERATOR_RESYNC:
2780 gst_iterator_resync (iter);
2781 if (fold_init)
2782 fold_init (urisrc, &fold_data);
2783 g_value_set_boolean (&ret, default_ret);
2784 break;
2785 case GST_ITERATOR_OK:
2786 case GST_ITERATOR_DONE:
2787 res = g_value_get_boolean (&ret);
2788 if (fold_done != NULL && res)
2789 fold_done (urisrc, &fold_data);
2790 goto done;
2791 default:
2792 res = FALSE;
2793 goto done;
2794 }
2795 }
2796 done:
2797 gst_iterator_free (iter);
2798
2799 return res;
2800 }
2801
2802 static GstStateChangeReturn
gst_uri_source_bin_change_state(GstElement * element,GstStateChange transition)2803 gst_uri_source_bin_change_state (GstElement * element,
2804 GstStateChange transition)
2805 {
2806 GstStateChangeReturn ret;
2807 GstURISourceBin *urisrc = GST_URI_SOURCE_BIN (element);
2808
2809 switch (transition) {
2810 case GST_STATE_CHANGE_READY_TO_PAUSED:
2811 GST_DEBUG ("ready to paused");
2812 if (!setup_source (urisrc))
2813 goto source_failed;
2814 break;
2815 default:
2816 break;
2817 }
2818
2819 ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
2820 if (ret == GST_STATE_CHANGE_FAILURE)
2821 goto setup_failed;
2822
2823 switch (transition) {
2824 case GST_STATE_CHANGE_READY_TO_PAUSED:
2825 break;
2826 case GST_STATE_CHANGE_PAUSED_TO_READY:
2827 GST_DEBUG ("paused to ready");
2828 remove_demuxer (urisrc);
2829 remove_source (urisrc);
2830 g_list_free_full (urisrc->buffering_status,
2831 (GDestroyNotify) gst_message_unref);
2832 urisrc->buffering_status = NULL;
2833 urisrc->last_buffering_pct = -1;
2834 break;
2835 case GST_STATE_CHANGE_READY_TO_NULL:
2836 GST_DEBUG ("ready to null");
2837 remove_demuxer (urisrc);
2838 remove_source (urisrc);
2839 break;
2840 default:
2841 break;
2842 }
2843 return ret;
2844
2845 /* ERRORS */
2846 source_failed:
2847 {
2848 return GST_STATE_CHANGE_FAILURE;
2849 }
2850 setup_failed:
2851 {
2852 /* clean up leftover groups */
2853 return GST_STATE_CHANGE_FAILURE;
2854 }
2855 }
2856
2857 gboolean
gst_uri_source_bin_plugin_init(GstPlugin * plugin)2858 gst_uri_source_bin_plugin_init (GstPlugin * plugin)
2859 {
2860 GST_DEBUG_CATEGORY_INIT (gst_uri_source_bin_debug, "urisourcebin", 0,
2861 "URI source element");
2862
2863 return gst_element_register (plugin, "urisourcebin", GST_RANK_NONE,
2864 GST_TYPE_URI_SOURCE_BIN);
2865 }
2866