gst-libs/gst/tag/gsttagdemux.c

   1 /* GStreamer Base Class for Tag Demuxing
   2  * Copyright (C) 2005 Jan Schmidt <thaytan@mad.scientist.com>
   3  * Copyright (C) 2006-2007 Tim-Philipp Müller <tim centricular net>
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18  * Boston, MA 02111-1307, USA.
  19  */
  20
  21 /**
  22  * SECTION:gsttagdemux
  23  * @see_also: GstApeDemux, GstID3Demux
  24  * @short_description: Base class for demuxing tags that are in chunks
  25  *                     directly at the beginning or at the end of a file
  26  *
  27  * <refsect2>
  28  * <para>
  29  * Provides a base class for demuxing tags at the beginning or end of a
  30  * stream and handles things like typefinding, querying, seeking, and
  31  * different modes of operation (chain-based, pull_range-based, and providing
  32  * downstream elements with random access if upstream supports that). The tag
  33  * is stripped from the output, and all offsets are adjusted for the tag
  34  * sizes, so that to the downstream element the stream will appear as if
  35  * there was no tag at all. Also, once the tag has been parsed, GstTagDemux
  36  * will try to determine the media type of the resulting stream and add a
  37  * source pad with the appropriate caps in order to facilitate auto-plugging.
  38  * </para>
  39  * <title>Deriving from GstTagDemux</title>
  40  * <para>
  41  * Subclasses have to do four things:
  42  * <itemizedlist>
  43  *  <listitem><para>
  44  *  In their base init function, they must add a pad template for the sink
  45  *  pad to the element class, describing the media type they can parse in
  46  *  the caps of the pad template.
  47  *  </para></listitem>
  48  *  <listitem><para>
  49  *  In their class init function, they must override
  50  *  GST_TAG_DEMUX_CLASS(demux_klass)->identify_tag with their own identify
  51  *  function.
  52  *  </para></listitem>
  53  *  <listitem><para>
  54  *  In their class init function, they must override
  55  *  GST_TAG_DEMUX_CLASS(demux_klass)->parse_tag with their own parse
  56  *  function.
  57  *  </para></listitem>
  58  *  <listitem><para>
  59  *  In their class init function, they must also set
  60  *  GST_TAG_DEMUX_CLASS(demux_klass)->min_start_size and/or
  61  *  GST_TAG_DEMUX_CLASS(demux_klass)->min_end_size to the minimum size required
  62  *  for the identify function to decide whether the stream has a supported tag
  63  *  or not. A class parsing ID3v1 tags, for example, would set min_end_size to
  64  *  128 bytes.
  65  *  </para></listitem>
  66  * </itemizedlist>
  67  * </para>
  68  * </refsect2>
  69  */
  70
  71 #ifdef HAVE_CONFIG_H
  72 #include "config.h"
  73 #endif
  74
  75 #include "gsttagdemux.h"
  76
  77 #include <gst/base/gsttypefindhelper.h>
  78 #include <gst/gst-i18n-plugin.h>
  79 #include <string.h>
  80
  81 typedef enum
  82 {
  83   GST_TAG_DEMUX_READ_START_TAG,
  84   GST_TAG_DEMUX_TYPEFINDING,
  85   GST_TAG_DEMUX_STREAMING
  86 } GstTagDemuxState;
  87
  88 struct _GstTagDemuxPrivate
  89 {
  90   GstPad *srcpad;
  91   GstPad *sinkpad;
  92
  93   /* Number of bytes to remove from the
  94    * start of file (tag at beginning) */
  95   guint strip_start;
  96
  97   /* Number of bytes to remove from the
  98    * end of file (tag at end) */
  99   guint strip_end;
 100
 101   gint64 upstream_size;
 102
 103   GstTagDemuxState state;
 104   GstBuffer *collect;
 105   gsize collect_size;
 106   GstCaps *src_caps;
 107
 108   GstTagList *event_tags;
 109   GstTagList *parsed_tags;
 110   gboolean send_tag_event;
 111
 112   GstSegment segment;
 113   gboolean need_newseg;
 114
 115   guint64 offset;
 116
 117   GList *pending_events;
 118 };
 119
 120 /* Require at least 8kB of data before we attempt typefind.
 121  * Seems a decent value based on test files
 122  * 40kB is massive overkill for the maximum, I think, but it
 123  * doesn't do any harm (tpm: increased to 64kB after watching
 124  * typefinding fail on a wavpack file that needed 42kB to succeed) */
 125 #define TYPE_FIND_MIN_SIZE 8192
 126 #define TYPE_FIND_MAX_SIZE 65536
 127
 128 #define DEFAULT_PULL_BLOCKSIZE 4096
 129
 130 GST_DEBUG_CATEGORY_STATIC (tagdemux_debug);
 131 #define GST_CAT_DEFAULT (tagdemux_debug)
 132
 133 static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",
 134     GST_PAD_SRC,
 135     GST_PAD_ALWAYS,
 136     GST_STATIC_CAPS ("ANY")
 137     );
 138
 139 static void gst_tag_demux_dispose (GObject * object);
 140
 141 static GstFlowReturn gst_tag_demux_chain (GstPad * pad, GstObject * parent,
 142     GstBuffer * buf);
 143 static gboolean gst_tag_demux_sink_event (GstPad * pad, GstObject * parent,
 144     GstEvent * event);
 145
 146 static gboolean gst_tag_demux_sink_activate_mode (GstPad * pad,
 147     GstObject * parent, GstPadMode mode, gboolean active);
 148 static gboolean gst_tag_demux_src_activate_mode (GstPad * pad,
 149     GstObject * parent, GstPadMode mode, gboolean active);
 150 static GstFlowReturn gst_tag_demux_read_range (GstTagDemux * tagdemux,
 151     GstObject * parent, guint64 offset, guint length, GstBuffer ** buffer);
 152
 153 static GstFlowReturn gst_tag_demux_src_getrange (GstPad * srcpad,
 154     GstObject * parent, guint64 offset, guint length, GstBuffer ** buffer);
 155
 156 static void gst_tag_demux_set_src_caps (GstTagDemux * tagdemux,
 157     GstCaps * new_caps);
 158
 159 static gboolean gst_tag_demux_srcpad_event (GstPad * pad, GstObject * parent,
 160     GstEvent * event);
 161 static gboolean gst_tag_demux_sink_activate (GstPad * sinkpad,
 162     GstObject * parent);
 163 static GstStateChangeReturn gst_tag_demux_change_state (GstElement * element,
 164     GstStateChange transition);
 165 static gboolean gst_tag_demux_pad_query (GstPad * pad, GstObject * parent,
 166     GstQuery * query);
 167 static gboolean gst_tag_demux_get_upstream_size (GstTagDemux * tagdemux);
 168 static void gst_tag_demux_send_pending_events (GstTagDemux * tagdemux);
 169 static void gst_tag_demux_send_tag_event (GstTagDemux * tagdemux);
 170 static gboolean gst_tag_demux_send_new_segment (GstTagDemux * tagdemux);
 171
 172 static void gst_tag_demux_base_init (gpointer g_class);
 173 static void gst_tag_demux_class_init (gpointer g_class, gpointer d);
 174 static void gst_tag_demux_init (GstTagDemux * obj, GstTagDemuxClass * klass);
 175
 176 static gpointer parent_class;   /* NULL */
 177
 178 GType
 179 gst_tag_demux_result_get_type (void)
 180 {
 181   static GType etype = 0;
 182   if (etype == 0) {
 183     static const GEnumValue values[] = {
 184       {GST_TAG_DEMUX_RESULT_BROKEN_TAG, "GST_TAG_DEMUX_RESULT_BROKEN_TAG",
 185           "broken-tag"},
 186       {GST_TAG_DEMUX_RESULT_AGAIN, "GST_TAG_DEMUX_RESULT_AGAIN", "again"},
 187       {GST_TAG_DEMUX_RESULT_OK, "GST_TAG_DEMUX_RESULT_OK", "ok"},
 188       {0, NULL, NULL}
 189     };
 190     etype = g_enum_register_static ("GstTagDemuxResult", values);
 191   }
 192   return etype;
 193 }
 194
 195 /* Cannot use boilerplate macros here because we want the abstract flag */
 196 GType
 197 gst_tag_demux_get_type (void)
 198 {
 199   static GType object_type;     /* 0 */
 200
 201   if (object_type == 0) {
 202     static const GTypeInfo object_info = {
 203       sizeof (GstTagDemuxClass),
 204       gst_tag_demux_base_init,
 205       NULL,                     /* base_finalize */
 206       gst_tag_demux_class_init,
 207       NULL,                     /* class_finalize */
 208       NULL,                     /* class_data */
 209       sizeof (GstTagDemux),
 210       0,                        /* n_preallocs */
 211       (GInstanceInitFunc) gst_tag_demux_init
 212     };
 213
 214     object_type = g_type_register_static (GST_TYPE_ELEMENT,
 215         "GstTagDemux", &object_info, G_TYPE_FLAG_ABSTRACT);
 216   }
 217
 218   return object_type;
 219 }
 220
 221 static void
 222 gst_tag_demux_base_init (gpointer klass)
 223 {
 224   GstElementClass *element_class = GST_ELEMENT_CLASS (klass);
 225
 226   gst_element_class_add_pad_template (element_class,
 227       gst_static_pad_template_get (&src_factory));
 228
 229   GST_DEBUG_CATEGORY_INIT (tagdemux_debug, "tagdemux", 0,
 230       "tag demux base class");
 231 }
 232
 233 static void
 234 gst_tag_demux_class_init (gpointer klass, gpointer d)
 235 {
 236   GstTagDemuxClass *tagdemux_class = GST_TAG_DEMUX_CLASS (klass);
 237   GstElementClass *element_class = GST_ELEMENT_CLASS (klass);
 238   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 239
 240   parent_class = g_type_class_peek_parent (klass);
 241
 242   gobject_class->dispose = gst_tag_demux_dispose;
 243
 244   element_class->change_state = GST_DEBUG_FUNCPTR (gst_tag_demux_change_state);
 245
 246   g_type_class_add_private (klass, sizeof (GstTagDemuxPrivate));
 247
 248   /* subclasses must set at least one of these */
 249   tagdemux_class->min_start_size = 0;
 250   tagdemux_class->min_end_size = 0;
 251 }
 252
 253 static void
 254 gst_tag_demux_reset (GstTagDemux * tagdemux)
 255 {
 256   GstBuffer **buffer_p = &tagdemux->priv->collect;
 257   GstCaps **caps_p = &tagdemux->priv->src_caps;
 258
 259   tagdemux->priv->strip_start = 0;
 260   tagdemux->priv->strip_end = 0;
 261   tagdemux->priv->upstream_size = -1;
 262   tagdemux->priv->state = GST_TAG_DEMUX_READ_START_TAG;
 263   tagdemux->priv->send_tag_event = FALSE;
 264
 265   gst_buffer_replace (buffer_p, NULL);
 266   tagdemux->priv->collect_size = 0;
 267   gst_caps_replace (caps_p, NULL);
 268
 269   if (tagdemux->priv->event_tags) {
 270     gst_tag_list_unref (tagdemux->priv->event_tags);
 271     tagdemux->priv->event_tags = NULL;
 272   }
 273   if (tagdemux->priv->parsed_tags) {
 274     gst_tag_list_unref (tagdemux->priv->parsed_tags);
 275     tagdemux->priv->parsed_tags = NULL;
 276   }
 277
 278   gst_segment_init (&tagdemux->priv->segment, GST_FORMAT_UNDEFINED);
 279   tagdemux->priv->need_newseg = TRUE;
 280
 281   g_list_foreach (tagdemux->priv->pending_events,
 282       (GFunc) gst_mini_object_unref, NULL);
 283   g_list_free (tagdemux->priv->pending_events);
 284   tagdemux->priv->pending_events = NULL;
 285 }
 286
 287 static void
 288 gst_tag_demux_init (GstTagDemux * demux, GstTagDemuxClass * gclass)
 289 {
 290   GstElementClass *element_klass = GST_ELEMENT_CLASS (gclass);
 291   GstPadTemplate *tmpl;
 292
 293   demux->priv = g_type_instance_get_private ((GTypeInstance *) demux,
 294       GST_TYPE_TAG_DEMUX);
 295
 296   /* sink pad */
 297   tmpl = gst_element_class_get_pad_template (element_klass, "sink");
 298   if (tmpl) {
 299     demux->priv->sinkpad = gst_pad_new_from_template (tmpl, "sink");
 300
 301     gst_pad_set_activatemode_function (demux->priv->sinkpad,
 302         GST_DEBUG_FUNCPTR (gst_tag_demux_sink_activate_mode));
 303     gst_pad_set_activate_function (demux->priv->sinkpad,
 304         GST_DEBUG_FUNCPTR (gst_tag_demux_sink_activate));
 305     gst_pad_set_event_function (demux->priv->sinkpad,
 306         GST_DEBUG_FUNCPTR (gst_tag_demux_sink_event));
 307     gst_pad_set_chain_function (demux->priv->sinkpad,
 308         GST_DEBUG_FUNCPTR (gst_tag_demux_chain));
 309     gst_element_add_pad (GST_ELEMENT (demux), demux->priv->sinkpad);
 310   } else {
 311     g_warning ("GstTagDemux subclass %s must provide a sink pad template",
 312         G_OBJECT_TYPE_NAME (demux));
 313   }
 314
 315   /* source pad */
 316   tmpl = gst_element_class_get_pad_template (element_klass, "src");
 317   demux->priv->srcpad = gst_pad_new_from_template (tmpl, "src");
 318   gst_pad_set_query_function (demux->priv->srcpad,
 319       GST_DEBUG_FUNCPTR (gst_tag_demux_pad_query));
 320   gst_pad_set_event_function (demux->priv->srcpad,
 321       GST_DEBUG_FUNCPTR (gst_tag_demux_srcpad_event));
 322   gst_pad_set_activatemode_function (demux->priv->srcpad,
 323       GST_DEBUG_FUNCPTR (gst_tag_demux_src_activate_mode));
 324   gst_pad_set_getrange_function (demux->priv->srcpad,
 325       GST_DEBUG_FUNCPTR (gst_tag_demux_src_getrange));
 326   gst_pad_use_fixed_caps (demux->priv->srcpad);
 327   gst_element_add_pad (GST_ELEMENT (demux), demux->priv->srcpad);
 328
 329   gst_tag_demux_reset (demux);
 330 }
 331
 332 static void
 333 gst_tag_demux_dispose (GObject * object)
 334 {
 335   GstTagDemux *tagdemux = GST_TAG_DEMUX (object);
 336
 337   gst_tag_demux_reset (tagdemux);
 338
 339   G_OBJECT_CLASS (parent_class)->dispose (object);
 340 }
 341
 342 // FIXME: convert to set_caps / sending a caps event
 343 static void
 344 gst_tag_demux_set_src_caps (GstTagDemux * tagdemux, GstCaps * new_caps)
 345 {
 346   GstCaps *old_caps = tagdemux->priv->src_caps;
 347
 348   if (old_caps == NULL || !gst_caps_is_equal (new_caps, old_caps)) {
 349     gchar *stream_id = gst_pad_create_stream_id (tagdemux->priv->srcpad,
 350         GST_ELEMENT_CAST (tagdemux), NULL);
 351
 352     gst_caps_replace (&tagdemux->priv->src_caps, new_caps);
 353
 354     GST_DEBUG_OBJECT (tagdemux, "Changing src pad caps to %" GST_PTR_FORMAT,
 355         tagdemux->priv->src_caps);
 356
 357     gst_pad_push_event (tagdemux->priv->srcpad,
 358         gst_event_new_stream_start (stream_id));
 359     g_free (stream_id);
 360     gst_pad_set_caps (tagdemux->priv->srcpad, tagdemux->priv->src_caps);
 361   } else {
 362     /* Caps never changed */
 363   }
 364 }
 365
 366 /* will return FALSE if buffer is beyond end of data; will return TRUE
 367  * if buffer was trimmed successfully or didn't need trimming, but may
 368  * also return TRUE and set *buf_ref to NULL if the buffer was before
 369  * the start of the data */
 370 static gboolean
 371 gst_tag_demux_trim_buffer (GstTagDemux * tagdemux, GstBuffer ** buf_ref,
 372     gsize * buf_size)
 373 {
 374   GstBuffer *buf = *buf_ref;
 375
 376   guint trim_start = 0;
 377   guint out_size, bsize;
 378   guint64 out_offset, boffset;
 379   gboolean need_sub = FALSE;
 380
 381   bsize = out_size = gst_buffer_get_size (buf);
 382   boffset = out_offset = GST_BUFFER_OFFSET (buf);
 383
 384   /* Adjust offset and length */
 385   if (!GST_BUFFER_OFFSET_IS_VALID (buf)) {
 386     /* Can't change anything without an offset */
 387     *buf_size = bsize;
 388     return TRUE;
 389   }
 390
 391   /* If the buffer crosses the tag at the end of file, trim it */
 392   if (tagdemux->priv->strip_end > 0) {
 393     if (gst_tag_demux_get_upstream_size (tagdemux)) {
 394       guint64 v1tag_offset =
 395           tagdemux->priv->upstream_size - tagdemux->priv->strip_end;
 396
 397       if (out_offset >= v1tag_offset) {
 398         GST_DEBUG_OBJECT (tagdemux, "Buffer is past the end of the data");
 399         goto no_out_buffer_end;
 400       }
 401
 402       if (out_offset + out_size > v1tag_offset) {
 403         out_size = v1tag_offset - out_offset;
 404         need_sub = TRUE;
 405       }
 406     }
 407   }
 408
 409   if (tagdemux->priv->strip_start > 0) {
 410     /* If the buffer crosses the tag at the start of file, trim it */
 411     if (out_offset <= tagdemux->priv->strip_start) {
 412       if (out_offset + out_size <= tagdemux->priv->strip_start) {
 413         GST_DEBUG_OBJECT (tagdemux, "Buffer is before the start of the data");
 414         goto no_out_buffer_start;
 415       }
 416
 417       trim_start = tagdemux->priv->strip_start - out_offset;
 418       out_size -= trim_start;
 419       out_offset = 0;
 420     } else {
 421       out_offset -= tagdemux->priv->strip_start;
 422     }
 423     need_sub = TRUE;
 424   }
 425
 426   if (need_sub == TRUE) {
 427     if (out_size != bsize || !gst_buffer_is_writable (buf)) {
 428       GstBuffer *sub;
 429
 430       GST_DEBUG_OBJECT (tagdemux, "Sub-buffering to trim size %d offset %"
 431           G_GINT64_FORMAT " to %d offset %" G_GINT64_FORMAT,
 432           bsize, boffset, out_size, out_offset);
 433
 434       sub =
 435           gst_buffer_copy_region (buf, GST_BUFFER_COPY_ALL, trim_start,
 436           out_size);
 437       g_return_val_if_fail (sub != NULL, FALSE);
 438       gst_buffer_unref (buf);
 439       *buf_ref = buf = sub;
 440       *buf_size = out_size;
 441     } else {
 442       GST_DEBUG_OBJECT (tagdemux, "Adjusting buffer from size %d offset %"
 443           G_GINT64_FORMAT " to %d offset %" G_GINT64_FORMAT,
 444           bsize, boffset, out_size, out_offset);
 445     }
 446
 447     GST_BUFFER_OFFSET (buf) = out_offset;
 448     GST_BUFFER_OFFSET_END (buf) = out_offset + out_size;
 449   }
 450
 451   return TRUE;
 452
 453 no_out_buffer_end:
 454   {
 455     gst_buffer_unref (buf);
 456     *buf_ref = NULL;
 457     return FALSE;
 458   }
 459 no_out_buffer_start:
 460   {
 461     gst_buffer_unref (buf);
 462     *buf_ref = NULL;
 463     return TRUE;
 464   }
 465 }
 466
 467 static void
 468 gst_tag_demux_chain_parse_tag (GstTagDemux * demux, GstBuffer * collect)
 469 {
 470   GstTagDemuxResult parse_ret;
 471   GstTagDemuxClass *klass;
 472   guint tagsize = 0;
 473   guint available;
 474
 475   g_assert (gst_buffer_is_writable (collect));
 476
 477   klass = GST_TAG_DEMUX_CLASS (G_OBJECT_GET_CLASS (demux));
 478
 479   /* If we receive a buffer that's from the middle of the file,
 480    * we can't read tags so move to typefinding */
 481   if (GST_BUFFER_OFFSET_IS_VALID (collect) && GST_BUFFER_OFFSET (collect) != 0) {
 482     GST_DEBUG_OBJECT (demux, "Received buffer from non-zero offset %"
 483         G_GINT64_FORMAT ". Can't read tags", GST_BUFFER_OFFSET (collect));
 484     demux->priv->state = GST_TAG_DEMUX_TYPEFINDING;
 485     return;
 486   }
 487
 488   g_assert (klass->identify_tag != NULL);
 489   g_assert (klass->parse_tag != NULL);
 490
 491   available = gst_buffer_get_size (collect);
 492
 493   if (available < klass->min_start_size) {
 494     GST_DEBUG_OBJECT (demux, "Only %u bytes available, but %u needed "
 495         "to identify tag", available, klass->min_start_size);
 496     return;                     /* wait for more data */
 497   }
 498
 499   if (!klass->identify_tag (demux, collect, TRUE, &tagsize)) {
 500     GST_DEBUG_OBJECT (demux, "Could not identify start tag");
 501     demux->priv->state = GST_TAG_DEMUX_TYPEFINDING;
 502     return;
 503   }
 504
 505   /* need to set offset of first buffer to 0 or trimming won't work */
 506   if (!GST_BUFFER_OFFSET_IS_VALID (collect)) {
 507     GST_WARNING_OBJECT (demux, "Fixing up first buffer without offset");
 508     GST_BUFFER_OFFSET (collect) = 0;
 509   }
 510
 511   GST_DEBUG_OBJECT (demux, "Identified tag, size = %u bytes", tagsize);
 512
 513   do {
 514     GstTagList *tags = NULL;
 515     guint newsize, saved_size;
 516
 517     demux->priv->strip_start = tagsize;
 518
 519     if (available < tagsize) {
 520       GST_DEBUG_OBJECT (demux, "Only %u bytes available, but %u needed "
 521           "to parse tag", available, tagsize);
 522       return;                   /* wait for more data */
 523     }
 524
 525     saved_size = gst_buffer_get_size (collect);
 526     gst_buffer_set_size (collect, tagsize);
 527     newsize = tagsize;
 528
 529     parse_ret = klass->parse_tag (demux, collect, TRUE, &newsize, &tags);
 530
 531     gst_buffer_set_size (collect, saved_size);
 532
 533     switch (parse_ret) {
 534       case GST_TAG_DEMUX_RESULT_OK:
 535         demux->priv->strip_start = newsize;
 536         demux->priv->parsed_tags = tags;
 537         GST_DEBUG_OBJECT (demux, "Read start tag of size %u", newsize);
 538         break;
 539       case GST_TAG_DEMUX_RESULT_BROKEN_TAG:
 540         demux->priv->strip_start = newsize;
 541         demux->priv->parsed_tags = tags;
 542         GST_WARNING_OBJECT (demux, "Ignoring broken start tag of size %d",
 543             demux->priv->strip_start);
 544         break;
 545       case GST_TAG_DEMUX_RESULT_AGAIN:
 546         GST_DEBUG_OBJECT (demux, "Re-parse, this time with %u bytes", newsize);
 547         g_assert (newsize != tagsize);
 548         tagsize = newsize;
 549         break;
 550     }
 551   } while (parse_ret == GST_TAG_DEMUX_RESULT_AGAIN);
 552
 553   GST_LOG_OBJECT (demux, "Parsed tag. Proceeding to typefinding");
 554   demux->priv->state = GST_TAG_DEMUX_TYPEFINDING;
 555   demux->priv->send_tag_event = TRUE;
 556 }
 557
 558 static GstFlowReturn
 559 gst_tag_demux_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
 560 {
 561   GstTagDemux *demux;
 562   gsize size;
 563
 564   demux = GST_TAG_DEMUX (parent);
 565
 566   size = gst_buffer_get_size (buf);
 567
 568   /* Update our segment position info */
 569   if (demux->priv->segment.format == GST_FORMAT_BYTES) {
 570     if (GST_BUFFER_OFFSET_IS_VALID (buf))
 571       demux->priv->segment.position = GST_BUFFER_OFFSET (buf);
 572     demux->priv->segment.position += size;
 573   } else if (demux->priv->segment.format == GST_FORMAT_TIME) {
 574     if (GST_BUFFER_TIMESTAMP_IS_VALID (buf))
 575       demux->priv->segment.position = GST_BUFFER_TIMESTAMP (buf);
 576     if (GST_BUFFER_DURATION_IS_VALID (buf))
 577       demux->priv->segment.position += GST_BUFFER_DURATION (buf);
 578   }
 579
 580   if (demux->priv->collect == NULL) {
 581     demux->priv->collect = buf;
 582   } else {
 583     demux->priv->collect = gst_buffer_append (demux->priv->collect, buf);
 584   }
 585   demux->priv->collect_size += size;
 586   buf = NULL;
 587
 588   switch (demux->priv->state) {
 589     case GST_TAG_DEMUX_READ_START_TAG:
 590       demux->priv->collect = gst_buffer_make_writable (demux->priv->collect);
 591       gst_tag_demux_chain_parse_tag (demux, demux->priv->collect);
 592       if (demux->priv->state != GST_TAG_DEMUX_TYPEFINDING)
 593         break;
 594       /* Fall-through */
 595     case GST_TAG_DEMUX_TYPEFINDING:{
 596       GstTypeFindProbability probability = 0;
 597       GstBuffer *typefind_buf = NULL;
 598       gsize typefind_size;
 599       GstCaps *caps;
 600
 601       if (demux->priv->collect_size <
 602           TYPE_FIND_MIN_SIZE + demux->priv->strip_start)
 603         break;                  /* Go get more data first */
 604
 605       GST_DEBUG_OBJECT (demux, "Typefinding with size %" G_GSIZE_FORMAT,
 606           demux->priv->collect_size);
 607
 608       /* Trim the buffer and adjust offset for typefinding */
 609       typefind_buf = demux->priv->collect;
 610       gst_buffer_ref (typefind_buf);
 611       if (!gst_tag_demux_trim_buffer (demux, &typefind_buf, &typefind_size))
 612         return GST_FLOW_EOS;
 613
 614       if (typefind_buf == NULL)
 615         break;                  /* Still need more data */
 616
 617       caps = gst_type_find_helper_for_buffer (GST_OBJECT (demux),
 618           typefind_buf, &probability);
 619
 620       if (caps == NULL) {
 621         if (typefind_size < TYPE_FIND_MAX_SIZE) {
 622           /* Just break for more data */
 623           gst_buffer_unref (typefind_buf);
 624           return GST_FLOW_OK;
 625         }
 626
 627         /* We failed typefind */
 628         GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND, (NULL),
 629             ("Could not detect type for contents within tag"));
 630         gst_buffer_unref (typefind_buf);
 631         gst_buffer_unref (demux->priv->collect);
 632         demux->priv->collect = NULL;
 633         demux->priv->collect_size = 0;
 634         return GST_FLOW_ERROR;
 635       }
 636       gst_buffer_unref (typefind_buf);
 637
 638       GST_DEBUG_OBJECT (demux, "Found type %" GST_PTR_FORMAT " with a "
 639           "probability of %u", caps, probability);
 640
 641       gst_tag_demux_set_src_caps (demux, caps);
 642       gst_caps_unref (caps);
 643
 644       /* Move onto streaming and fall-through to push out existing
 645        * data */
 646       demux->priv->state = GST_TAG_DEMUX_STREAMING;
 647       /* fall-through */
 648     }
 649     case GST_TAG_DEMUX_STREAMING:{
 650       GstBuffer *outbuf = NULL;
 651       gsize outbuf_size;
 652
 653       /* Trim the buffer and adjust offset */
 654       if (demux->priv->collect) {
 655         outbuf = demux->priv->collect;
 656         demux->priv->collect = NULL;
 657         demux->priv->collect_size = 0;
 658         if (!gst_tag_demux_trim_buffer (demux, &outbuf, &outbuf_size))
 659           return GST_FLOW_EOS;
 660       }
 661       if (outbuf) {
 662         /* Might need a new segment before the buffer */
 663         if (demux->priv->need_newseg) {
 664           if (!gst_tag_demux_send_new_segment (demux)) {
 665             GST_WARNING_OBJECT (demux, "Downstream did not handle newsegment "
 666                 "event as it should");
 667           }
 668           demux->priv->need_newseg = FALSE;
 669         }
 670
 671         /* send any pending events we cached */
 672         gst_tag_demux_send_pending_events (demux);
 673
 674         /* Send our own pending tag event */
 675         if (demux->priv->send_tag_event) {
 676           gst_tag_demux_send_tag_event (demux);
 677           demux->priv->send_tag_event = FALSE;
 678         }
 679
 680         GST_LOG_OBJECT (demux, "Pushing buffer %p", outbuf);
 681
 682         return gst_pad_push (demux->priv->srcpad, outbuf);
 683       }
 684     }
 685   }
 686   return GST_FLOW_OK;
 687 }
 688
 689 static gboolean
 690 gst_tag_demux_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
 691 {
 692   GstTagDemux *demux;
 693   gboolean ret;
 694
 695   demux = GST_TAG_DEMUX (parent);
 696
 697   switch (GST_EVENT_TYPE (event)) {
 698     case GST_EVENT_EOS:
 699       /* FIXME, detect this differently */
 700       if (demux->priv->srcpad == NULL) {
 701         GST_WARNING_OBJECT (demux, "EOS before we found a type");
 702         GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND, (NULL), (NULL));
 703       }
 704       ret = gst_pad_event_default (pad, parent, event);
 705       break;
 706     case GST_EVENT_SEGMENT:
 707     {
 708       gst_event_copy_segment (event, &demux->priv->segment);
 709
 710       demux->priv->need_newseg = TRUE;
 711       gst_event_unref (event);
 712       ret = TRUE;
 713       break;
 714     }
 715     case GST_EVENT_FLUSH_STOP:
 716     case GST_EVENT_FLUSH_START:
 717       ret = gst_pad_event_default (pad, parent, event);
 718       break;
 719     case GST_EVENT_CAPS:
 720       /* we drop the caps event. We do typefind and push a new caps event. */
 721       ret = gst_pad_event_default (pad, parent, event);
 722       break;
 723     default:
 724       if (demux->priv->need_newseg && GST_EVENT_IS_SERIALIZED (event)) {
 725         /* Cache all events if we have a pending segment, so they don't get
 726          * lost (esp. tag events) */
 727         GST_INFO_OBJECT (demux, "caching event: %" GST_PTR_FORMAT, event);
 728         GST_OBJECT_LOCK (demux);
 729         demux->priv->pending_events =
 730             g_list_append (demux->priv->pending_events, event);
 731         GST_OBJECT_UNLOCK (demux);
 732         ret = TRUE;
 733       } else {
 734         ret = gst_pad_event_default (pad, parent, event);
 735       }
 736       break;
 737   }
 738
 739   return ret;
 740 }
 741
 742 static gboolean
 743 gst_tag_demux_get_upstream_size (GstTagDemux * tagdemux)
 744 {
 745   gint64 len;
 746
 747   /* Short-cut if we already queried upstream */
 748   if (tagdemux->priv->upstream_size > 0)
 749     return TRUE;
 750
 751   if (!gst_pad_peer_query_duration (tagdemux->priv->sinkpad, GST_FORMAT_BYTES,
 752           &len) || len <= 0) {
 753     return FALSE;
 754   }
 755
 756   tagdemux->priv->upstream_size = len;
 757   return TRUE;
 758 }
 759
 760 static gboolean
 761 gst_tag_demux_srcpad_event (GstPad * pad, GstObject * parent, GstEvent * event)
 762 {
 763   GstTagDemux *tagdemux;
 764   gboolean res = FALSE;
 765
 766   tagdemux = GST_TAG_DEMUX (parent);
 767
 768   /* Handle SEEK events, with adjusted byte offsets and sizes. */
 769
 770   switch (GST_EVENT_TYPE (event)) {
 771     case GST_EVENT_SEEK:
 772     {
 773       gdouble rate;
 774       GstFormat format;
 775       GstSeekType start_type, stop_type;
 776       GstSeekFlags flags;
 777       gint64 start, stop;
 778
 779       gst_event_parse_seek (event, &rate, &format, &flags,
 780           &start_type, &start, &stop_type, &stop);
 781
 782       if (format == GST_FORMAT_BYTES &&
 783           tagdemux->priv->state == GST_TAG_DEMUX_STREAMING &&
 784           gst_pad_is_linked (tagdemux->priv->sinkpad)) {
 785         GstEvent *upstream;
 786
 787         switch (start_type) {
 788           case GST_SEEK_TYPE_SET:
 789             if (start == -1)
 790               start = 0;
 791             start += tagdemux->priv->strip_start;
 792             break;
 793           case GST_SEEK_TYPE_END:
 794             /* Adjust the seek to be relative to the start of any end tag
 795              * (note: 10 bytes before end is represented by stop=-10) */
 796             if (start > 0)
 797               start = 0;
 798             start -= tagdemux->priv->strip_end;
 799             break;
 800           case GST_SEEK_TYPE_NONE:
 801           default:
 802             break;
 803         }
 804         switch (stop_type) {
 805           case GST_SEEK_TYPE_SET:
 806             if (stop != -1) {
 807               /* -1 means the end of the file, pass it upstream intact */
 808               stop += tagdemux->priv->strip_start;
 809             }
 810             break;
 811           case GST_SEEK_TYPE_END:
 812             /* Adjust the seek to be relative to the start of any end tag
 813              * (note: 10 bytes before end is represented by stop=-10) */
 814             if (stop > 0)
 815               stop = 0;
 816             stop -= tagdemux->priv->strip_end;
 817             break;
 818           case GST_SEEK_TYPE_NONE:
 819           default:
 820             break;
 821         }
 822         upstream = gst_event_new_seek (rate, format, flags,
 823             start_type, start, stop_type, stop);
 824         res = gst_pad_push_event (tagdemux->priv->sinkpad, upstream);
 825       }
 826       break;
 827     }
 828     default:
 829       res = gst_pad_push_event (tagdemux->priv->sinkpad, event);
 830       event = NULL;
 831       break;
 832   }
 833
 834   if (event)
 835     gst_event_unref (event);
 836
 837   return res;
 838 }
 839
 840 /* Read and interpret any end tag when activating in pull_range.
 841  * Returns FALSE if pad activation should fail. */
 842 static gboolean
 843 gst_tag_demux_pull_end_tag (GstTagDemux * demux, GstTagList ** tags)
 844 {
 845   GstTagDemuxResult parse_ret;
 846   GstTagDemuxClass *klass;
 847   GstFlowReturn flow_ret;
 848   GstTagList *new_tags = NULL;
 849   GstBuffer *buffer = NULL;
 850   gboolean have_tag;
 851   gboolean res = FALSE;
 852   guint64 offset;
 853   guint tagsize;
 854   gsize bsize;
 855
 856   klass = GST_TAG_DEMUX_CLASS (G_OBJECT_GET_CLASS (demux));
 857
 858   g_assert (klass->identify_tag != NULL);
 859   g_assert (klass->parse_tag != NULL);
 860
 861   if (klass->min_end_size == 0) {
 862     GST_DEBUG_OBJECT (demux, "Not looking for tag at the end");
 863     return TRUE;
 864   }
 865
 866   if (demux->priv->upstream_size < klass->min_end_size) {
 867     GST_DEBUG_OBJECT (demux, "File too small");
 868     return TRUE;
 869   }
 870
 871   /* Pull enough to identify the tag and retrieve its total size */
 872   offset = demux->priv->upstream_size - klass->min_end_size;
 873
 874   flow_ret = gst_pad_pull_range (demux->priv->sinkpad, offset,
 875       klass->min_end_size, &buffer);
 876
 877   if (flow_ret != GST_FLOW_OK) {
 878     GST_DEBUG_OBJECT (demux, "Could not read tag header from end of file, "
 879         "ret = %s", gst_flow_get_name (flow_ret));
 880     goto done;
 881   }
 882
 883   bsize = gst_buffer_get_size (buffer);
 884
 885   if (bsize < klass->min_end_size) {
 886     GST_DEBUG_OBJECT (demux, "Only managed to read %" G_GSIZE_FORMAT " bytes"
 887         "from file (required: %u bytes)", bsize, klass->min_end_size);
 888     goto done;
 889   }
 890
 891   have_tag = klass->identify_tag (demux, buffer, FALSE, &tagsize);
 892
 893   if (!have_tag) {
 894     GST_DEBUG_OBJECT (demux, "Could not find tag at end");
 895     goto done;
 896   }
 897
 898   /* Now pull the entire tag */
 899   do {
 900     guint newsize, saved_size;
 901
 902     GST_DEBUG_OBJECT (demux, "Identified tag at end, size=%u bytes", tagsize);
 903
 904     demux->priv->strip_end = tagsize;
 905
 906     g_assert (tagsize >= klass->min_end_size);
 907
 908     /* Get buffer that's exactly the requested size */
 909     if (bsize != tagsize) {
 910       gst_buffer_unref (buffer);
 911       buffer = NULL;
 912
 913       offset = demux->priv->upstream_size - tagsize;
 914
 915       flow_ret = gst_pad_pull_range (demux->priv->sinkpad, offset,
 916           tagsize, &buffer);
 917
 918       if (flow_ret != GST_FLOW_OK) {
 919         GST_DEBUG_OBJECT (demux, "Could not read data from end of file at "
 920             "offset %" G_GUINT64_FORMAT ". ret = %s", offset,
 921             gst_flow_get_name (flow_ret));
 922         goto done;
 923       }
 924
 925       bsize = gst_buffer_get_size (buffer);
 926
 927       if (bsize < tagsize) {
 928         GST_DEBUG_OBJECT (demux, "Only managed to read %" G_GSIZE_FORMAT
 929             " bytes from file", bsize);
 930         goto done;
 931       }
 932     }
 933
 934     GST_BUFFER_OFFSET (buffer) = offset;
 935
 936     saved_size = bsize;
 937     gst_buffer_set_size (buffer, tagsize);
 938     newsize = tagsize;
 939
 940     parse_ret = klass->parse_tag (demux, buffer, FALSE, &newsize, &new_tags);
 941
 942     gst_buffer_set_size (buffer, saved_size);
 943
 944     switch (parse_ret) {
 945       case GST_TAG_DEMUX_RESULT_OK:
 946         res = TRUE;
 947         demux->priv->strip_end = newsize;
 948         GST_DEBUG_OBJECT (demux, "Read tag at end, size %d",
 949             demux->priv->strip_end);
 950         break;
 951       case GST_TAG_DEMUX_RESULT_BROKEN_TAG:
 952         res = TRUE;
 953         demux->priv->strip_end = newsize;
 954         GST_WARNING_OBJECT (demux, "Ignoring broken tag at end, size %d",
 955             demux->priv->strip_end);
 956         break;
 957       case GST_TAG_DEMUX_RESULT_AGAIN:
 958         GST_DEBUG_OBJECT (demux, "Re-parse, this time with %d bytes", newsize);
 959         g_assert (newsize != tagsize);
 960         tagsize = newsize;
 961         break;
 962     }
 963   } while (parse_ret == GST_TAG_DEMUX_RESULT_AGAIN);
 964
 965   *tags = new_tags;
 966   new_tags = NULL;
 967
 968 done:
 969   if (new_tags)
 970     gst_tag_list_unref (new_tags);
 971   if (buffer)
 972     gst_buffer_unref (buffer);
 973   return res;
 974 }
 975
 976 /* Read and interpret any tag at the start when activating in
 977  * pull_range. Returns FALSE if pad activation should fail. */
 978 static gboolean
 979 gst_tag_demux_pull_start_tag (GstTagDemux * demux, GstTagList ** tags)
 980 {
 981   GstTagDemuxResult parse_ret;
 982   GstTagDemuxClass *klass;
 983   GstFlowReturn flow_ret;
 984   GstTagList *new_tags = NULL;
 985   GstBuffer *buffer = NULL;
 986   gboolean have_tag;
 987   gboolean res = FALSE;
 988   guint req, tagsize;
 989   gsize bsize;
 990
 991   klass = GST_TAG_DEMUX_CLASS (G_OBJECT_GET_CLASS (demux));
 992
 993   g_assert (klass->identify_tag != NULL);
 994   g_assert (klass->parse_tag != NULL);
 995
 996   if (klass->min_start_size == 0) {
 997     GST_DEBUG_OBJECT (demux, "Not looking for tag at the beginning");
 998     return TRUE;
 999   }
1000
1001   /* Handle tag at start. Try with 4kB to start with */
1002   req = MAX (klass->min_start_size, 4096);
1003
1004   /* Pull enough to identify the tag and retrieve its total size */
1005   flow_ret = gst_pad_pull_range (demux->priv->sinkpad, 0, req, &buffer);
1006   if (flow_ret != GST_FLOW_OK) {
1007     GST_DEBUG_OBJECT (demux, "Could not read data from start of file ret=%s",
1008         gst_flow_get_name (flow_ret));
1009     goto done;
1010   }
1011
1012   bsize = gst_buffer_get_size (buffer);
1013
1014   if (bsize < klass->min_start_size) {
1015     GST_DEBUG_OBJECT (demux, "Only managed to read %" G_GSIZE_FORMAT
1016         " bytes from file - no tag in this file", bsize);
1017     goto done;
1018   }
1019
1020   have_tag = klass->identify_tag (demux, buffer, TRUE, &tagsize);
1021
1022   if (!have_tag) {
1023     GST_DEBUG_OBJECT (demux, "Could not find start tag");
1024     res = TRUE;
1025     goto done;
1026   }
1027
1028   GST_DEBUG_OBJECT (demux, "Identified start tag, size = %u bytes", tagsize);
1029
1030   do {
1031     guint newsize, saved_size;
1032
1033     demux->priv->strip_start = tagsize;
1034
1035     /* Now pull the entire tag */
1036     g_assert (tagsize >= klass->min_start_size);
1037
1038     if (bsize < tagsize) {
1039       gst_buffer_unref (buffer);
1040       buffer = NULL;
1041
1042       flow_ret = gst_pad_pull_range (demux->priv->sinkpad, 0, tagsize, &buffer);
1043       if (flow_ret != GST_FLOW_OK) {
1044         GST_DEBUG_OBJECT (demux, "Could not read data from start of file, "
1045             "ret = %s", gst_flow_get_name (flow_ret));
1046         goto done;
1047       }
1048
1049       bsize = gst_buffer_get_size (buffer);
1050
1051       if (bsize < tagsize) {
1052         GST_DEBUG_OBJECT (demux, "Only managed to read %" G_GSIZE_FORMAT
1053             " bytes from file", bsize);
1054         GST_ELEMENT_ERROR (demux, STREAM, DECODE,
1055             (_("Failed to read tag: not enough data")), (NULL));
1056         goto done;
1057       }
1058     }
1059
1060     saved_size = bsize;
1061     gst_buffer_set_size (buffer, tagsize);
1062     newsize = tagsize;
1063     parse_ret = klass->parse_tag (demux, buffer, TRUE, &newsize, &new_tags);
1064
1065     gst_buffer_set_size (buffer, saved_size);
1066
1067     switch (parse_ret) {
1068       case GST_TAG_DEMUX_RESULT_OK:
1069         res = TRUE;
1070         demux->priv->strip_start = newsize;
1071         GST_DEBUG_OBJECT (demux, "Read start tag of size %d", newsize);
1072         break;
1073       case GST_TAG_DEMUX_RESULT_BROKEN_TAG:
1074         res = TRUE;
1075         demux->priv->strip_start = newsize;
1076         GST_WARNING_OBJECT (demux, "Ignoring broken start tag of size %d",
1077             demux->priv->strip_start);
1078         break;
1079       case GST_TAG_DEMUX_RESULT_AGAIN:
1080         GST_DEBUG_OBJECT (demux, "Re-parse, this time with %d bytes", newsize);
1081         g_assert (newsize != tagsize);
1082         tagsize = newsize;
1083         break;
1084     }
1085   } while (parse_ret == GST_TAG_DEMUX_RESULT_AGAIN);
1086
1087   *tags = new_tags;
1088   new_tags = NULL;
1089
1090 done:
1091   if (new_tags)
1092     gst_tag_list_unref (new_tags);
1093   if (buffer)
1094     gst_buffer_unref (buffer);
1095   return res;
1096 }
1097
1098 /* This function operates similarly to gst_type_find_element_loop
1099  * in the typefind element
1100  * 1. try to read tags in pull mode
1101  * 2. typefind the contents
1102  * 3. if we didn't find any caps, fail.
1103  * 4. set caps on srcpad
1104  */
1105 static GstFlowReturn
1106 gst_tag_demux_element_find (GstTagDemux * demux)
1107 {
1108   GstTagDemuxClass *klass;
1109   GstTypeFindProbability probability = 0;
1110   GstFlowReturn ret = GST_FLOW_OK;
1111   GstTagList *start_tags = NULL;
1112   GstTagList *end_tags = NULL;
1113   gboolean e_tag_ok, s_tag_ok;
1114   GstCaps *caps = NULL;
1115
1116   /* Look for tags at start and end of file */
1117   GST_DEBUG_OBJECT (demux, "Activated pull mode. Looking for tags");
1118   if (!gst_tag_demux_get_upstream_size (demux))
1119     goto no_size;
1120
1121   demux->priv->strip_start = 0;
1122   demux->priv->strip_end = 0;
1123
1124   /* 1 - Read tags */
1125   s_tag_ok = gst_tag_demux_pull_start_tag (demux, &start_tags);
1126   e_tag_ok = gst_tag_demux_pull_end_tag (demux, &end_tags);
1127
1128   klass = GST_TAG_DEMUX_CLASS (G_OBJECT_GET_CLASS (demux));
1129
1130   if (klass->merge_tags != NULL) {
1131     demux->priv->parsed_tags = klass->merge_tags (demux, start_tags, end_tags);
1132   } else {
1133     /* we merge in REPLACE mode, so put the less important tags first, which
1134      * we'll just assume is the end tag (subclasses may change this behaviour
1135      * or make it configurable by overriding the merge_tags vfunc) */
1136     demux->priv->parsed_tags =
1137         gst_tag_list_merge (end_tags, start_tags, GST_TAG_MERGE_REPLACE);
1138   }
1139
1140   if (start_tags)
1141     gst_tag_list_unref (start_tags);
1142   if (end_tags)
1143     gst_tag_list_unref (end_tags);
1144
1145   if (!e_tag_ok && !s_tag_ok)
1146     goto no_tags;
1147
1148   if (demux->priv->parsed_tags != NULL) {
1149     demux->priv->send_tag_event = TRUE;
1150   }
1151
1152   if (demux->priv->upstream_size <=
1153       demux->priv->strip_start + demux->priv->strip_end)
1154     goto no_data;
1155
1156   /* 2 - Do typefinding on data, but not if downstream is in charge */
1157   if (GST_PAD_MODE (demux->priv->srcpad) == GST_PAD_MODE_PULL)
1158     goto skip_typefinding;
1159
1160   caps = gst_type_find_helper_get_range (GST_OBJECT (demux), NULL,
1161       (GstTypeFindHelperGetRangeFunction) gst_tag_demux_read_range,
1162       demux->priv->upstream_size
1163       - (demux->priv->strip_start + demux->priv->strip_end), NULL,
1164       &probability);
1165
1166   GST_INFO_OBJECT (demux, "Found type %" GST_PTR_FORMAT " with a "
1167       "probability of %u", caps, probability);
1168
1169   /* 3 - If we didn't find the caps, fail */
1170   if (caps == NULL)
1171     goto no_caps;
1172
1173   /* tag reading and typefinding were already done, don't do them again in
1174    * the chain function if we end up in push mode */
1175   demux->priv->state = GST_TAG_DEMUX_STREAMING;
1176
1177   /* 6 Set the srcpad caps now that we know them */
1178   gst_tag_demux_set_src_caps (demux, caps);
1179   gst_caps_unref (caps);
1180
1181 skip_typefinding:
1182
1183   /* set it again, in case we skipped typefinding */
1184   demux->priv->state = GST_TAG_DEMUX_STREAMING;
1185
1186   return ret;
1187
1188   /* ERRORS */
1189 no_size:
1190   {
1191     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND,
1192         ("Could not get stream size"), (NULL));
1193     return GST_FLOW_ERROR;
1194   }
1195 no_tags:
1196   {
1197     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND,
1198         ("Could not get start and/or end tag"), (NULL));
1199     return GST_FLOW_ERROR;
1200   }
1201 no_data:
1202   {
1203     /* There was no data (probably due to a truncated file) */
1204     /* so we don't know about type either */
1205     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND, ("No data in file"),
1206         (NULL));
1207     return GST_FLOW_ERROR;
1208   }
1209 no_caps:
1210   {
1211     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND,
1212         ("Could not detect type of contents"), (NULL));
1213     return GST_FLOW_ERROR;
1214   }
1215 }
1216
1217 /* This function operates similarly to gst_type_find_element_loop
1218  * in the typefind element
1219  * 1. try to read tags in pull mode
1220  * 2. typefind the contents
1221  * 3. if we didn't find any caps, fail.
1222  * 4. set caps on srcpad
1223  */
1224 static void
1225 gst_tag_demux_element_loop (GstTagDemux * demux)
1226 {
1227   GstFlowReturn ret;
1228
1229   switch (demux->priv->state) {
1230     case GST_TAG_DEMUX_READ_START_TAG:
1231     case GST_TAG_DEMUX_TYPEFINDING:
1232       ret = gst_tag_demux_element_find (demux);
1233       break;
1234     case GST_TAG_DEMUX_STREAMING:
1235     {
1236       GstBuffer *outbuf = NULL;
1237
1238       if (demux->priv->need_newseg) {
1239         demux->priv->need_newseg = FALSE;
1240         /* FIXME: check segment, should be 0-N for downstream */
1241         gst_tag_demux_send_new_segment (demux);
1242       }
1243
1244       /* Send our own pending tag event */
1245       if (demux->priv->send_tag_event) {
1246         gst_tag_demux_send_tag_event (demux);
1247         demux->priv->send_tag_event = FALSE;
1248       }
1249
1250       /* Pull data and push it downstream */
1251       ret = gst_pad_pull_range (demux->priv->sinkpad, demux->priv->offset,
1252           DEFAULT_PULL_BLOCKSIZE, &outbuf);
1253
1254       if (ret != GST_FLOW_OK)
1255         break;
1256
1257       demux->priv->offset += gst_buffer_get_size (outbuf);
1258
1259       ret = gst_pad_push (demux->priv->srcpad, outbuf);
1260       break;
1261     }
1262     default:
1263       ret = GST_FLOW_ERROR;
1264       break;
1265   }
1266   if (ret != GST_FLOW_OK)
1267     goto pause;
1268
1269   return;
1270
1271   /* ERRORS */
1272 pause:
1273   {
1274     const gchar *reason = gst_flow_get_name (ret);
1275     gboolean push_eos = FALSE;
1276
1277     GST_LOG_OBJECT (demux, "pausing task, reason %s", reason);
1278     gst_pad_pause_task (demux->priv->sinkpad);
1279
1280     if (ret == GST_FLOW_EOS) {
1281       /* perform EOS logic */
1282
1283       if (demux->priv->segment.flags & GST_SEEK_FLAG_SEGMENT) {
1284         gint64 stop;
1285
1286         /* for segment playback we need to post when (in stream time)
1287          * we stopped, this is either stop (when set) or the duration. */
1288         if ((stop = demux->priv->segment.stop) == -1)
1289           stop = demux->priv->offset;
1290
1291         GST_LOG_OBJECT (demux, "Sending segment done, at end of segment");
1292         gst_element_post_message (GST_ELEMENT_CAST (demux),
1293             gst_message_new_segment_done (GST_OBJECT_CAST (demux),
1294                 GST_FORMAT_BYTES, stop));
1295         gst_pad_push_event (demux->priv->srcpad,
1296             gst_event_new_segment_done (GST_FORMAT_BYTES, stop));
1297       } else {
1298         push_eos = TRUE;
1299       }
1300     } else if (ret == GST_FLOW_NOT_LINKED || ret < GST_FLOW_EOS) {
1301       /* for fatal errors we post an error message */
1302       GST_ELEMENT_ERROR (demux, STREAM, FAILED, (NULL),
1303           ("Stream stopped, reason %s", reason));
1304       push_eos = TRUE;
1305     }
1306     if (push_eos) {
1307       /* send EOS, and prevent hanging if no streams yet */
1308       GST_LOG_OBJECT (demux, "Sending EOS, at end of stream");
1309       gst_pad_push_event (demux->priv->srcpad, gst_event_new_eos ());
1310     }
1311     return;
1312   }
1313 }
1314
1315 static gboolean
1316 gst_tag_demux_sink_activate_mode (GstPad * pad, GstObject * parent,
1317     GstPadMode mode, gboolean active)
1318 {
1319   gboolean res;
1320
1321   switch (mode) {
1322     case GST_PAD_MODE_PULL:
1323       if (active)
1324         res = TRUE;
1325       else
1326         res = gst_pad_stop_task (pad);
1327       break;
1328     default:
1329       res = TRUE;
1330       break;
1331   }
1332
1333   if (active)
1334     GST_TAG_DEMUX (parent)->priv->state = GST_TAG_DEMUX_READ_START_TAG;
1335
1336   return res;
1337 }
1338
1339 static gboolean
1340 gst_tag_demux_sink_activate (GstPad * sinkpad, GstObject * parent)
1341 {
1342   GstTagDemux *demux;
1343   GstQuery *query;
1344   gboolean pull_mode;
1345
1346   demux = GST_TAG_DEMUX (parent);
1347
1348   /* 1: */
1349   /* If we can activate pull_range upstream, then read any end and start
1350    * tags, otherwise activate in push mode and the chain function will
1351    * collect buffers, read the start tag and output a buffer to end
1352    * preroll.
1353    */
1354   query = gst_query_new_scheduling ();
1355
1356   if (!gst_pad_peer_query (sinkpad, query)) {
1357     gst_query_unref (query);
1358     goto activate_push;
1359   }
1360
1361   pull_mode = gst_query_has_scheduling_mode_with_flags (query,
1362       GST_PAD_MODE_PULL, GST_SCHEDULING_FLAG_SEEKABLE);
1363   gst_query_unref (query);
1364
1365   if (!pull_mode)
1366     goto activate_push;
1367
1368   if (!gst_pad_activate_mode (sinkpad, GST_PAD_MODE_PULL, TRUE))
1369     goto activate_push;
1370
1371   /* only start our task if we ourselves decide to start in pull mode */
1372   return gst_pad_start_task (sinkpad,
1373       (GstTaskFunction) gst_tag_demux_element_loop, demux, NULL);
1374
1375 activate_push:
1376   {
1377     GST_DEBUG_OBJECT (demux, "No pull mode. Changing to push, but won't be "
1378         "able to read end tags");
1379     return gst_pad_activate_mode (sinkpad, GST_PAD_MODE_PUSH, TRUE);
1380   }
1381 }
1382
1383 static gboolean
1384 gst_tag_demux_src_activate_mode (GstPad * pad, GstObject * parent,
1385     GstPadMode mode, gboolean active)
1386 {
1387   gboolean res;
1388   GstTagDemux *demux = GST_TAG_DEMUX (parent);
1389
1390   switch (mode) {
1391     case GST_PAD_MODE_PULL:
1392       /* make sure our task stops pushing, we can't call _stop here
1393        * because this activation might happen from the streaming thread. */
1394       gst_pad_pause_task (demux->priv->sinkpad);
1395       res = gst_pad_activate_mode (demux->priv->sinkpad, mode, active);
1396       break;
1397     default:
1398       res = TRUE;
1399       break;
1400   }
1401   return res;
1402 }
1403
1404 static GstFlowReturn
1405 gst_tag_demux_read_range (GstTagDemux * demux, GstObject * parent,
1406     guint64 offset, guint length, GstBuffer ** buffer)
1407 {
1408   GstFlowReturn ret;
1409   guint64 in_offset;
1410   guint in_length;
1411   gsize size;
1412
1413   g_return_val_if_fail (buffer != NULL, GST_FLOW_ERROR);
1414
1415   /* Adjust offset and length of the request to trim off tag information.
1416    * For the returned buffer, adjust the output offset to match what downstream
1417    * should see */
1418   in_offset = offset + demux->priv->strip_start;
1419
1420   if (!gst_tag_demux_get_upstream_size (demux))
1421     return GST_FLOW_ERROR;
1422
1423   if (in_offset + length >= demux->priv->upstream_size - demux->priv->strip_end) {
1424     if (in_offset + demux->priv->strip_end >= demux->priv->upstream_size)
1425       return GST_FLOW_EOS;
1426     in_length = demux->priv->upstream_size - demux->priv->strip_end - in_offset;
1427   } else {
1428     in_length = length;
1429   }
1430
1431   ret = gst_pad_pull_range (demux->priv->sinkpad, in_offset, in_length, buffer);
1432
1433   if (ret == GST_FLOW_OK && *buffer) {
1434     if (!gst_tag_demux_trim_buffer (demux, buffer, &size))
1435       goto read_beyond_end;
1436
1437     /* this should only happen in streaming mode */
1438     g_assert (*buffer != NULL);
1439   }
1440
1441   return ret;
1442
1443 read_beyond_end:
1444   {
1445     GST_DEBUG_OBJECT (demux, "attempted read beyond end of file");
1446     if (*buffer != NULL) {
1447       gst_buffer_unref (*buffer);
1448       *buffer = NULL;
1449     }
1450     return GST_FLOW_EOS;
1451   }
1452 }
1453
1454 static GstFlowReturn
1455 gst_tag_demux_src_getrange (GstPad * srcpad, GstObject * parent,
1456     guint64 offset, guint length, GstBuffer ** buffer)
1457 {
1458   GstTagDemux *demux = GST_TAG_DEMUX (parent);
1459
1460   /* downstream in pull mode won't miss a newsegment event,
1461    * but it likely appreciates other (tag) events */
1462   if (demux->priv->need_newseg) {
1463     gst_tag_demux_send_pending_events (demux);
1464     demux->priv->need_newseg = FALSE;
1465   }
1466
1467   if (demux->priv->send_tag_event) {
1468     gst_tag_demux_send_tag_event (demux);
1469     demux->priv->send_tag_event = FALSE;
1470   }
1471
1472   return gst_tag_demux_read_range (demux, NULL, offset, length, buffer);
1473 }
1474
1475 static GstStateChangeReturn
1476 gst_tag_demux_change_state (GstElement * element, GstStateChange transition)
1477 {
1478   GstStateChangeReturn ret;
1479   GstTagDemux *demux = GST_TAG_DEMUX (element);
1480
1481   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1482
1483   switch (transition) {
1484     case GST_STATE_CHANGE_PAUSED_TO_READY:
1485       gst_tag_demux_reset (demux);
1486       break;
1487     default:
1488       break;
1489   }
1490   return ret;
1491 }
1492
1493 static gboolean
1494 gst_tag_demux_pad_query (GstPad * pad, GstObject * parent, GstQuery * query)
1495 {
1496   /* For a position or duration query, adjust the returned
1497    * bytes to strip off the end and start areas */
1498   GstTagDemux *demux = GST_TAG_DEMUX (parent);
1499   GstFormat format;
1500   gint64 result;
1501   gboolean res = TRUE;
1502
1503   /* FIXME: locking ? */
1504   switch (GST_QUERY_TYPE (query)) {
1505     case GST_QUERY_SCHEDULING:
1506       res = gst_pad_peer_query (demux->priv->sinkpad, query);
1507       break;
1508     case GST_QUERY_POSITION:
1509     {
1510       if (!(res = gst_pad_peer_query (demux->priv->sinkpad, query)))
1511         goto done;
1512
1513       gst_query_parse_position (query, &format, &result);
1514       if (format == GST_FORMAT_BYTES) {
1515         result -= demux->priv->strip_start;
1516         gst_query_set_position (query, format, result);
1517       }
1518       break;
1519     }
1520     case GST_QUERY_DURATION:
1521     {
1522       if (!(res = gst_pad_peer_query (demux->priv->sinkpad, query)))
1523         goto done;
1524
1525       gst_query_parse_duration (query, &format, &result);
1526       if (format == GST_FORMAT_BYTES) {
1527         /* if downstream activated us in pull mode right away, e.g. in case of
1528          * filesrc ! id3demux ! xyzparse ! .., read tags here, since we don't
1529          * have a streaming thread of our own to do that. We do it here and
1530          * not in get_range(), so we can return the right size in bytes.. */
1531         if (demux->priv->state == GST_TAG_DEMUX_READ_START_TAG &&
1532             GST_PAD_MODE (demux->priv->srcpad) == GST_PAD_MODE_PULL) {
1533           GstFlowReturn flow G_GNUC_UNUSED;
1534
1535           flow = gst_tag_demux_element_find (demux);
1536           GST_INFO_OBJECT (demux, "pulled tags: %s", gst_flow_get_name (flow));
1537         }
1538         result -= demux->priv->strip_start + demux->priv->strip_end;
1539         if (result < 0)
1540           result = 0;
1541         gst_query_set_duration (query, format, result);
1542       }
1543       break;
1544     }
1545     default:
1546       res = gst_pad_query_default (pad, parent, query);
1547       break;
1548   }
1549 done:
1550   return res;
1551 }
1552
1553 static void
1554 gst_tag_demux_send_pending_events (GstTagDemux * demux)
1555 {
1556   GList *events;
1557
1558   /* send any pending events we cached */
1559   GST_OBJECT_LOCK (demux);
1560   events = demux->priv->pending_events;
1561   demux->priv->pending_events = NULL;
1562   GST_OBJECT_UNLOCK (demux);
1563
1564   while (events != NULL) {
1565     GST_DEBUG_OBJECT (demux->priv->srcpad, "sending cached %s event: %"
1566         GST_PTR_FORMAT, GST_EVENT_TYPE_NAME (events->data), events->data);
1567     gst_pad_push_event (demux->priv->srcpad, GST_EVENT (events->data));
1568     events = g_list_delete_link (events, events);
1569   }
1570 }
1571
1572 static void
1573 gst_tag_demux_send_tag_event (GstTagDemux * demux)
1574 {
1575   /* FIXME: what's the correct merge mode? Docs need to tell... */
1576   GstTagList *merged = gst_tag_list_merge (demux->priv->event_tags,
1577       demux->priv->parsed_tags, GST_TAG_MERGE_KEEP);
1578
1579   if (merged) {
1580     GstEvent *event = gst_event_new_tag (merged);
1581
1582     GST_EVENT_TIMESTAMP (event) = 0;
1583     GST_DEBUG_OBJECT (demux, "Sending tag event on src pad");
1584     gst_pad_push_event (demux->priv->srcpad, event);
1585   }
1586 }
1587
1588 static gboolean
1589 gst_tag_demux_send_new_segment (GstTagDemux * tagdemux)
1590 {
1591   GstEvent *event;
1592   gint64 start, stop, position;
1593   GstSegment *seg = &tagdemux->priv->segment;
1594   GstSegment newseg;
1595
1596   if (seg->format == GST_FORMAT_UNDEFINED) {
1597     GST_LOG_OBJECT (tagdemux,
1598         "No new segment received before first buffer. Using default");
1599     gst_segment_init (seg, GST_FORMAT_BYTES);
1600     seg->start = tagdemux->priv->strip_start;
1601     seg->time = tagdemux->priv->strip_start;
1602   }
1603
1604   /* Can't adjust segments in non-BYTES formats */
1605   if (tagdemux->priv->segment.format != GST_FORMAT_BYTES) {
1606     event = gst_event_new_segment (seg);
1607     return gst_pad_push_event (tagdemux->priv->srcpad, event);
1608   }
1609
1610   start = seg->start;
1611   stop = seg->stop;
1612   position = seg->time;
1613
1614   g_return_val_if_fail (start != -1, FALSE);
1615   g_return_val_if_fail (position != -1, FALSE);
1616
1617   if (tagdemux->priv->strip_end > 0) {
1618     if (gst_tag_demux_get_upstream_size (tagdemux)) {
1619       guint64 v1tag_offset =
1620           tagdemux->priv->upstream_size - tagdemux->priv->strip_end;
1621
1622       if (start >= v1tag_offset) {
1623         /* Segment is completely within the end tag, output an open-ended
1624          * segment, even though all the buffers will get trimmed away */
1625         start = v1tag_offset;
1626         stop = -1;
1627       }
1628
1629       if (stop != -1 && stop >= v1tag_offset) {
1630         GST_DEBUG_OBJECT (tagdemux,
1631             "Segment crosses the end tag. Trimming end");
1632         stop = v1tag_offset;
1633       }
1634     }
1635   }
1636
1637   if (tagdemux->priv->strip_start > 0) {
1638     if (start > tagdemux->priv->strip_start)
1639       start -= tagdemux->priv->strip_start;
1640     else
1641       start = 0;
1642
1643     if (position > tagdemux->priv->strip_start)
1644       position -= tagdemux->priv->strip_start;
1645     else
1646       position = 0;
1647
1648     if (stop != -1) {
1649       if (stop > tagdemux->priv->strip_start)
1650         stop -= tagdemux->priv->strip_start;
1651       else
1652         stop = 0;
1653     }
1654   }
1655
1656   GST_DEBUG_OBJECT (tagdemux, "Sending segment %" GST_SEGMENT_FORMAT, seg);
1657
1658   gst_segment_copy_into (seg, &newseg);
1659   newseg.start = start;
1660   newseg.stop = stop;
1661   newseg.position = position;
1662   event = gst_event_new_segment (&newseg);
1663
1664   return gst_pad_push_event (tagdemux->priv->srcpad, event);
1665 }