gstoggdemux.c

/* GStreamer
 * Copyright (C) 2004 Wim Taymans <wim@fluendo.com>
 *
 * gstoggdemux.c: ogg stream demuxer
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 02111-1307, USA.
 */

/**
 * SECTION:element-oggdemux
 * @see_also: <link linkend="gst-plugins-base-plugins-oggmux">oggmux</link>
 *
 * This element demuxes ogg files into their encoded audio and video components.
 *
 * <refsect2>
 * <title>Example pipelines</title>
 * |[
 * gst-launch -v filesrc location=test.ogg ! oggdemux ! vorbisdec ! audioconvert ! alsasink
 * ]| Decodes the vorbis audio stored inside an ogg container.
 * </refsect2>
 *
 * Last reviewed on 2006-12-30 (0.10.5)
 */


#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <string.h>
#include <gst/gst-i18n-plugin.h>
#include <gst/tag/tag.h>

#include "gstoggdemux.h"

#define CHUNKSIZE (8500)        /* this is out of vorbisfile */

#define GST_FLOW_LIMIT GST_FLOW_CUSTOM_ERROR

#define GST_CHAIN_LOCK(ogg)     g_mutex_lock((ogg)->chain_lock)
#define GST_CHAIN_UNLOCK(ogg)   g_mutex_unlock((ogg)->chain_lock)

GST_DEBUG_CATEGORY (gst_ogg_demux_debug);
GST_DEBUG_CATEGORY (gst_ogg_demux_setup_debug);
#define GST_CAT_DEFAULT gst_ogg_demux_debug


static ogg_packet *
_ogg_packet_copy (const ogg_packet * packet)
{
  ogg_packet *ret = g_new0 (ogg_packet, 1);

  *ret = *packet;
  ret->packet = g_memdup (packet->packet, packet->bytes);

  return ret;
}

static void
_ogg_packet_free (ogg_packet * packet)
{
  g_free (packet->packet);
  g_free (packet);
}

static ogg_page *
gst_ogg_page_copy (ogg_page * page)
{
  ogg_page *p = g_new0 (ogg_page, 1);

  /* make a copy of the page */
  p->header = g_memdup (page->header, page->header_len);
  p->header_len = page->header_len;
  p->body = g_memdup (page->body, page->body_len);
  p->body_len = page->body_len;

  return p;
}

static void
gst_ogg_page_free (ogg_page * page)
{
  g_free (page->header);
  g_free (page->body);
  g_free (page);
}

static gboolean gst_ogg_demux_collect_chain_info (GstOggDemux * ogg,
    GstOggChain * chain);
static gboolean gst_ogg_demux_activate_chain (GstOggDemux * ogg,
    GstOggChain * chain, GstEvent * event);
static void gst_ogg_chain_mark_discont (GstOggChain * chain);

static gboolean gst_ogg_demux_perform_seek (GstOggDemux * ogg,
    GstEvent * event);
static gboolean gst_ogg_demux_receive_event (GstElement * element,
    GstEvent * event);

static void gst_ogg_pad_dispose (GObject * object);
static void gst_ogg_pad_finalize (GObject * object);

static const GstQueryType *gst_ogg_pad_query_types (GstPad * pad);
static gboolean gst_ogg_pad_src_query (GstPad * pad, GstQuery * query);
static gboolean gst_ogg_pad_event (GstPad * pad, GstEvent * event);
static GstCaps *gst_ogg_pad_getcaps (GstPad * pad);
static GstOggPad *gst_ogg_chain_get_stream (GstOggChain * chain,
    glong serialno);

static GstFlowReturn gst_ogg_demux_combine_flows (GstOggDemux * ogg,
    GstOggPad * pad, GstFlowReturn ret);
static void gst_ogg_demux_sync_streams (GstOggDemux * ogg);

GstCaps *gst_ogg_demux_set_header_on_caps (GstOggDemux * ogg,
    GstCaps * caps, GList * headers);

GType gst_ogg_pad_get_type (void);
G_DEFINE_TYPE (GstOggPad, gst_ogg_pad, GST_TYPE_PAD);

static void
gst_ogg_pad_class_init (GstOggPadClass * klass)
{
  GObjectClass *gobject_class;

  gobject_class = (GObjectClass *) klass;

  gobject_class->dispose = gst_ogg_pad_dispose;
  gobject_class->finalize = gst_ogg_pad_finalize;
}

static void
gst_ogg_pad_init (GstOggPad * pad)
{
  gst_pad_set_event_function (GST_PAD (pad),
      GST_DEBUG_FUNCPTR (gst_ogg_pad_event));
  gst_pad_set_getcaps_function (GST_PAD (pad),
      GST_DEBUG_FUNCPTR (gst_ogg_pad_getcaps));
  gst_pad_set_query_type_function (GST_PAD (pad),
      GST_DEBUG_FUNCPTR (gst_ogg_pad_query_types));
  gst_pad_set_query_function (GST_PAD (pad),
      GST_DEBUG_FUNCPTR (gst_ogg_pad_src_query));

  pad->mode = GST_OGG_PAD_MODE_INIT;

  pad->current_granule = -1;
  pad->keyframe_granule = -1;

  pad->start_time = GST_CLOCK_TIME_NONE;

  pad->last_stop = GST_CLOCK_TIME_NONE;

  pad->have_type = FALSE;
  pad->continued = NULL;
  pad->map.headers = NULL;
  pad->map.queued = NULL;
}

static void
gst_ogg_pad_dispose (GObject * object)
{
  GstOggPad *pad = GST_OGG_PAD (object);

  pad->chain = NULL;
  pad->ogg = NULL;

  g_list_foreach (pad->map.headers, (GFunc) _ogg_packet_free, NULL);
  g_list_free (pad->map.headers);
  pad->map.headers = NULL;
  g_list_foreach (pad->map.queued, (GFunc) _ogg_packet_free, NULL);
  g_list_free (pad->map.queued);
  pad->map.queued = NULL;

  g_free (pad->map.index);
  pad->map.index = NULL;

  /* clear continued pages */
  g_list_foreach (pad->continued, (GFunc) gst_ogg_page_free, NULL);
  g_list_free (pad->continued);
  pad->continued = NULL;

  if (pad->map.caps) {
    gst_caps_unref (pad->map.caps);
    pad->map.caps = NULL;
  }

  if (pad->map.taglist) {
    gst_tag_list_free (pad->map.taglist);
    pad->map.taglist = NULL;
  }

  ogg_stream_reset (&pad->map.stream);

  G_OBJECT_CLASS (gst_ogg_pad_parent_class)->dispose (object);
}

static void
gst_ogg_pad_finalize (GObject * object)
{
  GstOggPad *pad = GST_OGG_PAD (object);

  ogg_stream_clear (&pad->map.stream);

  G_OBJECT_CLASS (gst_ogg_pad_parent_class)->finalize (object);
}

static const GstQueryType *
gst_ogg_pad_query_types (GstPad * pad)
{
  static const GstQueryType query_types[] = {
    GST_QUERY_DURATION,
    GST_QUERY_SEEKING,
    0
  };

  return query_types;
}

static GstCaps *
gst_ogg_pad_getcaps (GstPad * pad)
{
  return gst_caps_ref (GST_PAD_CAPS (pad));
}

static gboolean
gst_ogg_pad_src_query (GstPad * pad, GstQuery * query)
{
  gboolean res = TRUE;
  GstOggDemux *ogg;

  ogg = GST_OGG_DEMUX (gst_pad_get_parent (pad));

  switch (GST_QUERY_TYPE (query)) {
    case GST_QUERY_DURATION:
    {
      GstFormat format;
      gint64 total_time = -1;

      gst_query_parse_duration (query, &format, NULL);
      /* can only get position in time */
      if (format != GST_FORMAT_TIME)
        goto wrong_format;

      if (ogg->total_time != -1) {
        /* we can return the total length */
        total_time = ogg->total_time;
      } else {
        gint bitrate = ogg->bitrate;

        /* try with length and bitrate */
        if (bitrate > 0) {
          GstQuery *uquery;

          /* ask upstream for total length in bytes */
          uquery = gst_query_new_duration (GST_FORMAT_BYTES);
          if (gst_pad_peer_query (ogg->sinkpad, uquery)) {
            gint64 length;

            gst_query_parse_duration (uquery, NULL, &length);

            /* estimate using the bitrate */
            total_time =
                gst_util_uint64_scale (length, 8 * GST_SECOND, bitrate);

            GST_LOG_OBJECT (ogg,
                "length: %" G_GINT64_FORMAT ", bitrate %d, total_time %"
                GST_TIME_FORMAT, length, bitrate, GST_TIME_ARGS (total_time));
          }
          gst_query_unref (uquery);
        }
      }

      gst_query_set_duration (query, GST_FORMAT_TIME, total_time);
      break;
    }
    case GST_QUERY_SEEKING:
    {
      GstFormat format;

      gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
      if (format == GST_FORMAT_TIME) {
        gboolean seekable = FALSE;
        gint64 stop = -1;

        if (ogg->pullmode) {
          seekable = TRUE;
          stop = ogg->total_time;
        } else if (ogg->current_chain->streams->len) {
          gint i;

          seekable = FALSE;
          for (i = 0; i < ogg->current_chain->streams->len; i++) {
            GstOggPad *pad =
                g_array_index (ogg->current_chain->streams, GstOggPad *, i);

            seekable |= (pad->map.index != NULL && pad->map.n_index != 0);

            if (pad->map.index != NULL && pad->map.n_index != 0) {
              GstOggIndex *idx;
              GstClockTime idx_time;

              idx = &pad->map.index[pad->map.n_index - 1];
              idx_time =
                  gst_util_uint64_scale (idx->timestamp, GST_SECOND,
                  pad->map.kp_denom);
              if (stop == -1)
                stop = idx_time;
              else
                stop = MAX (idx_time, stop);
            }
          }
        }

        gst_query_set_seeking (query, GST_FORMAT_TIME, seekable, 0, stop);
      } else {
        res = FALSE;
      }
      break;
    }

    default:
      res = gst_pad_query_default (pad, query);
      break;
  }
done:
  gst_object_unref (ogg);

  return res;

  /* ERRORS */
wrong_format:
  {
    GST_DEBUG_OBJECT (ogg, "only query duration on TIME is supported");
    res = FALSE;
    goto done;
  }
}

static gboolean
gst_ogg_demux_receive_event (GstElement * element, GstEvent * event)
{
  gboolean res;
  GstOggDemux *ogg;

  ogg = GST_OGG_DEMUX (element);

  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_SEEK:
      /* now do the seek */
      res = gst_ogg_demux_perform_seek (ogg, event);
      gst_event_unref (event);
      break;
    default:
      GST_DEBUG_OBJECT (ogg, "We only handle seek events here");
      goto error;
  }

  return res;

  /* ERRORS */
error:
  {
    GST_DEBUG_OBJECT (ogg, "error handling event");
    gst_event_unref (event);
    return FALSE;
  }
}

static gboolean
gst_ogg_pad_event (GstPad * pad, GstEvent * event)
{
  gboolean res;
  GstOggDemux *ogg;

  ogg = GST_OGG_DEMUX (gst_pad_get_parent (pad));

  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_SEEK:
      /* now do the seek */
      res = gst_ogg_demux_perform_seek (ogg, event);
      gst_event_unref (event);
      break;
    default:
      res = gst_pad_event_default (pad, event);
      break;
  }
  gst_object_unref (ogg);

  return res;
}

static void
gst_ogg_pad_reset (GstOggPad * pad)
{
  ogg_stream_reset (&pad->map.stream);

  GST_DEBUG_OBJECT (pad, "doing reset");

  /* clear continued pages */
  g_list_foreach (pad->continued, (GFunc) gst_ogg_page_free, NULL);
  g_list_free (pad->continued);
  pad->continued = NULL;

  pad->last_ret = GST_FLOW_OK;
  pad->last_stop = GST_CLOCK_TIME_NONE;
  pad->current_granule = -1;
  pad->keyframe_granule = -1;
  pad->is_eos = FALSE;
}

/* queue data, basically takes the packet, puts it in a buffer and store the
 * buffer in the queued list.  */
static GstFlowReturn
gst_ogg_demux_queue_data (GstOggPad * pad, ogg_packet * packet)
{
#ifndef GST_DISABLE_GST_DEBUG
  GstOggDemux *ogg = pad->ogg;
#endif

  GST_DEBUG_OBJECT (ogg, "%p queueing data serial %08lx", pad,
      pad->map.serialno);

  pad->map.queued = g_list_append (pad->map.queued, _ogg_packet_copy (packet));

  /* we are ok now */
  return GST_FLOW_OK;
}

static GstFlowReturn
gst_ogg_demux_chain_peer (GstOggPad * pad, ogg_packet * packet,
    gboolean push_headers)
{
  GstBuffer *buf = NULL;
  GstFlowReturn ret, cret;
  GstOggDemux *ogg = pad->ogg;
  gint64 current_time;
  GstOggChain *chain;
  gint64 duration;
  gint offset;
  gint trim;
  GstClockTime out_timestamp, out_duration;
  guint64 out_offset, out_offset_end;
  gboolean delta_unit = FALSE;

  cret = GST_FLOW_OK;

  GST_DEBUG_OBJECT (ogg,
      "%p streaming to peer serial %08lx", pad, pad->map.serialno);

  if (pad->map.is_ogm) {
    const guint8 *data;
    long bytes;

    data = packet->packet;
    bytes = packet->bytes;

    if (bytes < 1)
      goto empty_packet;

    if ((data[0] & 1) || (data[0] & 3 && pad->map.is_ogm_text)) {
      /* We don't push header packets for OGM */
      goto done;
    }

    offset = 1 + (((data[0] & 0xc0) >> 6) | ((data[0] & 0x02) << 1));
    delta_unit = (((data[0] & 0x08) >> 3) == 0);

    trim = 0;

    /* Strip trailing \0 for subtitles */
    if (pad->map.is_ogm_text) {
      while (bytes && data[bytes - 1] == 0) {
        trim++;
        bytes--;
      }
    }
  } else if (pad->map.is_vp8) {
    if ((packet->bytes >= 7 && memcmp (packet->packet, "OVP80\2 ", 7) == 0) ||
        packet->b_o_s ||
        (packet->bytes >= 5 && memcmp (packet->packet, "OVP80", 5) == 0)) {
      /* We don't push header packets for VP8 */
      goto done;
    }
    offset = 0;
    trim = 0;
  } else {
    offset = 0;
    trim = 0;
  }

  /* get timing info for the packet */
  duration = gst_ogg_stream_get_packet_duration (&pad->map, packet);
  GST_DEBUG_OBJECT (ogg, "packet duration %" G_GUINT64_FORMAT, duration);

  if (packet->b_o_s) {
    out_timestamp = GST_CLOCK_TIME_NONE;
    out_duration = GST_CLOCK_TIME_NONE;
    out_offset = 0;
    out_offset_end = -1;
  } else {
    if (packet->granulepos != -1) {
      pad->current_granule = gst_ogg_stream_granulepos_to_granule (&pad->map,
          packet->granulepos);
      pad->keyframe_granule =
          gst_ogg_stream_granulepos_to_key_granule (&pad->map,
          packet->granulepos);
      GST_DEBUG_OBJECT (ogg, "new granule %" G_GUINT64_FORMAT,
          pad->current_granule);
    } else if (ogg->segment.rate > 0.0 && pad->current_granule != -1) {
      pad->current_granule += duration;
      GST_DEBUG_OBJECT (ogg, "interpollating granule %" G_GUINT64_FORMAT,
          pad->current_granule);
    }
    if (ogg->segment.rate < 0.0 && packet->granulepos == -1) {
      /* negative rates, only set timestamp on the packets with a granulepos */
      out_timestamp = -1;
      out_duration = -1;
      out_offset = -1;
      out_offset_end = -1;
    } else {
      /* we only push buffers after we have a valid granule. This is done so that
       * we nicely skip packets without a timestamp after a seek. This is ok
       * because we base or seek on the packet after the page with the smaller
       * timestamp. */
      if (pad->current_granule == -1)
        goto no_timestamp;

      if (pad->map.is_ogm) {
        out_timestamp = gst_ogg_stream_granule_to_time (&pad->map,
            pad->current_granule);
        out_duration = gst_util_uint64_scale (duration,
            GST_SECOND * pad->map.granulerate_d, pad->map.granulerate_n);
      } else if (pad->map.is_sparse) {
        out_timestamp = gst_ogg_stream_granule_to_time (&pad->map,
            pad->current_granule);
        if (duration == GST_CLOCK_TIME_NONE) {
          out_duration = GST_CLOCK_TIME_NONE;
        } else {
          out_duration = gst_util_uint64_scale (duration,
              GST_SECOND * pad->map.granulerate_d, pad->map.granulerate_n);
        }
      } else {
        out_timestamp = gst_ogg_stream_granule_to_time (&pad->map,
            pad->current_granule - duration);
        out_duration =
            gst_ogg_stream_granule_to_time (&pad->map,
            pad->current_granule) - out_timestamp;
      }
      out_offset_end =
          gst_ogg_stream_granule_to_granulepos (&pad->map,
          pad->current_granule, pad->keyframe_granule);
      out_offset =
          gst_ogg_stream_granule_to_time (&pad->map, pad->current_granule);
    }
  }

  if (pad->map.is_ogm_text) {
    /* check for invalid buffer sizes */
    if (G_UNLIKELY (offset + trim >= packet->bytes))
      goto empty_packet;
  }

  if (!pad->added)
    goto not_added;

  buf = gst_buffer_new_and_alloc (packet->bytes - offset - trim);
  gst_buffer_set_caps (buf, GST_PAD_CAPS (pad));

  /* set delta flag for OGM content */
  if (delta_unit)
    GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT);

  /* copy packet in buffer */
  memcpy (buf->data, packet->packet + offset, packet->bytes - offset - trim);

  GST_BUFFER_TIMESTAMP (buf) = out_timestamp;
  GST_BUFFER_DURATION (buf) = out_duration;
  GST_BUFFER_OFFSET (buf) = out_offset;
  GST_BUFFER_OFFSET_END (buf) = out_offset_end;

  /* Mark discont on the buffer */
  if (pad->discont) {
    GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT);
    pad->discont = FALSE;
  }

  pad->last_stop = ogg->segment.last_stop;

  /* don't push the header packets when we are asked to skip them */
  if (!packet->b_o_s || push_headers) {
    ret = gst_pad_push (GST_PAD_CAST (pad), buf);
    buf = NULL;

    /* combine flows */
    cret = gst_ogg_demux_combine_flows (ogg, pad, ret);
  }

  /* we're done with skeleton stuff */
  if (pad->map.is_skeleton)
    goto done;

  /* check if valid granulepos, then we can calculate the current
   * position. We know the granule for each packet but we only want to update
   * the last_stop when we have a valid granulepos on the packet because else
   * our time jumps around for the different streams. */
  if (packet->granulepos < 0)
    goto done;

  /* convert to time */
  current_time = gst_ogg_stream_get_end_time_for_granulepos (&pad->map,
      packet->granulepos);

  /* convert to stream time */
  if ((chain = pad->chain)) {
    gint64 chain_start = 0;

    if (chain->segment_start != GST_CLOCK_TIME_NONE)
      chain_start = chain->segment_start;

    current_time = current_time - chain_start + chain->begin_time;
  }

  /* and store as the current position */
  gst_segment_set_last_stop (&ogg->segment, GST_FORMAT_TIME, current_time);

  GST_DEBUG_OBJECT (ogg, "ogg current time %" GST_TIME_FORMAT,
      GST_TIME_ARGS (current_time));

  /* check stream eos */
  if ((ogg->segment.rate > 0.0 && ogg->segment.stop != GST_CLOCK_TIME_NONE &&
          current_time > ogg->segment.stop) ||
      (ogg->segment.rate < 0.0 && ogg->segment.start != GST_CLOCK_TIME_NONE &&
          current_time < ogg->segment.start)) {
    GST_DEBUG_OBJECT (ogg, "marking pad %p EOS", pad);
    pad->is_eos = TRUE;
  }

done:
  if (buf)
    gst_buffer_unref (buf);
  /* return combined flow result */
  return cret;

  /* special cases */
empty_packet:
  {
    GST_DEBUG_OBJECT (ogg, "Skipping empty packet");
    goto done;
  }

no_timestamp:
  {
    GST_DEBUG_OBJECT (ogg, "skipping packet: no valid granule found yet");
    goto done;
  }
not_added:
  {
    GST_DEBUG_OBJECT (ogg, "pad not added yet");
    goto done;
  }
}

static guint64
gst_ogg_demux_collect_start_time (GstOggDemux * ogg, GstOggChain * chain)
{
  gint i;
  guint64 start_time = G_MAXUINT64;

  for (i = 0; i < chain->streams->len; i++) {
    GstOggPad *pad = g_array_index (chain->streams, GstOggPad *, i);

    if (pad->map.is_sparse)
      continue;

    /*  can do this if the pad start time is not defined */
    if (pad->start_time == GST_CLOCK_TIME_NONE) {
      start_time = G_MAXUINT64;
      break;
    } else {
      start_time = MIN (start_time, pad->start_time);
    }
  }
  return start_time;
}

/* submit a packet to the oggpad, this function will run the
 * typefind code for the pad if this is the first packet for this
 * stream 
 */
static GstFlowReturn
gst_ogg_pad_submit_packet (GstOggPad * pad, ogg_packet * packet)
{
  gint64 granule;
  GstFlowReturn ret = GST_FLOW_OK;

  GstOggDemux *ogg = pad->ogg;

  GST_DEBUG_OBJECT (ogg, "%p submit packet serial %08lx", pad,
      pad->map.serialno);

  if (!pad->have_type) {
    pad->have_type = gst_ogg_stream_setup_map (&pad->map, packet);
    if (!pad->have_type) {
      pad->map.caps = gst_caps_new_simple ("application/x-unknown", NULL);
    }
    if (pad->map.is_skeleton) {
      GST_DEBUG_OBJECT (ogg, "we have a fishead");
      /* copy values over to global ogg level */
      ogg->basetime = pad->map.basetime;
      ogg->prestime = pad->map.prestime;

      /* use total time to update the total ogg time */
      if (ogg->total_time == -1) {
        ogg->total_time = pad->map.total_time;
      } else if (pad->map.total_time > 0) {
        ogg->total_time = MAX (ogg->total_time, pad->map.total_time);
      }
    }
    if (pad->map.caps) {
      gst_pad_set_caps (GST_PAD (pad), pad->map.caps);
    } else {
      GST_WARNING_OBJECT (ogg, "stream parser didn't create src pad caps");
    }
  }

  if (pad->map.is_skeleton) {
    guint32 serialno;
    GstOggPad *skel_pad;
    GstOggSkeleton type;

    /* try to parse the serialno first */
    if (gst_ogg_map_parse_fisbone (&pad->map, packet->packet, packet->bytes,
            &serialno, &type)) {

      GST_WARNING_OBJECT (pad->ogg,
          "got skeleton packet for stream 0x%08x", serialno);

      skel_pad = gst_ogg_chain_get_stream (pad->chain, serialno);
      if (skel_pad) {
        switch (type) {
          case GST_OGG_SKELETON_FISBONE:
            /* parse the remainder of the fisbone in the pad with the serialno,
             * note that we ignore the start_time as this is usually wrong for
             * live streams */
            gst_ogg_map_add_fisbone (&skel_pad->map, &pad->map, packet->packet,
                packet->bytes, NULL);
            break;
          case GST_OGG_SKELETON_INDEX:
            gst_ogg_map_add_index (&skel_pad->map, &pad->map, packet->packet,
                packet->bytes);

            /* use total time to update the total ogg time */
            if (ogg->total_time == -1) {
              ogg->total_time = skel_pad->map.total_time;
            } else if (skel_pad->map.total_time > 0) {
              ogg->total_time = MAX (ogg->total_time, skel_pad->map.total_time);
            }
            break;
          default:
            break;
        }

      } else {
        GST_WARNING_OBJECT (pad->ogg,
            "found skeleton fisbone for an unknown stream 0x%08x", serialno);
      }
    }
  }

  granule = gst_ogg_stream_granulepos_to_granule (&pad->map,
      packet->granulepos);
  if (granule != -1) {
    GST_DEBUG_OBJECT (ogg, "%p has granulepos %" G_GINT64_FORMAT, pad, granule);
    pad->current_granule = granule;
  }

  /* restart header packet count when seeing a b_o_s page;
   * particularly useful following a seek or even following chain finding */
  if (packet->b_o_s) {
    GST_DEBUG_OBJECT (ogg, "b_o_s packet, resetting header packet count");
    pad->map.n_header_packets_seen = 0;
    if (!pad->map.have_headers) {
      GST_DEBUG_OBJECT (ogg, "clearing header packets");
      g_list_foreach (pad->map.headers, (GFunc) _ogg_packet_free, NULL);
      g_list_free (pad->map.headers);
      pad->map.headers = NULL;
    }
  }

  /* Overload the value of b_o_s in ogg_packet with a flag whether or
   * not this is a header packet.  Maybe some day this could be cleaned
   * up.  */
  packet->b_o_s = gst_ogg_stream_packet_is_header (&pad->map, packet);
  if (!packet->b_o_s) {
    GST_DEBUG ("found non-header packet");
    pad->map.have_headers = TRUE;
    if (pad->start_time == GST_CLOCK_TIME_NONE) {
      gint64 duration = gst_ogg_stream_get_packet_duration (&pad->map, packet);
      GST_DEBUG ("duration %" G_GINT64_FORMAT, duration);
      if (duration != -1) {
        pad->map.accumulated_granule += duration;
        GST_DEBUG ("accumulated granule %" G_GINT64_FORMAT,
            pad->map.accumulated_granule);
      }

      if (packet->granulepos != -1) {
        ogg_int64_t start_granule;
        gint64 granule;

        granule = gst_ogg_stream_granulepos_to_granule (&pad->map,
            packet->granulepos);

        if (granule > pad->map.accumulated_granule)
          start_granule = granule - pad->map.accumulated_granule;
        else
          start_granule = 0;

        pad->start_time = gst_ogg_stream_granule_to_time (&pad->map,
            start_granule);
        GST_DEBUG ("start time %" G_GINT64_FORMAT, pad->start_time);
      } else {
        packet->granulepos = gst_ogg_stream_granule_to_granulepos (&pad->map,
            pad->map.accumulated_granule, pad->keyframe_granule);
      }
    }
  } else {
    /* look for tags in header packet (before inc header count) */
    gst_ogg_stream_extract_tags (&pad->map, packet);
    pad->map.n_header_packets_seen++;
    if (!pad->map.have_headers) {
      pad->map.headers =
          g_list_append (pad->map.headers, _ogg_packet_copy (packet));
      GST_DEBUG ("keeping header packet %d", pad->map.n_header_packets_seen);
    }
  }

  /* we know the start_time of the pad data, see if we
   * can activate the complete chain if this is a dynamic
   * chain. We need all the headers too for this. */
  if (pad->start_time != GST_CLOCK_TIME_NONE && pad->map.have_headers) {
    GstOggChain *chain = pad->chain;

    /* check if complete chain has start time */
    if (chain == ogg->building_chain) {
      GstEvent *event = NULL;

      if (ogg->resync) {
        guint64 start_time;

        GST_DEBUG_OBJECT (ogg, "need to resync");

        /* when we need to resync after a seek, we wait until we have received
         * timestamps on all streams */
        start_time = gst_ogg_demux_collect_start_time (ogg, chain);

        if (start_time != G_MAXUINT64) {
          gint64 segment_time;

          GST_DEBUG_OBJECT (ogg, "start_time:  %" GST_TIME_FORMAT,
              GST_TIME_ARGS (start_time));

          if (chain->segment_start < start_time)
            segment_time =
                (start_time - chain->segment_start) + chain->begin_time;
          else
            segment_time = chain->begin_time;

          /* create the newsegment event we are going to send out */
          event = gst_event_new_new_segment (FALSE, ogg->segment.rate,
              GST_FORMAT_TIME, start_time, chain->segment_stop, segment_time);

          ogg->resync = FALSE;
        }
      } else {
        /* see if we have enough info to activate the chain, we have enough info
         * when all streams have a valid start time. */
        if (gst_ogg_demux_collect_chain_info (ogg, chain)) {

          GST_DEBUG_OBJECT (ogg, "segment_start: %" GST_TIME_FORMAT,
              GST_TIME_ARGS (chain->segment_start));
          GST_DEBUG_OBJECT (ogg, "segment_stop:  %" GST_TIME_FORMAT,
              GST_TIME_ARGS (chain->segment_stop));
          GST_DEBUG_OBJECT (ogg, "segment_time:  %" GST_TIME_FORMAT,
              GST_TIME_ARGS (chain->begin_time));

          /* create the newsegment event we are going to send out */
          event = gst_event_new_new_segment (FALSE, ogg->segment.rate,
              GST_FORMAT_TIME, chain->segment_start, chain->segment_stop,
              chain->begin_time);
        }
      }

      if (event) {
        gst_event_set_seqnum (event, ogg->seqnum);

        gst_ogg_demux_activate_chain (ogg, chain, event);

        ogg->building_chain = NULL;
      }
    }
  }

  /* if we are building a chain, store buffer for when we activate
   * it. This path is taken if we operate in streaming mode. */
  if (ogg->building_chain) {
    /* bos packets where stored in the header list so we can discard
     * them here*/
    if (!packet->b_o_s)
      ret = gst_ogg_demux_queue_data (pad, packet);
  }
  /* else we are completely streaming to the peer */
  else {
    ret = gst_ogg_demux_chain_peer (pad, packet, !ogg->pullmode);
  }
  return ret;
}

/* flush at most @npackets from the stream layer. All packets if 
 * @npackets is 0;
 */
static GstFlowReturn
gst_ogg_pad_stream_out (GstOggPad * pad, gint npackets)
{
  GstFlowReturn result = GST_FLOW_OK;
  gboolean done = FALSE;
  GstOggDemux *ogg;

  ogg = pad->ogg;

  while (!done) {
    int ret;
    ogg_packet packet;

    ret = ogg_stream_packetout (&pad->map.stream, &packet);
    switch (ret) {
      case 0:
        GST_LOG_OBJECT (ogg, "packetout done");
        done = TRUE;
        break;
      case -1:
        GST_LOG_OBJECT (ogg, "packetout discont");
        gst_ogg_chain_mark_discont (pad->chain);
        break;
      case 1:
        GST_LOG_OBJECT (ogg, "packetout gave packet of size %ld", packet.bytes);
        result = gst_ogg_pad_submit_packet (pad, &packet);
        /* not linked is not a problem, it's possible that we are still
         * collecting headers and that we don't have exposed the pads yet */
        if (result == GST_FLOW_NOT_LINKED)
          break;
        else if (result <= GST_FLOW_UNEXPECTED)
          goto could_not_submit;
        break;
      default:
        GST_WARNING_OBJECT (ogg,
            "invalid return value %d for ogg_stream_packetout, resetting stream",
            ret);
        gst_ogg_pad_reset (pad);
        break;
    }
    if (npackets > 0) {
      npackets--;
      done = (npackets == 0);
    }
  }
  return result;

  /* ERRORS */
could_not_submit:
  {
    GST_WARNING_OBJECT (ogg,
        "could not submit packet for stream %08lx, error: %d",
        pad->map.serialno, result);
    gst_ogg_pad_reset (pad);
    return result;
  }
}

/* submit a page to an oggpad, this function will then submit all
 * the packets in the page.
 */
static GstFlowReturn
gst_ogg_pad_submit_page (GstOggPad * pad, ogg_page * page)
{
  GstFlowReturn result = GST_FLOW_OK;
  GstOggDemux *ogg;
  gboolean continued = FALSE;

  ogg = pad->ogg;