basebarse: Add detect vfunc to allow subclasses to do format detection before anythin...
[gstreamer-omap:gstreamer.git] / libs / gst / base / gstbaseparse.c
1 /* GStreamer
2  * Copyright (C) 2008 Nokia Corporation. All rights reserved.
3  *   Contact: Stefan Kost <stefan.kost@nokia.com>
4  * Copyright (C) 2008 Sebastian Dröge <sebastian.droege@collabora.co.uk>.
5  * Copyright (C) 2011, Hewlett-Packard Development Company, L.P.
6  *   Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>, Collabora Ltd.
7  *
8  * This library is free software; you can redistribute it and/or
9  * modify it under the terms of the GNU Library General Public
10  * License as published by the Free Software Foundation; either
11  * version 2 of the License, or (at your option) any later version.
12  *
13  * This library is distributed in the hope that it will be useful,
14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
16  * Library General Public License for more details.
17  *
18  * You should have received a copy of the GNU Library General Public
19  * License along with this library; if not, write to the
20  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
21  * Boston, MA 02111-1307, USA.
22  */
23
24 /**
25  * SECTION:gstbaseparse
26  * @short_description: Base class for stream parsers
27  * @see_also: #GstBaseTransform
28  *
29  * This base class is for parser elements that process data and splits it
30  * into separate audio/video/whatever frames.
31  *
32  * It provides for:
33  * <itemizedlist>
34  *   <listitem><para>provides one sink pad and one source pad</para></listitem>
35  *   <listitem><para>handles state changes</para></listitem>
36  *   <listitem><para>can operate in pull mode or push mode</para></listitem>
37  *   <listitem><para>handles seeking in both modes</para></listitem>
38  *   <listitem><para>handles events (NEWSEGMENT/EOS/FLUSH)</para></listitem>
39  *   <listitem><para>
40  *        handles queries (POSITION/DURATION/SEEKING/FORMAT/CONVERT)
41  *   </para></listitem>
42  *   <listitem><para>handles flushing</para></listitem>
43  * </itemizedlist>
44  *
45  * The purpose of this base class is to provide the basic functionality of
46  * a parser and share a lot of rather complex code.
47  *
48  * Description of the parsing mechanism:
49  * <orderedlist>
50  * <listitem>
51  *   <itemizedlist><title>Set-up phase</title>
52  *   <listitem><para>
53  *     GstBaseParse class calls @set_sink_caps to inform the subclass about
54  *     incoming sinkpad caps. Subclass should set the srcpad caps accordingly.
55  *   </para></listitem>
56  *   <listitem><para>
57  *     GstBaseParse calls @start to inform subclass that data processing is
58  *     about to start now.
59  *   </para></listitem>
60  *   <listitem><para>
61  *      At least at this point subclass needs to tell the GstBaseParse class
62  *      how big data chunks it wants to receive (min_frame_size). It can do
63  *      this with gst_base_parse_set_min_frame_size().
64  *   </para></listitem>
65  *   <listitem><para>
66  *      GstBaseParse class sets up appropriate data passing mode (pull/push)
67  *      and starts to process the data.
68  *   </para></listitem>
69  *   </itemizedlist>
70  * </listitem>
71  * <listitem>
72  *   <itemizedlist>
73  *   <title>Parsing phase</title>
74  *     <listitem><para>
75  *       GstBaseParse gathers at least min_frame_size bytes of data either
76  *       by pulling it from upstream or collecting buffers in an internal
77  *       #GstAdapter.
78  *     </para></listitem>
79  *     <listitem><para>
80  *       A buffer of (at least) min_frame_size bytes is passed to subclass with
81  *       @check_valid_frame. Subclass checks the contents and returns TRUE
82  *       if the buffer contains a valid frame. It also needs to set the
83  *       @framesize according to the detected frame size. If buffer didn't
84  *       contain a valid frame, this call must return FALSE and optionally
85  *       set the @skipsize value to inform base class that how many bytes
86  *       it needs to skip in order to find a valid frame. @framesize can always
87  *       indicate a new minimum for current frame parsing.  Indicating G_MAXUINT
88  *       for requested amount means subclass simply needs best available
89  *       subsequent data.  In push mode this amounts to an additional input buffer
90  *       (thus minimal additional latency), in pull mode this amounts to some
91  *       arbitrary reasonable buffer size increase.  The passed buffer
92  *       is read-only.  Note that @check_valid_frame might receive any small
93  *       amount of input data when leftover data is being drained (e.g. at EOS).
94  *     </para></listitem>
95  *     <listitem><para>
96  *       After valid frame is found, it will be passed again to subclass with
97  *       @parse_frame call. Now subclass is responsible for parsing the
98  *       frame contents and setting the caps, and buffer metadata (e.g.
99  *       buffer timestamp and duration, or keyframe if applicable).
100  *       (although the latter can also be done by GstBaseParse if it is
101  *       appropriately configured, see below).  Frame is provided with
102  *       timestamp derived from upstream (as much as generally possible),
103  *       duration obtained from configuration (see below), and offset
104  *       if meaningful (in pull mode).
105  *     </para></listitem>
106  *     <listitem><para>
107  *       Finally the buffer can be pushed downstream and the parsing loop starts
108  *       over again.  Just prior to actually pushing the buffer in question,
109  *       it is passed to @pre_push_buffer which gives subclass yet one
110  *       last chance to examine buffer metadata, or to send some custom (tag)
111  *       events, or to perform custom (segment) filtering.
112  *     </para></listitem>
113  *     <listitem><para>
114  *       During the parsing process GstBaseParseClass will handle both srcpad
115  *       and sinkpad events. They will be passed to subclass if @event or
116  *       @src_event callbacks have been provided.
117  *     </para></listitem>
118  *   </itemizedlist>
119  * </listitem>
120  * <listitem>
121  *   <itemizedlist><title>Shutdown phase</title>
122  *   <listitem><para>
123  *     GstBaseParse class calls @stop to inform the subclass that data
124  *     parsing will be stopped.
125  *   </para></listitem>
126  *   </itemizedlist>
127  * </listitem>
128  * </orderedlist>
129  *
130  * Subclass is responsible for providing pad template caps for
131  * source and sink pads. The pads need to be named "sink" and "src". It also
132  * needs to set the fixed caps on srcpad, when the format is ensured (e.g.
133  * when base class calls subclass' @set_sink_caps function).
134  *
135  * This base class uses #GST_FORMAT_DEFAULT as a meaning of frames. So,
136  * subclass conversion routine needs to know that conversion from
137  * #GST_FORMAT_TIME to #GST_FORMAT_DEFAULT must return the
138  * frame number that can be found from the given byte position.
139  *
140  * GstBaseParse uses subclasses conversion methods also for seeking (or
141  * otherwise uses its own default one, see also below).
142  *
143  * Subclass @start and @stop functions will be called to inform the beginning
144  * and end of data processing.
145  *
146  * Things that subclass need to take care of:
147  * <itemizedlist>
148  *   <listitem><para>Provide pad templates</para></listitem>
149  *   <listitem><para>
150  *      Fixate the source pad caps when appropriate
151  *   </para></listitem>
152  *   <listitem><para>
153  *      Inform base class how big data chunks should be retrieved. This is
154  *      done with gst_base_parse_set_min_frame_size() function.
155  *   </para></listitem>
156  *   <listitem><para>
157  *      Examine data chunks passed to subclass with @check_valid_frame
158  *      and tell if they contain a valid frame
159  *   </para></listitem>
160  *   <listitem><para>
161  *      Set the caps and timestamp to frame that is passed to subclass with
162  *      @parse_frame function.
163  *   </para></listitem>
164  *   <listitem><para>Provide conversion functions</para></listitem>
165  *   <listitem><para>
166  *      Update the duration information with gst_base_parse_set_duration()
167  *   </para></listitem>
168  *   <listitem><para>
169  *      Optionally passthrough using gst_base_parse_set_passthrough()
170  *   </para></listitem>
171  *   <listitem><para>
172  *      Configure various baseparse parameters using
173  *      gst_base_parse_set_average_bitrate(), gst_base_parse_set_syncable()
174  *      and gst_base_parse_set_frame_rate().
175  *   </para></listitem>
176  *   <listitem><para>
177  *      In particular, if subclass is unable to determine a duration, but
178  *      parsing (or specs) yields a frames per seconds rate, then this can be
179  *      provided to GstBaseParse to enable it to cater for
180  *      buffer time metadata (which will be taken from upstream as much as
181  *      possible). Internally keeping track of frame durations and respective
182  *      sizes that have been pushed provides GstBaseParse with an estimated
183  *      bitrate. A default @convert (used if not overriden) will then use these
184  *      rates to perform obvious conversions.  These rates are also used to
185  *      update (estimated) duration at regular frame intervals.
186  *   </para></listitem>
187  * </itemizedlist>
188  *
189  */
190
191 /* TODO:
192  *  - In push mode provide a queue of adapter-"queued" buffers for upstream
193  *    buffer metadata
194  *  - Queue buffers/events until caps are set
195  */
196
197 #ifdef HAVE_CONFIG_H
198 #  include "config.h"
199 #endif
200
201 #include <stdlib.h>
202 #include <string.h>
203
204 #include <gst/base/gstadapter.h>
205
206 #include "gstbaseparse.h"
207
208 #define GST_BASE_PARSE_FRAME_PRIVATE_FLAG_NOALLOC  (1 << 0)
209
210 #define MIN_FRAMES_TO_POST_BITRATE 10
211 #define TARGET_DIFFERENCE          (20 * GST_SECOND)
212
213 GST_DEBUG_CATEGORY_STATIC (gst_base_parse_debug);
214 #define GST_CAT_DEFAULT gst_base_parse_debug
215
216 /* Supported formats */
217 static const GstFormat fmtlist[] = {
218   GST_FORMAT_DEFAULT,
219   GST_FORMAT_BYTES,
220   GST_FORMAT_TIME,
221   0
222 };
223
224 #define GST_BASE_PARSE_GET_PRIVATE(obj)  \
225     (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_BASE_PARSE, GstBaseParsePrivate))
226
227 struct _GstBaseParsePrivate
228 {
229   GstActivateMode pad_mode;
230
231   GstAdapter *adapter;
232
233   gint64 duration;
234   GstFormat duration_fmt;
235   gint64 estimated_duration;
236
237   guint min_frame_size;
238   gboolean passthrough;
239   gboolean syncable;
240   gboolean has_timing_info;
241   guint fps_num, fps_den;
242   gint update_interval;
243   guint bitrate;
244   guint lead_in, lead_out;
245   GstClockTime lead_in_ts, lead_out_ts;
246   GstClockTime min_latency, max_latency;
247
248   gboolean discont;
249   gboolean flushing;
250   gboolean drain;
251
252   gint64 offset;
253   gint64 sync_offset;
254   GstClockTime next_ts;
255   GstClockTime prev_ts;
256   GstClockTime frame_duration;
257   gboolean seen_keyframe;
258   gboolean is_video;
259
260   guint64 framecount;
261   guint64 bytecount;
262   guint64 data_bytecount;
263   guint64 acc_duration;
264   GstClockTime first_frame_ts;
265   gint64 first_frame_offset;
266
267   gboolean post_min_bitrate;
268   gboolean post_avg_bitrate;
269   gboolean post_max_bitrate;
270   guint min_bitrate;
271   guint avg_bitrate;
272   guint max_bitrate;
273   guint posted_avg_bitrate;
274
275   GList *pending_events;
276
277   /* frames/buffers that are queued and ready to go on OK */
278   GQueue queued_frames;
279
280   GstBuffer *cache;
281
282   /* index entry storage, either ours or provided */
283   GstIndex *index;
284   gint index_id;
285   gboolean own_index;
286   GStaticMutex index_lock;
287
288   /* seek table entries only maintained if upstream is BYTE seekable */
289   gboolean upstream_seekable;
290   gboolean upstream_has_duration;
291   gint64 upstream_size;
292   /* minimum distance between two index entries */
293   GstClockTimeDiff idx_interval;
294   /* ts and offset of last entry added */
295   GstClockTime index_last_ts;
296   gint64 index_last_offset;
297   gboolean index_last_valid;
298
299   /* timestamps currently produced are accurate, e.g. started from 0 onwards */
300   gboolean exact_position;
301   /* seek events are temporarily kept to match them with newsegments */
302   GSList *pending_seeks;
303
304   /* reverse playback */
305   GSList *buffers_pending;
306   GSList *buffers_queued;
307   GSList *buffers_send;
308   GstClockTime last_ts;
309   gint64 last_offset;
310
311   /* Newsegment event to be sent after SEEK */
312   GstEvent *pending_segment;
313
314   /* Segment event that closes the running segment prior to SEEK */
315   GstEvent *close_segment;
316
317   /* push mode helper frame */
318   GstBaseParseFrame frame;
319
320   /* TRUE if we're still detecting the format, i.e.
321    * if ::detect() is still called for future buffers */
322   gboolean detecting;
323   GList *detect_buffers;
324   guint detect_buffers_size;
325 };
326
327 typedef struct _GstBaseParseSeek
328 {
329   GstSegment segment;
330   gboolean accurate;
331   gint64 offset;
332   GstClockTime start_ts;
333 } GstBaseParseSeek;
334
335 static GstElementClass *parent_class = NULL;
336
337 static void gst_base_parse_class_init (GstBaseParseClass * klass);
338 static void gst_base_parse_init (GstBaseParse * parse,
339     GstBaseParseClass * klass);
340
341 GType
342 gst_base_parse_get_type (void)
343 {
344   static volatile gsize base_parse_type = 0;
345
346   if (g_once_init_enter (&base_parse_type)) {
347     static const GTypeInfo base_parse_info = {
348       sizeof (GstBaseParseClass),
349       (GBaseInitFunc) NULL,
350       (GBaseFinalizeFunc) NULL,
351       (GClassInitFunc) gst_base_parse_class_init,
352       NULL,
353       NULL,
354       sizeof (GstBaseParse),
355       0,
356       (GInstanceInitFunc) gst_base_parse_init,
357     };
358     GType _type;
359
360     _type = g_type_register_static (GST_TYPE_ELEMENT,
361         "GstBaseParse", &base_parse_info, G_TYPE_FLAG_ABSTRACT);
362     g_once_init_leave (&base_parse_type, _type);
363   }
364   return (GType) base_parse_type;
365 }
366
367 static void gst_base_parse_finalize (GObject * object);
368
369 static GstStateChangeReturn gst_base_parse_change_state (GstElement * element,
370     GstStateChange transition);
371 static void gst_base_parse_reset (GstBaseParse * parse);
372
373 static void gst_base_parse_set_index (GstElement * element, GstIndex * index);
374 static GstIndex *gst_base_parse_get_index (GstElement * element);
375
376 static gboolean gst_base_parse_sink_activate (GstPad * sinkpad);
377 static gboolean gst_base_parse_sink_activate_push (GstPad * pad,
378     gboolean active);
379 static gboolean gst_base_parse_sink_activate_pull (GstPad * pad,
380     gboolean active);
381 static gboolean gst_base_parse_handle_seek (GstBaseParse * parse,
382     GstEvent * event);
383 static void gst_base_parse_handle_tag (GstBaseParse * parse, GstEvent * event);
384
385 static gboolean gst_base_parse_src_event (GstPad * pad, GstEvent * event);
386 static gboolean gst_base_parse_sink_event (GstPad * pad, GstEvent * event);
387 static gboolean gst_base_parse_query (GstPad * pad, GstQuery * query);
388 static gboolean gst_base_parse_sink_setcaps (GstPad * pad, GstCaps * caps);
389 static GstCaps *gst_base_parse_sink_getcaps (GstPad * pad);
390 static const GstQueryType *gst_base_parse_get_querytypes (GstPad * pad);
391
392 static GstFlowReturn gst_base_parse_chain (GstPad * pad, GstBuffer * buffer);
393 static void gst_base_parse_loop (GstPad * pad);
394
395 static gboolean gst_base_parse_check_frame (GstBaseParse * parse,
396     GstBaseParseFrame * frame, guint * framesize, gint * skipsize);
397 static GstFlowReturn gst_base_parse_parse_frame (GstBaseParse * parse,
398     GstBaseParseFrame * frame);
399
400 static gboolean gst_base_parse_sink_eventfunc (GstBaseParse * parse,
401     GstEvent * event);
402
403 static gboolean gst_base_parse_src_eventfunc (GstBaseParse * parse,
404     GstEvent * event);
405
406 static void gst_base_parse_drain (GstBaseParse * parse);
407
408 static void gst_base_parse_post_bitrates (GstBaseParse * parse,
409     gboolean post_min, gboolean post_avg, gboolean post_max);
410
411 static gint64 gst_base_parse_find_offset (GstBaseParse * parse,
412     GstClockTime time, gboolean before, GstClockTime * _ts);
413 static GstFlowReturn gst_base_parse_locate_time (GstBaseParse * parse,
414     GstClockTime * _time, gint64 * _offset);
415
416 static GstFlowReturn gst_base_parse_process_fragment (GstBaseParse * parse,
417     gboolean push_only);
418
419 static gboolean gst_base_parse_is_seekable (GstBaseParse * parse);
420
421 static void
422 gst_base_parse_clear_queues (GstBaseParse * parse)
423 {
424   g_slist_foreach (parse->priv->buffers_queued, (GFunc) gst_buffer_unref, NULL);
425   g_slist_free (parse->priv->buffers_queued);
426   parse->priv->buffers_queued = NULL;
427   g_slist_foreach (parse->priv->buffers_pending, (GFunc) gst_buffer_unref,
428       NULL);
429   g_slist_free (parse->priv->buffers_pending);
430   parse->priv->buffers_pending = NULL;
431   g_slist_foreach (parse->priv->buffers_send, (GFunc) gst_buffer_unref, NULL);
432   g_slist_free (parse->priv->buffers_send);
433   parse->priv->buffers_send = NULL;
434
435   g_list_foreach (parse->priv->detect_buffers, (GFunc) gst_buffer_unref, NULL);
436   g_list_free (parse->priv->detect_buffers);
437   parse->priv->detect_buffers = NULL;
438   parse->priv->detect_buffers_size = 0;
439 }
440
441 static void
442 gst_base_parse_finalize (GObject * object)
443 {
444   GstBaseParse *parse = GST_BASE_PARSE (object);
445   GstEvent **p_ev;
446
447   g_object_unref (parse->priv->adapter);
448
449   if (parse->priv->pending_segment) {
450     p_ev = &parse->priv->pending_segment;
451     gst_event_replace (p_ev, NULL);
452   }
453   if (parse->priv->close_segment) {
454     p_ev = &parse->priv->close_segment;
455     gst_event_replace (p_ev, NULL);
456   }
457
458   if (parse->priv->cache) {
459     gst_buffer_unref (parse->priv->cache);
460     parse->priv->cache = NULL;
461   }
462
463   g_list_foreach (parse->priv->pending_events, (GFunc) gst_mini_object_unref,
464       NULL);
465   g_list_free (parse->priv->pending_events);
466   parse->priv->pending_events = NULL;
467
468   g_queue_foreach (&parse->priv->queued_frames,
469       (GFunc) gst_base_parse_frame_free, NULL);
470   g_queue_clear (&parse->priv->queued_frames);
471
472   if (parse->priv->index) {
473     gst_object_unref (parse->priv->index);
474     parse->priv->index = NULL;
475   }
476
477   g_static_mutex_free (&parse->priv->index_lock);
478
479   gst_base_parse_clear_queues (parse);
480
481   G_OBJECT_CLASS (parent_class)->finalize (object);
482 }
483
484 static void
485 gst_base_parse_class_init (GstBaseParseClass * klass)
486 {
487   GObjectClass *gobject_class;
488   GstElementClass *gstelement_class;
489
490   gobject_class = G_OBJECT_CLASS (klass);
491   g_type_class_add_private (klass, sizeof (GstBaseParsePrivate));
492   parent_class = g_type_class_peek_parent (klass);
493   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_base_parse_finalize);
494
495   gstelement_class = (GstElementClass *) klass;
496   gstelement_class->change_state =
497       GST_DEBUG_FUNCPTR (gst_base_parse_change_state);
498   gstelement_class->set_index = GST_DEBUG_FUNCPTR (gst_base_parse_set_index);
499   gstelement_class->get_index = GST_DEBUG_FUNCPTR (gst_base_parse_get_index);
500
501   /* Default handlers */
502   klass->check_valid_frame = gst_base_parse_check_frame;
503   klass->parse_frame = gst_base_parse_parse_frame;
504   klass->src_event = gst_base_parse_src_eventfunc;
505   klass->convert = gst_base_parse_convert_default;
506
507   GST_DEBUG_CATEGORY_INIT (gst_base_parse_debug, "baseparse", 0,
508       "baseparse element");
509 }
510
511 static void
512 gst_base_parse_init (GstBaseParse * parse, GstBaseParseClass * bclass)
513 {
514   GstPadTemplate *pad_template;
515
516   GST_DEBUG_OBJECT (parse, "gst_base_parse_init");
517
518   parse->priv = GST_BASE_PARSE_GET_PRIVATE (parse);
519
520   pad_template =
521       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass), "sink");
522   g_return_if_fail (pad_template != NULL);
523   parse->sinkpad = gst_pad_new_from_template (pad_template, "sink");
524   gst_pad_set_event_function (parse->sinkpad,
525       GST_DEBUG_FUNCPTR (gst_base_parse_sink_event));
526   gst_pad_set_setcaps_function (parse->sinkpad,
527       GST_DEBUG_FUNCPTR (gst_base_parse_sink_setcaps));
528   gst_pad_set_getcaps_function (parse->sinkpad,
529       GST_DEBUG_FUNCPTR (gst_base_parse_sink_getcaps));
530   gst_pad_set_chain_function (parse->sinkpad,
531       GST_DEBUG_FUNCPTR (gst_base_parse_chain));
532   gst_pad_set_activate_function (parse->sinkpad,
533       GST_DEBUG_FUNCPTR (gst_base_parse_sink_activate));
534   gst_pad_set_activatepush_function (parse->sinkpad,
535       GST_DEBUG_FUNCPTR (gst_base_parse_sink_activate_push));
536   gst_pad_set_activatepull_function (parse->sinkpad,
537       GST_DEBUG_FUNCPTR (gst_base_parse_sink_activate_pull));
538   gst_element_add_pad (GST_ELEMENT (parse), parse->sinkpad);
539
540   GST_DEBUG_OBJECT (parse, "sinkpad created");
541
542   pad_template =
543       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass), "src");
544   g_return_if_fail (pad_template != NULL);
545   parse->srcpad = gst_pad_new_from_template (pad_template, "src");
546   gst_pad_set_event_function (parse->srcpad,
547       GST_DEBUG_FUNCPTR (gst_base_parse_src_event));
548   gst_pad_set_query_type_function (parse->srcpad,
549       GST_DEBUG_FUNCPTR (gst_base_parse_get_querytypes));
550   gst_pad_set_query_function (parse->srcpad,
551       GST_DEBUG_FUNCPTR (gst_base_parse_query));
552   gst_pad_use_fixed_caps (parse->srcpad);
553   gst_element_add_pad (GST_ELEMENT (parse), parse->srcpad);
554   GST_DEBUG_OBJECT (parse, "src created");
555
556   g_queue_init (&parse->priv->queued_frames);
557
558   parse->priv->adapter = gst_adapter_new ();
559
560   parse->priv->pad_mode = GST_ACTIVATE_NONE;
561
562   g_static_mutex_init (&parse->priv->index_lock);
563
564   /* init state */
565   gst_base_parse_reset (parse);
566   GST_DEBUG_OBJECT (parse, "init ok");
567 }
568
569 static GstBaseParseFrame *
570 gst_base_parse_frame_copy (GstBaseParseFrame * frame)
571 {
572   GstBaseParseFrame *copy;
573
574   copy = g_slice_dup (GstBaseParseFrame, frame);
575   copy->buffer = gst_buffer_ref (frame->buffer);
576   copy->_private_flags &= ~GST_BASE_PARSE_FRAME_PRIVATE_FLAG_NOALLOC;
577
578   GST_TRACE ("copied frame %p -> %p", frame, copy);
579
580   return copy;
581 }
582
583 void
584 gst_base_parse_frame_free (GstBaseParseFrame * frame)
585 {
586   GST_TRACE ("freeing frame %p", frame);
587
588   if (frame->buffer) {
589     gst_buffer_unref (frame->buffer);
590     frame->buffer = NULL;
591   }
592
593   if (!(frame->_private_flags & GST_BASE_PARSE_FRAME_PRIVATE_FLAG_NOALLOC)) {
594     g_slice_free (GstBaseParseFrame, frame);
595   } else {
596     memset (frame, 0, sizeof (*frame));
597   }
598 }
599
600 GType
601 gst_base_parse_frame_get_type (void)
602 {
603   static volatile gsize frame_type = 0;
604
605   if (g_once_init_enter (&frame_type)) {
606     GType _type;
607
608     _type = g_boxed_type_register_static ("GstBaseParseFrame",
609         (GBoxedCopyFunc) gst_base_parse_frame_copy,
610         (GBoxedFreeFunc) gst_base_parse_frame_free);
611     g_once_init_leave (&frame_type, _type);
612   }
613   return (GType) frame_type;
614 }
615
616 /**
617  * gst_base_parse_frame_init:
618  * @frame: #GstBaseParseFrame.
619  *
620  * Sets a #GstBaseParseFrame to initial state.  Currently this means
621  * all public fields are zero-ed and a private flag is set to make
622  * sure gst_base_parse_frame_free() only frees the contents but not
623  * the actual frame. Use this function to initialise a #GstBaseParseFrame
624  * allocated on the stack.
625  *
626  * Since: 0.10.33
627  */
628 void
629 gst_base_parse_frame_init (GstBaseParseFrame * frame)
630 {
631   memset (frame, 0, sizeof (GstBaseParseFrame));
632   frame->_private_flags = GST_BASE_PARSE_FRAME_PRIVATE_FLAG_NOALLOC;
633   GST_TRACE ("inited frame %p", frame);
634 }
635
636 /**
637  * gst_base_parse_frame_new:
638  * @buffer: (transfer none): a #GstBuffer
639  * @flags: the flags
640  * @overhead: number of bytes in this frame which should be counted as
641  *     metadata overhead, ie. not used to calculate the average bitrate.
642  *     Set to -1 to mark the entire frame as metadata. If in doubt, set to 0.
643  *
644  * Allocates a new #GstBaseParseFrame. This function is mainly for bindings,
645  * elements written in C should usually allocate the frame on the stack and
646  * then use gst_base_parse_frame_init() to initialise it.
647  *
648  * Returns: a newly-allocated #GstBaseParseFrame. Free with
649  *     gst_base_parse_frame_free() when no longer needed, unless you gave
650  *     away ownership to gst_base_parse_push_frame().
651  *
652  * Since: 0.10.33
653  */
654 GstBaseParseFrame *
655 gst_base_parse_frame_new (GstBuffer * buffer, GstBaseParseFrameFlags flags,
656     gint overhead)
657 {
658   GstBaseParseFrame *frame;
659
660   frame = g_slice_new0 (GstBaseParseFrame);
661   frame->buffer = gst_buffer_ref (buffer);
662
663   GST_TRACE ("created frame %p", frame);
664   return frame;
665 }
666
667 static inline void
668 gst_base_parse_frame_update (GstBaseParse * parse, GstBaseParseFrame * frame,
669     GstBuffer * buf)
670 {
671   gst_buffer_replace (&frame->buffer, buf);
672
673   parse->flags = 0;
674
675   /* set flags one by one for clarity */
676   if (G_UNLIKELY (parse->priv->drain))
677     parse->flags |= GST_BASE_PARSE_FLAG_DRAINING;
678
679   /* losing sync is pretty much a discont (and vice versa), no ? */
680   if (G_UNLIKELY (parse->priv->discont))
681     parse->flags |= GST_BASE_PARSE_FLAG_LOST_SYNC;
682 }
683
684 static void
685 gst_base_parse_reset (GstBaseParse * parse)
686 {
687   GST_OBJECT_LOCK (parse);
688   gst_segment_init (&parse->segment, GST_FORMAT_TIME);
689   parse->priv->duration = -1;
690   parse->priv->min_frame_size = 1;
691   parse->priv->discont = TRUE;
692   parse->priv->flushing = FALSE;
693   parse->priv->offset = 0;
694   parse->priv->sync_offset = 0;
695   parse->priv->update_interval = -1;
696   parse->priv->fps_num = parse->priv->fps_den = 0;
697   parse->priv->frame_duration = GST_CLOCK_TIME_NONE;
698   parse->priv->lead_in = parse->priv->lead_out = 0;
699   parse->priv->lead_in_ts = parse->priv->lead_out_ts = 0;
700   parse->priv->bitrate = 0;
701   parse->priv->framecount = 0;
702   parse->priv->bytecount = 0;
703   parse->priv->acc_duration = 0;
704   parse->priv->first_frame_ts = GST_CLOCK_TIME_NONE;
705   parse->priv->first_frame_offset = -1;
706   parse->priv->estimated_duration = -1;
707   parse->priv->next_ts = 0;
708   parse->priv->syncable = TRUE;
709   parse->priv->passthrough = FALSE;
710   parse->priv->has_timing_info = FALSE;
711   parse->priv->post_min_bitrate = TRUE;
712   parse->priv->post_avg_bitrate = TRUE;
713   parse->priv->post_max_bitrate = TRUE;
714   parse->priv->min_bitrate = G_MAXUINT;
715   parse->priv->max_bitrate = 0;
716   parse->priv->avg_bitrate = 0;
717   parse->priv->posted_avg_bitrate = 0;
718
719   parse->priv->index_last_ts = GST_CLOCK_TIME_NONE;
720   parse->priv->index_last_offset = -1;
721   parse->priv->index_last_valid = TRUE;
722   parse->priv->upstream_seekable = FALSE;
723   parse->priv->upstream_size = 0;
724   parse->priv->upstream_has_duration = FALSE;
725   parse->priv->idx_interval = 0;
726   parse->priv->exact_position = TRUE;
727   parse->priv->seen_keyframe = FALSE;
728
729   parse->priv->last_ts = GST_CLOCK_TIME_NONE;
730   parse->priv->last_offset = 0;
731
732   if (parse->priv->pending_segment) {
733     gst_event_unref (parse->priv->pending_segment);
734     parse->priv->pending_segment = NULL;
735   }
736
737   g_list_foreach (parse->priv->pending_events, (GFunc) gst_mini_object_unref,
738       NULL);
739   g_list_free (parse->priv->pending_events);
740   parse->priv->pending_events = NULL;
741
742   if (parse->priv->cache) {
743     gst_buffer_unref (parse->priv->cache);
744     parse->priv->cache = NULL;
745   }
746
747   g_slist_foreach (parse->priv->pending_seeks, (GFunc) g_free, NULL);
748   g_slist_free (parse->priv->pending_seeks);
749   parse->priv->pending_seeks = NULL;
750
751   /* we know it is not alloc'ed, but maybe other stuff to free, some day ... */
752   parse->priv->frame._private_flags |=
753       GST_BASE_PARSE_FRAME_PRIVATE_FLAG_NOALLOC;
754   gst_base_parse_frame_free (&parse->priv->frame);
755
756   g_list_foreach (parse->priv->detect_buffers, (GFunc) gst_buffer_unref, NULL);
757   g_list_free (parse->priv->detect_buffers);
758   parse->priv->detect_buffers = NULL;
759   parse->priv->detect_buffers_size = 0;
760   GST_OBJECT_UNLOCK (parse);
761 }
762
763 /* gst_base_parse_check_frame:
764  * @parse: #GstBaseParse.
765  * @buffer: GstBuffer.
766  * @framesize: This will be set to tell the found frame size in bytes.
767  * @skipsize: Output parameter that tells how much data needs to be skipped
768  *            in order to find the following frame header.
769  *
770  * Default callback for check_valid_frame.
771  *
772  * Returns: Always TRUE.
773  */
774 static gboolean
775 gst_base_parse_check_frame (GstBaseParse * parse,
776     GstBaseParseFrame * frame, guint * framesize, gint * skipsize)
777 {
778   *framesize = GST_BUFFER_SIZE (frame->buffer);
779   *skipsize = 0;
780   return TRUE;
781 }
782
783
784 /* gst_base_parse_parse_frame:
785  * @parse: #GstBaseParse.
786  * @buffer: #GstBuffer.
787  *
788  * Default callback for parse_frame.
789  */
790 static GstFlowReturn
791 gst_base_parse_parse_frame (GstBaseParse * parse, GstBaseParseFrame * frame)
792 {
793   GstBuffer *buffer = frame->buffer;
794
795   if (!GST_BUFFER_TIMESTAMP_IS_VALID (buffer) &&
796       GST_CLOCK_TIME_IS_VALID (parse->priv->next_ts)) {
797     GST_BUFFER_TIMESTAMP (buffer) = parse->priv->next_ts;
798   }
799   if (!GST_BUFFER_DURATION_IS_VALID (buffer) &&
800       GST_CLOCK_TIME_IS_VALID (parse->priv->frame_duration)) {
801     GST_BUFFER_DURATION (buffer) = parse->priv->frame_duration;
802   }
803   return GST_FLOW_OK;
804 }
805
806 /* gst_base_parse_convert:
807  * @parse: #GstBaseParse.
808  * @src_format: #GstFormat describing the source format.
809  * @src_value: Source value to be converted.
810  * @dest_format: #GstFormat defining the converted format.
811  * @dest_value: Pointer where the conversion result will be put.
812  *
813  * Converts using configured "convert" vmethod in #GstBaseParse class.
814  *
815  * Returns: TRUE if conversion was successful.
816  */
817 static gboolean
818 gst_base_parse_convert (GstBaseParse * parse,
819     GstFormat src_format,
820     gint64 src_value, GstFormat dest_format, gint64 * dest_value)
821 {
822   GstBaseParseClass *klass = GST_BASE_PARSE_GET_CLASS (parse);
823   gboolean ret;
824
825   g_return_val_if_fail (dest_value != NULL, FALSE);
826
827   if (!klass->convert)
828     return FALSE;
829
830   ret = klass->convert (parse, src_format, src_value, dest_format, dest_value);
831
832 #ifndef GST_DISABLE_GST_DEBUG
833   {
834     if (ret) {
835       if (src_format == GST_FORMAT_TIME && dest_format == GST_FORMAT_BYTES) {
836         GST_LOG_OBJECT (parse,
837             "TIME -> BYTES: %" GST_TIME_FORMAT " -> %" G_GINT64_FORMAT,
838             GST_TIME_ARGS (src_value), *dest_value);
839       } else if (dest_format == GST_FORMAT_TIME &&
840           src_format == GST_FORMAT_BYTES) {
841         GST_LOG_OBJECT (parse,
842             "BYTES -> TIME: %" G_GINT64_FORMAT " -> %" GST_TIME_FORMAT,
843             src_value, GST_TIME_ARGS (*dest_value));
844       } else {
845         GST_LOG_OBJECT (parse,
846             "%s -> %s: %" G_GINT64_FORMAT " -> %" G_GINT64_FORMAT,
847             GST_STR_NULL (gst_format_get_name (src_format)),
848             GST_STR_NULL (gst_format_get_name (dest_format)),
849             src_value, *dest_value);
850       }
851     } else {
852       GST_DEBUG_OBJECT (parse, "conversion failed");
853     }
854   }
855 #endif
856
857   return ret;
858 }
859
860 /* gst_base_parse_sink_event:
861  * @pad: #GstPad that received the event.
862  * @event: #GstEvent to be handled.
863  *
864  * Handler for sink pad events.
865  *
866  * Returns: TRUE if the event was handled.
867  */
868 static gboolean
869 gst_base_parse_sink_event (GstPad * pad, GstEvent * event)
870 {
871   GstBaseParse *parse;
872   GstBaseParseClass *bclass;
873   gboolean handled = FALSE;
874   gboolean ret = TRUE;
875
876   parse = GST_BASE_PARSE (gst_pad_get_parent (pad));
877   bclass = GST_BASE_PARSE_GET_CLASS (parse);
878
879   GST_DEBUG_OBJECT (parse, "handling event %d, %s", GST_EVENT_TYPE (event),
880       GST_EVENT_TYPE_NAME (event));
881
882   /* Cache all events except EOS, NEWSEGMENT and FLUSH_STOP if we have a
883    * pending segment */
884   if (parse->priv->pending_segment && GST_EVENT_TYPE (event) != GST_EVENT_EOS
885       && GST_EVENT_TYPE (event) != GST_EVENT_NEWSEGMENT
886       && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_START
887       && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP) {
888
889     if (GST_EVENT_TYPE (event) == GST_EVENT_TAG)
890       /* See if any bitrate tags were posted */
891       gst_base_parse_handle_tag (parse, event);
892
893     parse->priv->pending_events =
894         g_list_append (parse->priv->pending_events, event);
895     ret = TRUE;
896   } else {
897
898     if (GST_EVENT_TYPE (event) == GST_EVENT_EOS &&
899         parse->priv->framecount < MIN_FRAMES_TO_POST_BITRATE)
900       /* We've not posted bitrate tags yet - do so now */
901       gst_base_parse_post_bitrates (parse, TRUE, TRUE, TRUE);
902
903     if (bclass->event)
904       handled = bclass->event (parse, event);
905
906     if (!handled)
907       handled = gst_base_parse_sink_eventfunc (parse, event);
908
909     if (!handled)
910       ret = gst_pad_event_default (pad, event);
911   }
912
913   gst_object_unref (parse);
914   GST_DEBUG_OBJECT (parse, "event handled");
915   return ret;
916 }
917
918
919 /* gst_base_parse_sink_eventfunc:
920  * @parse: #GstBaseParse.
921  * @event: #GstEvent to be handled.
922  *
923  * Element-level event handler function.
924  *
925  * The event will be unreffed only if it has been handled and this
926  * function returns %TRUE
927  *
928  * Returns: %TRUE if the event was handled and not need forwarding.
929  */
930 static gboolean
931 gst_base_parse_sink_eventfunc (GstBaseParse * parse, GstEvent * event)
932 {
933   gboolean handled = FALSE;
934   GstEvent **eventp;
935
936   switch (GST_EVENT_TYPE (event)) {
937     case GST_EVENT_NEWSEGMENT:
938     {
939       gdouble rate, applied_rate;
940       GstFormat format;
941       gint64 start, stop, pos, next_ts, offset = 0;
942       gboolean update;
943
944       gst_event_parse_new_segment_full (event, &update, &rate, &applied_rate,
945           &format, &start, &stop, &pos);
946
947       GST_DEBUG_OBJECT (parse, "newseg rate %g, applied rate %g, "
948           "format %d, start = %" GST_TIME_FORMAT ", stop = %" GST_TIME_FORMAT
949           ", pos = %" GST_TIME_FORMAT, rate, applied_rate, format,
950           GST_TIME_ARGS (start), GST_TIME_ARGS (stop), GST_TIME_ARGS (pos));
951
952       if (format == GST_FORMAT_BYTES) {
953         GstClockTime seg_start, seg_stop;
954         GstBaseParseSeek *seek = NULL;
955         GSList *node;
956
957         /* stop time is allowed to be open-ended, but not start & pos */
958         seg_stop = GST_CLOCK_TIME_NONE;
959         seg_start = 0;
960         offset = pos;
961
962         GST_OBJECT_LOCK (parse);
963         for (node = parse->priv->pending_seeks; node; node = node->next) {
964           GstBaseParseSeek *tmp = node->data;
965
966           if (tmp->offset == pos) {
967             seek = tmp;
968             break;
969           }
970         }
971         parse->priv->pending_seeks =
972             g_slist_remove (parse->priv->pending_seeks, seek);
973         GST_OBJECT_UNLOCK (parse);
974
975         if (seek) {
976           GST_DEBUG_OBJECT (parse,
977               "Matched newsegment to%s seek: %" GST_SEGMENT_FORMAT,
978               seek->accurate ? " accurate" : "", &seek->segment);
979           seg_start = seek->segment.start;
980           seg_stop = seek->segment.stop;
981           next_ts = seek->start_ts;
982           parse->priv->exact_position = seek->accurate;
983           g_free (seek);
984         } else {
985           /* best attempt convert */
986           /* as these are only estimates, stop is kept open-ended to avoid
987            * premature cutting */
988           gst_base_parse_convert (parse, GST_FORMAT_BYTES, start,
989               GST_FORMAT_TIME, (gint64 *) & seg_start);
990           parse->priv->exact_position = (start == 0);
991           next_ts = seg_start;
992         }
993
994         gst_event_unref (event);
995         event = gst_event_new_new_segment_full (update, rate, applied_rate,
996             GST_FORMAT_TIME, seg_start, seg_stop, seg_start);
997         format = GST_FORMAT_TIME;
998         start = seg_start;
999         stop = seg_stop;
1000         GST_DEBUG_OBJECT (parse, "Converted incoming segment to TIME. "
1001             "start = %" GST_TIME_FORMAT ", stop = %" GST_TIME_FORMAT,
1002             GST_TIME_ARGS (seg_start), GST_TIME_ARGS (seg_stop));
1003       } else if (format != GST_FORMAT_TIME) {
1004         /* Unknown incoming segment format. Output a default open-ended
1005          * TIME segment */
1006         gst_event_unref (event);
1007         event = gst_event_new_new_segment_full (update, rate, applied_rate,
1008             GST_FORMAT_TIME, 0, GST_CLOCK_TIME_NONE, 0);
1009         format = GST_FORMAT_TIME;
1010         next_ts = start = 0;
1011         stop = GST_CLOCK_TIME_NONE;
1012       } else {
1013         /* not considered BYTE seekable if it is talking to us in TIME,
1014          * whatever else it might claim */
1015         parse->priv->upstream_seekable = FALSE;
1016         next_ts = start;
1017       }
1018
1019       gst_segment_set_newsegment_full (&parse->segment, update, rate,
1020           applied_rate, format, start, stop, start);
1021
1022       /* save the segment for later, right before we push a new buffer so that
1023        * the caps are fixed and the next linked element can receive
1024        * the segment. */
1025       eventp = &parse->priv->pending_segment;
1026       gst_event_replace (eventp, event);
1027       gst_event_unref (event);
1028       handled = TRUE;
1029
1030       /* but finish the current segment */
1031       GST_DEBUG_OBJECT (parse, "draining current segment");
1032       if (parse->segment.rate > 0.0)
1033         gst_base_parse_drain (parse);
1034       else
1035         gst_base_parse_process_fragment (parse, FALSE);
1036       gst_adapter_clear (parse->priv->adapter);
1037       parse->priv->offset = offset;
1038       parse->priv->sync_offset = offset;
1039       parse->priv->next_ts = next_ts;
1040       parse->priv->last_ts = GST_CLOCK_TIME_NONE;
1041       parse->priv->discont = TRUE;
1042       parse->priv->seen_keyframe = FALSE;
1043       break;
1044     }
1045
1046     case GST_EVENT_FLUSH_START:
1047       parse->priv->flushing = TRUE;
1048       handled = gst_pad_push_event (parse->srcpad, gst_event_ref (event));
1049       if (handled)
1050         gst_event_unref (event);
1051       /* Wait for _chain() to exit by taking the srcpad STREAM_LOCK */
1052       GST_PAD_STREAM_LOCK (parse->srcpad);
1053       GST_PAD_STREAM_UNLOCK (parse->srcpad);
1054
1055       break;
1056
1057     case GST_EVENT_FLUSH_STOP:
1058       gst_adapter_clear (parse->priv->adapter);
1059       gst_base_parse_clear_queues (parse);
1060       parse->priv->flushing = FALSE;
1061       parse->priv->discont = TRUE;
1062       parse->priv->last_ts = GST_CLOCK_TIME_NONE;
1063       parse->priv->frame._private_flags |=
1064           GST_BASE_PARSE_FRAME_PRIVATE_FLAG_NOALLOC;
1065       gst_base_parse_frame_free (&parse->priv->frame);
1066       break;
1067
1068     case GST_EVENT_EOS:
1069       if (parse->segment.rate > 0.0)
1070         gst_base_parse_drain (parse);
1071       else
1072         gst_base_parse_process_fragment (parse, FALSE);
1073
1074       /* If we STILL have zero frames processed, fire an error */
1075       if (parse->priv->framecount == 0) {
1076         GST_ELEMENT_ERROR (parse, STREAM, WRONG_TYPE,
1077             ("No valid frames found before end of stream"), (NULL));
1078       }
1079       /* newsegment before eos */
1080       if (parse->priv->pending_segment) {
1081         gst_pad_push_event (parse->srcpad, parse->priv->pending_segment);
1082         parse->priv->pending_segment = NULL;
1083       }
1084       break;
1085
1086     default:
1087       break;
1088   }
1089
1090   return handled;
1091 }
1092
1093
1094 /* gst_base_parse_src_event:
1095  * @pad: #GstPad that received the event.
1096  * @event: #GstEvent that was received.
1097  *
1098  * Handler for source pad events.
1099  *
1100  * Returns: TRUE if the event was handled.
1101  */
1102 static gboolean
1103 gst_base_parse_src_event (GstPad * pad, GstEvent * event)
1104 {
1105   GstBaseParse *parse;
1106   GstBaseParseClass *bclass;
1107   gboolean handled = FALSE;
1108   gboolean ret = TRUE;
1109
1110   parse = GST_BASE_PARSE (gst_pad_get_parent (pad));
1111   bclass = GST_BASE_PARSE_GET_CLASS (parse);
1112
1113   GST_DEBUG_OBJECT (parse, "event %d, %s", GST_EVENT_TYPE (event),
1114       GST_EVENT_TYPE_NAME (event));
1115
1116   if (bclass->src_event)
1117     handled = bclass->src_event (parse, event);
1118
1119   if (!handled)
1120     ret = gst_pad_event_default (pad, event);
1121
1122   gst_object_unref (parse);
1123   return ret;
1124 }
1125
1126 static gboolean
1127 gst_base_parse_is_seekable (GstBaseParse * parse)
1128 {
1129   /* FIXME: could do more here, e.g. check index or just send data from 0
1130    * in pull mode and let decoder/sink clip */
1131   return parse->priv->syncable;
1132 }
1133
1134 /* gst_base_parse_src_eventfunc:
1135  * @parse: #GstBaseParse.
1136  * @event: #GstEvent that was received.
1137  *
1138  * Default srcpad event handler.
1139  *
1140  * Returns: TRUE if the event was handled and can be dropped.
1141  */
1142 static gboolean
1143 gst_base_parse_src_eventfunc (GstBaseParse * parse, GstEvent * event)
1144 {
1145   gboolean handled = FALSE;
1146
1147   switch (GST_EVENT_TYPE (event)) {
1148     case GST_EVENT_SEEK:
1149     {
1150       if (gst_base_parse_is_seekable (parse)) {
1151         handled = gst_base_parse_handle_seek (parse, event);
1152       }
1153       break;
1154     }
1155     default:
1156       break;
1157   }
1158   return handled;
1159 }
1160
1161
1162 /**
1163  * gst_base_parse_convert_default:
1164  * @parse: #GstBaseParse.
1165  * @src_format: #GstFormat describing the source format.
1166  * @src_value: Source value to be converted.
1167  * @dest_format: #GstFormat defining the converted format.
1168  * @dest_value: Pointer where the conversion result will be put.
1169  *
1170  * Default implementation of "convert" vmethod in #GstBaseParse class.
1171  *
1172  * Returns: TRUE if conversion was successful.
1173  *
1174  * Since: 0.10.33
1175  */
1176 gboolean
1177 gst_base_parse_convert_default (GstBaseParse * parse,
1178     GstFormat src_format,
1179     gint64 src_value, GstFormat dest_format, gint64 * dest_value)
1180 {
1181   gboolean ret = FALSE;
1182   guint64 bytes, duration;
1183
1184   if (G_UNLIKELY (src_format == dest_format)) {
1185     *dest_value = src_value;
1186     return TRUE;
1187   }
1188
1189   if (G_UNLIKELY (src_value == -1)) {
1190     *dest_value = -1;
1191     return TRUE;
1192   }
1193
1194   if (G_UNLIKELY (src_value == 0)) {
1195     *dest_value = 0;
1196     return TRUE;
1197   }
1198
1199   /* need at least some frames */
1200   if (!parse->priv->framecount)
1201     return FALSE;
1202
1203   duration = parse->priv->acc_duration / GST_MSECOND;
1204   bytes = parse->priv->bytecount;
1205
1206   if (G_UNLIKELY (!duration || !bytes))
1207     return FALSE;
1208
1209   if (src_format == GST_FORMAT_BYTES) {
1210     if (dest_format == GST_FORMAT_TIME) {
1211       /* BYTES -> TIME conversion */
1212       GST_DEBUG_OBJECT (parse, "converting bytes -> time");
1213       *dest_value = gst_util_uint64_scale (src_value, duration, bytes);
1214       *dest_value *= GST_MSECOND;
1215       GST_DEBUG_OBJECT (parse, "conversion result: %" G_GINT64_FORMAT " ms",
1216           *dest_value / GST_MSECOND);
1217       ret = TRUE;
1218     }
1219   } else if (src_format == GST_FORMAT_TIME) {
1220     if (dest_format == GST_FORMAT_BYTES) {
1221       GST_DEBUG_OBJECT (parse, "converting time -> bytes");
1222       *dest_value = gst_util_uint64_scale (src_value / GST_MSECOND, bytes,
1223           duration);
1224       GST_DEBUG_OBJECT (parse,
1225           "time %" G_GINT64_FORMAT " ms in bytes = %" G_GINT64_FORMAT,
1226           src_value / GST_MSECOND, *dest_value);
1227       ret = TRUE;
1228     }
1229   } else if (src_format == GST_FORMAT_DEFAULT) {
1230     /* DEFAULT == frame-based */
1231     if (dest_format == GST_FORMAT_TIME) {
1232       if (parse->priv->fps_den) {
1233         *dest_value = gst_util_uint64_scale (src_value,
1234             GST_SECOND * parse->priv->fps_den, parse->priv->fps_num);
1235         ret = TRUE;
1236       }
1237     } else if (dest_format == GST_FORMAT_BYTES) {
1238     }
1239   }
1240
1241   return ret;
1242 }
1243
1244 static void
1245 gst_base_parse_update_duration (GstBaseParse * baseparse)
1246 {
1247   GstPad *peer;
1248   GstBaseParse *parse;
1249
1250   parse = GST_BASE_PARSE (baseparse);
1251
1252   peer = gst_pad_get_peer (parse->sinkpad);
1253   if (peer) {
1254     GstFormat pformat = GST_FORMAT_BYTES;
1255     gboolean qres = FALSE;
1256     gint64 ptot, dest_value;
1257
1258     qres = gst_pad_query_duration (peer, &pformat, &ptot);
1259     gst_object_unref (GST_OBJECT (peer));
1260     if (qres) {
1261       if (gst_base_parse_convert (parse, pformat, ptot,
1262               GST_FORMAT_TIME, &dest_value)) {
1263         parse->priv->estimated_duration = dest_value;
1264         GST_LOG_OBJECT (parse,
1265             "updated estimated duration to %" GST_TIME_FORMAT,
1266             GST_TIME_ARGS (dest_value));
1267       }
1268     }
1269   }
1270 }
1271
1272 static void
1273 gst_base_parse_post_bitrates (GstBaseParse * parse, gboolean post_min,
1274     gboolean post_avg, gboolean post_max)
1275 {
1276   GstTagList *taglist = NULL;
1277
1278   if (post_min && parse->priv->post_min_bitrate) {
1279     taglist = gst_tag_list_new ();
1280
1281     gst_tag_list_add (taglist, GST_TAG_MERGE_REPLACE,
1282         GST_TAG_MINIMUM_BITRATE, parse->priv->min_bitrate, NULL);
1283   }
1284
1285   if (post_avg && parse->priv->post_avg_bitrate) {
1286     if (taglist == NULL)
1287       taglist = gst_tag_list_new ();
1288
1289     parse->priv->posted_avg_bitrate = parse->priv->avg_bitrate;
1290     gst_tag_list_add (taglist, GST_TAG_MERGE_REPLACE, GST_TAG_BITRATE,
1291         parse->priv->avg_bitrate, NULL);
1292   }
1293
1294   if (post_max && parse->priv->post_max_bitrate) {
1295     if (taglist == NULL)
1296       taglist = gst_tag_list_new ();
1297
1298     gst_tag_list_add (taglist, GST_TAG_MERGE_REPLACE,
1299         GST_TAG_MAXIMUM_BITRATE, parse->priv->max_bitrate, NULL);
1300   }
1301
1302   GST_DEBUG_OBJECT (parse, "Updated bitrates. Min: %u, Avg: %u, Max: %u",
1303       parse->priv->min_bitrate, parse->priv->avg_bitrate,
1304       parse->priv->max_bitrate);
1305
1306   if (taglist != NULL) {
1307     gst_element_found_tags_for_pad (GST_ELEMENT_CAST (parse), parse->srcpad,
1308         taglist);
1309   }
1310 }
1311
1312 /* gst_base_parse_update_bitrates:
1313  * @parse: #GstBaseParse.
1314  * @buffer: Current frame as a #GstBuffer
1315  *
1316  * Keeps track of the minimum and maximum bitrates, and also maintains a
1317  * running average bitrate of the stream so far.
1318  */
1319 static void
1320 gst_base_parse_update_bitrates (GstBaseParse * parse, GstBaseParseFrame * frame)
1321 {
1322   /* Only update the tag on a 10 kbps delta */
1323   static const gint update_threshold = 10000;
1324
1325   guint64 data_len, frame_dur;
1326   gint overhead, frame_bitrate, old_avg_bitrate;
1327   gboolean update_min = FALSE, update_avg = FALSE, update_max = FALSE;
1328   GstBuffer *buffer = frame->buffer;
1329
1330   overhead = frame->overhead;
1331   if (overhead == -1)
1332     return;
1333
1334   data_len = GST_BUFFER_SIZE (buffer) - overhead;
1335   parse->priv->data_bytecount += data_len;
1336
1337   /* duration should be valid by now,
1338    * either set by subclass or maybe based on fps settings */
1339   if (GST_BUFFER_DURATION_IS_VALID (buffer) && parse->priv->acc_duration != 0) {
1340     /* Calculate duration of a frame from buffer properties */
1341     frame_dur = GST_BUFFER_DURATION (buffer);
1342     parse->priv->avg_bitrate = (8 * parse->priv->data_bytecount * GST_SECOND) /
1343         parse->priv->acc_duration;
1344
1345   } else {
1346     /* No way to figure out frame duration (is this even possible?) */
1347     return;
1348   }
1349
1350   /* override if subclass provided bitrate, e.g. metadata based */
1351   if (parse->priv->bitrate) {
1352     parse->priv->avg_bitrate = parse->priv->bitrate;
1353     /* spread this (confirmed) info ASAP */
1354     if (parse->priv->posted_avg_bitrate != parse->priv->avg_bitrate)
1355       gst_base_parse_post_bitrates (parse, FALSE, TRUE, FALSE);
1356   }
1357
1358   if (frame_dur)
1359     frame_bitrate = (8 * data_len * GST_SECOND) / frame_dur;
1360   else
1361     return;
1362
1363   GST_LOG_OBJECT (parse, "frame bitrate %u, avg bitrate %u", frame_bitrate,
1364       parse->priv->avg_bitrate);
1365
1366   if (parse->priv->framecount < MIN_FRAMES_TO_POST_BITRATE) {
1367     goto exit;
1368   } else if (parse->priv->framecount == MIN_FRAMES_TO_POST_BITRATE) {
1369     /* always post all at threshold time */
1370     update_min = update_max = update_avg = TRUE;
1371   }
1372
1373   if (G_LIKELY (parse->priv->framecount >= MIN_FRAMES_TO_POST_BITRATE)) {
1374     if (frame_bitrate < parse->priv->min_bitrate) {
1375       parse->priv->min_bitrate = frame_bitrate;
1376       update_min = TRUE;
1377     }
1378
1379     if (frame_bitrate > parse->priv->max_bitrate) {
1380       parse->priv->max_bitrate = frame_bitrate;
1381       update_max = TRUE;
1382     }
1383
1384     old_avg_bitrate = parse->priv->posted_avg_bitrate;
1385     if ((gint) (old_avg_bitrate - parse->priv->avg_bitrate) > update_threshold
1386         || (gint) (parse->priv->avg_bitrate - old_avg_bitrate) >
1387         update_threshold)
1388       update_avg = TRUE;
1389   }
1390
1391   if ((update_min || update_avg || update_max))
1392     gst_base_parse_post_bitrates (parse, update_min, update_avg, update_max);
1393
1394   /* If average bitrate changes that much and no valid (time) duration provided,
1395    * then post a new duration message so applications can update their cached
1396    * values */
1397   if (update_avg && !(parse->priv->duration_fmt == GST_FORMAT_TIME &&
1398           GST_CLOCK_TIME_IS_VALID (parse->priv->duration)))
1399     gst_element_post_message (GST_ELEMENT (parse),
1400         gst_message_new_duration (GST_OBJECT (parse), GST_FORMAT_TIME, -1));
1401
1402 exit:
1403   return;
1404 }
1405
1406 /**
1407  * gst_base_parse_add_index_entry:
1408  * @parse: #GstBaseParse.
1409  * @offset: offset of entry
1410  * @ts: timestamp associated with offset
1411  * @key: whether entry refers to keyframe
1412  * @force: add entry disregarding sanity checks
1413  *
1414  * Adds an entry to the index associating @offset to @ts.  It is recommended
1415  * to only add keyframe entries.  @force allows to bypass checks, such as
1416  * whether the stream is (upstream) seekable, another entry is already "close"
1417  * to the new entry, etc.
1418  *
1419  * Returns: #gboolean indicating whether entry was added
1420  *
1421  * Since: 0.10.33
1422  */
1423 gboolean
1424 gst_base_parse_add_index_entry (GstBaseParse * parse, guint64 offset,
1425     GstClockTime ts, gboolean key, gboolean force)
1426 {
1427   gboolean ret = FALSE;
1428   GstIndexAssociation associations[2];
1429
1430   GST_LOG_OBJECT (parse, "Adding key=%d index entry %" GST_TIME_FORMAT
1431       " @ offset 0x%08" G_GINT64_MODIFIER "x", key, GST_TIME_ARGS (ts), offset);
1432
1433   if (G_LIKELY (!force)) {
1434
1435     if (!parse->priv->upstream_seekable) {
1436       GST_DEBUG_OBJECT (parse, "upstream not seekable; discarding");
1437       goto exit;
1438     }
1439
1440     /* FIXME need better helper data structure that handles these issues
1441      * related to ongoing collecting of index entries */
1442     if (parse->priv->index_last_offset >= (gint64) offset) {
1443       GST_DEBUG_OBJECT (parse, "already have entries up to offset "
1444           "0x%08" G_GINT64_MODIFIER "x", parse->priv->index_last_offset);
1445       goto exit;
1446     }
1447
1448     if (GST_CLOCK_TIME_IS_VALID (parse->priv->index_last_ts) &&
1449         GST_CLOCK_DIFF (parse->priv->index_last_ts, ts) <
1450         parse->priv->idx_interval) {
1451       GST_DEBUG_OBJECT (parse, "entry too close to last time %" GST_TIME_FORMAT,
1452           GST_TIME_ARGS (parse->priv->index_last_ts));
1453       goto exit;
1454     }
1455
1456     /* if last is not really the last one */
1457     if (!parse->priv->index_last_valid) {
1458       GstClockTime prev_ts;
1459
1460       gst_base_parse_find_offset (parse, ts, TRUE, &prev_ts);
1461       if (GST_CLOCK_DIFF (prev_ts, ts) < parse->priv->idx_interval) {
1462         GST_DEBUG_OBJECT (parse,
1463             "entry too close to existing entry %" GST_TIME_FORMAT,
1464             GST_TIME_ARGS (prev_ts));
1465         parse->priv->index_last_offset = offset;
1466         parse->priv->index_last_ts = ts;
1467         goto exit;
1468       }
1469     }
1470   }
1471
1472   associations[0].format = GST_FORMAT_TIME;
1473   associations[0].value = ts;
1474   associations[1].format = GST_FORMAT_BYTES;
1475   associations[1].value = offset;
1476
1477   /* index might change on-the-fly, although that would be nutty app ... */
1478   g_static_mutex_lock (&parse->priv->index_lock);
1479   gst_index_add_associationv (parse->priv->index, parse->priv->index_id,
1480       (key) ? GST_ASSOCIATION_FLAG_KEY_UNIT : GST_ASSOCIATION_FLAG_DELTA_UNIT,
1481       2, (const GstIndexAssociation *) &associations);
1482   g_static_mutex_unlock (&parse->priv->index_lock);
1483
1484   if (key) {
1485     parse->priv->index_last_offset = offset;
1486     parse->priv->index_last_ts = ts;
1487   }
1488
1489   ret = TRUE;
1490
1491 exit:
1492   return ret;
1493 }
1494
1495 /* check for seekable upstream, above and beyond a mere query */
1496 static void
1497 gst_base_parse_check_seekability (GstBaseParse * parse)
1498 {
1499   GstQuery *query;
1500   gboolean seekable = FALSE;
1501   gint64 start = -1, stop = -1;
1502   guint idx_interval = 0;
1503
1504   query = gst_query_new_seeking (GST_FORMAT_BYTES);
1505   if (!gst_pad_peer_query (parse->sinkpad, query)) {
1506     GST_DEBUG_OBJECT (parse, "seeking query failed");
1507     goto done;
1508   }
1509
1510   gst_query_parse_seeking (query, NULL, &seekable, &start, &stop);
1511
1512   /* try harder to query upstream size if we didn't get it the first time */
1513   if (seekable && stop == -1) {
1514     GstFormat fmt = GST_FORMAT_BYTES;
1515
1516     GST_DEBUG_OBJECT (parse, "doing duration query to fix up unset stop");
1517     gst_pad_query_peer_duration (parse->sinkpad, &fmt, &stop);
1518   }
1519
1520   /* if upstream doesn't know the size, it's likely that it's not seekable in
1521    * practice even if it technically may be seekable */
1522   if (seekable && (start != 0 || stop <= start)) {
1523     GST_DEBUG_OBJECT (parse, "seekable but unknown start/stop -> disable");
1524     seekable = FALSE;
1525   }
1526
1527   /* let's not put every single frame into our index */
1528   if (seekable) {
1529     if (stop < 10 * 1024 * 1024)
1530       idx_interval = 100;
1531     else if (stop < 100 * 1024 * 1024)
1532       idx_interval = 500;
1533     else
1534       idx_interval = 1000;
1535   }
1536
1537 done:
1538   gst_query_unref (query);
1539
1540   GST_DEBUG_OBJECT (parse, "seekable: %d (%" G_GUINT64_FORMAT " - %"
1541       G_GUINT64_FORMAT ")", seekable, start, stop);
1542   parse->priv->upstream_seekable = seekable;
1543   parse->priv->upstream_size = seekable ? stop : 0;
1544
1545   GST_DEBUG_OBJECT (parse, "idx_interval: %ums", idx_interval);
1546   parse->priv->idx_interval = idx_interval * GST_MSECOND;
1547 }
1548
1549 /* some misc checks on upstream */
1550 static void
1551 gst_base_parse_check_upstream (GstBaseParse * parse)
1552 {
1553   GstFormat fmt = GST_FORMAT_TIME;
1554   gint64 stop;
1555
1556   if (gst_pad_query_peer_duration (parse->sinkpad, &fmt, &stop))
1557     if (GST_CLOCK_TIME_IS_VALID (stop) && stop) {
1558       /* upstream has one, accept it also, and no further updates */
1559       gst_base_parse_set_duration (parse, GST_FORMAT_TIME, stop, 0);
1560       parse->priv->upstream_has_duration = TRUE;
1561     }
1562
1563   GST_DEBUG_OBJECT (parse, "upstream_has_duration: %d",
1564       parse->priv->upstream_has_duration);
1565 }
1566
1567 /* checks src caps to determine if dealing with audio or video */
1568 /* TODO maybe forego automagic stuff and let subclass configure it ? */
1569 static void
1570 gst_base_parse_check_media (GstBaseParse * parse)
1571 {
1572   GstCaps *caps;
1573   GstStructure *s;
1574
1575   caps = GST_PAD_CAPS (parse->srcpad);
1576   if (G_LIKELY (caps) && (s = gst_caps_get_structure (caps, 0))) {
1577     parse->priv->is_video =
1578         g_str_has_prefix (gst_structure_get_name (s), "video");
1579   } else {
1580     /* historical default */
1581     parse->priv->is_video = FALSE;
1582   }
1583
1584   GST_DEBUG_OBJECT (parse, "media is video == %d", parse->priv->is_video);
1585 }
1586
1587 /* takes ownership of frame */
1588 static void
1589 gst_base_parse_queue_frame (GstBaseParse * parse, GstBaseParseFrame * frame)
1590 {
1591   if (!(frame->_private_flags & GST_BASE_PARSE_FRAME_PRIVATE_FLAG_NOALLOC)) {
1592     /* frame allocated on the heap, we can just take ownership */
1593     g_queue_push_tail (&parse->priv->queued_frames, frame);
1594     GST_TRACE ("queued frame %p", frame);
1595   } else {
1596     GstBaseParseFrame *copy;
1597
1598     /* probably allocated on the stack, must make a proper copy */
1599     copy = gst_base_parse_frame_copy (frame);
1600     g_queue_push_tail (&parse->priv->queued_frames, copy);
1601     GST_TRACE ("queued frame %p (copy of %p)", copy, frame);
1602     gst_base_parse_frame_free (frame);
1603   }
1604 }
1605
1606 /* gst_base_parse_handle_and_push_buffer:
1607  * @parse: #GstBaseParse.
1608  * @klass: #GstBaseParseClass.
1609  * @frame: (transfer full): a #GstBaseParseFrame
1610  *
1611  * Parses the frame from given buffer and pushes it forward. Also performs
1612  * timestamp handling and checks the segment limits.
1613  *
1614  * This is called with srcpad STREAM_LOCK held.
1615  *
1616  * Returns: #GstFlowReturn
1617  */
1618 static GstFlowReturn
1619 gst_base_parse_handle_and_push_frame (GstBaseParse * parse,
1620     GstBaseParseClass * klass, GstBaseParseFrame * frame)
1621 {
1622   GstFlowReturn ret;
1623   gint64 offset;
1624   GstBuffer *buffer;
1625
1626   g_return_val_if_fail (frame != NULL, GST_FLOW_ERROR);
1627
1628   buffer = frame->buffer;
1629
1630   if (parse->priv->discont) {
1631     GST_DEBUG_OBJECT (parse, "marking DISCONT");
1632     GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT);
1633     parse->priv->discont = FALSE;
1634   }
1635
1636   /* some one-time start-up */
1637   if (G_UNLIKELY (!parse->priv->framecount)) {
1638     gst_base_parse_check_seekability (parse);
1639     gst_base_parse_check_upstream (parse);
1640   }
1641
1642   GST_LOG_OBJECT (parse,
1643       "parsing frame at offset %" G_GUINT64_FORMAT
1644       " (%#" G_GINT64_MODIFIER "x) of size %d",
1645       GST_BUFFER_OFFSET (buffer), GST_BUFFER_OFFSET (buffer),
1646       GST_BUFFER_SIZE (buffer));
1647
1648   /* use default handler to provide initial (upstream) metadata */
1649   gst_base_parse_parse_frame (parse, frame);
1650
1651   /* store offset as it might get overwritten */
1652   offset = GST_BUFFER_OFFSET (buffer);
1653   ret = klass->parse_frame (parse, frame);
1654   /* sync */
1655   buffer = frame->buffer;
1656   /* subclass must play nice */
1657   g_return_val_if_fail (buffer != NULL, GST_FLOW_ERROR);
1658
1659   /* check if subclass/format can provide ts.
1660    * If so, that allows and enables extra seek and duration determining options */
1661   if (G_UNLIKELY (parse->priv->first_frame_offset < 0 && ret == GST_FLOW_OK)) {
1662     if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer) && parse->priv->has_timing_info
1663         && parse->priv->pad_mode == GST_ACTIVATE_PULL) {
1664       parse->priv->first_frame_offset = offset;
1665       parse->priv->first_frame_ts = GST_BUFFER_TIMESTAMP (buffer);
1666       GST_DEBUG_OBJECT (parse, "subclass provided ts %" GST_TIME_FORMAT
1667           " for first frame at offset %" G_GINT64_FORMAT,
1668           GST_TIME_ARGS (parse->priv->first_frame_ts),
1669           parse->priv->first_frame_offset);
1670       if (!GST_CLOCK_TIME_IS_VALID (parse->priv->duration)) {
1671         gint64 off;
1672         GstClockTime last_ts = G_MAXINT64;
1673
1674         GST_DEBUG_OBJECT (parse, "no duration; trying scan to determine");
1675         gst_base_parse_locate_time (parse, &last_ts, &off);
1676         if (GST_CLOCK_TIME_IS_VALID (last_ts))
1677           gst_base_parse_set_duration (parse, GST_FORMAT_TIME, last_ts, 0);
1678       }
1679     } else {
1680       /* disable further checks */
1681       parse->priv->first_frame_offset = 0;
1682     }
1683   }
1684
1685   /* again use default handler to add missing metadata;
1686    * we may have new information on frame properties */
1687   gst_base_parse_parse_frame (parse, frame);
1688   if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer) &&
1689       GST_BUFFER_DURATION_IS_VALID (buffer)) {
1690     parse->priv->next_ts =
1691         GST_BUFFER_TIMESTAMP (buffer) + GST_BUFFER_DURATION (buffer);
1692   } else {
1693     /* we lost track, do not produce bogus time next time around
1694      * (probably means parser subclass has given up on parsing as well) */
1695     GST_DEBUG_OBJECT (parse, "no next fallback timestamp");
1696     parse->priv->next_ts = GST_CLOCK_TIME_NONE;
1697   }
1698
1699   if (parse->priv->upstream_seekable && parse->priv->exact_position &&
1700       GST_BUFFER_TIMESTAMP_IS_VALID (buffer))
1701     gst_base_parse_add_index_entry (parse, offset,
1702         GST_BUFFER_TIMESTAMP (buffer),
1703         !GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT), FALSE);
1704
1705   /* First buffers are dropped, this means that the subclass needs more
1706    * frames to decide on the format and queues them internally */
1707   /* convert internal flow to OK and mark discont for the next buffer. */
1708   if (ret == GST_BASE_PARSE_FLOW_DROPPED) {
1709     gst_base_parse_frame_free (frame);
1710     return GST_FLOW_OK;
1711   } else if (ret == GST_BASE_PARSE_FLOW_QUEUED) {
1712     gst_base_parse_queue_frame (parse, frame);
1713     return GST_FLOW_OK;
1714   } else if (ret != GST_FLOW_OK) {
1715     return ret;
1716   }
1717
1718   /* All OK, push queued frames if there are any */
1719   if (G_UNLIKELY (!g_queue_is_empty (&parse->priv->queued_frames))) {
1720     GstBaseParseFrame *queued_frame;
1721
1722     while ((queued_frame = g_queue_pop_head (&parse->priv->queued_frames))) {
1723       queued_frame->buffer =
1724           gst_buffer_make_metadata_writable (queued_frame->buffer);
1725       gst_buffer_set_caps (queued_frame->buffer,
1726           GST_PAD_CAPS (GST_BASE_PARSE_SRC_PAD (parse)));
1727       gst_base_parse_push_frame (parse, queued_frame);
1728       gst_base_parse_frame_free (queued_frame);
1729     }
1730   }
1731
1732   return gst_base_parse_push_frame (parse, frame);
1733 }
1734
1735 /**
1736  * gst_base_parse_push_frame:
1737  * @parse: #GstBaseParse.
1738  * @frame: (transfer full): a #GstBaseParseFrame
1739  *
1740  * Pushes the frame downstream, sends any pending events and
1741  * does some timestamp and segment handling. Takes ownership
1742  * of @frame and will clear it (if it was initialised with
1743  * gst_base_parse_frame_init()) or free it.
1744  *
1745  * This must be called with sinkpad STREAM_LOCK held.
1746  *
1747  * Returns: #GstFlowReturn
1748  *
1749  * Since: 0.10.33
1750  */
1751 GstFlowReturn
1752 gst_base_parse_push_frame (GstBaseParse * parse, GstBaseParseFrame * frame)
1753 {
1754   GstFlowReturn ret = GST_FLOW_OK;
1755   GstClockTime last_start = GST_CLOCK_TIME_NONE;
1756   GstClockTime last_stop = GST_CLOCK_TIME_NONE;
1757   GstBaseParseClass *klass = GST_BASE_PARSE_GET_CLASS (parse);
1758   GstBuffer *buffer;
1759
1760   g_return_val_if_fail (frame != NULL, GST_FLOW_ERROR);
1761   g_return_val_if_fail (frame->buffer != NULL, GST_FLOW_ERROR);
1762
1763   GST_TRACE_OBJECT (parse, "pushing frame %p", frame);
1764
1765   buffer = frame->buffer;
1766
1767   GST_LOG_OBJECT (parse,
1768       "processing buffer of size %d with ts %" GST_TIME_FORMAT
1769       ", duration %" GST_TIME_FORMAT, GST_BUFFER_SIZE (buffer),
1770       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
1771       GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));
1772
1773   /* update stats */
1774   parse->priv->bytecount += GST_BUFFER_SIZE (buffer);
1775   if (G_LIKELY (!(frame->flags & GST_BASE_PARSE_FRAME_FLAG_NO_FRAME))) {
1776     parse->priv->framecount++;
1777     if (GST_BUFFER_DURATION_IS_VALID (buffer)) {
1778       parse->priv->acc_duration += GST_BUFFER_DURATION (buffer);
1779     }
1780   }
1781   /* 0 means disabled */
1782   if (parse->priv->update_interval < 0)
1783     parse->priv->update_interval = 50;
1784   else if (parse->priv->update_interval > 0 &&
1785       (parse->priv->framecount % parse->priv->update_interval) == 0)
1786     gst_base_parse_update_duration (parse);
1787
1788   if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer))
1789     last_start = last_stop = GST_BUFFER_TIMESTAMP (buffer);
1790   if (last_start != GST_CLOCK_TIME_NONE
1791       && GST_BUFFER_DURATION_IS_VALID (buffer))
1792     last_stop = last_start + GST_BUFFER_DURATION (buffer);
1793
1794   /* should have caps by now */
1795   g_return_val_if_fail (GST_PAD_CAPS (parse->srcpad), GST_FLOW_ERROR);
1796
1797   /* segment adjustment magic; only if we are running the whole show */
1798   if (!parse->priv->passthrough && parse->segment.rate > 0.0 &&
1799       (parse->priv->pad_mode == GST_ACTIVATE_PULL ||
1800           parse->priv->upstream_seekable)) {
1801     /* segment times are typically estimates,
1802      * actual frame data might lead subclass to different timestamps,
1803      * so override segment start from what is supplied there */
1804     if (G_UNLIKELY (parse->priv->pending_segment && !parse->priv->exact_position
1805             && GST_CLOCK_TIME_IS_VALID (last_start))) {
1806       gst_event_unref (parse->priv->pending_segment);
1807       parse->segment.start =
1808           MIN ((guint64) last_start, (guint64) parse->segment.stop);
1809       GST_DEBUG_OBJECT (parse,
1810           "adjusting pending segment start to %" GST_TIME_FORMAT,
1811           GST_TIME_ARGS (parse->segment.start));
1812       parse->priv->pending_segment =
1813           gst_event_new_new_segment (FALSE, parse->segment.rate,
1814           parse->segment.format, parse->segment.start,
1815           parse->segment.stop, parse->segment.start);
1816     }
1817     /* handle gaps, e.g. non-zero start-time, in as much not handled by above */
1818     if (GST_CLOCK_TIME_IS_VALID (parse->segment.last_stop) &&
1819         GST_CLOCK_TIME_IS_VALID (last_start)) {
1820       GstClockTimeDiff diff;
1821
1822       /* only send newsegments with increasing start times,
1823        * otherwise if these go back and forth downstream (sinks) increase
1824        * accumulated time and running_time */
1825       diff = GST_CLOCK_DIFF (parse->segment.last_stop, last_start);
1826       if (G_UNLIKELY (diff > 2 * GST_SECOND
1827               && last_start > parse->segment.start
1828               && (!GST_CLOCK_TIME_IS_VALID (parse->segment.stop)
1829                   || last_start < parse->segment.stop))) {
1830         GST_DEBUG_OBJECT (parse,
1831             "Gap of %" G_GINT64_FORMAT " ns detected in stream " "(%"
1832             GST_TIME_FORMAT " -> %" GST_TIME_FORMAT "). "
1833             "Sending updated NEWSEGMENT events", diff,
1834             GST_TIME_ARGS (parse->segment.last_stop),
1835             GST_TIME_ARGS (last_start));
1836         if (G_UNLIKELY (parse->priv->pending_segment)) {
1837           gst_event_unref (parse->priv->pending_segment);
1838           parse->segment.start = last_start;
1839           parse->priv->pending_segment =
1840               gst_event_new_new_segment (FALSE, parse->segment.rate,
1841               parse->segment.format, parse->segment.start,
1842               parse->segment.stop, parse->segment.start);
1843         } else {
1844           /* send newsegment events such that the gap is not accounted in
1845            * accum time, hence running_time */
1846           /* close ahead of gap */
1847           gst_pad_push_event (parse->srcpad,
1848               gst_event_new_new_segment (TRUE, parse->segment.rate,
1849                   parse->segment.format, parse->segment.last_stop,
1850                   parse->segment.last_stop, parse->segment.last_stop));
1851           /* skip gap */
1852           gst_pad_push_event (parse->srcpad,
1853               gst_event_new_new_segment (FALSE, parse->segment.rate,
1854                   parse->segment.format, last_start,
1855                   parse->segment.stop, last_start));
1856         }
1857         /* align segment view with downstream,
1858          * prevents double-counting accum when closing segment */
1859         gst_segment_set_newsegment (&parse->segment, FALSE,
1860             parse->segment.rate, parse->segment.format, last_start,
1861             parse->segment.stop, last_start);
1862         parse->segment.last_stop = last_start;
1863       }
1864     }
1865   }
1866
1867   /* and should then also be linked downstream, so safe to send some events */
1868   if (G_UNLIKELY (parse->priv->close_segment)) {
1869     /* only set up by loop */
1870     GST_DEBUG_OBJECT (parse, "loop sending close segment");
1871     gst_pad_push_event (parse->srcpad, parse->priv->close_segment);
1872     parse->priv->close_segment = NULL;
1873   }
1874   if (G_UNLIKELY (parse->priv->pending_segment)) {
1875     GST_DEBUG_OBJECT (parse, "%s push pending segment",
1876         parse->priv->pad_mode == GST_ACTIVATE_PULL ? "loop" : "chain");
1877     gst_pad_push_event (parse->srcpad, parse->priv->pending_segment);
1878     parse->priv->pending_segment = NULL;
1879
1880     /* have caps; check identity */
1881     gst_base_parse_check_media (parse);
1882   }
1883
1884   /* update bitrates and optionally post corresponding tags
1885    * (following newsegment) */
1886   gst_base_parse_update_bitrates (parse, frame);
1887
1888   if (G_UNLIKELY (parse->priv->pending_events)) {
1889     GList *l;
1890
1891     for (l = parse->priv->pending_events; l != NULL; l = l->next) {
1892       gst_pad_push_event (parse->srcpad, GST_EVENT (l->data));
1893     }
1894     g_list_free (parse->priv->pending_events);
1895     parse->priv->pending_events = NULL;
1896   }
1897
1898   if (klass->pre_push_frame) {
1899     ret = klass->pre_push_frame (parse, frame);
1900   } else {
1901     frame->flags |= GST_BASE_PARSE_FRAME_FLAG_CLIP;
1902   }
1903
1904   /* take final ownership of frame buffer */
1905   buffer = frame->buffer;
1906   frame->buffer = NULL;
1907
1908   /* subclass must play nice */
1909   g_return_val_if_fail (buffer != NULL, GST_FLOW_ERROR);
1910
1911   /* decorate */
1912   buffer = gst_buffer_make_metadata_writable (buffer);
1913   gst_buffer_set_caps (buffer, GST_PAD_CAPS (parse->srcpad));
1914
1915   parse->priv->seen_keyframe |= parse->priv->is_video &&
1916       !GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
1917
1918   if (frame->flags & GST_BASE_PARSE_FRAME_FLAG_CLIP) {
1919     if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer) &&
1920         GST_CLOCK_TIME_IS_VALID (parse->segment.stop) &&
1921         GST_BUFFER_TIMESTAMP (buffer) >
1922         parse->segment.stop + parse->priv->lead_out_ts) {
1923       GST_LOG_OBJECT (parse, "Dropped frame, after segment");
1924       ret = GST_FLOW_UNEXPECTED;
1925     } else if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer) &&
1926         GST_BUFFER_DURATION_IS_VALID (buffer) &&
1927         GST_CLOCK_TIME_IS_VALID (parse->segment.start) &&
1928         GST_BUFFER_TIMESTAMP (buffer) + GST_BUFFER_DURATION (buffer) +
1929         parse->priv->lead_in_ts < parse->segment.start) {
1930       if (parse->priv->seen_keyframe) {
1931         GST_LOG_OBJECT (parse, "Frame before segment, after keyframe");
1932         ret = GST_FLOW_OK;
1933       } else {
1934         GST_LOG_OBJECT (parse, "Dropped frame, before segment");
1935         ret = GST_BASE_PARSE_FLOW_DROPPED;
1936       }
1937     } else {
1938       ret = GST_FLOW_OK;
1939     }
1940   }
1941
1942   if (ret == GST_BASE_PARSE_FLOW_DROPPED) {
1943     GST_LOG_OBJECT (parse, "frame (%d bytes) dropped",
1944         GST_BUFFER_SIZE (buffer));
1945     gst_buffer_unref (buffer);
1946     ret = GST_FLOW_OK;
1947   } else if (ret == GST_FLOW_OK) {
1948     if (parse->segment.rate > 0.0) {
1949       ret = gst_pad_push (parse->srcpad, buffer);
1950       GST_LOG_OBJECT (parse, "frame (%d bytes) pushed: %s",
1951           GST_BUFFER_SIZE (buffer), gst_flow_get_name (ret));
1952     } else {
1953       GST_LOG_OBJECT (parse, "frame (%d bytes) queued for now",
1954           GST_BUFFER_SIZE (buffer));
1955       parse->priv->buffers_queued =
1956           g_slist_prepend (parse->priv->buffers_queued, buffer);
1957       ret = GST_FLOW_OK;
1958     }
1959   } else {
1960     gst_buffer_unref (buffer);
1961     GST_LOG_OBJECT (parse, "frame (%d bytes) not pushed: %s",
1962         GST_BUFFER_SIZE (buffer), gst_flow_get_name (ret));
1963     /* if we are not sufficiently in control, let upstream decide on EOS */
1964     if (ret == GST_FLOW_UNEXPECTED &&
1965         (parse->priv->passthrough ||
1966             (parse->priv->pad_mode == GST_ACTIVATE_PUSH &&
1967                 !parse->priv->upstream_seekable)))
1968       ret = GST_FLOW_OK;
1969   }
1970
1971   /* Update current running segment position */
1972   if (ret == GST_FLOW_OK && last_stop != GST_CLOCK_TIME_NONE &&
1973       parse->segment.last_stop < last_stop)
1974     gst_segment_set_last_stop (&parse->segment, GST_FORMAT_TIME, last_stop);
1975
1976   gst_base_parse_frame_free (frame);
1977
1978   return ret;
1979 }
1980
1981
1982 /* gst_base_parse_drain:
1983  *
1984  * Drains the adapter until it is empty. It decreases the min_frame_size to
1985  * match the current adapter size and calls chain method until the adapter
1986  * is emptied or chain returns with error.
1987  */
1988 static void
1989 gst_base_parse_drain (GstBaseParse * parse)
1990 {
1991   guint avail;
1992
1993   GST_DEBUG_OBJECT (parse, "draining");
1994   parse->priv->drain = TRUE;
1995
1996   for (;;) {
1997     avail = gst_adapter_available (parse->priv->adapter);
1998     if (!avail)
1999       break;
2000
2001     if (gst_base_parse_chain (parse->sinkpad, NULL) != GST_FLOW_OK) {
2002       break;
2003     }
2004
2005     /* nothing changed, maybe due to truncated frame; break infinite loop */
2006     if (avail == gst_adapter_available (parse->priv->adapter)) {
2007       GST_DEBUG_OBJECT (parse, "no change during draining; flushing");
2008       gst_adapter_clear (parse->priv->adapter);
2009     }
2010   }
2011
2012   parse->priv->drain = FALSE;
2013 }
2014
2015 /* gst_base_parse_send_buffers
2016  *
2017  * Sends buffers collected in send_buffers downstream, and ensures that list
2018  * is empty at the end (errors or not).
2019  */
2020 static GstFlowReturn
2021 gst_base_parse_send_buffers (GstBaseParse * parse)
2022 {
2023   GSList *send = NULL;
2024   GstBuffer *buf;
2025   GstFlowReturn ret = GST_FLOW_OK;
2026
2027   send = parse->priv->buffers_send;
2028
2029   /* send buffers */
2030   while (send) {
2031     buf = GST_BUFFER_CAST (send->data);
2032     GST_LOG_OBJECT (parse, "pushing buffer %p, timestamp %"
2033         GST_TIME_FORMAT ", duration %" GST_TIME_FORMAT
2034         ", offset %" G_GINT64_FORMAT, buf,
2035         GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)),
2036         GST_TIME_ARGS (GST_BUFFER_DURATION (buf)), GST_BUFFER_OFFSET (buf));
2037
2038     /* iterate output queue an push downstream */
2039     ret = gst_pad_push (parse->srcpad, buf);
2040     send = g_slist_delete_link (send, send);
2041
2042     /* clear any leftover if error */
2043     if (G_UNLIKELY (ret != GST_FLOW_OK)) {
2044       while (send) {
2045         buf = GST_BUFFER_CAST (send->data);
2046         gst_buffer_unref (buf);
2047         send = g_slist_delete_link (send, send);
2048       }
2049     }
2050   }
2051
2052   parse->priv->buffers_send = send;
2053
2054   return ret;
2055 }
2056
2057 /* gst_base_parse_process_fragment:
2058  *
2059  * Processes a reverse playback (forward) fragment:
2060  * - append head of last fragment that was skipped to current fragment data
2061  * - drain the resulting current fragment data (i.e. repeated chain)
2062  * - add time/duration (if needed) to frames queued by chain
2063  * - push queued data
2064  */
2065 static GstFlowReturn
2066 gst_base_parse_process_fragment (GstBaseParse * parse, gboolean push_only)
2067 {
2068   GstBuffer *buf;
2069   GstFlowReturn ret = GST_FLOW_OK;
2070   gboolean seen_key = FALSE, seen_delta = FALSE;
2071
2072   if (push_only)
2073     goto push;
2074
2075   /* restore order */
2076   parse->priv->buffers_pending = g_slist_reverse (parse->priv->buffers_pending);
2077   while (parse->priv->buffers_pending) {
2078     buf = GST_BUFFER_CAST (parse->priv->buffers_pending->data);
2079     GST_LOG_OBJECT (parse, "adding pending buffer (size %d)",
2080         GST_BUFFER_SIZE (buf));
2081     gst_adapter_push (parse->priv->adapter, buf);
2082     parse->priv->buffers_pending =
2083         g_slist_delete_link (parse->priv->buffers_pending,
2084         parse->priv->buffers_pending);
2085   }
2086
2087   /* invalidate so no fall-back timestamping is performed;
2088    * ok if taken from subclass or upstream */
2089   parse->priv->next_ts = GST_CLOCK_TIME_NONE;
2090   /* prevent it hanging around stop all the time */
2091   parse->segment.last_stop = GST_CLOCK_TIME_NONE;
2092   /* mark next run */
2093   parse->priv->discont = TRUE;
2094
2095   /* chain looks for frames and queues resulting ones (in stead of pushing) */
2096   /* initial skipped data is added to buffers_pending */
2097   gst_base_parse_drain (parse);
2098
2099 push:
2100   if (parse->priv->buffers_send) {
2101     buf = GST_BUFFER_CAST (parse->priv->buffers_send->data);
2102     seen_key |= !GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT);
2103   }
2104
2105   /* add metadata (if needed to queued buffers */
2106   GST_LOG_OBJECT (parse, "last timestamp: %" GST_TIME_FORMAT,
2107       GST_TIME_ARGS (parse->priv->last_ts));
2108   while (parse->priv->buffers_queued) {
2109     buf = GST_BUFFER_CAST (parse->priv->buffers_queued->data);
2110
2111     /* no touching if upstream or parsing provided time */
2112     if (GST_BUFFER_TIMESTAMP_IS_VALID (buf)) {
2113       GST_LOG_OBJECT (parse, "buffer has time %" GST_TIME_FORMAT,
2114           GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
2115     } else if (GST_CLOCK_TIME_IS_VALID (parse->priv->last_ts) &&
2116         GST_BUFFER_DURATION_IS_VALID (buf)) {
2117       if (G_LIKELY (GST_BUFFER_DURATION (buf) <= parse->priv->last_ts))
2118         parse->priv->last_ts -= GST_BUFFER_DURATION (buf);
2119       else
2120         parse->priv->last_ts = 0;
2121       GST_BUFFER_TIMESTAMP (buf) = parse->priv->last_ts;
2122       GST_LOG_OBJECT (parse, "applied time %" GST_TIME_FORMAT,
2123           GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
2124     } else {
2125       /* no idea, very bad */
2126       GST_WARNING_OBJECT (parse, "could not determine time for buffer");
2127     }
2128
2129     parse->priv->last_ts = GST_BUFFER_TIMESTAMP (buf);
2130
2131     /* reverse order for ascending sending */
2132     /* send downstream at keyframe not preceded by a keyframe
2133      * (e.g. that should identify start of collection of IDR nals) */
2134     if (GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT)) {
2135       if (seen_key) {
2136         ret = gst_base_parse_send_buffers (parse);
2137         /* if a problem, throw all to sending */
2138         if (ret != GST_FLOW_OK) {
2139           parse->priv->buffers_send =
2140               g_slist_reverse (parse->priv->buffers_queued);
2141           parse->priv->buffers_queued = NULL;
2142           break;
2143         }
2144         seen_key = FALSE;
2145       }
2146     } else {
2147       seen_delta = TRUE;
2148     }
2149
2150     seen_key |= !GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT);
2151
2152     parse->priv->buffers_send =
2153         g_slist_prepend (parse->priv->buffers_send, buf);
2154     parse->priv->buffers_queued =
2155         g_slist_delete_link (parse->priv->buffers_queued,
2156         parse->priv->buffers_queued);
2157   }
2158
2159   /* audio may have all marked as keyframe, so arrange to send here */
2160   if (!seen_delta)
2161     ret = gst_base_parse_send_buffers (parse);
2162
2163   /* any trailing unused no longer usable (ideally none) */
2164   if (G_UNLIKELY (gst_adapter_available (parse->priv->adapter))) {
2165     GST_DEBUG_OBJECT (parse, "discarding %d trailing bytes",
2166         gst_adapter_available (parse->priv->adapter));
2167     gst_adapter_clear (parse->priv->adapter);
2168   }
2169
2170   return ret;
2171 }
2172
2173 /* small helper that checks whether we have been trying to resync too long */
2174 static inline GstFlowReturn
2175 gst_base_parse_check_sync (GstBaseParse * parse)
2176 {
2177   if (G_UNLIKELY (parse->priv->discont &&
2178           parse->priv->offset - parse->priv->sync_offset > 2 * 1024 * 1024)) {
2179     GST_ELEMENT_ERROR (parse, STREAM, DECODE,
2180         ("Failed to parse stream"), (NULL));
2181     return GST_FLOW_ERROR;
2182   }
2183
2184   return GST_FLOW_OK;
2185 }
2186
2187 static GstFlowReturn
2188 gst_base_parse_chain (GstPad * pad, GstBuffer * buffer)
2189 {
2190   GstBaseParseClass *bclass;
2191   GstBaseParse *parse;
2192   GstFlowReturn ret = GST_FLOW_OK;
2193   GstBuffer *outbuf = NULL;
2194   GstBuffer *tmpbuf = NULL;
2195   guint fsize = 1;
2196   gint skip = -1;
2197   const guint8 *data;
2198   guint old_min_size = 0, min_size, av;
2199   GstClockTime timestamp;
2200   GstBaseParseFrame *frame;
2201
2202   parse = GST_BASE_PARSE (GST_OBJECT_PARENT (pad));
2203   bclass = GST_BASE_PARSE_GET_CLASS (parse);
2204
2205   if (parse->priv->detecting) {
2206     GstBuffer *detect_buf;
2207
2208     if (parse->priv->detect_buffers_size == 0) {
2209       detect_buf = gst_buffer_ref (buffer);
2210     } else {
2211       GList *l;
2212       guint offset = 0;
2213
2214       detect_buf =
2215           gst_buffer_new_and_alloc (parse->priv->detect_buffers_size +
2216           (buffer ? GST_BUFFER_SIZE (buffer) : 0));
2217       for (l = parse->priv->detect_buffers; l; l = l->next) {
2218         memcpy (GST_BUFFER_DATA (detect_buf) + offset,
2219             GST_BUFFER_DATA (l->data), GST_BUFFER_SIZE (l->data));
2220         offset += GST_BUFFER_SIZE (l->data);
2221       }
2222       if (buffer)
2223         memcpy (GST_BUFFER_DATA (detect_buf) + offset, GST_BUFFER_DATA (buffer),
2224             GST_BUFFER_SIZE (buffer));
2225     }
2226
2227     ret = bclass->detect (parse, detect_buf);
2228     gst_buffer_unref (detect_buf);
2229
2230     if (ret == GST_FLOW_OK) {
2231       GList *l;
2232
2233       /* Detected something */
2234       parse->priv->detecting = FALSE;
2235
2236       for (l = parse->priv->detect_buffers; l; l = l->next) {
2237         if (ret == GST_FLOW_OK && !parse->priv->flushing)
2238           ret =
2239               gst_base_parse_chain (GST_BASE_PARSE_SINK_PAD (parse),
2240               GST_BUFFER_CAST (l->data));
2241         else
2242           gst_buffer_unref (GST_BUFFER_CAST (l->data));
2243       }
2244       g_list_free (parse->priv->detect_buffers);
2245       parse->priv->detect_buffers = NULL;
2246       parse->priv->detect_buffers_size = 0;
2247
2248       if (ret != GST_FLOW_OK) {
2249         return ret;
2250       }
2251
2252       /* Handle the current buffer */
2253     } else if (ret == GST_FLOW_NOT_NEGOTIATED) {
2254       /* Still detecting, append buffer or error out if draining */
2255
2256       if (parse->priv->drain) {
2257         GST_DEBUG_OBJECT (parse, "Draining but did not detect format yet");
2258         return GST_FLOW_ERROR;
2259       } else if (parse->priv->flushing) {
2260         g_list_foreach (parse->priv->detect_buffers, (GFunc) gst_buffer_unref,
2261             NULL);
2262         g_list_free (parse->priv->detect_buffers);
2263         parse->priv->detect_buffers = NULL;
2264         parse->priv->detect_buffers_size = 0;
2265       } else {
2266         parse->priv->detect_buffers =
2267             g_list_append (parse->priv->detect_buffers, buffer);
2268         parse->priv->detect_buffers_size += GST_BUFFER_SIZE (buffer);
2269         return GST_FLOW_OK;
2270       }
2271     } else {
2272       /* Something went wrong, subclass responsible for error reporting */
2273       return ret;
2274     }
2275
2276     /* And now handle the current buffer if detection worked */
2277   }
2278
2279   frame = &parse->priv->frame;
2280
2281   if (G_LIKELY (buffer)) {
2282     GST_LOG_OBJECT (parse, "buffer size: %d, offset = %" G_GINT64_FORMAT,
2283         GST_BUFFER_SIZE (buffer), GST_BUFFER_OFFSET (buffer));
2284     if (G_UNLIKELY (parse->priv->passthrough)) {
2285       gst_base_parse_frame_init (frame);
2286       frame->buffer = gst_buffer_make_metadata_writable (buffer);
2287       return gst_base_parse_push_frame (parse, frame);
2288     }
2289     /* upstream feeding us in reverse playback;
2290      * gather each fragment, then process it in single run */
2291     if (parse->segment.rate < 0.0) {
2292       if (G_UNLIKELY (GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DISCONT))) {
2293         GST_DEBUG_OBJECT (parse, "buffer starts new reverse playback fragment");
2294         ret = gst_base_parse_process_fragment (parse, FALSE);
2295       }
2296       gst_adapter_push (parse->priv->adapter, buffer);
2297       return ret;
2298     }
2299     gst_adapter_push (parse->priv->adapter, buffer);
2300   }
2301
2302   if (G_UNLIKELY (buffer &&
2303           GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DISCONT))) {
2304     frame->_private_flags |= GST_BASE_PARSE_FRAME_PRIVATE_FLAG_NOALLOC;
2305     gst_base_parse_frame_free (frame);
2306   }
2307
2308   /* Parse and push as many frames as possible */
2309   /* Stop either when adapter is empty or we are flushing */
2310   while (!parse->priv->flushing) {
2311     gboolean res;
2312
2313     /* maintain frame state for a single frame parsing round across _chain calls,
2314      * so only init when needed */
2315     if (!frame->_private_flags)
2316       gst_base_parse_frame_init (frame);
2317
2318     tmpbuf = gst_buffer_new ();
2319
2320     old_min_size = 0;
2321     /* Synchronization loop */
2322     for (;;) {
2323       /* note: if subclass indicates MAX fsize,
2324        * this will not likely be available anyway ... */
2325       min_size = MAX (parse->priv->min_frame_size, fsize);
2326       av = gst_adapter_available (parse->priv->adapter);
2327
2328       /* loop safety check */
2329       if (G_UNLIKELY (old_min_size >= min_size))
2330         goto invalid_min;
2331       old_min_size = min_size;
2332
2333       if (G_UNLIKELY (parse->priv->drain)) {
2334         min_size = av;
2335         GST_DEBUG_OBJECT (parse, "draining, data left: %d", min_size);
2336         if (G_UNLIKELY (!min_size)) {
2337           gst_buffer_unref (tmpbuf);
2338           goto done;
2339         }
2340       }
2341
2342       /* Collect at least min_frame_size bytes */
2343       if (av < min_size) {
2344         GST_DEBUG_OBJECT (parse, "not enough data available (only %d bytes)",
2345             av);
2346         gst_buffer_unref (tmpbuf);
2347         goto done;
2348       }
2349
2350       /* always pass all available data */
2351       data = gst_adapter_peek (parse->priv->adapter, av);
2352       GST_BUFFER_DATA (tmpbuf) = (guint8 *) data;
2353       GST_BUFFER_SIZE (tmpbuf) = av;
2354       GST_BUFFER_OFFSET (tmpbuf) = parse->priv->offset;
2355       GST_BUFFER_FLAG_SET (tmpbuf, GST_MINI_OBJECT_FLAG_READONLY);
2356
2357       if (parse->priv->discont) {
2358         GST_DEBUG_OBJECT (parse, "marking DISCONT");
2359         GST_BUFFER_FLAG_SET (tmpbuf, GST_BUFFER_FLAG_DISCONT);
2360       }
2361
2362       skip = -1;
2363       gst_base_parse_frame_update (parse, frame, tmpbuf);
2364       res = bclass->check_valid_frame (parse, frame, &fsize, &skip);
2365       gst_buffer_replace (&frame->buffer, NULL);
2366       if (res) {
2367         if (gst_adapter_available (parse->priv->adapter) < fsize) {
2368           GST_DEBUG_OBJECT (parse,
2369               "found valid frame but not enough data available (only %d bytes)",
2370               gst_adapter_available (parse->priv->adapter));
2371           gst_buffer_unref (tmpbuf);
2372           goto done;
2373         }
2374         GST_LOG_OBJECT (parse, "valid frame of size %d at pos %d", fsize, skip);
2375         break;
2376       }
2377       if (skip == -1) {
2378         /* subclass didn't touch this value. By default we skip 1 byte */
2379         skip = 1;
2380       }
2381       if (skip > 0) {
2382         GST_LOG_OBJECT (parse, "finding sync, skipping %d bytes", skip);
2383         if (parse->segment.rate < 0.0 && !parse->priv->buffers_queued) {
2384           /* reverse playback, and no frames found yet, so we are skipping
2385            * the leading part of a fragment, which may form the tail of
2386            * fragment coming later, hopefully subclass skips efficiently ... */
2387           timestamp = gst_adapter_prev_timestamp (parse->priv->adapter, NULL);
2388           outbuf = gst_adapter_take_buffer (parse->priv->adapter, skip);
2389           outbuf = gst_buffer_make_metadata_writable (outbuf);
2390           GST_BUFFER_TIMESTAMP (outbuf) = timestamp;
2391           parse->priv->buffers_pending =
2392               g_slist_prepend (parse->priv->buffers_pending, outbuf);
2393           outbuf = NULL;
2394         } else {
2395           gst_adapter_flush (parse->priv->adapter, skip);
2396         }
2397         parse->priv->offset += skip;
2398         if (!parse->priv->discont)
2399           parse->priv->sync_offset = parse->priv->offset;
2400         parse->priv->discont = TRUE;
2401         /* something changed least; nullify loop check */
2402         old_min_size = 0;
2403       }
2404       /* skip == 0 should imply subclass set min_size to need more data;
2405        * we check this shortly */
2406       if ((ret = gst_base_parse_check_sync (parse)) != GST_FLOW_OK) {
2407         gst_buffer_unref (tmpbuf);
2408         goto done;
2409       }
2410     }
2411     gst_buffer_unref (tmpbuf);
2412     tmpbuf = NULL;
2413
2414     if (skip > 0) {
2415       /* Subclass found the sync, but still wants to skip some data */
2416       GST_LOG_OBJECT (parse, "skipping %d bytes", skip);
2417       gst_adapter_flush (parse->priv->adapter, skip);
2418       parse->priv->offset += skip;
2419     }
2420
2421     /* Grab lock to prevent a race with FLUSH_START handler */
2422     GST_PAD_STREAM_LOCK (parse->srcpad);
2423
2424     /* FLUSH_START event causes the "flushing" flag to be set. In this
2425      * case we can leave the frame pushing loop */
2426     if (parse->priv->flushing) {
2427       GST_PAD_STREAM_UNLOCK (parse->srcpad);
2428       break;
2429     }
2430
2431     /* move along with upstream timestamp (if any),
2432      * but interpolate in between */
2433     timestamp = gst_adapter_prev_timestamp (parse->priv->adapter, NULL);
2434     if (GST_CLOCK_TIME_IS_VALID (timestamp) &&
2435         (parse->priv->prev_ts != timestamp)) {
2436       parse->priv->prev_ts = parse->priv->next_ts = timestamp;
2437     }
2438
2439     /* FIXME: Would it be more efficient to make a subbuffer instead? */
2440     outbuf = gst_adapter_take_buffer (parse->priv->adapter, fsize);
2441     outbuf = gst_buffer_make_metadata_writable (outbuf);
2442
2443     /* Subclass may want to know the data offset */
2444     GST_BUFFER_OFFSET (outbuf) = parse->priv->offset;
2445     parse->priv->offset += fsize;
2446     GST_BUFFER_TIMESTAMP (outbuf) = GST_CLOCK_TIME_NONE;
2447     GST_BUFFER_DURATION (outbuf) = GST_CLOCK_TIME_NONE;
2448
2449     frame->buffer = outbuf;
2450     ret = gst_base_parse_handle_and_push_frame (parse, bclass, frame);
2451     GST_PAD_STREAM_UNLOCK (parse->srcpad);
2452
2453     if (ret != GST_FLOW_OK) {
2454       GST_LOG_OBJECT (parse, "push returned %d", ret);
2455       break;
2456     }
2457   }
2458
2459 done:
2460   GST_LOG_OBJECT (parse, "chain leaving");
2461   return ret;
2462
2463   /* ERRORS */
2464 invalid_min:
2465   {
2466     GST_ELEMENT_ERROR (parse, STREAM, FAILED, (NULL),
2467         ("min_size evolution %d -> %d; breaking to avoid looping",
2468             old_min_size, min_size));
2469     return GST_FLOW_ERROR;
2470   }
2471 }
2472
2473 /* pull @size bytes at current offset,
2474  * i.e. at least try to and possibly return a shorter buffer if near the end */
2475 static GstFlowReturn
2476 gst_base_parse_pull_range (GstBaseParse * parse, guint size,
2477     GstBuffer ** buffer)
2478 {
2479   GstFlowReturn ret = GST_FLOW_OK;
2480
2481   g_return_val_if_fail (buffer != NULL, GST_FLOW_ERROR);
2482
2483   /* Caching here actually makes much less difference than one would expect.
2484    * We do it mainly to avoid pulling buffers of 1 byte all the time */
2485   if (parse->priv->cache) {
2486     gint64 cache_offset = GST_BUFFER_OFFSET (parse->priv->cache);
2487     gint cache_size = GST_BUFFER_SIZE (parse->priv->cache);
2488
2489     if (cache_offset <= parse->priv->offset &&
2490         (parse->priv->offset + size) <= (cache_offset + cache_size)) {
2491       *buffer = gst_buffer_create_sub (parse->priv->cache,
2492           parse->priv->offset - cache_offset, size);
2493       GST_BUFFER_OFFSET (*buffer) = parse->priv->offset;
2494       return GST_FLOW_OK;
2495     }
2496     /* not enough data in the cache, free cache and get a new one */
2497     gst_buffer_unref (parse->priv->cache);
2498     parse->priv->cache = NULL;
2499   }
2500
2501   /* refill the cache */
2502   ret =
2503       gst_pad_pull_range (parse->sinkpad, parse->priv->offset, MAX (size,
2504           64 * 1024), &parse->priv->cache);
2505   if (ret != GST_FLOW_OK) {
2506     parse->priv->cache = NULL;
2507     return ret;
2508   }
2509
2510   if (GST_BUFFER_SIZE (parse->priv->cache) >= size) {
2511     *buffer = gst_buffer_create_sub (parse->priv->cache, 0, size);
2512     GST_BUFFER_OFFSET (*buffer) = parse->priv->offset;
2513     return GST_FLOW_OK;
2514   }
2515
2516   /* Not possible to get enough data, try a last time with
2517    * requesting exactly the size we need */
2518   gst_buffer_unref (parse->priv->cache);
2519   parse->priv->cache = NULL;
2520
2521   ret = gst_pad_pull_range (parse->sinkpad, parse->priv->offset, size,
2522       &parse->priv->cache);
2523
2524   if (ret != GST_FLOW_OK) {
2525     GST_DEBUG_OBJECT (parse, "pull_range returned %d", ret);
2526     *buffer = NULL;
2527     return ret;
2528   }
2529
2530   if (GST_BUFFER_SIZE (parse->priv->cache) < size) {
2531     GST_DEBUG_OBJECT (parse, "Returning short buffer at offset %"
2532         G_GUINT64_FORMAT ": wanted %u bytes, got %u bytes", parse->priv->offset,
2533         size, GST_BUFFER_SIZE (parse->priv->cache));
2534
2535     *buffer = parse->priv->cache;
2536     parse->priv->cache = NULL;
2537
2538     return GST_FLOW_OK;
2539   }
2540
2541   *buffer = gst_buffer_create_sub (parse->priv->cache, 0, size);
2542   GST_BUFFER_OFFSET (*buffer) = parse->priv->offset;
2543
2544   return GST_FLOW_OK;
2545 }
2546
2547 static GstFlowReturn
2548 gst_base_parse_handle_previous_fragment (GstBaseParse * parse)
2549 {
2550   gint64 offset = 0;
2551   GstClockTime ts = 0;
2552   GstBuffer *buffer;
2553   GstFlowReturn ret;
2554
2555   GST_DEBUG_OBJECT (parse, "fragment ended; last_ts = %" GST_TIME_FORMAT
2556       ", last_offset = %" G_GINT64_FORMAT, GST_TIME_ARGS (parse->priv->last_ts),
2557       parse->priv->last_offset);
2558
2559   if (!parse->priv->last_offset || parse->priv->last_ts <= parse->segment.start) {
2560     GST_DEBUG_OBJECT (parse, "past start of segment %" GST_TIME_FORMAT,
2561         GST_TIME_ARGS (parse->segment.start));
2562     ret = GST_FLOW_UNEXPECTED;
2563     goto exit;
2564   }
2565
2566   /* last fragment started at last_offset / last_ts;
2567    * seek back 10s capped at 1MB */
2568   if (parse->priv->last_ts >= 10 * GST_SECOND)
2569     ts = parse->priv->last_ts - 10 * GST_SECOND;
2570   /* if we are exact now, we will be more so going backwards */
2571   if (parse->priv->exact_position) {
2572     offset = gst_base_parse_find_offset (parse, ts, TRUE, NULL);
2573   } else {
2574     GstFormat dstformat = GST_FORMAT_BYTES;
2575
2576     if (!gst_pad_query_convert (parse->srcpad, GST_FORMAT_TIME, ts,
2577             &dstformat, &offset)) {
2578       GST_DEBUG_OBJECT (parse, "conversion failed, only BYTE based");
2579     }
2580   }
2581   offset = CLAMP (offset, parse->priv->last_offset - 1024 * 1024,
2582       parse->priv->last_offset - 1024);
2583   offset = MAX (0, offset);
2584
2585   GST_DEBUG_OBJECT (parse, "next fragment from offset %" G_GINT64_FORMAT,
2586       offset);
2587   parse->priv->offset = offset;
2588
2589   ret = gst_base_parse_pull_range (parse, parse->priv->last_offset - offset,
2590       &buffer);
2591   if (ret != GST_FLOW_OK)
2592     goto exit;
2593
2594   /* offset will increase again as fragment is processed/parsed */
2595   parse->priv->last_offset = offset;
2596
2597   gst_adapter_push (parse->priv->adapter, buffer);
2598   ret = gst_base_parse_process_fragment (parse, FALSE);
2599   if (ret != GST_FLOW_OK)
2600     goto exit;
2601
2602   /* force previous fragment */
2603   parse->priv->offset = -1;
2604
2605 exit:
2606   return ret;
2607 }
2608
2609 /* PULL mode:
2610  * pull and scan for next frame starting from current offset
2611  * ajusts sync, drain and offset going along */
2612 static GstFlowReturn
2613 gst_base_parse_scan_frame (GstBaseParse * parse, GstBaseParseClass * klass,
2614     GstBaseParseFrame * frame, gboolean full)
2615 {
2616   GstBuffer *buffer, *outbuf;
2617   GstFlowReturn ret = GST_FLOW_OK;
2618   guint fsize = 1, min_size, old_min_size = 0;
2619   gint skip = 0;
2620
2621   g_return_val_if_fail (frame != NULL, GST_FLOW_ERROR);
2622
2623   GST_LOG_OBJECT (parse, "scanning for frame at offset %" G_GUINT64_FORMAT
2624       " (%#" G_GINT64_MODIFIER "x)", parse->priv->offset, parse->priv->offset);
2625
2626   /* let's make this efficient for all subclass once and for all;
2627    * maybe it does not need this much, but in the latter case, we know we are
2628    * in pull mode here and might as well try to read and supply more anyway
2629    * (so does the buffer caching mechanism) */
2630   fsize = 64 * 1024;
2631
2632   while (TRUE) {
2633     gboolean res;
2634
2635     min_size = MAX (parse->priv->min_frame_size, fsize);
2636     /* loop safety check */
2637     if (G_UNLIKELY (old_min_size >= min_size))
2638       goto invalid_min;
2639     old_min_size = min_size;
2640
2641     ret = gst_base_parse_pull_range (parse, min_size, &buffer);
2642     if (ret != GST_FLOW_OK)
2643       goto done;
2644
2645     if (parse->priv->discont) {
2646       GST_DEBUG_OBJECT (parse, "marking DISCONT");
2647       GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT);
2648     }
2649
2650     /* if we got a short read, inform subclass we are draining leftover
2651      * and no more is to be expected */
2652     if (GST_BUFFER_SIZE (buffer) < min_size)
2653       parse->priv->drain = TRUE;
2654
2655     if (parse->priv->detecting) {
2656       ret = klass->detect (parse, buffer);
2657       if (ret == GST_FLOW_NOT_NEGOTIATED) {
2658         /* If draining we error out, otherwise request a buffer
2659          * with 64kb more */
2660         if (parse->priv->drain) {
2661           gst_buffer_unref (buffer);
2662           GST_ERROR_OBJECT (parse, "Failed to detect format but draining");
2663           return GST_FLOW_ERROR;
2664         } else {
2665           fsize += 64 * 1024;
2666           gst_buffer_unref (buffer);
2667           continue;
2668         }
2669       } else if (ret != GST_FLOW_OK) {
2670         gst_buffer_unref (buffer);
2671         GST_ERROR_OBJECT (parse, "detect() returned %s",
2672             gst_flow_get_name (ret));
2673         return ret;
2674       }
2675
2676       /* Else handle this buffer normally */
2677     }
2678
2679     skip = -1;
2680     gst_base_parse_frame_update (parse, frame, buffer);
2681     res = klass->check_valid_frame (parse, frame, &fsize, &skip);
2682     gst_buffer_replace (&frame->buffer, NULL);
2683     if (res) {
2684       parse->priv->drain = FALSE;
2685       GST_LOG_OBJECT (parse, "valid frame of size %d at pos %d", fsize, skip);
2686       break;
2687     }
2688     parse->priv->drain = FALSE;
2689     if (skip == -1)
2690       skip = 1;
2691     if (skip > 0) {
2692       GST_LOG_OBJECT (parse, "finding sync, skipping %d bytes", skip);
2693       if (full && parse->segment.rate < 0.0 && !parse->priv->buffers_queued) {
2694         /* reverse playback, and no frames found yet, so we are skipping
2695          * the leading part of a fragment, which may form the tail of
2696          * fragment coming later, hopefully subclass skips efficiently ... */
2697         outbuf = gst_buffer_create_sub (buffer, 0, skip);
2698         parse->priv->buffers_pending =
2699             g_slist_prepend (parse->priv->buffers_pending, outbuf);
2700         outbuf = NULL;
2701       }
2702       parse->priv->offset += skip;
2703       if (!parse->priv->discont)
2704         parse->priv->sync_offset = parse->priv->offset;
2705       parse->priv->discont = TRUE;
2706       /* something changed at least; nullify loop check */
2707       if (fsize == G_MAXUINT)
2708         fsize = old_min_size + 64 * 1024;
2709       old_min_size = 0;
2710     }
2711     /* skip == 0 should imply subclass set min_size to need more data;
2712      * we check this shortly */
2713     GST_DEBUG_OBJECT (parse, "finding sync...");
2714     gst_buffer_unref (buffer);
2715     if ((ret = gst_base_parse_check_sync (parse)) != GST_FLOW_OK) {
2716       goto done;
2717     }
2718   }
2719
2720   /* Does the subclass want to skip too? */
2721   if (skip > 0)
2722     parse->priv->offset += skip;
2723   else if (skip < 0)
2724     skip = 0;
2725
2726   if (fsize + skip <= GST_BUFFER_SIZE (buffer)) {
2727     outbuf = gst_buffer_create_sub (buffer, skip, fsize);
2728     GST_BUFFER_OFFSET (outbuf) = GST_BUFFER_OFFSET (buffer) + skip;
2729     GST_BUFFER_TIMESTAMP (outbuf) = GST_CLOCK_TIME_NONE;
2730     gst_buffer_unref (buffer);
2731   } else {
2732     gst_buffer_unref (buffer);
2733     ret = gst_base_parse_pull_range (parse, fsize, &outbuf);
2734     if (ret != GST_FLOW_OK)
2735       goto done;
2736     if (GST_BUFFER_SIZE (outbuf) < fsize) {
2737       gst_buffer_unref (outbuf);
2738       ret = GST_FLOW_UNEXPECTED;
2739     }
2740   }
2741
2742   parse->priv->offset += fsize;
2743
2744   frame->buffer = outbuf;
2745
2746 done:
2747   return ret;
2748
2749   /* ERRORS */
2750 invalid_min:
2751   {
2752     GST_ELEMENT_ERROR (parse, STREAM, FAILED, (NULL),
2753         ("min_size evolution %d -> %d; breaking to avoid looping",
2754             old_min_size, min_size));
2755     return GST_FLOW_ERROR;
2756   }
2757 }
2758
2759 /* Loop that is used in pull mode to retrieve data from upstream */
2760 static void
2761 gst_base_parse_loop (GstPad * pad)
2762 {
2763   GstBaseParse *parse;
2764   GstBaseParseClass *klass;
2765   GstFlowReturn ret = GST_FLOW_OK;
2766   GstBaseParseFrame frame;
2767
2768   parse = GST_BASE_PARSE (gst_pad_get_parent (pad));
2769   klass = GST_BASE_PARSE_GET_CLASS (parse);
2770
2771   /* reverse playback:
2772    * first fragment (closest to stop time) is handled normally below,
2773    * then we pull in fragments going backwards */
2774   if (parse->segment.rate < 0.0) {
2775     /* check if we jumped back to a previous fragment,
2776      * which is a post-first fragment */
2777     if (parse->priv->offset < 0) {
2778       ret = gst_base_parse_handle_previous_fragment (parse);
2779       goto done;
2780     }
2781   }
2782
2783   gst_base_parse_frame_init (&frame);
2784   ret = gst_base_parse_scan_frame (parse, klass, &frame, TRUE);
2785   if (ret != GST_FLOW_OK)
2786     goto done;
2787
2788   /* This always cleans up frame, even if error occurs */
2789   ret = gst_base_parse_handle_and_push_frame (parse, klass, &frame);
2790
2791   /* eat expected eos signalling past segment in reverse playback */
2792   if (parse->segment.rate < 0.0 && ret == GST_FLOW_UNEXPECTED &&
2793       parse->segment.last_stop >= parse->segment.stop) {
2794     GST_DEBUG_OBJECT (parse, "downstream has reached end of segment");
2795     /* push what was accumulated during loop run */
2796     gst_base_parse_process_fragment (parse, TRUE);
2797     /* force previous fragment */
2798     parse->priv->offset = -1;
2799     ret = GST_FLOW_OK;
2800   }
2801
2802 done:
2803   if (ret == GST_FLOW_UNEXPECTED)
2804     goto eos;
2805   else if (ret != GST_FLOW_OK)
2806     goto pause;