* <itemizedlist>
* <listitem><para>The format is set to #GST_FORMAT_BYTES (default).</para>
* </listitem>
- * <listitem><para>#GstBaseSrc::is_seekable returns %TRUE.</para>
+ * <listitem><para>#GstBaseSrcClass.is_seekable() returns %TRUE.</para>
* </listitem>
* </itemizedlist>
*
- * Since 0.10.9, any #GstBaseSrc can enable pull based scheduling at any
- * time by overriding #GstBaseSrc::check_get_range so that it returns %TRUE.
+ * Since 0.10.9, any #GstBaseSrc can enable pull based scheduling at any time
+ * by overriding #GstBaseSrcClass.check_get_range() so that it returns %TRUE.
*
* If all the conditions are met for operating in pull mode, #GstBaseSrc is
* automatically seekable in push mode as well. The following conditions must
* #GST_FORMAT_BYTES:
* <itemizedlist>
* <listitem><para>
- * #GstBaseSrc::is_seekable returns %TRUE.
+ * #GstBaseSrcClass.is_seekable() returns %TRUE.
* </para></listitem>
* <listitem><para>
- * #GstBaseSrc::query can convert all supported seek formats to the
+ * #GstBaseSrcClass.query() can convert all supported seek formats to the
* internal format as set with gst_base_src_set_format().
* </para></listitem>
* <listitem><para>
- * #GstBaseSrc::do_seek is implemented, performs the seek and returns %TRUE.
+ * #GstBaseSrcClass.do_seek() is implemented, performs the seek and returns
+ * %TRUE.
* </para></listitem>
* </itemizedlist>
*
- * When the element does not meet the requirements to operate in pull mode,
- * the offset and length in the #GstBaseSrc::create method should be ignored.
+ * When the element does not meet the requirements to operate in pull mode, the
+ * offset and length in the #GstBaseSrcClass.create() method should be ignored.
* It is recommended to subclass #GstPushSrc instead, in this situation. If the
* element can operate in pull mode but only with specific offsets and
* lengths, it is allowed to generate an error when the wrong values are passed
- * to the #GstBaseSrc::create function.
+ * to the #GstBaseSrcClass.create() function.
*
* #GstBaseSrc has support for live sources. Live sources are sources that when
* paused discard data, such as audio or video capture devices. A typical live
* Use gst_base_src_set_live() to activate the live source mode.
*
* A live source does not produce data in the PAUSED state. This means that the
- * #GstBaseSrc::create method will not be called in PAUSED but only in PLAYING.
- * To signal the pipeline that the element will not produce data, the return
- * value from the READY to PAUSED state will be #GST_STATE_CHANGE_NO_PREROLL.
+ * #GstBaseSrcClass.create() method will not be called in PAUSED but only in
+ * PLAYING. To signal the pipeline that the element will not produce data, the
+ * return value from the READY to PAUSED state will be
+ * #GST_STATE_CHANGE_NO_PREROLL.
*
* A typical live source will timestamp the buffers it creates with the
* current running time of the pipeline. This is one reason why a live source
* distributed and running.
*
* Live sources that synchronize and block on the clock (an audio source, for
- * example) can since 0.10.12 use gst_base_src_wait_playing() when the ::create
- * function was interrupted by a state change to PAUSED.
+ * example) can since 0.10.12 use gst_base_src_wait_playing() when the
+ * #GstBaseSrcClass.create() function was interrupted by a state change to
+ * PAUSED.
*
- * The #GstBaseSrc::get_times method can be used to implement pseudo-live
- * sources.
- * It only makes sense to implement the ::get_times function if the source is
- * a live source. The ::get_times function should return timestamps starting
- * from 0, as if it were a non-live source. The base class will make sure that
- * the timestamps are transformed into the current running_time.
- * The base source will then wait for the calculated running_time before pushing
- * out the buffer.
+ * The #GstBaseSrcClass.get_times() method can be used to implement pseudo-live
+ * sources. It only makes sense to implement the #GstBaseSrcClass.get_times()
+ * function if the source is a live source. The #GstBaseSrcClass.get_times()
+ * function should return timestamps starting from 0, as if it were a non-live
+ * source. The base class will make sure that the timestamps are transformed
+ * into the current running_time. The base source will then wait for the
+ * calculated running_time before pushing out the buffer.
*
* For live sources, the base class will by default report a latency of 0.
* For pseudo live sources, the base class will by default measure the difference
* There is only support in #GstBaseSrc for exactly one source pad, which
* should be named "src". A source implementation (subclass of #GstBaseSrc)
* should install a pad template in its class_init function, like so:
- * <programlisting>
+ * |[
* static void
* my_element_class_init (GstMyElementClass *klass)
* {
* // see #GstElementDetails
* gst_element_class_set_details (gstelement_class, &details);
* }
- * </programlisting>
+ * ]|
*
* <refsect2>
* <title>Controlled shutdown of live sources in applications</title>
#include <stdlib.h>
#include <string.h>
+#include <gst/gst_private.h>
+
#include "gstbasesrc.h"
#include "gsttypefindhelper.h"
#include <gst/gstmarshal.h>
/* two segments to be sent in the streaming thread with STREAM_LOCK */
GstEvent *close_segment;
GstEvent *start_segment;
+ gboolean newsegment_pending;
/* if EOS is pending (atomic) */
gint pending_eos;
GstClockTimeDiff ts_offset;
gboolean do_timestamp;
+ volatile gint dynamic_size;
/* stream sequence number */
guint32 seqnum;
- /* pending tags to be pushed in the data stream */
- GList *pending_tags;
+ /* pending events (TAG, CUSTOM_BOTH, CUSTOM_DOWNSTREAM) to be
+ * pushed in the data stream */
+ GList *pending_events;
+ volatile gint have_events;
+
+ /* QoS *//* with LOCK */
+ gboolean qos_enabled;
+ gdouble proportion;
+ GstClockTime earliest_time;
};
static GstElementClass *parent_class = NULL;
static GstFlowReturn gst_base_src_get_range (GstBaseSrc * src, guint64 offset,
guint length, GstBuffer ** buf);
static gboolean gst_base_src_seekable (GstBaseSrc * src);
+static gboolean gst_base_src_update_length (GstBaseSrc * src, guint64 offset,
+ guint * length);
static void
gst_base_src_base_init (gpointer g_class)
parent_class = g_type_class_peek_parent (klass);
- gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_base_src_finalize);
- gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_base_src_set_property);
- gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_base_src_get_property);
+ gobject_class->finalize = gst_base_src_finalize;
+ gobject_class->set_property = gst_base_src_set_property;
+ gobject_class->get_property = gst_base_src_get_property;
+/* FIXME 0.11: blocksize property should be int, not ulong (min is >max here) */
g_object_class_install_property (gobject_class, PROP_BLOCKSIZE,
g_param_spec_ulong ("blocksize", "Block size",
"Size in bytes to read per buffer (-1 = default)", 0, G_MAXULONG,
GST_DEBUG_FUNCPTR (gst_base_src_default_check_get_range);
klass->prepare_seek_segment =
GST_DEBUG_FUNCPTR (gst_base_src_default_prepare_seek_segment);
+
+ /* Registering debug symbols for function pointers */
+ GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_activate_push);
+ GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_activate_pull);
+ GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_event_handler);
+ GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_query);
+ GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_pad_get_range);
+ GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_pad_check_get_range);
+ GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_getcaps);
+ GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_setcaps);
+ GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_fixate);
}
static void
pad = gst_pad_new_from_template (pad_template, "src");
GST_DEBUG_OBJECT (basesrc, "setting functions on src pad");
- gst_pad_set_activatepush_function (pad,
- GST_DEBUG_FUNCPTR (gst_base_src_activate_push));
- gst_pad_set_activatepull_function (pad,
- GST_DEBUG_FUNCPTR (gst_base_src_activate_pull));
- gst_pad_set_event_function (pad,
- GST_DEBUG_FUNCPTR (gst_base_src_event_handler));
- gst_pad_set_query_function (pad, GST_DEBUG_FUNCPTR (gst_base_src_query));
- gst_pad_set_checkgetrange_function (pad,
- GST_DEBUG_FUNCPTR (gst_base_src_pad_check_get_range));
- gst_pad_set_getrange_function (pad,
- GST_DEBUG_FUNCPTR (gst_base_src_pad_get_range));
- gst_pad_set_getcaps_function (pad, GST_DEBUG_FUNCPTR (gst_base_src_getcaps));
- gst_pad_set_setcaps_function (pad, GST_DEBUG_FUNCPTR (gst_base_src_setcaps));
- gst_pad_set_fixatecaps_function (pad,
- GST_DEBUG_FUNCPTR (gst_base_src_fixate));
+ gst_pad_set_activatepush_function (pad, gst_base_src_activate_push);
+ gst_pad_set_activatepull_function (pad, gst_base_src_activate_pull);
+ gst_pad_set_event_function (pad, gst_base_src_event_handler);
+ gst_pad_set_query_function (pad, gst_base_src_query);
+ gst_pad_set_checkgetrange_function (pad, gst_base_src_pad_check_get_range);
+ gst_pad_set_getrange_function (pad, gst_base_src_pad_get_range);
+ gst_pad_set_getcaps_function (pad, gst_base_src_getcaps);
+ gst_pad_set_setcaps_function (pad, gst_base_src_setcaps);
+ gst_pad_set_fixatecaps_function (pad, gst_base_src_fixate);
/* hold pointer to pad */
basesrc->srcpad = pad;
gst_base_src_set_format (basesrc, GST_FORMAT_BYTES);
basesrc->data.ABI.typefind = DEFAULT_TYPEFIND;
basesrc->priv->do_timestamp = DEFAULT_DO_TIMESTAMP;
+ g_atomic_int_set (&basesrc->priv->have_events, FALSE);
GST_OBJECT_FLAG_UNSET (basesrc, GST_BASE_SRC_STARTED);
+ GST_OBJECT_FLAG_SET (basesrc, GST_ELEMENT_IS_SOURCE);
GST_DEBUG_OBJECT (basesrc, "init done");
}
event_p = &basesrc->data.ABI.pending_seek;
gst_event_replace (event_p, NULL);
- if (basesrc->priv->pending_tags) {
- g_list_foreach (basesrc->priv->pending_tags, (GFunc) gst_event_unref, NULL);
- g_list_free (basesrc->priv->pending_tags);
+ if (basesrc->priv->pending_events) {
+ g_list_foreach (basesrc->priv->pending_events, (GFunc) gst_event_unref,
+ NULL);
+ g_list_free (basesrc->priv->pending_events);
}
G_OBJECT_CLASS (parent_class)->finalize (object);
* gst_base_src_wait_playing:
* @src: the src
*
- * If the #GstBaseSrcClass::create method performs its own synchronisation against
- * the clock it must unblock when going from PLAYING to the PAUSED state and call
- * this method before continuing to produce the remaining data.
+ * If the #GstBaseSrcClass.create() method performs its own synchronisation
+ * against the clock it must unblock when going from PLAYING to the PAUSED state
+ * and call this method before continuing to produce the remaining data.
*
* This function will block until a state change to PLAYING happens (in which
* case this function returns #GST_FLOW_OK) or the processing must be stopped due
{
g_return_val_if_fail (GST_IS_BASE_SRC (src), GST_FLOW_ERROR);
- /* block until the state changes, or we get a flush, or something */
- GST_DEBUG_OBJECT (src, "live source waiting for running state");
- GST_LIVE_WAIT (src);
- if (src->priv->flushing)
- goto flushing;
- GST_DEBUG_OBJECT (src, "live source unlocked");
+ do {
+ /* block until the state changes, or we get a flush, or something */
+ GST_DEBUG_OBJECT (src, "live source waiting for running state");
+ GST_LIVE_WAIT (src);
+ GST_DEBUG_OBJECT (src, "live source unlocked");
+ if (src->priv->flushing)
+ goto flushing;
+ } while (G_UNLIKELY (!src->live_running));
return GST_FLOW_OK;
* for sending NEW_SEGMENT events and for performing seeks.
*
* If a format of GST_FORMAT_BYTES is set, the element will be able to
- * operate in pull mode if the #GstBaseSrc::is_seekable returns TRUE.
+ * operate in pull mode if the #GstBaseSrcClass.is_seekable() returns TRUE.
+ *
+ * This function must only be called in states < %GST_STATE_PAUSED.
*
* Since: 0.10.1
*/
gst_base_src_set_format (GstBaseSrc * src, GstFormat format)
{
g_return_if_fail (GST_IS_BASE_SRC (src));
+ g_return_if_fail (GST_STATE (src) <= GST_STATE_READY);
+ GST_OBJECT_LOCK (src);
gst_segment_init (&src->segment, format);
+ GST_OBJECT_UNLOCK (src);
+}
+
+/**
+ * gst_base_src_set_dynamic_size:
+ * @src: base source instance
+ * @dynamic: new dynamic size mode
+ *
+ * If not @dynamic, size is only updated when needed, such as when trying to
+ * read past current tracked size. Otherwise, size is checked for upon each
+ * read.
+ *
+ * Since: 0.10.35
+ */
+void
+gst_base_src_set_dynamic_size (GstBaseSrc * src, gboolean dynamic)
+{
+ g_return_if_fail (GST_IS_BASE_SRC (src));
+
+ g_atomic_int_set (&src->priv->dynamic_size, dynamic);
}
/**
* gst_base_src_query_latency:
* @src: the source
- * @live: if the source is live
- * @min_latency: the min latency of the source
- * @max_latency: the max latency of the source
+ * @live: (out) (allow-none): if the source is live
+ * @min_latency: (out) (allow-none): the min latency of the source
+ * @max_latency: (out) (allow-none): the max latency of the source
*
* Query the source for the latency parameters. @live will be TRUE when @src is
* configured as a live source. @min_latency will be set to the difference
*
* Since: 0.10.22
*/
+/* FIXME 0.11: blocksize property should be int, not ulong */
void
gst_base_src_set_blocksize (GstBaseSrc * src, gulong blocksize)
{
*
* Since: 0.10.22
*/
+/* FIXME 0.11: blocksize property should be int, not ulong */
gulong
gst_base_src_get_blocksize (GstBaseSrc * src)
{
return res;
}
+/**
+ * gst_base_src_new_seamless_segment:
+ * @src: The source
+ * @start: The new start value for the segment
+ * @stop: Stop value for the new segment
+ * @position: The position value for the new segent
+ *
+ * Prepare a new seamless segment for emission downstream. This function must
+ * only be called by derived sub-classes, and only from the create() function,
+ * as the stream-lock needs to be held.
+ *
+ * The format for the new segment will be the current format of the source, as
+ * configured with gst_base_src_set_format()
+ *
+ * Returns: %TRUE if preparation of the seamless segment succeeded.
+ *
+ * Since: 0.10.26
+ */
+gboolean
+gst_base_src_new_seamless_segment (GstBaseSrc * src, gint64 start, gint64 stop,
+ gint64 position)
+{
+ gboolean res = TRUE;
+
+ GST_DEBUG_OBJECT (src,
+ "Starting new seamless segment. Start %" GST_TIME_FORMAT " stop %"
+ GST_TIME_FORMAT " position %" GST_TIME_FORMAT, GST_TIME_ARGS (start),
+ GST_TIME_ARGS (stop), GST_TIME_ARGS (position));
+
+ GST_OBJECT_LOCK (src);
+ if (src->data.ABI.running && !src->priv->newsegment_pending) {
+ if (src->priv->close_segment)
+ gst_event_unref (src->priv->close_segment);
+ src->priv->close_segment =
+ gst_event_new_new_segment_full (TRUE,
+ src->segment.rate, src->segment.applied_rate, src->segment.format,
+ src->segment.start, src->segment.last_stop, src->segment.time);
+ }
+
+ gst_segment_set_newsegment_full (&src->segment, FALSE, src->segment.rate,
+ src->segment.applied_rate, src->segment.format, start, stop, position);
+
+ if (src->priv->start_segment)
+ gst_event_unref (src->priv->start_segment);
+ if (src->segment.rate >= 0.0) {
+ /* forward, we send data from last_stop to stop */
+ src->priv->start_segment =
+ gst_event_new_new_segment_full (FALSE,
+ src->segment.rate, src->segment.applied_rate, src->segment.format,
+ src->segment.last_stop, stop, src->segment.time);
+ } else {
+ /* reverse, we send data from last_stop to start */
+ src->priv->start_segment =
+ gst_event_new_new_segment_full (FALSE,
+ src->segment.rate, src->segment.applied_rate, src->segment.format,
+ src->segment.start, src->segment.last_stop, src->segment.time);
+ }
+ GST_OBJECT_UNLOCK (src);
+
+ src->priv->discont = TRUE;
+ src->data.ABI.running = TRUE;
+
+ return res;
+}
+
static gboolean
gst_base_src_setcaps (GstPad * pad, GstCaps * caps)
{
GstFormat format;
gst_query_parse_position (query, &format, NULL);
+
+ GST_DEBUG_OBJECT (src, "position query in format %s",
+ gst_format_get_name (format));
+
switch (format) {
case GST_FORMAT_PERCENT:
{
gint64 position;
gint64 duration;
+ GST_OBJECT_LOCK (src);
position = src->segment.last_stop;
duration = src->segment.duration;
+ GST_OBJECT_UNLOCK (src);
if (position != -1 && duration != -1) {
if (position < duration)
default:
{
gint64 position;
+ GstFormat seg_format;
- position = src->segment.last_stop;
+ GST_OBJECT_LOCK (src);
+ position =
+ gst_segment_to_stream_time (&src->segment, src->segment.format,
+ src->segment.last_stop);
+ seg_format = src->segment.format;
+ GST_OBJECT_UNLOCK (src);
if (position != -1) {
/* convert to requested format */
res =
- gst_pad_query_convert (src->srcpad, src->segment.format,
+ gst_pad_query_convert (src->srcpad, seg_format,
position, &format, &position);
} else
res = TRUE;
default:
{
gint64 duration;
+ GstFormat seg_format;
+ guint length = 0;
+
+ /* may have to refresh duration */
+ if (g_atomic_int_get (&src->priv->dynamic_size))
+ gst_base_src_update_length (src, 0, &length);
/* this is the duration as configured by the subclass. */
+ GST_OBJECT_LOCK (src);
duration = src->segment.duration;
+ seg_format = src->segment.format;
+ GST_OBJECT_UNLOCK (src);
+
+ GST_LOG_OBJECT (src, "duration %" G_GINT64_FORMAT ", format %s",
+ duration, gst_format_get_name (seg_format));
if (duration != -1) {
/* convert to requested format, if this fails, we have a duration
* but we cannot answer the query, we must return FALSE. */
res =
- gst_pad_query_convert (src->srcpad, src->segment.format,
+ gst_pad_query_convert (src->srcpad, seg_format,
duration, &format, &duration);
} else {
/* The subclass did not configure a duration, we assume that the
case GST_QUERY_SEEKING:
{
- GstFormat format;
+ GstFormat format, seg_format;
+ gint64 duration;
+
+ GST_OBJECT_LOCK (src);
+ duration = src->segment.duration;
+ seg_format = src->segment.format;
+ GST_OBJECT_UNLOCK (src);
gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
- if (format == src->segment.format) {
- gst_query_set_seeking (query, src->segment.format,
- gst_base_src_seekable (src), 0, src->segment.duration);
+ if (format == seg_format) {
+ gst_query_set_seeking (query, seg_format,
+ gst_base_src_seekable (src), 0, duration);
res = TRUE;
} else {
/* FIXME 0.11: return TRUE + seekable=FALSE for SEEKING query here */
{
gint64 start, stop;
+ GST_OBJECT_LOCK (src);
/* no end segment configured, current duration then */
if ((stop = src->segment.stop) == -1)
stop = src->segment.duration;
if (stop != -1)
stop -= src->segment.time;
}
+
gst_query_set_segment (query, src->segment.rate, src->segment.format,
start, stop);
+ GST_OBJECT_UNLOCK (src);
res = TRUE;
break;
}
GstClockTime min, max;
gboolean live;
- /* Subclasses should override and implement something usefull */
+ /* Subclasses should override and implement something useful */
res = gst_base_src_query_latency (src, &live, &min, &max);
GST_LOG_OBJECT (src, "report latency: live %d, min %" GST_TIME_FORMAT
break;
case GST_QUERY_BUFFERING:
{
- GstFormat format;
+ GstFormat format, seg_format;
gint64 start, stop, estimated;
gst_query_parse_buffering_range (query, &format, NULL, NULL, NULL);
GST_DEBUG_OBJECT (src, "buffering query in format %s",
gst_format_get_name (format));
+ GST_OBJECT_LOCK (src);
if (src->random_access) {
estimated = 0;
start = 0;
start = -1;
stop = -1;
}
+ seg_format = src->segment.format;
+ GST_OBJECT_UNLOCK (src);
+
/* convert to required format. When the conversion fails, we can't answer
* the query. When the value is unknown, we can don't perform conversion
* but report TRUE. */
if (format != GST_FORMAT_PERCENT && stop != -1) {
- res = gst_pad_query_convert (src->srcpad, src->segment.format,
+ res = gst_pad_query_convert (src->srcpad, seg_format,
stop, &format, &stop);
} else {
res = TRUE;
}
if (res && format != GST_FORMAT_PERCENT && start != -1)
- res = gst_pad_query_convert (src->srcpad, src->segment.format,
+ res = gst_pad_query_convert (src->srcpad, seg_format,
start, &format, &start);
gst_query_set_buffering_range (query, format, start, stop, estimated);
gboolean result = FALSE;
src = GST_BASE_SRC (gst_pad_get_parent (pad));
+ if (G_UNLIKELY (src == NULL))
+ return FALSE;
bclass = GST_BASE_SRC_GET_CLASS (src);
guint32 seqnum;
GstEvent *tevent;
- GST_DEBUG_OBJECT (src, "doing seek");
+ GST_DEBUG_OBJECT (src, "doing seek: %" GST_PTR_FORMAT, event);
+ GST_OBJECT_LOCK (src);
dest_format = src->segment.format;
+ GST_OBJECT_UNLOCK (src);
if (event) {
gst_event_parse_seek (event, &rate, &seek_format, &flags,
/* If we configured the seeksegment above, don't overwrite it now. Otherwise
* copy the current segment info into the temp segment that we can actually
- * attempt the seek with. We only update the real segment if the seek suceeds. */
+ * attempt the seek with. We only update the real segment if the seek succeeds. */
if (!seekseg_configured) {
memcpy (&seeksegment, &src->segment, sizeof (GstSegment));
/* now configure the final seek segment */
if (event) {
- if (src->segment.format != seek_format) {
+ if (seeksegment.format != seek_format) {
/* OK, here's where we give the subclass a chance to convert the relative
* seek into an absolute one in the processing format. We set up any
* absolute seek above, before taking the stream lock. */
res = FALSE;
}
- /* if successfull seek, we update our real segment and push
+ /* if the seek was successful, we update our real segment and push
* out the new segment. */
if (res) {
+ GST_OBJECT_LOCK (src);
memcpy (&src->segment, &seeksegment, sizeof (GstSegment));
+ GST_OBJECT_UNLOCK (src);
- if (src->segment.flags & GST_SEEK_FLAG_SEGMENT) {
+ if (seeksegment.flags & GST_SEEK_FLAG_SEGMENT) {
GstMessage *message;
message = gst_message_new_segment_start (GST_OBJECT (src),
- src->segment.format, src->segment.last_stop);
+ seeksegment.format, seeksegment.last_stop);
gst_message_set_seqnum (message, seqnum);
gst_element_post_message (GST_ELEMENT (src), message);
/* for deriving a stop position for the playback segment from the seek
* segment, we must take the duration when the stop is not set */
- if ((stop = src->segment.stop) == -1)
- stop = src->segment.duration;
+ if ((stop = seeksegment.stop) == -1)
+ stop = seeksegment.duration;
GST_DEBUG_OBJECT (src, "Sending newsegment from %" G_GINT64_FORMAT
- " to %" G_GINT64_FORMAT, src->segment.start, stop);
+ " to %" G_GINT64_FORMAT, seeksegment.start, stop);
/* now replace the old segment so that we send it in the stream thread the
* next time it is scheduled. */
if (src->priv->start_segment)
gst_event_unref (src->priv->start_segment);
- if (src->segment.rate >= 0.0) {
+ if (seeksegment.rate >= 0.0) {
/* forward, we send data from last_stop to stop */
src->priv->start_segment =
gst_event_new_new_segment_full (FALSE,
- src->segment.rate, src->segment.applied_rate, src->segment.format,
- src->segment.last_stop, stop, src->segment.time);
+ seeksegment.rate, seeksegment.applied_rate, seeksegment.format,
+ seeksegment.last_stop, stop, seeksegment.time);
} else {
/* reverse, we send data from last_stop to start */
src->priv->start_segment =
gst_event_new_new_segment_full (FALSE,
- src->segment.rate, src->segment.applied_rate, src->segment.format,
- src->segment.start, src->segment.last_stop, src->segment.time);
+ seeksegment.rate, seeksegment.applied_rate, seeksegment.format,
+ seeksegment.start, seeksegment.last_stop, seeksegment.time);
}
gst_event_set_seqnum (src->priv->start_segment, seqnum);
+ src->priv->newsegment_pending = TRUE;
}
src->priv->discont = TRUE;
src->data.ABI.running = TRUE;
- /* and restart the task in case it got paused explicitely or by
+ /* and restart the task in case it got paused explicitly or by
* the FLUSH_START event we pushed out. */
tres = gst_pad_start_task (src->srcpad, (GstTaskFunction) gst_base_src_loop,
src->srcpad);
src = GST_BASE_SRC (element);
- GST_DEBUG_OBJECT (src, "reveived %s event", GST_EVENT_TYPE_NAME (event));
+ GST_DEBUG_OBJECT (src, "handling event %p %" GST_PTR_FORMAT, event, event);
switch (GST_EVENT_TYPE (event)) {
/* bidirectional events */
* first and do EOS instead of entering it.
* - If we are in the _create function or we did not manage to set the
* flag fast enough and we are about to enter the _create function,
- * we unlock it so that we exit with WRONG_STATE immediatly. We then
+ * we unlock it so that we exit with WRONG_STATE immediately. We then
* check the EOS flag and do the EOS logic.
*/
g_atomic_int_set (&src->priv->pending_eos, TRUE);
/* sending random NEWSEGMENT downstream can break sync. */
break;
case GST_EVENT_TAG:
- /* Insert tag in the dataflow */
+ case GST_EVENT_CUSTOM_DOWNSTREAM:
+ case GST_EVENT_CUSTOM_BOTH:
+ /* Insert TAG, CUSTOM_DOWNSTREAM, CUSTOM_BOTH in the dataflow */
GST_OBJECT_LOCK (src);
- src->priv->pending_tags = g_list_append (src->priv->pending_tags, event);
+ src->priv->pending_events =
+ g_list_append (src->priv->pending_events, event);
+ g_atomic_int_set (&src->priv->have_events, TRUE);
GST_OBJECT_UNLOCK (src);
event = NULL;
result = TRUE;
case GST_EVENT_CUSTOM_UPSTREAM:
/* override send_event if you want this */
break;
- case GST_EVENT_CUSTOM_DOWNSTREAM:
- case GST_EVENT_CUSTOM_BOTH:
- /* FIXME, insert event in the dataflow */
- break;
case GST_EVENT_CUSTOM_DOWNSTREAM_OOB:
case GST_EVENT_CUSTOM_BOTH_OOB:
/* insert a random custom event into the pipeline */
return FALSE;
}
+static void
+gst_base_src_update_qos (GstBaseSrc * src,
+ gdouble proportion, GstClockTimeDiff diff, GstClockTime timestamp)
+{
+ GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, src,
+ "qos: proportion: %lf, diff %" G_GINT64_FORMAT ", timestamp %"
+ GST_TIME_FORMAT, proportion, diff, GST_TIME_ARGS (timestamp));
+
+ GST_OBJECT_LOCK (src);
+ src->priv->proportion = proportion;
+ src->priv->earliest_time = timestamp + diff;
+ GST_OBJECT_UNLOCK (src);
+}
+
+
static gboolean
gst_base_src_default_event (GstBaseSrc * src, GstEvent * event)
{
gboolean result;
+ GST_DEBUG_OBJECT (src, "handle event %" GST_PTR_FORMAT, event);
+
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_SEEK:
/* is normally called when in push mode */
case GST_EVENT_FLUSH_STOP:
result = gst_base_src_set_flushing (src, FALSE, TRUE, TRUE, NULL);
break;
- default:
+ case GST_EVENT_QOS:
+ {
+ gdouble proportion;
+ GstClockTimeDiff diff;
+ GstClockTime timestamp;
+
+ gst_event_parse_qos (event, &proportion, &diff, ×tamp);
+ gst_base_src_update_qos (src, proportion, diff, timestamp);
result = TRUE;
break;
+ }
+ default:
+ result = FALSE;
+ break;
}
return result;
gboolean result = FALSE;
src = GST_BASE_SRC (gst_pad_get_parent (pad));
+ if (G_UNLIKELY (src == NULL)) {
+ gst_event_unref (event);
+ return FALSE;
+ }
+
bclass = GST_BASE_SRC_GET_CLASS (src);
if (bclass->event) {
}
}
+/* Called with STREAM_LOCK and LIVE_LOCK */
static gboolean
gst_base_src_update_length (GstBaseSrc * src, guint64 offset, guint * length)
{
guint64 size, maxsize;
GstBaseSrcClass *bclass;
+ GstFormat format;
+ gint64 stop;
+ gboolean dynamic;
bclass = GST_BASE_SRC_GET_CLASS (src);
- /* only operate if we are working with bytes */
- if (src->segment.format != GST_FORMAT_BYTES)
- return TRUE;
-
+ format = src->segment.format;
+ stop = src->segment.stop;
/* get total file size */
size = (guint64) src->segment.duration;
+ /* only operate if we are working with bytes */
+ if (format != GST_FORMAT_BYTES)
+ return TRUE;
+
/* the max amount of bytes to read is the total size or
* up to the segment.stop if present. */
- if (src->segment.stop != -1)
- maxsize = MIN (size, src->segment.stop);
+ if (stop != -1)
+ maxsize = MIN (size, stop);
else
maxsize = size;
GST_DEBUG_OBJECT (src,
"reading offset %" G_GUINT64_FORMAT ", length %u, size %" G_GINT64_FORMAT
", segment.stop %" G_GINT64_FORMAT ", maxsize %" G_GINT64_FORMAT, offset,
- *length, size, src->segment.stop, maxsize);
+ *length, size, stop, maxsize);
+
+ dynamic = g_atomic_int_get (&src->priv->dynamic_size);
+ GST_DEBUG_OBJECT (src, "dynamic size: %d", dynamic);
/* check size if we have one */
if (maxsize != -1) {
/* if we run past the end, check if the file became bigger and
* retry. */
- if (G_UNLIKELY (offset + *length >= maxsize)) {
+ if (G_UNLIKELY (offset + *length >= maxsize || dynamic)) {
/* see if length of the file changed */
if (bclass->get_size)
if (!bclass->get_size (src, &size))
size = -1;
- gst_segment_set_duration (&src->segment, GST_FORMAT_BYTES, size);
-
/* make sure we don't exceed the configured segment stop
* if it was set */
- if (src->segment.stop != -1)
- maxsize = MIN (size, src->segment.stop);
+ if (stop != -1)
+ maxsize = MIN (size, stop);
else
maxsize = size;
}
}
- /* keep track of current position. segment is in bytes, we checked
- * that above. */
- gst_segment_set_last_stop (&src->segment, GST_FORMAT_BYTES, offset);
+ /* keep track of current position and update duration.
+ * segment is in bytes, we checked that above. */
+ GST_OBJECT_LOCK (src);
+ gst_segment_set_duration (&src->segment, GST_FORMAT_BYTES, size);
+ GST_OBJECT_UNLOCK (src);
return TRUE;
bclass = GST_BASE_SRC_GET_CLASS (src);
+again:
if (src->is_live) {
- while (G_UNLIKELY (!src->live_running)) {
+ if (G_UNLIKELY (!src->live_running)) {
ret = gst_base_src_wait_playing (src);
if (ret != GST_FLOW_OK)
goto stopped;
if (G_UNLIKELY (!gst_base_src_update_length (src, offset, &length)))
goto unexpected_length;
+ /* track position */
+ GST_OBJECT_LOCK (src);
+ if (src->segment.format == GST_FORMAT_BYTES)
+ gst_segment_set_last_stop (&src->segment, GST_FORMAT_BYTES, offset);
+ GST_OBJECT_UNLOCK (src);
+
/* normally we don't count buffers */
if (G_UNLIKELY (src->num_buffers_left >= 0)) {
if (src->num_buffers_left == 0)
/* no timestamp set and we are at offset 0, we can timestamp with 0 */
if (offset == 0 && src->segment.time == 0
- && GST_BUFFER_TIMESTAMP (*buf) == -1)
+ && GST_BUFFER_TIMESTAMP (*buf) == -1 && !src->is_live) {
+ *buf = gst_buffer_make_metadata_writable (*buf);
GST_BUFFER_TIMESTAMP (*buf) = 0;
+ }
/* set pad caps on the buffer if the buffer had no caps */
- if (GST_BUFFER_CAPS (*buf) == NULL)
+ if (GST_BUFFER_CAPS (*buf) == NULL) {
+ *buf = gst_buffer_make_metadata_writable (*buf);
gst_buffer_set_caps (*buf, GST_PAD_CAPS (src->srcpad));
+ }
/* now sync before pushing the buffer */
status = gst_base_src_do_sync (src, *buf);
break;
case GST_CLOCK_UNSCHEDULED:
/* this case is triggered when we were waiting for the clock and
- * it got unlocked because we did a state change. We return
- * WRONG_STATE in this case to stop the dataflow also get rid of the
- * produced buffer. */
- GST_DEBUG_OBJECT (src,
- "clock was unscheduled (%d), returning WRONG_STATE", status);
+ * it got unlocked because we did a state change. In any case, get rid of
+ * the buffer. */
gst_buffer_unref (*buf);
*buf = NULL;
- ret = GST_FLOW_WRONG_STATE;
+ if (!src->live_running) {
+ /* We return WRONG_STATE when we are not running to stop the dataflow also
+ * get rid of the produced buffer. */
+ GST_DEBUG_OBJECT (src,
+ "clock was unscheduled (%d), returning WRONG_STATE", status);
+ ret = GST_FLOW_WRONG_STATE;
+ } else {
+ /* If we are running when this happens, we quickly switched between
+ * pause and playing. We try to produce a new buffer */
+ GST_DEBUG_OBJECT (src,
+ "clock was unscheduled (%d), but we are running", status);
+ goto again;
+ }
break;
default:
/* all other result values are unexpected and errors */
GstBaseSrc *src;
GstFlowReturn res;
- src = GST_BASE_SRC (gst_pad_get_parent (pad));
+ src = GST_BASE_SRC_CAST (gst_object_ref (GST_OBJECT_PARENT (pad)));
GST_LIVE_LOCK (src);
if (G_UNLIKELY (src->priv->flushing))
gint64 position;
gboolean eos;
gulong blocksize;
- GList *tags, *tmp;
+ GList *pending_events = NULL, *tmp;
eos = FALSE;
gst_pad_push_event (pad, src->priv->start_segment);
src->priv->start_segment = NULL;
}
+ src->priv->newsegment_pending = FALSE;
- GST_OBJECT_LOCK (src);
- /* take the tags */
- tags = src->priv->pending_tags;
- src->priv->pending_tags = NULL;
- GST_OBJECT_UNLOCK (src);
+ if (g_atomic_int_get (&src->priv->have_events)) {
+ GST_OBJECT_LOCK (src);
+ /* take the events */
+ pending_events = src->priv->pending_events;
+ src->priv->pending_events = NULL;
+ g_atomic_int_set (&src->priv->have_events, FALSE);
+ GST_OBJECT_UNLOCK (src);
+ }
- /* Push out pending tags if any */
- if (G_UNLIKELY (tags != NULL)) {
- for (tmp = tags; tmp; tmp = g_list_next (tmp)) {
+ /* Push out pending events if any */
+ if (G_UNLIKELY (pending_events != NULL)) {
+ for (tmp = pending_events; tmp; tmp = g_list_next (tmp)) {
GstEvent *ev = (GstEvent *) tmp->data;
gst_pad_push_event (pad, ev);
}
- g_list_free (tags);
+ g_list_free (pending_events);
}
/* figure out the new position */
/* when going reverse, all buffers are DISCONT */
src->priv->discont = TRUE;
}
+ GST_OBJECT_LOCK (src);
gst_segment_set_last_stop (&src->segment, src->segment.format, position);
+ GST_OBJECT_UNLOCK (src);
}
if (G_UNLIKELY (src->priv->discont)) {
GST_DEBUG_OBJECT (src, "pausing task, reason %s", reason);
src->data.ABI.running = FALSE;
gst_pad_pause_task (pad);
- if (GST_FLOW_IS_FATAL (ret) || ret == GST_FLOW_NOT_LINKED) {
- if (ret == GST_FLOW_UNEXPECTED) {
- /* perform EOS logic */
- if (src->segment.flags & GST_SEEK_FLAG_SEGMENT) {
- GstMessage *message;
-
- message = gst_message_new_segment_done (GST_OBJECT_CAST (src),
- src->segment.format, src->segment.last_stop);
- gst_message_set_seqnum (message, src->priv->seqnum);
- gst_element_post_message (GST_ELEMENT_CAST (src), message);
- } else {
- event = gst_event_new_eos ();
- gst_event_set_seqnum (event, src->priv->seqnum);
- gst_pad_push_event (pad, event);
- src->priv->last_sent_eos = TRUE;
- }
+ if (ret == GST_FLOW_UNEXPECTED) {
+ gboolean flag_segment;
+ GstFormat format;
+ gint64 last_stop;
+
+ /* perform EOS logic */
+ flag_segment = (src->segment.flags & GST_SEEK_FLAG_SEGMENT) != 0;
+ format = src->segment.format;
+ last_stop = src->segment.last_stop;
+
+ if (flag_segment) {
+ GstMessage *message;
+
+ message = gst_message_new_segment_done (GST_OBJECT_CAST (src),
+ format, last_stop);
+ gst_message_set_seqnum (message, src->priv->seqnum);
+ gst_element_post_message (GST_ELEMENT_CAST (src), message);
} else {
event = gst_event_new_eos ();
gst_event_set_seqnum (event, src->priv->seqnum);
- /* for fatal errors we post an error message, post the error
- * first so the app knows about the error first. */
- GST_ELEMENT_ERROR (src, STREAM, FAILED,
- (_("Internal data flow error.")),
- ("streaming task paused, reason %s (%d)", reason, ret));
gst_pad_push_event (pad, event);
src->priv->last_sent_eos = TRUE;
}
+ } else if (ret == GST_FLOW_NOT_LINKED || ret <= GST_FLOW_UNEXPECTED) {
+ event = gst_event_new_eos ();
+ gst_event_set_seqnum (event, src->priv->seqnum);
+ /* for fatal errors we post an error message, post the error
+ * first so the app knows about the error first.
+ * Also don't do this for WRONG_STATE because it happens
+ * due to flushing and posting an error message because of
+ * that is the wrong thing to do, e.g. when we're doing
+ * a flushing seek. */
+ GST_ELEMENT_ERROR (src, STREAM, FAILED,
+ (_("Internal data flow error.")),
+ ("streaming task paused, reason %s (%d)", reason, ret));
+ gst_pad_push_event (pad, event);
+ src->priv->last_sent_eos = TRUE;
}
goto done;
}
GST_ELEMENT_ERROR (src, STREAM, FAILED,
(_("Internal data flow error.")), ("element returned NULL buffer"));
GST_LIVE_UNLOCK (src);
- /* we finished the segment on error */
- ret = GST_FLOW_ERROR;
goto done;
}
}
gboolean result = FALSE;
/* first see what is possible on our source pad */
- thiscaps = gst_pad_get_caps (GST_BASE_SRC_PAD (basesrc));
+ thiscaps = gst_pad_get_caps_reffed (GST_BASE_SRC_PAD (basesrc));
GST_DEBUG_OBJECT (basesrc, "caps of src: %" GST_PTR_FORMAT, thiscaps);
/* nothing or anything is allowed, we're done */
if (thiscaps == NULL || gst_caps_is_any (thiscaps))
goto no_caps;
/* get the peer caps */
- peercaps = gst_pad_peer_get_caps (GST_BASE_SRC_PAD (basesrc));
+ peercaps = gst_pad_peer_get_caps_reffed (GST_BASE_SRC_PAD (basesrc));
GST_DEBUG_OBJECT (basesrc, "caps of peer: %" GST_PTR_FORMAT, peercaps);
if (peercaps) {
- GstCaps *icaps;
-
/* get intersection */
- icaps = gst_caps_intersect (thiscaps, peercaps);
- GST_DEBUG_OBJECT (basesrc, "intersect: %" GST_PTR_FORMAT, icaps);
- gst_caps_unref (thiscaps);
+ caps =
+ gst_caps_intersect_full (peercaps, thiscaps, GST_CAPS_INTERSECT_FIRST);
+ GST_DEBUG_OBJECT (basesrc, "intersect: %" GST_PTR_FORMAT, caps);
gst_caps_unref (peercaps);
- if (icaps) {
- /* take first (and best, since they are sorted) possibility */
- caps = gst_caps_copy_nth (icaps, 0);
- gst_caps_unref (icaps);
- }
} else {
/* no peer, work with our own caps then */
- caps = thiscaps;
+ caps = gst_caps_copy (thiscaps);
}
+ gst_caps_unref (thiscaps);
if (caps) {
- caps = gst_caps_make_writable (caps);
+ /* take first (and best, since they are sorted) possibility */
gst_caps_truncate (caps);
/* now fixate */
gboolean result;
guint64 size;
gboolean seekable;
+ GstFormat format;
if (GST_OBJECT_FLAG_IS_SET (basesrc, GST_BASE_SRC_STARTED))
return TRUE;
basesrc->num_buffers_left = basesrc->num_buffers;
+ GST_OBJECT_LOCK (basesrc);
gst_segment_init (&basesrc->segment, basesrc->segment.format);
+ GST_OBJECT_UNLOCK (basesrc);
+
basesrc->data.ABI.running = FALSE;
+ basesrc->priv->newsegment_pending = FALSE;
bclass = GST_BASE_SRC_GET_CLASS (basesrc);
if (bclass->start)
GST_OBJECT_FLAG_SET (basesrc, GST_BASE_SRC_STARTED);
+ format = basesrc->segment.format;
+
/* figure out the size */
- if (basesrc->segment.format == GST_FORMAT_BYTES) {
+ if (format == GST_FORMAT_BYTES) {
if (bclass->get_size) {
if (!(result = bclass->get_size (basesrc, &size)))
size = -1;
GST_DEBUG_OBJECT (basesrc, "setting size %" G_GUINT64_FORMAT, size);
/* only update the size when operating in bytes, subclass is supposed
* to set duration in the start method for other formats */
+ GST_OBJECT_LOCK (basesrc);
gst_segment_set_duration (&basesrc->segment, GST_FORMAT_BYTES, size);
+ GST_OBJECT_UNLOCK (basesrc);
} else {
size = -1;
}
GST_DEBUG_OBJECT (basesrc,
- "format: %d, have size: %d, size: %" G_GUINT64_FORMAT ", duration: %"
- G_GINT64_FORMAT, basesrc->segment.format, result, size,
+ "format: %s, have size: %d, size: %" G_GUINT64_FORMAT ", duration: %"
+ G_GINT64_FORMAT, gst_format_get_name (format), result, size,
basesrc->segment.duration);
seekable = gst_base_src_seekable (basesrc);
GST_DEBUG_OBJECT (basesrc, "is seekable: %d", seekable);
/* update for random access flag */
- basesrc->random_access = seekable &&
- basesrc->segment.format == GST_FORMAT_BYTES;
+ basesrc->random_access = seekable && format == GST_FORMAT_BYTES;
GST_DEBUG_OBJECT (basesrc, "is random_access: %d", basesrc->random_access);
} else {
/* signal the live source that it can start playing */
basesrc->live_running = live_play;
+
+ /* When unlocking drop all delayed events */
+ if (unlock) {
+ GST_OBJECT_LOCK (basesrc);
+ if (basesrc->priv->pending_events) {
+ g_list_foreach (basesrc->priv->pending_events, (GFunc) gst_event_unref,
+ NULL);
+ g_list_free (basesrc->priv->pending_events);
+ basesrc->priv->pending_events = NULL;
+ g_atomic_int_set (&basesrc->priv->have_events, FALSE);
+ }
+ GST_OBJECT_UNLOCK (basesrc);
+ }
}
GST_LIVE_SIGNAL (basesrc);
GST_LIVE_UNLOCK (basesrc);
seek_failed:
{
GST_ERROR_OBJECT (basesrc, "Failed to perform initial seek");
+ /* flush all */
+ gst_base_src_set_flushing (basesrc, TRUE, FALSE, TRUE, NULL);
+ /* stop the task */
+ gst_pad_stop_task (pad);
+ /* Stop the basesrc */
gst_base_src_stop (basesrc);
if (event)
gst_event_unref (event);
* already did this */
/* FIXME, deprecate this behaviour, it is very dangerous.
- * the prefered way of sending EOS downstream is by sending
+ * the preferred way of sending EOS downstream is by sending
* the EOS event to the element */
if (!basesrc->priv->last_sent_eos) {
GST_DEBUG_OBJECT (basesrc, "Sending EOS event");
projects / platform / upstream / gstreamer.git / blobdiff