/* GStreamer * Copyright (C) 2008 Nokia Corporation. All rights reserved. * Contact: Stefan Kost * Copyright (C) 2008 Sebastian Dröge . * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Library General Public * License as published by the Free Software Foundation; either * version 2 of the License, or (at your option) any later version. * * This library is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * Library General Public License for more details. * * You should have received a copy of the GNU Library General Public * License along with this library; if not, write to the * Free Software Foundation, Inc., 59 Temple Place - Suite 330, * Boston, MA 02111-1307, USA. */ /** * SECTION:gstbaseparse * @short_description: Base class for stream parsers * @see_also: #GstBaseTransform * * This base class is for parser elements that process data and splits it * into separate audio/video/whatever frames. * * It provides for: * * One sinkpad and one srcpad * Handles state changes * Does flushing * Push mode * Pull mode * Handles events (NEWSEGMENT/EOS/FLUSH) * Handles seeking in both modes * * Handles POSITION/DURATION/SEEKING/FORMAT/CONVERT queries * * * * The purpose of this base class is to provide a basic functionality of * a parser and share a lot of rather complex code. * * Description of the parsing mechanism: * * * Set-up phase * * GstBaseParse class calls @set_sink_caps to inform the subclass about * incoming sinkpad caps. Subclass should set the srcpad caps accordingly. * * * GstBaseParse calls @start to inform subclass that data processing is * about to start now. * * * At least in this point subclass needs to tell the GstBaseParse class * how big data chunks it wants to receive (min_frame_size). It can do * this with @gst_base_parse_set_min_frame_size. * * * GstBaseParse class sets up appropriate data passing mode (pull/push) * and starts to process the data. * * * * * * Parsing phase * * GstBaseParse gathers at least min_frame_size bytes of data either * by pulling it from upstream or collecting buffers into internal * #GstAdapter. * * * A buffer of min_frame_size bytes is passed to subclass with * @check_valid_frame. Subclass checks the contents and returns TRUE * if the buffer contains a valid frame. It also needs to set the * @framesize according to the detected frame size. If buffer didn't * contain a valid frame, this call must return FALSE and optionally * set the @skipsize value to inform base class that how many bytes * it needs to skip in order to find a valid frame. The passed buffer * is read-only. * * * After valid frame is found, it will be passed again to subclass with * @parse_frame call. Now subclass is responsible for parsing the * frame contents and setting the buffer timestamp, duration and caps. * * * Finally the buffer can be pushed downstream and parsing loop starts * over again. * * * During the parsing process GstBaseClass will handle both srcpad and * sinkpad events. They will be passed to subclass if @event or * @src_event callbacks have been provided. * * * * * Shutdown phase * * GstBaseParse class calls @stop to inform the subclass that data * parsing will be stopped. * * * * * * Subclass is responsible for providing pad template caps for * source and sink pads. The pads need to be named "sink" and "src". It also * needs to set the fixed caps on srcpad, when the format is ensured (e.g. * when base class calls subclass' @set_sink_caps function). * * This base class uses GST_FORMAT_DEFAULT as a meaning of frames. So, * subclass conversion routine needs to know that conversion from * GST_FORMAT_TIME to GST_FORMAT_DEFAULT must return the * frame number that can be found from the given byte position. * * GstBaseParse uses subclasses conversion methods also for seeking. If * subclass doesn't provide @convert function, seeking will get disabled. * * Subclass @start and @stop functions will be called to inform the beginning * and end of data processing. * * Things that subclass need to take care of: * * Provide pad templates * * Fixate the source pad caps when appropriate * * * Inform base class how big data chunks should be retrieved. This is * done with @gst_base_parse_set_min_frame_size function. * * * Examine data chunks passed to subclass with @check_valid_frame * and tell if they contain a valid frame * * * Set the caps and timestamp to frame that is passed to subclass with * @parse_frame function. * * Provide conversion functions * * Update the duration information with @gst_base_parse_set_duration * * * */ /* TODO: * - Better segment handling: * - NEWSEGMENT for gaps * - Not NEWSEGMENT starting at 0 but at first frame timestamp * - Seek table generation and subclass seek entry injection * - Accurate seeking * - In push mode provide a queue of adapter-"queued" buffers for upstream * buffer metadata * - Handle upstream timestamps */ #ifdef HAVE_CONFIG_H # include "config.h" #endif #include #include #include "gstbaseparse.h" GST_DEBUG_CATEGORY_STATIC (gst_base_parse_debug); #define GST_CAT_DEFAULT gst_base_parse_debug /* Supported formats */ static GstFormat fmtlist[] = { GST_FORMAT_DEFAULT, GST_FORMAT_BYTES, GST_FORMAT_TIME, 0 }; #define GST_BASE_PARSE_GET_PRIVATE(obj) \ (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_BASE_PARSE, GstBaseParsePrivate)) struct _GstBaseParsePrivate { GstActivateMode pad_mode; gint64 duration; GstFormat duration_fmt; guint min_frame_size; gboolean discont; gboolean flushing; gint64 offset; gint64 pending_offset; GList *pending_events; GstBuffer *cache; GstAdapter *adapter; guint64 upstream_size; guint64 upstream_duration; guint64 avg_bitrate; guint64 estimated_size; guint64 estimated_duration; #ifndef GST_DISABLE_INDEX GstIndex *index; gint index_id; #endif }; struct _GstBaseParseClassPrivate { gpointer _padding; }; static GstElementClass *parent_class = NULL; static void gst_base_parse_base_init (gpointer g_class); static void gst_base_parse_base_finalize (gpointer g_class); static void gst_base_parse_class_init (GstBaseParseClass * klass); static void gst_base_parse_init (GstBaseParse * parse, GstBaseParseClass * klass); GType gst_base_parse_get_type (void) { static GType base_parse_type = 0; if (!base_parse_type) { static const GTypeInfo base_parse_info = { sizeof (GstBaseParseClass), (GBaseInitFunc) gst_base_parse_base_init, (GBaseFinalizeFunc) gst_base_parse_base_finalize, (GClassInitFunc) gst_base_parse_class_init, NULL, NULL, sizeof (GstBaseParse), 0, (GInstanceInitFunc) gst_base_parse_init, }; base_parse_type = g_type_register_static (GST_TYPE_ELEMENT, "GstFlacBaseParse", &base_parse_info, G_TYPE_FLAG_ABSTRACT); } return base_parse_type; } static void gst_base_parse_finalize (GObject * object); static gboolean gst_base_parse_sink_activate (GstPad * sinkpad); static gboolean gst_base_parse_sink_activate_push (GstPad * pad, gboolean active); static gboolean gst_base_parse_sink_activate_pull (GstPad * pad, gboolean active); static gboolean gst_base_parse_handle_seek (GstBaseParse * parse, GstEvent * event); static GstStateChangeReturn gst_base_parse_change_state (GstElement * element, GstStateChange transition); #ifndef GST_DISABLE_INDEX static void gst_base_parse_set_index (GstElement * element, GstIndex * index); static GstIndex *gst_base_parse_get_index (GstElement * element); #endif static gboolean gst_base_parse_src_event (GstPad * pad, GstEvent * event); static gboolean gst_base_parse_sink_event (GstPad * pad, GstEvent * event); static gboolean gst_base_parse_query (GstPad * pad, GstQuery * query); static gboolean gst_base_parse_sink_setcaps (GstPad * pad, GstCaps * caps); static const GstQueryType *gst_base_parse_get_querytypes (GstPad * pad); static GstFlowReturn gst_base_parse_chain (GstPad * pad, GstBuffer * buffer); static void gst_base_parse_loop (GstPad * pad); static gboolean gst_base_parse_sink_eventfunc (GstBaseParse * parse, GstEvent * event); static gboolean gst_base_parse_src_eventfunc (GstBaseParse * parse, GstEvent * event); static gboolean gst_base_parse_is_seekable (GstBaseParse * parse); static gboolean gst_base_parse_convert (GstBaseParse * parse, GstFormat src_format, gint64 src_value, GstFormat dest_format, gint64 * dest_value); static gboolean gst_base_parse_frame_in_segment (GstBaseParse * parse, GstBuffer * buffer, GstSegment * segment); static void gst_base_parse_drain (GstBaseParse * parse); static void gst_base_parse_base_init (gpointer g_class) { GstBaseParseClass *klass = GST_BASE_PARSE_CLASS (g_class); GstBaseParseClassPrivate *priv; GST_DEBUG_CATEGORY_INIT (gst_base_parse_debug, "flacbaseparse", 0, "baseparse element"); /* TODO: Remove this once GObject supports class private data */ priv = g_slice_new0 (GstBaseParseClassPrivate); if (klass->priv) memcpy (priv, klass->priv, sizeof (GstBaseParseClassPrivate)); klass->priv = priv; } static void gst_base_parse_base_finalize (gpointer g_class) { GstBaseParseClass *klass = GST_BASE_PARSE_CLASS (g_class); g_slice_free (GstBaseParseClassPrivate, klass->priv); klass->priv = NULL; } static void gst_base_parse_finalize (GObject * object) { GstBaseParse *parse = GST_BASE_PARSE (object); g_mutex_free (parse->parse_lock); g_object_unref (parse->priv->adapter); if (parse->pending_segment) { gst_event_replace (&parse->pending_segment, NULL); } if (parse->close_segment) { gst_event_replace (&parse->close_segment, NULL); } if (parse->priv->cache) { gst_buffer_unref (parse->priv->cache); parse->priv->cache = NULL; } g_list_foreach (parse->priv->pending_events, (GFunc) gst_mini_object_unref, NULL); g_list_free (parse->priv->pending_events); parse->priv->pending_events = NULL; #ifndef GST_DISABLE_INDEX gst_base_parse_set_index (GST_ELEMENT (object), NULL); #endif G_OBJECT_CLASS (parent_class)->finalize (object); } static void gst_base_parse_class_init (GstBaseParseClass * klass) { GObjectClass *gobject_class; GstElementClass *gstelement_class; gobject_class = G_OBJECT_CLASS (klass); gstelement_class = GST_ELEMENT_CLASS (klass); g_type_class_add_private (klass, sizeof (GstBaseParsePrivate)); parent_class = g_type_class_peek_parent (klass); gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_base_parse_finalize); gstelement_class->change_state = gst_base_parse_change_state; #ifndef GST_DISABLE_INDEX gstelement_class->set_index = gst_base_parse_set_index; gstelement_class->get_index = gst_base_parse_get_index; #endif /* Default handlers */ klass->frame_in_segment = gst_base_parse_frame_in_segment; klass->event = gst_base_parse_sink_eventfunc; klass->src_event = gst_base_parse_src_eventfunc; klass->is_seekable = gst_base_parse_is_seekable; klass->convert = gst_base_parse_convert; } static void gst_base_parse_init (GstBaseParse * parse, GstBaseParseClass * bclass) { GstPadTemplate *pad_template; GST_DEBUG_OBJECT (parse, "gst_base_parse_init"); parse->priv = GST_BASE_PARSE_GET_PRIVATE (parse); pad_template = gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass), "sink"); g_return_if_fail (pad_template != NULL); parse->sinkpad = gst_pad_new_from_template (pad_template, "sink"); gst_pad_set_event_function (parse->sinkpad, GST_DEBUG_FUNCPTR (gst_base_parse_sink_event)); gst_pad_set_setcaps_function (parse->sinkpad, GST_DEBUG_FUNCPTR (gst_base_parse_sink_setcaps)); gst_pad_set_chain_function (parse->sinkpad, GST_DEBUG_FUNCPTR (gst_base_parse_chain)); gst_pad_set_activate_function (parse->sinkpad, GST_DEBUG_FUNCPTR (gst_base_parse_sink_activate)); gst_pad_set_activatepush_function (parse->sinkpad, GST_DEBUG_FUNCPTR (gst_base_parse_sink_activate_push)); gst_pad_set_activatepull_function (parse->sinkpad, GST_DEBUG_FUNCPTR (gst_base_parse_sink_activate_pull)); gst_element_add_pad (GST_ELEMENT (parse), parse->sinkpad); GST_DEBUG_OBJECT (parse, "sinkpad created"); pad_template = gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass), "src"); g_return_if_fail (pad_template != NULL); parse->srcpad = gst_pad_new_from_template (pad_template, "src"); gst_pad_set_event_function (parse->srcpad, GST_DEBUG_FUNCPTR (gst_base_parse_src_event)); gst_pad_set_query_type_function (parse->srcpad, GST_DEBUG_FUNCPTR (gst_base_parse_get_querytypes)); gst_pad_set_query_function (parse->srcpad, GST_DEBUG_FUNCPTR (gst_base_parse_query)); gst_element_add_pad (GST_ELEMENT (parse), parse->srcpad); GST_DEBUG_OBJECT (parse, "src created"); parse->parse_lock = g_mutex_new (); parse->priv->adapter = gst_adapter_new (); parse->pending_segment = NULL; parse->close_segment = NULL; parse->priv->pad_mode = GST_ACTIVATE_NONE; parse->priv->duration = -1; parse->priv->min_frame_size = 1; parse->priv->discont = FALSE; parse->priv->flushing = FALSE; parse->priv->offset = 0; parse->priv->avg_bitrate = 0; parse->priv->upstream_duration = -1; parse->priv->upstream_size = -1; parse->priv->estimated_size = -1; parse->priv->estimated_duration = -1; GST_DEBUG_OBJECT (parse, "init ok"); } /** * gst_base_parse_bytepos_to_time: * @parse: #GstBaseParse. * @bytepos: Position (in bytes) to be converted. * @pos_in_time: #GstClockTime pointer where the result is set. * * Convert given byte position into #GstClockTime format. * * Returns: TRUE if conversion succeeded. */ static gboolean gst_base_parse_bytepos_to_time (GstBaseParse * parse, gint64 bytepos, GstClockTime * pos_in_time) { GstBaseParseClass *klass; gboolean res = FALSE; klass = GST_BASE_PARSE_GET_CLASS (parse); if (klass->convert) { res = klass->convert (parse, GST_FORMAT_BYTES, bytepos, GST_FORMAT_TIME, (gint64 *) pos_in_time); } return res; } /** * gst_base_parse_sink_event: * @pad: #GstPad that received the event. * @event: #GstEvent to be handled. * * Handler for sink pad events. * * Returns: TRUE if the event was handled. */ static gboolean gst_base_parse_sink_event (GstPad * pad, GstEvent * event) { GstBaseParse *parse; GstBaseParseClass *bclass; gboolean handled = FALSE; gboolean ret = TRUE; parse = GST_BASE_PARSE (gst_pad_get_parent (pad)); bclass = GST_BASE_PARSE_GET_CLASS (parse); GST_DEBUG_OBJECT (parse, "handling event %d", GST_EVENT_TYPE (event)); /* Cache all events except EOS, NEWSEGMENT and FLUSH_STOP if we have a * pending segment */ if (parse->pending_segment && GST_EVENT_TYPE (event) != GST_EVENT_EOS && GST_EVENT_TYPE (event) != GST_EVENT_NEWSEGMENT && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_START && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP) { parse->priv->pending_events = g_list_append (parse->priv->pending_events, event); ret = TRUE; } else { if (bclass->event) handled = bclass->event (parse, event); if (!handled) ret = gst_pad_event_default (pad, event); } gst_object_unref (parse); GST_DEBUG_OBJECT (parse, "event handled"); return ret; } /** * gst_base_parse_sink_eventfunc: * @parse: #GstBaseParse. * @event: #GstEvent to be handled. * * Element-level event handler function. * * Returns: TRUE if the event was handled and not need forwarding. */ static gboolean gst_base_parse_sink_eventfunc (GstBaseParse * parse, GstEvent * event) { gboolean handled = FALSE; GstEvent **eventp; switch (GST_EVENT_TYPE (event)) { case GST_EVENT_NEWSEGMENT: { gdouble rate, applied_rate; GstFormat format; gint64 start, stop, pos; gboolean update; gst_event_parse_new_segment_full (event, &update, &rate, &applied_rate, &format, &start, &stop, &pos); if (format == GST_FORMAT_BYTES) { GstClockTime seg_start, seg_stop, seg_pos; /* stop time is allowed to be open-ended, but not start & pos */ seg_stop = GST_CLOCK_TIME_NONE; parse->priv->pending_offset = pos; if (gst_base_parse_bytepos_to_time (parse, start, &seg_start) && gst_base_parse_bytepos_to_time (parse, pos, &seg_pos)) { gst_event_unref (event); event = gst_event_new_new_segment_full (update, rate, applied_rate, GST_FORMAT_TIME, seg_start, seg_stop, seg_pos); format = GST_FORMAT_TIME; GST_DEBUG_OBJECT (parse, "Converted incoming segment to TIME. " "start = %" GST_TIME_FORMAT ", stop = %" GST_TIME_FORMAT ", pos = %" GST_TIME_FORMAT, GST_TIME_ARGS (seg_start), GST_TIME_ARGS (seg_stop), GST_TIME_ARGS (seg_pos)); } } if (format != GST_FORMAT_TIME) { /* Unknown incoming segment format. Output a default open-ended * TIME segment */ gst_event_unref (event); event = gst_event_new_new_segment_full (update, rate, applied_rate, GST_FORMAT_TIME, 0, GST_CLOCK_TIME_NONE, 0); } gst_event_parse_new_segment_full (event, &update, &rate, &applied_rate, &format, &start, &stop, &pos); gst_segment_set_newsegment_full (&parse->segment, update, rate, applied_rate, format, start, stop, pos); GST_DEBUG_OBJECT (parse, "Created newseg rate %g, applied rate %g, " "format %d, start = %" GST_TIME_FORMAT ", stop = %" GST_TIME_FORMAT ", pos = %" GST_TIME_FORMAT, rate, applied_rate, format, GST_TIME_ARGS (start), GST_TIME_ARGS (stop), GST_TIME_ARGS (pos)); /* save the segment for later, right before we push a new buffer so that * the caps are fixed and the next linked element can receive * the segment. */ eventp = &parse->pending_segment; gst_event_replace (eventp, event); gst_event_unref (event); handled = TRUE; break; } case GST_EVENT_FLUSH_START: parse->priv->flushing = TRUE; /* Wait for _chain() to exit by taking the srcpad STREAM_LOCK */ GST_PAD_STREAM_LOCK (parse->srcpad); handled = gst_pad_push_event (parse->srcpad, event); GST_PAD_STREAM_UNLOCK (parse->srcpad); break; case GST_EVENT_FLUSH_STOP: gst_adapter_clear (parse->priv->adapter); parse->priv->flushing = FALSE; parse->priv->discont = TRUE; break; case GST_EVENT_EOS: gst_base_parse_drain (parse); break; default: break; } return handled; } /** * gst_base_parse_src_event: * @pad: #GstPad that received the event. * @event: #GstEvent that was received. * * Handler for source pad events. * * Returns: TRUE if the event was handled. */ static gboolean gst_base_parse_src_event (GstPad * pad, GstEvent * event) { GstBaseParse *parse; GstBaseParseClass *bclass; gboolean handled = FALSE; gboolean ret = TRUE; parse = GST_BASE_PARSE (gst_pad_get_parent (pad)); bclass = GST_BASE_PARSE_GET_CLASS (parse); GST_DEBUG_OBJECT (parse, "event %d, %s", GST_EVENT_TYPE (event), GST_EVENT_TYPE_NAME (event)); if (bclass->src_event) handled = bclass->src_event (parse, event); if (!handled) ret = gst_pad_event_default (pad, event); gst_object_unref (parse); return ret; } /** * gst_base_parse_src_eventfunc: * @parse: #GstBaseParse. * @event: #GstEvent that was received. * * Default srcpad event handler. * * Returns: TRUE if the event was handled and can be dropped. */ static gboolean gst_base_parse_src_eventfunc (GstBaseParse * parse, GstEvent * event) { gboolean handled = FALSE; GstBaseParseClass *bclass; bclass = GST_BASE_PARSE_GET_CLASS (parse); switch (GST_EVENT_TYPE (event)) { case GST_EVENT_SEEK: { if (bclass->is_seekable (parse)) { handled = gst_base_parse_handle_seek (parse, event); gst_event_unref (event); } break; } default: break; } return handled; } /** * gst_base_parse_is_seekable: * @parse: #GstBaseParse. * * Default handler for is_seekable. * * Returns: Always TRUE. */ static gboolean gst_base_parse_is_seekable (GstBaseParse * parse) { return TRUE; } /** * gst_base_parse_push_buffer: * @parse: #GstBaseParse. * @buffer: #GstBuffer. * * Pushes @buffer on the srcpad and updates the internal state. * * This must be called with srcpad STREAM_LOCK held. * * Returns: #GstFlowReturn */ GstFlowReturn gst_base_parse_push_buffer (GstBaseParse * parse, GstBuffer * buffer) { GstClockTime last_stop = GST_CLOCK_TIME_NONE; GstFlowReturn ret = GST_FLOW_OK; GstBaseParseClass *klass = GST_BASE_PARSE_GET_CLASS (parse); g_return_val_if_fail (GST_PAD_CAPS (parse->srcpad), GST_FLOW_ERROR); if (parse->priv->pad_mode == GST_ACTIVATE_PULL) { if (parse->close_segment) { GST_DEBUG_OBJECT (parse, "loop sending close segment"); gst_pad_push_event (parse->srcpad, parse->close_segment); parse->close_segment = NULL; } if (parse->pending_segment) { GST_DEBUG_OBJECT (parse, "loop push pending segment"); gst_pad_push_event (parse->srcpad, parse->pending_segment); parse->pending_segment = NULL; } } else { if (parse->pending_segment) { GST_DEBUG_OBJECT (parse, "chain pushing a pending segment"); gst_pad_push_event (parse->srcpad, parse->pending_segment); parse->pending_segment = NULL; } } if (parse->priv->pending_events) { GList *l; for (l = parse->priv->pending_events; l != NULL; l = l->next) { gst_pad_push_event (parse->srcpad, GST_EVENT (l->data)); } g_list_free (parse->priv->pending_events); parse->priv->pending_events = NULL; } if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer)) last_stop = GST_BUFFER_TIMESTAMP (buffer); if (last_stop != GST_CLOCK_TIME_NONE && GST_BUFFER_DURATION_IS_VALID (buffer)) last_stop += GST_BUFFER_DURATION (buffer); #ifndef GST_DISABLE_INDEX if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer)) { gint64 offset, timestamp; g_assert (parse->priv->offset >= GST_BUFFER_SIZE (buffer)); offset = parse->priv->offset - GST_BUFFER_SIZE (buffer); timestamp = GST_BUFFER_TIMESTAMP (buffer); gst_index_add_association (parse->priv->index, parse->priv->index_id, GST_ASSOCIATION_FLAG_NONE, GST_FORMAT_BYTES, offset, GST_FORMAT_TIME, timestamp, NULL); } #endif gst_buffer_set_caps (buffer, GST_PAD_CAPS (parse->srcpad)); if (!klass->frame_in_segment (parse, buffer, &parse->segment)) { GST_LOG_OBJECT (parse, "Dropped frame, outside configured segment"); gst_buffer_unref (buffer); } else { guint size = GST_BUFFER_SIZE (buffer); ret = gst_pad_push (parse->srcpad, buffer); GST_LOG_OBJECT (parse, "frame (%d bytes) pushed: %s", size, gst_flow_get_name (ret)); } /* Update current running segment position */ if (ret == GST_FLOW_OK && last_stop != GST_CLOCK_TIME_NONE) gst_segment_set_last_stop (&parse->segment, GST_FORMAT_TIME, last_stop); if (parse->priv->upstream_size && (!parse->priv->upstream_duration && parse->priv->duration == -1)) { parse->priv->avg_bitrate = gst_util_uint64_scale (parse->priv->offset, 8 * GST_SECOND, parse->segment.last_stop); GST_DEBUG_OBJECT (parse, "Approximate average bitrate: %" G_GUINT64_FORMAT " bps", parse->priv->avg_bitrate); parse->priv->estimated_duration = gst_util_uint64_scale (parse->priv->upstream_size, 8 * GST_SECOND, parse->priv->avg_bitrate); GST_DEBUG_OBJECT (parse, "Estimated duration: %" GST_TIME_FORMAT, parse->priv->estimated_duration); } else if (!parse->priv->upstream_size && parse->priv->upstream_duration) { parse->priv->avg_bitrate = gst_util_uint64_scale (parse->priv->offset, 8 * GST_SECOND, parse->segment.last_stop); GST_DEBUG_OBJECT (parse, "Approximate average bitrate: %" G_GUINT64_FORMAT " bps", parse->priv->avg_bitrate); parse->priv->estimated_size = gst_util_uint64_scale (parse->priv->upstream_duration, parse->priv->avg_bitrate, 8 * GST_SECOND); GST_DEBUG_OBJECT (parse, "Estimated size: %" G_GUINT64_FORMAT, parse->priv->estimated_size); } else if (!parse->priv->upstream_size && parse->priv->duration != -1) { parse->priv->avg_bitrate = gst_util_uint64_scale (parse->priv->offset, 8 * GST_SECOND, parse->segment.last_stop); GST_DEBUG_OBJECT (parse, "Approximate average bitrate: %" G_GUINT64_FORMAT " bps", parse->priv->avg_bitrate); parse->priv->estimated_size = gst_util_uint64_scale (parse->priv->duration, parse->priv->avg_bitrate, 8 * GST_SECOND); GST_DEBUG_OBJECT (parse, "Estimated size: %" G_GUINT64_FORMAT, parse->priv->estimated_size); } return ret; } /** * gst_base_parse_handle_and_push_buffer: * @parse: #GstBaseParse. * @klass: #GstBaseParseClass. * @buffer: #GstBuffer. * * Parses the frame from given buffer and pushes it forward. Also performs * timestamp handling and checks the segment limits. * * This is called with srcpad STREAM_LOCK held. * * Returns: #GstFlowReturn */ static GstFlowReturn gst_base_parse_handle_and_push_buffer (GstBaseParse * parse, GstBaseParseClass * klass, GstBuffer * buffer) { GstFlowReturn ret; if (parse->priv->discont) { GST_DEBUG_OBJECT (parse, "marking DISCONT"); GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT); parse->priv->discont = FALSE; } /* FIXME: Set upstream timestamp if possible */ ret = klass->parse_frame (parse, buffer); /* convert internal flow to OK and mark discont for the next buffer. */ if (ret == GST_BASE_PARSE_FLOW_DROPPED) { parse->priv->discont = TRUE; ret = GST_FLOW_OK; return ret; } g_return_val_if_fail (GST_PAD_CAPS (parse->srcpad), GST_FLOW_ERROR); ret = gst_base_parse_push_buffer (parse, buffer); return ret; } /** * gst_base_parse_drain: * @parse: #GstBaseParse. * * Drains the adapter until it is empty. It decreases the min_frame_size to * match the current adapter size and calls chain method until the adapter * is emptied or chain returns with error. */ static void gst_base_parse_drain (GstBaseParse * parse) { guint avail; for (;;) { avail = gst_adapter_available (parse->priv->adapter); if (!avail) break; gst_base_parse_set_min_frame_size (parse, avail); if (gst_base_parse_chain (parse->sinkpad, NULL) != GST_FLOW_OK) { break; } } } static void gst_base_parse_update_upstream_durations (GstBaseParse * parse) { /* Get/update upstream size and duration if possible */ if (parse->priv->upstream_duration == -1 || (parse->priv->upstream_duration > 0 && parse->segment.last_stop > parse->priv->upstream_duration)) { GstFormat fmt = GST_FORMAT_TIME; gint64 duration; if (gst_pad_query_peer_duration (parse->sinkpad, &fmt, &duration) && fmt == GST_FORMAT_TIME && duration != -1) { parse->priv->upstream_duration = duration; GST_DEBUG_OBJECT (parse, "Upstream duration: %" GST_TIME_FORMAT, parse->priv->upstream_duration); } else { GST_DEBUG_OBJECT (parse, "Failed to get upstream duration"); parse->priv->upstream_duration = 0; } } if (parse->priv->upstream_size == -1 || (parse->priv->upstream_size > 0 && parse->priv->offset > parse->priv->upstream_size)) { GstFormat fmt = GST_FORMAT_BYTES; gint64 duration; if (gst_pad_query_peer_duration (parse->sinkpad, &fmt, &duration) && fmt == GST_FORMAT_BYTES && duration != -1) { parse->priv->upstream_size = duration; GST_DEBUG_OBJECT (parse, "Upstream size: %" G_GUINT64_FORMAT " bytes", parse->priv->upstream_size); } else { GST_DEBUG_OBJECT (parse, "Failed to get upstream size in bytes"); parse->priv->upstream_size = 0; } } if (parse->priv->upstream_size && parse->priv->upstream_duration) { parse->priv->avg_bitrate = gst_util_uint64_scale (parse->priv->upstream_size, 8 * GST_SECOND, parse->priv->upstream_duration); GST_DEBUG_OBJECT (parse, "Approximate average bitrate: %" G_GUINT64_FORMAT " bps", parse->priv->avg_bitrate); } else if (parse->priv->upstream_size && parse->priv->duration != -1) { parse->priv->avg_bitrate = gst_util_uint64_scale (parse->priv->upstream_size, 8 * GST_SECOND, parse->priv->duration); GST_DEBUG_OBJECT (parse, "Approximate average bitrate: %" G_GUINT64_FORMAT " bps", parse->priv->avg_bitrate); } } /** * gst_base_parse_chain: * @pad: #GstPad. * @buffer: #GstBuffer. * * Returns: #GstFlowReturn. */ static GstFlowReturn gst_base_parse_chain (GstPad * pad, GstBuffer * buffer) { GstBaseParseClass *bclass; GstBaseParse *parse; GstFlowReturn ret = GST_FLOW_OK; GstBuffer *outbuf = NULL; GstBuffer *tmpbuf = NULL; guint fsize = 0; gint skip = -1; const guint8 *data; guint min_size; parse = GST_BASE_PARSE (GST_OBJECT_PARENT (pad)); bclass = GST_BASE_PARSE_GET_CLASS (parse); /* Make sure that adapter doesn't have any old data after newsegment has been received and update our offset */ if (parse->pending_segment) { gst_adapter_clear (parse->priv->adapter); parse->priv->offset = parse->priv->pending_offset; } gst_base_parse_update_upstream_durations (parse); if (buffer) { GST_LOG_OBJECT (parse, "buffer size: %d, offset = %lld", GST_BUFFER_SIZE (buffer), GST_BUFFER_OFFSET (buffer)); gst_adapter_push (parse->priv->adapter, buffer); } /* Parse and push as many frames as possible */ /* Stop either when adapter is empty or we are flushing */ while (!parse->priv->flushing) { tmpbuf = gst_buffer_new (); /* Synchronization loop */ for (;;) { GST_BASE_PARSE_LOCK (parse); min_size = parse->priv->min_frame_size; GST_BASE_PARSE_UNLOCK (parse); /* Collect at least min_frame_size bytes */ if (gst_adapter_available (parse->priv->adapter) < min_size) { GST_DEBUG_OBJECT (parse, "not enough data available (only %d bytes)", gst_adapter_available (parse->priv->adapter)); gst_buffer_unref (tmpbuf); goto done; } data = gst_adapter_peek (parse->priv->adapter, min_size); GST_BUFFER_DATA (tmpbuf) = (guint8 *) data; GST_BUFFER_SIZE (tmpbuf) = min_size; GST_BUFFER_OFFSET (tmpbuf) = parse->priv->offset; GST_BUFFER_FLAG_SET (tmpbuf, GST_MINI_OBJECT_FLAG_READONLY); if (parse->priv->discont) { GST_DEBUG_OBJECT (parse, "marking DISCONT"); GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT); } skip = -1; if (bclass->check_valid_frame (parse, tmpbuf, &fsize, &skip)) { break; } if (skip > 0) { GST_LOG_OBJECT (parse, "finding sync, skipping %d bytes", skip); gst_adapter_flush (parse->priv->adapter, skip); parse->priv->offset += skip; } else if (skip == -1) { /* subclass didn't touch this value. By default we skip 1 byte */ GST_LOG_OBJECT (parse, "finding sync, skipping 1 byte"); gst_adapter_flush (parse->priv->adapter, 1); parse->priv->offset++; } /* There is a possibility that subclass set the skip value to zero. This means that it has probably found a frame but wants to ask more data (by increasing the min_size) to be sure of this. */ } gst_buffer_unref (tmpbuf); tmpbuf = NULL; if (skip > 0) { /* Subclass found the sync, but still wants to skip some data */ GST_LOG_OBJECT (parse, "skipping %d bytes", skip); gst_adapter_flush (parse->priv->adapter, skip); parse->priv->offset += skip; } /* Grab lock to prevent a race with FLUSH_START handler */ GST_PAD_STREAM_LOCK (parse->srcpad); /* FLUSH_START event causes the "flushing" flag to be set. In this * case we can leave the frame pushing loop */ if (parse->priv->flushing) { GST_PAD_STREAM_UNLOCK (parse->srcpad); break; } /* FIXME: Would it be more efficient to make a subbuffer instead? */ outbuf = gst_adapter_take_buffer (parse->priv->adapter, fsize); /* Subclass may want to know the data offset */ GST_BUFFER_OFFSET (outbuf) = parse->priv->offset; parse->priv->offset += fsize; ret = gst_base_parse_handle_and_push_buffer (parse, bclass, outbuf); GST_PAD_STREAM_UNLOCK (parse->srcpad); if (ret != GST_FLOW_OK) { GST_LOG_OBJECT (parse, "push returned %d", ret); break; } } done: GST_LOG_OBJECT (parse, "chain leaving"); return ret; } static GstFlowReturn gst_base_parse_pull_range (GstBaseParse * parse, guint size, GstBuffer ** buffer) { GstFlowReturn ret = GST_FLOW_OK; g_return_val_if_fail (buffer != NULL, GST_FLOW_ERROR); /* Caching here actually makes much less difference than one would expect. * We do it mainly to avoid pulling buffers of 1 byte all the time */ if (parse->priv->cache) { guint64 cache_offset = GST_BUFFER_OFFSET (parse->priv->cache); guint cache_size = GST_BUFFER_SIZE (parse->priv->cache); if (cache_offset <= parse->priv->offset && (parse->priv->offset + size) < (cache_offset + cache_size)) { *buffer = gst_buffer_create_sub (parse->priv->cache, parse->priv->offset - cache_offset, size); GST_BUFFER_OFFSET (*buffer) = parse->priv->offset; return GST_FLOW_OK; } /* not enough data in the cache, free cache and get a new one */ gst_buffer_unref (parse->priv->cache); parse->priv->cache = NULL; } /* refill the cache */ ret = gst_pad_pull_range (parse->sinkpad, parse->priv->offset, MAX (size, 64 * 1024), &parse->priv->cache); if (ret != GST_FLOW_OK) { parse->priv->cache = NULL; return ret; } if (GST_BUFFER_SIZE (parse->priv->cache) >= size) { *buffer = gst_buffer_create_sub (parse->priv->cache, 0, size); GST_BUFFER_OFFSET (*buffer) = parse->priv->offset; return GST_FLOW_OK; } /* Not possible to get enough data, try a last time with * requesting exactly the size we need */ gst_buffer_unref (parse->priv->cache); parse->priv->cache = NULL; ret = gst_pad_pull_range (parse->sinkpad, parse->priv->offset, size, &parse->priv->cache); if (ret != GST_FLOW_OK) { GST_DEBUG_OBJECT (parse, "pull_range returned %d", ret); *buffer = NULL; return ret; } if (GST_BUFFER_SIZE (parse->priv->cache) < size) { GST_WARNING_OBJECT (parse, "Dropping short buffer at offset %" G_GUINT64_FORMAT ": wanted %u bytes, got %u bytes", parse->priv->offset, size, GST_BUFFER_SIZE (parse->priv->cache)); gst_buffer_unref (parse->priv->cache); parse->priv->cache = NULL; *buffer = NULL; return GST_FLOW_UNEXPECTED; } *buffer = gst_buffer_create_sub (parse->priv->cache, 0, size); GST_BUFFER_OFFSET (*buffer) = parse->priv->offset; return GST_FLOW_OK; } /** * gst_base_parse_loop: * @pad: GstPad * * Loop that is used in pull mode to retrieve data from upstream. */ static void gst_base_parse_loop (GstPad * pad) { GstBaseParse *parse; GstBaseParseClass *klass; GstBuffer *buffer, *outbuf; gboolean ret = FALSE; guint fsize = 0, min_size; gint skip = 0; parse = GST_BASE_PARSE (gst_pad_get_parent (pad)); klass = GST_BASE_PARSE_GET_CLASS (parse); gst_base_parse_update_upstream_durations (parse); /* TODO: Check if we reach segment stop limit */ while (TRUE) { GST_BASE_PARSE_LOCK (parse); min_size = parse->priv->min_frame_size; GST_BASE_PARSE_UNLOCK (parse); ret = gst_base_parse_pull_range (parse, min_size, &buffer); if (ret == GST_FLOW_UNEXPECTED) goto eos; else if (ret != GST_FLOW_OK) goto need_pause; if (parse->priv->discont) { GST_DEBUG_OBJECT (parse, "marking DISCONT"); GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT); } skip = -1; if (klass->check_valid_frame (parse, buffer, &fsize, &skip)) { break; } if (skip > 0) { GST_LOG_OBJECT (parse, "finding sync, skipping %d bytes", skip); parse->priv->offset += skip; } else if (skip == -1) { GST_LOG_OBJECT (parse, "finding sync, skipping 1 byte"); parse->priv->offset++; } GST_DEBUG_OBJECT (parse, "finding sync..."); gst_buffer_unref (buffer); } if (fsize <= GST_BUFFER_SIZE (buffer)) { outbuf = gst_buffer_create_sub (buffer, 0, fsize); GST_BUFFER_OFFSET (outbuf) = GST_BUFFER_OFFSET (buffer); gst_buffer_unref (buffer); } else { gst_buffer_unref (buffer); ret = gst_base_parse_pull_range (parse, fsize, &outbuf); if (ret == GST_FLOW_UNEXPECTED) goto eos; else if (ret != GST_FLOW_OK) goto need_pause; } parse->priv->offset += fsize; /* Does the subclass want to skip too? */ if (skip > 0) parse->priv->offset += skip; /* This always unrefs the outbuf, even if error occurs */ ret = gst_base_parse_handle_and_push_buffer (parse, klass, outbuf); if (ret != GST_FLOW_OK) { GST_DEBUG_OBJECT (parse, "flow: %s", gst_flow_get_name (ret)); if (GST_FLOW_IS_FATAL (ret)) { GST_ELEMENT_ERROR (parse, STREAM, FAILED, (NULL), ("streaming task paused, reason: %s", gst_flow_get_name (ret))); gst_pad_push_event (parse->srcpad, gst_event_new_eos ()); } goto need_pause; } gst_object_unref (parse); return; need_pause: { GST_LOG_OBJECT (parse, "pausing task"); gst_pad_pause_task (pad); gst_object_unref (parse); return; } eos: { GST_LOG_OBJECT (parse, "pausing task %d", ret); gst_pad_push_event (parse->srcpad, gst_event_new_eos ()); gst_pad_pause_task (pad); gst_object_unref (parse); return; } } /** * gst_base_parse_sink_activate: * @sinkpad: #GstPad to be activated. * * Returns: TRUE if activation succeeded. */ static gboolean gst_base_parse_sink_activate (GstPad * sinkpad) { GstBaseParse *parse; gboolean result = TRUE; parse = GST_BASE_PARSE (gst_pad_get_parent (sinkpad)); GST_DEBUG_OBJECT (parse, "sink activate"); if (gst_pad_check_pull_range (sinkpad)) { GST_DEBUG_OBJECT (parse, "trying to activate in pull mode"); result = gst_pad_activate_pull (sinkpad, TRUE); } else { GST_DEBUG_OBJECT (parse, "trying to activate in push mode"); result = gst_pad_activate_push (sinkpad, TRUE); } GST_DEBUG_OBJECT (parse, "sink activate return %d", result); gst_object_unref (parse); return result; } /** * gst_base_parse_activate: * @parse: #GstBaseParse. * @active: TRUE if element will be activated, FALSE if disactivated. * * Returns: TRUE if the operation succeeded. */ static gboolean gst_base_parse_activate (GstBaseParse * parse, gboolean active) { GstBaseParseClass *klass; gboolean result = FALSE; GST_DEBUG_OBJECT (parse, "activate"); klass = GST_BASE_PARSE_GET_CLASS (parse); if (active) { if (parse->priv->pad_mode == GST_ACTIVATE_NONE && klass->start) result = klass->start (parse); GST_OBJECT_LOCK (parse); gst_segment_init (&parse->segment, GST_FORMAT_TIME); parse->priv->duration = -1; parse->priv->discont = FALSE; parse->priv->flushing = FALSE; parse->priv->offset = 0; parse->priv->avg_bitrate = 0; parse->priv->upstream_duration = -1; parse->priv->upstream_size = -1; parse->priv->estimated_size = -1; parse->priv->estimated_duration = -1; if (parse->pending_segment) gst_event_unref (parse->pending_segment); parse->pending_segment = gst_event_new_new_segment (FALSE, parse->segment.rate, parse->segment.format, parse->segment.start, parse->segment.stop, parse->segment.last_stop); GST_OBJECT_UNLOCK (parse); } else { /* We must make sure streaming has finished before resetting things * and calling the ::stop vfunc */ GST_PAD_STREAM_LOCK (parse->sinkpad); GST_PAD_STREAM_UNLOCK (parse->sinkpad); if (parse->priv->pad_mode != GST_ACTIVATE_NONE && klass->stop) result = klass->stop (parse); g_list_foreach (parse->priv->pending_events, (GFunc) gst_mini_object_unref, NULL); g_list_free (parse->priv->pending_events); parse->priv->pending_events = NULL; if (parse->priv->cache) { gst_buffer_unref (parse->priv->cache); parse->priv->cache = NULL; } parse->priv->pad_mode = GST_ACTIVATE_NONE; } GST_DEBUG_OBJECT (parse, "activate: %d", result); return result; } /** * gst_base_parse_sink_activate_push: * @pad: #GstPad to be (de)activated. * @active: TRUE when activating, FALSE when deactivating. * * Returns: TRUE if (de)activation succeeded. */ static gboolean gst_base_parse_sink_activate_push (GstPad * pad, gboolean active) { gboolean result = TRUE; GstBaseParse *parse; parse = GST_BASE_PARSE (gst_pad_get_parent (pad)); GST_DEBUG_OBJECT (parse, "sink activate push"); result = gst_base_parse_activate (parse, active); if (result) parse->priv->pad_mode = active ? GST_ACTIVATE_PUSH : GST_ACTIVATE_NONE; GST_DEBUG_OBJECT (parse, "sink activate push: %d", result); gst_object_unref (parse); return result; } /** * gst_base_parse_sink_activate_pull: * @sinkpad: #GstPad to be (de)activated. * @active: TRUE when activating, FALSE when deactivating. * * Returns: TRUE if (de)activation succeeded. */ static gboolean gst_base_parse_sink_activate_pull (GstPad * sinkpad, gboolean active) { gboolean result = FALSE; GstBaseParse *parse; parse = GST_BASE_PARSE (gst_pad_get_parent (sinkpad)); GST_DEBUG_OBJECT (parse, "activate pull"); result = gst_base_parse_activate (parse, active); if (result) { if (active) { result &= gst_pad_start_task (sinkpad, (GstTaskFunction) gst_base_parse_loop, sinkpad); } else { result &= gst_pad_stop_task (sinkpad); } } if (result) parse->priv->pad_mode = active ? GST_ACTIVATE_PULL : GST_ACTIVATE_NONE; GST_DEBUG_OBJECT (parse, "sink activate pull: %d", result); gst_object_unref (parse); return result; } static gboolean gst_base_parse_convert (GstBaseParse * parse, GstFormat src_format, gint64 src_value, GstFormat dest_format, gint64 * dest_value) { gboolean res = FALSE; if (G_UNLIKELY (src_format == dest_format)) { *dest_value = src_value; return TRUE; } if (src_value == -1) { *dest_value = -1; return TRUE; } /* TODO: Use seek table for accurate conversions */ if (parse->priv->avg_bitrate && src_format == GST_FORMAT_BYTES && dest_format == GST_FORMAT_TIME) { *dest_value = gst_util_uint64_scale (src_value, 8 * GST_SECOND, parse->priv->avg_bitrate); res = TRUE; } else if (parse->priv->avg_bitrate && src_format == GST_FORMAT_TIME && dest_format == GST_FORMAT_BYTES) { *dest_value = gst_util_uint64_scale (src_value, parse->priv->avg_bitrate, 8 * GST_SECOND); res = TRUE; } else { res = FALSE; } return res; } /** * gst_base_parse_set_duration: * @parse: #GstBaseParse. * @fmt: #GstFormat. * @duration: duration value. * * Sets the duration of the currently playing media. Subclass can use this * when it notices a change in the media duration. */ void gst_base_parse_set_duration (GstBaseParse * parse, GstFormat fmt, gint64 duration) { g_return_if_fail (parse != NULL); GST_BASE_PARSE_LOCK (parse); if (duration != parse->priv->duration) { GstMessage *m; m = gst_message_new_duration (GST_OBJECT (parse), fmt, duration); gst_element_post_message (GST_ELEMENT (parse), m); /* TODO: what about duration tag? */ } parse->priv->duration = duration; parse->priv->duration_fmt = fmt; GST_DEBUG_OBJECT (parse, "set duration: %lld", duration); GST_BASE_PARSE_UNLOCK (parse); } /** * gst_base_parse_set_min_frame_size: * @parse: #GstBaseParse. * @min_size: Minimum size of the data that this base class should give to * subclass. * * Subclass can use this function to tell the base class that it needs to * give at least #min_size buffers. */ void gst_base_parse_set_min_frame_size (GstBaseParse * parse, guint min_size) { g_return_if_fail (parse != NULL); GST_BASE_PARSE_LOCK (parse); parse->priv->min_frame_size = min_size; GST_LOG_OBJECT (parse, "set frame_min_size: %d", min_size); GST_BASE_PARSE_UNLOCK (parse); } /** * gst_base_parse_get_querytypes: * @pad: GstPad * * Returns: A table of #GstQueryType items describing supported query types. */ static const GstQueryType * gst_base_parse_get_querytypes (GstPad * pad) { static const GstQueryType list[] = { GST_QUERY_POSITION, GST_QUERY_DURATION, GST_QUERY_FORMATS, GST_QUERY_SEEKING, GST_QUERY_CONVERT, 0 }; return list; } /** * gst_base_parse_query: * @pad: #GstPad. * @query: #GstQuery. * * Returns: TRUE on success. */ static gboolean gst_base_parse_query (GstPad * pad, GstQuery * query) { GstBaseParse *parse; GstBaseParseClass *klass; gboolean res = FALSE; parse = GST_BASE_PARSE (GST_PAD_PARENT (pad)); klass = GST_BASE_PARSE_GET_CLASS (parse); switch (GST_QUERY_TYPE (query)) { case GST_QUERY_POSITION: { gint64 dest_value; GstFormat format; GST_DEBUG_OBJECT (parse, "position query"); gst_query_parse_position (query, &format, NULL); g_mutex_lock (parse->parse_lock); if (format == GST_FORMAT_BYTES) { dest_value = parse->priv->offset; res = TRUE; } else if (format == parse->segment.format && GST_CLOCK_TIME_IS_VALID (parse->segment.last_stop)) { dest_value = parse->segment.last_stop; res = TRUE; } else { /* priv->offset is updated in both PUSH/PULL modes */ res = klass->convert (parse, GST_FORMAT_BYTES, parse->priv->offset, format, &dest_value); } g_mutex_unlock (parse->parse_lock); if (res) gst_query_set_position (query, format, dest_value); else res = gst_pad_peer_query (parse->sinkpad, query); break; } case GST_QUERY_DURATION: { GstFormat format; gint64 dest_value; GST_DEBUG_OBJECT (parse, "duration query"); gst_query_parse_duration (query, &format, NULL); g_mutex_lock (parse->parse_lock); switch (format) { case GST_FORMAT_BYTES: if (parse->priv->upstream_duration != -1 && parse->priv->upstream_duration) { dest_value = parse->priv->upstream_duration; res = TRUE; } else if (parse->priv->estimated_duration != -1 && parse->priv->estimated_duration) { dest_value = parse->priv->estimated_duration; res = TRUE; } break; case GST_FORMAT_TIME: if (parse->priv->duration != -1 && format == parse->priv->duration_fmt) { dest_value = parse->priv->duration; res = TRUE; } else if (parse->priv->duration != -1) { res = klass->convert (parse, parse->priv->duration_fmt, parse->priv->duration, format, &dest_value); } else if (parse->priv->upstream_duration != -1 && parse->priv->upstream_duration > 0) { dest_value = parse->priv->upstream_duration; res = TRUE; } else if (parse->priv->estimated_duration != -1 && parse->priv->estimated_duration > 0) { dest_value = parse->priv->estimated_duration; res = TRUE; } break; default: break; } g_mutex_unlock (parse->parse_lock); if (res) gst_query_set_duration (query, format, dest_value); else res = gst_pad_peer_query (parse->sinkpad, query); break; } case GST_QUERY_SEEKING: { GstFormat fmt; gboolean seekable = FALSE; GST_DEBUG_OBJECT (parse, "seeking query"); gst_query_parse_seeking (query, &fmt, NULL, NULL, NULL); if (fmt != GST_FORMAT_TIME) { return gst_pad_query_default (pad, query); } seekable = klass->is_seekable (parse); /* TODO: could this duration be calculated/converted if subclass hasn't given it? */ gst_query_set_seeking (query, GST_FORMAT_TIME, seekable, 0, (parse->priv->duration == -1) ? GST_CLOCK_TIME_NONE : parse->priv->duration); GST_DEBUG_OBJECT (parse, "seekable: %d", seekable); res = TRUE; break; } case GST_QUERY_FORMATS: gst_query_set_formatsv (query, 3, fmtlist); res = TRUE; break; case GST_QUERY_CONVERT: { GstFormat src_format, dest_format; gint64 src_value, dest_value; gst_query_parse_convert (query, &src_format, &src_value, &dest_format, &dest_value); res = klass->convert (parse, src_format, src_value, dest_format, &dest_value); if (res) { gst_query_set_convert (query, src_format, src_value, dest_format, dest_value); } else { res = gst_pad_peer_query (parse->sinkpad, query); } break; } default: res = gst_pad_peer_query (parse->sinkpad, query); break; } return res; } static GstStateChangeReturn gst_base_parse_change_state (GstElement * element, GstStateChange transition) { GstBaseParse *parse = GST_BASE_PARSE (element); GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS; switch (transition) { case GST_STATE_CHANGE_READY_TO_PAUSED: #ifndef GST_DISABLE_INDEX if (!parse->priv->index) gst_base_parse_set_index (element, gst_index_factory_make ("memindex")); #endif default: break; } if (GST_ELEMENT_CLASS (parent_class)->change_state) ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition); switch (transition) { case GST_STATE_CHANGE_PAUSED_TO_READY: break; default: break; } return ret; } #ifndef GST_DISABLE_INDEX static void gst_base_parse_set_index (GstElement * element, GstIndex * index) { GstBaseParse *parse = GST_BASE_PARSE (element); if (index) { parse->priv->index = index; gst_index_get_writer_id (index, GST_OBJECT (element), &parse->priv->index_id); } else { parse->priv->index = NULL; } } static GstIndex * gst_base_parse_get_index (GstElement * element) { GstBaseParse *parse = GST_BASE_PARSE (element); return parse->priv->index; } #endif static gboolean gst_base_parse_frame_in_segment (GstBaseParse * parse, GstBuffer * buffer, GstSegment * segment) { if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer) && GST_CLOCK_TIME_IS_VALID (segment->stop) && GST_BUFFER_TIMESTAMP (buffer) > segment->stop) { return FALSE; } else if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer) && GST_BUFFER_DURATION_IS_VALID (buffer) && GST_CLOCK_TIME_IS_VALID (segment->start) && GST_BUFFER_TIMESTAMP (buffer) + GST_BUFFER_DURATION_IS_VALID (buffer) < segment->start) { return FALSE; } return TRUE; } /** * gst_base_parse_handle_seek: * @parse: #GstBaseParse. * @event: #GstEvent. * * Returns: TRUE if seek succeeded. */ static gboolean gst_base_parse_handle_seek (GstBaseParse * parse, GstEvent * event) { GstBaseParseClass *klass; gdouble rate; GstFormat format; GstSeekFlags flags; GstSeekType cur_type = GST_SEEK_TYPE_NONE, stop_type; gboolean flush, update, res = TRUE; gint64 cur, stop, seekpos; GstSegment seeksegment = { 0, }; GstFormat dstformat; klass = GST_BASE_PARSE_GET_CLASS (parse); gst_event_parse_seek (event, &rate, &format, &flags, &cur_type, &cur, &stop_type, &stop); /* no negative rates yet */ if (rate < 0.0) goto negative_rate; if (cur_type != GST_SEEK_TYPE_SET) goto wrong_type; /* For any format other than TIME, see if upstream handles * it directly or fail. For TIME, try upstream, but do it ourselves if * it fails upstream */ if (format != GST_FORMAT_TIME) { gst_event_ref (event); return gst_pad_push_event (parse->sinkpad, event); } else { gst_event_ref (event); if (gst_pad_push_event (parse->sinkpad, event)) return TRUE; } /* get flush flag */ flush = flags & GST_SEEK_FLAG_FLUSH; dstformat = GST_FORMAT_BYTES; if (!gst_pad_query_convert (parse->srcpad, format, cur, &dstformat, &seekpos)) { GST_DEBUG_OBJECT (parse, "conversion failed"); return FALSE; } GST_DEBUG_OBJECT (parse, "seek position %lld in bytes: %lld", cur, seekpos); if (parse->priv->pad_mode == GST_ACTIVATE_PULL) { gint64 last_stop; GST_DEBUG_OBJECT (parse, "seek in PULL mode"); if (flush) { if (parse->srcpad) { GST_DEBUG_OBJECT (parse, "sending flush start"); gst_pad_push_event (parse->srcpad, gst_event_new_flush_start ()); } } else { gst_pad_pause_task (parse->sinkpad); } /* we should now be able to grab the streaming thread because we stopped it * with the above flush/pause code */ GST_PAD_STREAM_LOCK (parse->sinkpad); /* save current position */ last_stop = parse->segment.last_stop; GST_DEBUG_OBJECT (parse, "stopped streaming at %" G_GINT64_FORMAT, last_stop); /* copy segment, we need this because we still need the old * segment when we close the current segment. */ memcpy (&seeksegment, &parse->segment, sizeof (GstSegment)); GST_DEBUG_OBJECT (parse, "configuring seek"); gst_segment_set_seek (&seeksegment, rate, format, flags, cur_type, cur, stop_type, stop, &update); /* figure out the last position we need to play. If it's configured (stop != * -1), use that, else we play until the total duration of the file */ if ((stop = seeksegment.stop) == -1) stop = seeksegment.duration; parse->priv->offset = seekpos; /* prepare for streaming again */ if (flush) { GST_DEBUG_OBJECT (parse, "sending flush stop"); gst_pad_push_event (parse->srcpad, gst_event_new_flush_stop ()); } else { if (parse->close_segment) gst_event_unref (parse->close_segment); parse->close_segment = gst_event_new_new_segment (TRUE, parse->segment.rate, parse->segment.format, parse->segment.accum, parse->segment.last_stop, parse->segment.accum); /* keep track of our last_stop */ seeksegment.accum = parse->segment.last_stop; GST_DEBUG_OBJECT (parse, "Created close seg format %d, " "start = %" GST_TIME_FORMAT ", stop = %" GST_TIME_FORMAT ", pos = %" GST_TIME_FORMAT, format, GST_TIME_ARGS (parse->segment.accum), GST_TIME_ARGS (parse->segment.last_stop), GST_TIME_ARGS (parse->segment.accum)); } memcpy (&parse->segment, &seeksegment, sizeof (GstSegment)); /* store the newsegment event so it can be sent from the streaming thread. */ if (parse->pending_segment) gst_event_unref (parse->pending_segment); /* This will be sent later in _loop() */ parse->pending_segment = gst_event_new_new_segment (FALSE, parse->segment.rate, parse->segment.format, parse->segment.last_stop, stop, parse->segment.last_stop); GST_DEBUG_OBJECT (parse, "Created newseg format %d, " "start = %" GST_TIME_FORMAT ", stop = %" GST_TIME_FORMAT ", pos = %" GST_TIME_FORMAT, format, GST_TIME_ARGS (parse->segment.last_stop), GST_TIME_ARGS (stop), GST_TIME_ARGS (parse->segment.last_stop)); /* mark discont if we are going to stream from another position. */ if (last_stop != parse->segment.last_stop) { GST_DEBUG_OBJECT (parse, "mark DISCONT, we did a seek to another position"); parse->priv->discont = TRUE; } /* Start streaming thread if paused */ gst_pad_start_task (parse->sinkpad, (GstTaskFunction) gst_base_parse_loop, parse->sinkpad); GST_PAD_STREAM_UNLOCK (parse->sinkpad); } else { GstEvent *new_event; /* The only thing we need to do in PUSH-mode is to send the seek event (in bytes) to upstream. Segment / flush handling happens in corresponding src event handlers */ GST_DEBUG_OBJECT (parse, "seek in PUSH mode"); new_event = gst_event_new_seek (rate, GST_FORMAT_BYTES, flush, GST_SEEK_TYPE_SET, seekpos, stop_type, stop); res = gst_pad_push_event (parse->sinkpad, new_event); } done: return res; /* ERRORS */ negative_rate: { GST_DEBUG_OBJECT (parse, "negative playback rates are not supported yet."); res = FALSE; goto done; } wrong_type: { GST_DEBUG_OBJECT (parse, "unsupported seek type."); res = FALSE; goto done; } } /** * gst_base_parse_sink_setcaps: * @pad: #GstPad. * @caps: #GstCaps. * * Returns: TRUE if caps were accepted. */ static gboolean gst_base_parse_sink_setcaps (GstPad * pad, GstCaps * caps) { GstBaseParse *parse; GstBaseParseClass *klass; gboolean res = TRUE; gchar *caps_str = gst_caps_to_string (caps); g_free (caps_str); parse = GST_BASE_PARSE (gst_pad_get_parent (pad)); klass = GST_BASE_PARSE_GET_CLASS (parse); GST_DEBUG_OBJECT (parse, "setcaps: %s", caps_str); if (klass->set_sink_caps) res = klass->set_sink_caps (parse, caps); parse->negotiated = res; gst_object_unref (parse); return gst_pad_set_caps (pad, caps); }