ext/ogg/gstoggmux.c

   1 /* OGG muxer plugin for GStreamer
   2  * Copyright (C) 2004 Wim Taymans <wim@fluendo.com>
   3  * Copyright (C) 2006 Thomas Vander Stichele <thomas at apestaart dot org>
   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:element-oggmux
  23  * @see_also: <link linkend="gst-plugins-base-plugins-oggdemux">oggdemux</link>
  24  *
  25  * This element merges streams (audio and video) into ogg files.
  26  *
  27  * <refsect2>
  28  * <title>Example pipelines</title>
  29  * |[
  30  * gst-launch v4l2src num-buffers=500 ! video/x-raw,width=320,height=240 ! videoconvert ! theoraenc ! oggmux ! filesink location=video.ogg
  31  * ]| Encodes a video stream captured from a v4l2-compatible camera to Ogg/Theora
  32  * (the encoding will stop automatically after 500 frames)
  33  * </refsect2>
  34  *
  35  * Last reviewed on 2008-02-06 (0.10.17)
  36  */
  37
  38 #ifdef HAVE_CONFIG_H
  39 #include "config.h"
  40 #endif
  41
  42 #include <gst/gst.h>
  43 #include <gst/base/gstbytewriter.h>
  44 #include <gst/tag/tag.h>
  45
  46 #include "gstoggmux.h"
  47
  48 /* memcpy - if someone knows a way to get rid of it, please speak up
  49  * note: the ogg docs even say you need this... */
  50 #include <string.h>
  51 #include <time.h>
  52 #include <stdlib.h>             /* rand, srand, atoi */
  53
  54 GST_DEBUG_CATEGORY_STATIC (gst_ogg_mux_debug);
  55 #define GST_CAT_DEFAULT gst_ogg_mux_debug
  56
  57 /* This isn't generally what you'd want with an end-time macro, because
  58    technically the end time of a buffer with invalid duration is invalid. But
  59    for sorting ogg pages this is what we want. */
  60 #define GST_BUFFER_END_TIME(buf) \
  61     (GST_BUFFER_DURATION_IS_VALID (buf) \
  62     ? GST_BUFFER_TIMESTAMP (buf) + GST_BUFFER_DURATION (buf) \
  63     : GST_BUFFER_TIMESTAMP (buf))
  64
  65 #define GST_GP_FORMAT "[gp %8" G_GINT64_FORMAT "]"
  66 #define GST_GP_CAST(_gp) ((gint64) _gp)
  67
  68 typedef enum
  69 {
  70   GST_OGG_FLAG_BOS = GST_ELEMENT_FLAG_LAST,
  71   GST_OGG_FLAG_EOS
  72 }
  73 GstOggFlag;
  74
  75 /* OggMux signals and args */
  76 enum
  77 {
  78   /* FILL ME */
  79   LAST_SIGNAL
  80 };
  81
  82 /* set to 0.5 seconds by default */
  83 #define DEFAULT_MAX_DELAY       G_GINT64_CONSTANT(500000000)
  84 #define DEFAULT_MAX_PAGE_DELAY  G_GINT64_CONSTANT(500000000)
  85 #define DEFAULT_MAX_TOLERANCE   G_GINT64_CONSTANT(40000000)
  86 #define DEFAULT_SKELETON        FALSE
  87
  88 enum
  89 {
  90   ARG_0,
  91   ARG_MAX_DELAY,
  92   ARG_MAX_PAGE_DELAY,
  93   ARG_MAX_TOLERANCE,
  94   ARG_SKELETON
  95 };
  96
  97 static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",
  98     GST_PAD_SRC,
  99     GST_PAD_ALWAYS,
 100     GST_STATIC_CAPS ("application/ogg")
 101     );
 102
 103 static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink_%u",
 104     GST_PAD_SINK,
 105     GST_PAD_REQUEST,
 106     GST_STATIC_CAPS ("video/x-theora; "
 107         "audio/x-vorbis; audio/x-flac; audio/x-speex; audio/x-celt; "
 108         "application/x-ogm-video; application/x-ogm-audio; video/x-dirac; "
 109         "video/x-smoke; video/x-vp8; text/x-cmml, encoded = (boolean) TRUE; "
 110         "subtitle/x-kate; application/x-kate")
 111     );
 112
 113 static void gst_ogg_mux_finalize (GObject * object);
 114
 115 static GstFlowReturn
 116 gst_ogg_mux_collected (GstCollectPads2 * pads, GstOggMux * ogg_mux);
 117 static gboolean gst_ogg_mux_handle_src_event (GstPad * pad, GstEvent * event);
 118 static GstPad *gst_ogg_mux_request_new_pad (GstElement * element,
 119     GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
 120 static void gst_ogg_mux_release_pad (GstElement * element, GstPad * pad);
 121
 122 static void gst_ogg_mux_set_property (GObject * object,
 123     guint prop_id, const GValue * value, GParamSpec * pspec);
 124 static void gst_ogg_mux_get_property (GObject * object,
 125     guint prop_id, GValue * value, GParamSpec * pspec);
 126 static GstStateChangeReturn gst_ogg_mux_change_state (GstElement * element,
 127     GstStateChange transition);
 128
 129 /*static guint gst_ogg_mux_signals[LAST_SIGNAL] = { 0 }; */
 130 #define gst_ogg_mux_parent_class parent_class
 131 G_DEFINE_TYPE_WITH_CODE (GstOggMux, gst_ogg_mux, GST_TYPE_ELEMENT,
 132     G_IMPLEMENT_INTERFACE (GST_TYPE_PRESET, NULL));
 133
 134 static void
 135 gst_ogg_mux_class_init (GstOggMuxClass * klass)
 136 {
 137   GObjectClass *gobject_class;
 138   GstElementClass *gstelement_class;
 139
 140   gobject_class = (GObjectClass *) klass;
 141   gstelement_class = (GstElementClass *) klass;
 142
 143   gobject_class->finalize = gst_ogg_mux_finalize;
 144   gobject_class->get_property = gst_ogg_mux_get_property;
 145   gobject_class->set_property = gst_ogg_mux_set_property;
 146
 147   gst_element_class_add_pad_template (gstelement_class,
 148       gst_static_pad_template_get (&src_factory));
 149   gst_element_class_add_pad_template (gstelement_class,
 150       gst_static_pad_template_get (&sink_factory));
 151
 152   gst_element_class_set_details_simple (gstelement_class,
 153       "Ogg muxer", "Codec/Muxer",
 154       "mux ogg streams (info about ogg: http://xiph.org)",
 155       "Wim Taymans <wim@fluendo.com>");
 156
 157   gstelement_class->request_new_pad = gst_ogg_mux_request_new_pad;
 158   gstelement_class->release_pad = gst_ogg_mux_release_pad;
 159
 160   g_object_class_install_property (gobject_class, ARG_MAX_DELAY,
 161       g_param_spec_uint64 ("max-delay", "Max delay",
 162           "Maximum delay in multiplexing streams", 0, G_MAXUINT64,
 163           DEFAULT_MAX_DELAY,
 164           (GParamFlags) G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 165   g_object_class_install_property (gobject_class, ARG_MAX_PAGE_DELAY,
 166       g_param_spec_uint64 ("max-page-delay", "Max page delay",
 167           "Maximum delay for sending out a page", 0, G_MAXUINT64,
 168           DEFAULT_MAX_PAGE_DELAY,
 169           (GParamFlags) G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 170   g_object_class_install_property (gobject_class, ARG_MAX_TOLERANCE,
 171       g_param_spec_uint64 ("max-tolerance", "Max time tolerance",
 172           "Maximum timestamp difference for maintaining perfect granules",
 173           0, G_MAXUINT64, DEFAULT_MAX_TOLERANCE,
 174           (GParamFlags) G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 175   g_object_class_install_property (gobject_class, ARG_SKELETON,
 176       g_param_spec_boolean ("skeleton", "Skeleton",
 177           "Whether to include a Skeleton track",
 178           DEFAULT_SKELETON,
 179           (GParamFlags) G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 180
 181   gstelement_class->change_state = gst_ogg_mux_change_state;
 182
 183 }
 184
 185 #if 0
 186 static const GstEventMask *
 187 gst_ogg_mux_get_sink_event_masks (GstPad * pad)
 188 {
 189   static const GstEventMask gst_ogg_mux_sink_event_masks[] = {
 190     {GST_EVENT_EOS, 0},
 191     {GST_EVENT_DISCONTINUOUS, 0},
 192     {0,}
 193   };
 194
 195   return gst_ogg_mux_sink_event_masks;
 196 }
 197 #endif
 198
 199 static void
 200 gst_ogg_mux_clear (GstOggMux * ogg_mux)
 201 {
 202   ogg_mux->pulling = NULL;
 203   ogg_mux->need_headers = TRUE;
 204   ogg_mux->delta_pad = NULL;
 205   ogg_mux->offset = 0;
 206   ogg_mux->next_ts = 0;
 207   ogg_mux->last_ts = GST_CLOCK_TIME_NONE;
 208 }
 209
 210 static void
 211 gst_ogg_mux_init (GstOggMux * ogg_mux)
 212 {
 213   GstElementClass *klass = GST_ELEMENT_GET_CLASS (ogg_mux);
 214
 215   ogg_mux->srcpad =
 216       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
 217           "src"), "src");
 218   gst_pad_set_event_function (ogg_mux->srcpad, gst_ogg_mux_handle_src_event);
 219   gst_element_add_pad (GST_ELEMENT (ogg_mux), ogg_mux->srcpad);
 220
 221   GST_OBJECT_FLAG_SET (GST_ELEMENT (ogg_mux), GST_OGG_FLAG_BOS);
 222
 223   /* seed random number generator for creation of serial numbers */
 224   srand (time (NULL));
 225
 226   ogg_mux->collect = gst_collect_pads2_new ();
 227   gst_collect_pads2_set_function (ogg_mux->collect,
 228       (GstCollectPads2Function) GST_DEBUG_FUNCPTR (gst_ogg_mux_collected),
 229       ogg_mux);
 230
 231   ogg_mux->max_delay = DEFAULT_MAX_DELAY;
 232   ogg_mux->max_page_delay = DEFAULT_MAX_PAGE_DELAY;
 233   ogg_mux->max_tolerance = DEFAULT_MAX_TOLERANCE;
 234
 235   gst_ogg_mux_clear (ogg_mux);
 236 }
 237
 238 static void
 239 gst_ogg_mux_finalize (GObject * object)
 240 {
 241   GstOggMux *ogg_mux;
 242
 243   ogg_mux = GST_OGG_MUX (object);
 244
 245   if (ogg_mux->collect) {
 246     gst_object_unref (ogg_mux->collect);
 247     ogg_mux->collect = NULL;
 248   }
 249
 250   G_OBJECT_CLASS (parent_class)->finalize (object);
 251 }
 252
 253 static void
 254 gst_ogg_mux_ogg_pad_destroy_notify (GstCollectData2 * data)
 255 {
 256   GstOggPadData *oggpad = (GstOggPadData *) data;
 257   GstBuffer *buf;
 258
 259   ogg_stream_clear (&oggpad->map.stream);
 260   gst_caps_replace (&oggpad->map.caps, NULL);
 261
 262   if (oggpad->pagebuffers) {
 263     while ((buf = g_queue_pop_head (oggpad->pagebuffers)) != NULL) {
 264       gst_buffer_unref (buf);
 265     }
 266     g_queue_free (oggpad->pagebuffers);
 267     oggpad->pagebuffers = NULL;
 268   }
 269 }
 270
 271 static GstPadLinkReturn
 272 gst_ogg_mux_sinkconnect (GstPad * pad, GstPad * peer)
 273 {
 274   GstOggMux *ogg_mux;
 275
 276   ogg_mux = GST_OGG_MUX (gst_pad_get_parent (pad));
 277
 278   GST_DEBUG_OBJECT (ogg_mux, "sinkconnect triggered on %s", GST_PAD_NAME (pad));
 279
 280   gst_object_unref (ogg_mux);
 281
 282   return GST_PAD_LINK_OK;
 283 }
 284
 285 static gboolean
 286 gst_ogg_mux_sink_event (GstPad * pad, GstEvent * event)
 287 {
 288   GstOggMux *ogg_mux = GST_OGG_MUX (gst_pad_get_parent (pad));
 289   GstOggPadData *ogg_pad = (GstOggPadData *) gst_pad_get_element_private (pad);
 290   gboolean ret = FALSE;
 291
 292   GST_DEBUG_OBJECT (pad, "Got %s event", GST_EVENT_TYPE_NAME (event));
 293
 294   switch (GST_EVENT_TYPE (event)) {
 295     case GST_EVENT_SEGMENT:
 296     {
 297       const GstSegment *segment;
 298
 299       gst_event_parse_segment (event, &segment);
 300
 301       /* We don't support non time NEWSEGMENT events */
 302       if (segment->format != GST_FORMAT_TIME) {
 303         gst_event_unref (event);
 304         event = NULL;
 305         break;
 306       }
 307
 308       gst_segment_copy_into (segment, &ogg_pad->segment);
 309       break;
 310     }
 311     case GST_EVENT_FLUSH_STOP:{
 312       gst_segment_init (&ogg_pad->segment, GST_FORMAT_TIME);
 313       break;
 314     }
 315     case GST_EVENT_TAG:{
 316       GstTagList *tags;
 317
 318       gst_event_parse_tag (event, &tags);
 319       tags = gst_tag_list_merge (ogg_pad->tags, tags, GST_TAG_MERGE_APPEND);
 320       if (ogg_pad->tags)
 321         gst_tag_list_free (ogg_pad->tags);
 322       ogg_pad->tags = tags;
 323
 324       GST_DEBUG_OBJECT (ogg_mux, "Got tags %" GST_PTR_FORMAT, ogg_pad->tags);
 325       break;
 326     }
 327     default:
 328       break;
 329   }
 330
 331   /* now GstCollectPads can take care of the rest, e.g. EOS */
 332   if (event != NULL)
 333     ret = ogg_pad->collect_event (pad, event);
 334
 335   gst_object_unref (ogg_mux);
 336   return ret;
 337 }
 338
 339 static gboolean
 340 gst_ogg_mux_is_serialno_present (GstOggMux * ogg_mux, guint32 serialno)
 341 {
 342   GSList *walk;
 343
 344   walk = ogg_mux->collect->data;
 345   while (walk) {
 346     GstOggPadData *pad = (GstOggPadData *) walk->data;
 347     if (pad->map.serialno == serialno)
 348       return TRUE;
 349     walk = walk->next;
 350   }
 351
 352   return FALSE;
 353 }
 354
 355 static guint32
 356 gst_ogg_mux_generate_serialno (GstOggMux * ogg_mux)
 357 {
 358   guint32 serialno;
 359
 360   do {
 361     serialno = g_random_int_range (0, G_MAXINT32);
 362   } while (gst_ogg_mux_is_serialno_present (ogg_mux, serialno));
 363
 364   return serialno;
 365 }
 366
 367 static GstPad *
 368 gst_ogg_mux_request_new_pad (GstElement * element,
 369     GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
 370 {
 371   GstOggMux *ogg_mux;
 372   GstPad *newpad;
 373   GstElementClass *klass;
 374
 375   g_return_val_if_fail (templ != NULL, NULL);
 376
 377   if (templ->direction != GST_PAD_SINK)
 378     goto wrong_direction;
 379
 380   g_return_val_if_fail (GST_IS_OGG_MUX (element), NULL);
 381   ogg_mux = GST_OGG_MUX (element);
 382
 383   klass = GST_ELEMENT_GET_CLASS (element);
 384
 385   if (templ != gst_element_class_get_pad_template (klass, "sink_%u"))
 386     goto wrong_template;
 387
 388   {
 389     guint32 serial;
 390     gchar *name;
 391
 392     if (req_name == NULL || strlen (req_name) < 6) {
 393       /* no name given when requesting the pad, use random serial number */
 394       serial = gst_ogg_mux_generate_serialno (ogg_mux);
 395     } else {
 396       /* parse serial number from requested padname */
 397       unsigned long long_serial;
 398       char *endptr = NULL;
 399       long_serial = strtoul (&req_name[5], &endptr, 10);
 400       if ((endptr && *endptr) || (long_serial & ~0xffffffff)) {
 401         GST_WARNING_OBJECT (ogg_mux, "Invalid serial number specification: %s",
 402             req_name + 5);
 403         return NULL;
 404       }
 405       serial = (guint32) long_serial;
 406     }
 407     /* create new pad with the name */
 408     GST_DEBUG_OBJECT (ogg_mux, "Creating new pad for serial %d", serial);
 409     name = g_strdup_printf ("sink_%u", serial);
 410     newpad = gst_pad_new_from_template (templ, name);
 411     g_free (name);
 412
 413     /* construct our own wrapper data structure for the pad to
 414      * keep track of its status */
 415     {
 416       GstOggPadData *oggpad;
 417
 418       oggpad = (GstOggPadData *)
 419           gst_collect_pads2_add_pad_full (ogg_mux->collect, newpad,
 420           sizeof (GstOggPadData), gst_ogg_mux_ogg_pad_destroy_notify, TRUE);
 421       ogg_mux->active_pads++;
 422
 423       oggpad->map.serialno = serial;
 424       ogg_stream_init (&oggpad->map.stream, oggpad->map.serialno);
 425       oggpad->packetno = 0;
 426       oggpad->pageno = 0;
 427       oggpad->eos = FALSE;
 428       /* we assume there will be some control data first for this pad */
 429       oggpad->state = GST_OGG_PAD_STATE_CONTROL;
 430       oggpad->new_page = TRUE;
 431       oggpad->first_delta = FALSE;
 432       oggpad->prev_delta = FALSE;
 433       oggpad->data_pushed = FALSE;
 434       oggpad->pagebuffers = g_queue_new ();
 435       oggpad->map.headers = NULL;
 436       oggpad->map.queued = NULL;
 437       oggpad->next_granule = 0;
 438       oggpad->keyframe_granule = -1;
 439
 440       gst_segment_init (&oggpad->segment, GST_FORMAT_TIME);
 441
 442       oggpad->collect_event = (GstPadEventFunction) GST_PAD_EVENTFUNC (newpad);
 443       gst_pad_set_event_function (newpad,
 444           GST_DEBUG_FUNCPTR (gst_ogg_mux_sink_event));
 445     }
 446   }
 447
 448   /* setup some pad functions */
 449   gst_pad_set_link_function (newpad, gst_ogg_mux_sinkconnect);
 450
 451   /* dd the pad to the element */
 452   gst_element_add_pad (element, newpad);
 453
 454   return newpad;
 455
 456   /* ERRORS */
 457 wrong_direction:
 458   {
 459     g_warning ("ogg_mux: request pad that is not a SINK pad\n");
 460     return NULL;
 461   }
 462 wrong_template:
 463   {
 464     g_warning ("ogg_mux: this is not our template!\n");
 465     return NULL;
 466   }
 467 }
 468
 469 static void
 470 gst_ogg_mux_release_pad (GstElement * element, GstPad * pad)
 471 {
 472   GstOggMux *ogg_mux;
 473
 474   ogg_mux = GST_OGG_MUX (gst_pad_get_parent (pad));
 475
 476   gst_collect_pads2_remove_pad (ogg_mux->collect, pad);
 477   gst_element_remove_pad (element, pad);
 478
 479   gst_object_unref (ogg_mux);
 480 }
 481
 482 /* handle events */
 483 static gboolean
 484 gst_ogg_mux_handle_src_event (GstPad * pad, GstEvent * event)
 485 {
 486   GstEventType type;
 487
 488   type = event ? GST_EVENT_TYPE (event) : GST_EVENT_UNKNOWN;
 489
 490   switch (type) {
 491     case GST_EVENT_SEEK:
 492       /* disable seeking for now */
 493       return FALSE;
 494     default:
 495       break;
 496   }
 497
 498   return gst_pad_event_default (pad, event);
 499 }
 500
 501 static GstBuffer *
 502 gst_ogg_mux_buffer_from_page (GstOggMux * mux, ogg_page * page, gboolean delta)
 503 {
 504   GstBuffer *buffer;
 505
 506   /* allocate space for header and body */
 507   buffer = gst_buffer_new_and_alloc (page->header_len + page->body_len);
 508   gst_buffer_fill (buffer, 0, page->header, page->header_len);
 509   gst_buffer_fill (buffer, page->header_len, page->body, page->body_len);
 510
 511   /* Here we set granulepos as our OFFSET_END to give easy direct access to
 512    * this value later. Before we push it, we reset this to OFFSET + SIZE
 513    * (see gst_ogg_mux_push_buffer). */
 514   GST_BUFFER_OFFSET_END (buffer) = ogg_page_granulepos (page);
 515   if (delta)
 516     GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
 517
 518   GST_LOG_OBJECT (mux, GST_GP_FORMAT
 519       " created buffer %p from ogg page",
 520       GST_GP_CAST (ogg_page_granulepos (page)), buffer);
 521
 522   return buffer;
 523 }
 524
 525 static GstFlowReturn
 526 gst_ogg_mux_push_buffer (GstOggMux * mux, GstBuffer * buffer,
 527     GstOggPadData * oggpad)
 528 {
 529   /* fix up OFFSET and OFFSET_END again */
 530   GST_BUFFER_OFFSET (buffer) = mux->offset;
 531   mux->offset += gst_buffer_get_size (buffer);
 532   GST_BUFFER_OFFSET_END (buffer) = mux->offset;
 533
 534   /* Ensure we have monotonically increasing timestamps in the output. */
 535   if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer)) {
 536     gint64 run_time = GST_BUFFER_TIMESTAMP (buffer);
 537     if (mux->last_ts != GST_CLOCK_TIME_NONE && run_time < mux->last_ts)
 538       GST_BUFFER_TIMESTAMP (buffer) = mux->last_ts;
 539     else
 540       mux->last_ts = run_time;
 541   }
 542
 543   return gst_pad_push (mux->srcpad, buffer);
 544 }
 545
 546 /* if all queues have at least one page, dequeue the page with the lowest
 547  * timestamp */
 548 static gboolean
 549 gst_ogg_mux_dequeue_page (GstOggMux * mux, GstFlowReturn * flowret)
 550 {
 551   GSList *walk;
 552   GstOggPadData *opad = NULL;   /* "oldest" pad */
 553   GstClockTime oldest = GST_CLOCK_TIME_NONE;
 554   GstBuffer *buf = NULL;
 555   gboolean ret = FALSE;
 556
 557   *flowret = GST_FLOW_OK;
 558
 559   walk = mux->collect->data;
 560   while (walk) {
 561     GstOggPadData *pad = (GstOggPadData *) walk->data;
 562
 563     /* We need each queue to either be at EOS, or have one or more pages
 564      * available with a set granulepos (i.e. not -1), otherwise we don't have
 565      * enough data yet to determine which stream needs to go next for correct
 566      * time ordering. */
 567     if (pad->pagebuffers->length == 0) {
 568       if (pad->eos) {
 569         GST_LOG_OBJECT (pad->collect.pad,
 570             "pad is EOS, skipping for dequeue decision");
 571       } else {
 572         GST_LOG_OBJECT (pad->collect.pad,
 573             "no pages in this queue, can't dequeue");
 574         return FALSE;
 575       }
 576     } else {
 577       /* We then need to check for a non-negative granulepos */
 578       int i;
 579       gboolean valid = FALSE;
 580
 581       for (i = 0; i < pad->pagebuffers->length; i++) {
 582         buf = g_queue_peek_nth (pad->pagebuffers, i);
 583         /* Here we check the OFFSET_END, which is actually temporarily the
 584          * granulepos value for this buffer */
 585         if (GST_BUFFER_OFFSET_END (buf) != -1) {
 586           valid = TRUE;
 587           break;
 588         }
 589       }
 590       if (!valid) {
 591         GST_LOG_OBJECT (pad->collect.pad,
 592             "No page timestamps in queue, can't dequeue");
 593         return FALSE;
 594       }
 595     }
 596
 597     walk = g_slist_next (walk);
 598   }
 599
 600   walk = mux->collect->data;
 601   while (walk) {
 602     GstOggPadData *pad = (GstOggPadData *) walk->data;
 603
 604     /* any page with a granulepos of -1 can be pushed immediately.
 605      * TODO: it CAN be, but it seems silly to do so? */
 606     buf = g_queue_peek_head (pad->pagebuffers);
 607     while (buf && GST_BUFFER_OFFSET_END (buf) == -1) {
 608       GST_LOG_OBJECT (pad->collect.pad, "[gp        -1] pushing page");
 609       g_queue_pop_head (pad->pagebuffers);
 610       *flowret = gst_ogg_mux_push_buffer (mux, buf, pad);
 611       buf = g_queue_peek_head (pad->pagebuffers);
 612       ret = TRUE;
 613     }
 614
 615     if (buf) {
 616       /* if no oldest buffer yet, take this one */
 617       if (oldest == GST_CLOCK_TIME_NONE) {
 618         GST_LOG_OBJECT (mux, "no oldest yet, taking buffer %p from pad %"
 619             GST_PTR_FORMAT " with gp time %" GST_TIME_FORMAT,
 620             buf, pad->collect.pad, GST_TIME_ARGS (GST_BUFFER_OFFSET (buf)));
 621         oldest = GST_BUFFER_OFFSET (buf);
 622         opad = pad;
 623       } else {
 624         /* if we have an oldest, compare with this one */
 625         if (GST_BUFFER_OFFSET (buf) < oldest) {
 626           GST_LOG_OBJECT (mux, "older buffer %p, taking from pad %"
 627               GST_PTR_FORMAT " with gp time %" GST_TIME_FORMAT,
 628               buf, pad->collect.pad, GST_TIME_ARGS (GST_BUFFER_OFFSET (buf)));
 629           oldest = GST_BUFFER_OFFSET (buf);
 630           opad = pad;
 631         }
 632       }
 633     }
 634     walk = g_slist_next (walk);
 635   }
 636
 637   if (oldest != GST_CLOCK_TIME_NONE) {
 638     g_assert (opad);
 639     buf = g_queue_pop_head (opad->pagebuffers);
 640     GST_LOG_OBJECT (opad->collect.pad,
 641         GST_GP_FORMAT " pushing oldest page buffer %p (granulepos time %"
 642         GST_TIME_FORMAT ")", GST_BUFFER_OFFSET_END (buf), buf,
 643         GST_TIME_ARGS (GST_BUFFER_OFFSET (buf)));
 644     *flowret = gst_ogg_mux_push_buffer (mux, buf, opad);
 645     ret = TRUE;
 646   }
 647
 648   return ret;
 649 }
 650
 651 /* put the given ogg page on a per-pad queue, timestamping it correctly.
 652  * after that, dequeue and push as many pages as possible.
 653  * Caller should make sure:
 654  * pad->timestamp     was set with the timestamp of the first packet put
 655  *                    on the page
 656  * pad->timestamp_end was set with the timestamp + duration of the last packet
 657  *                    put on the page
 658  * pad->gp_time       was set with the time matching the gp of the last
 659  *                    packet put on the page
 660  *
 661  * will also reset timestamp and timestamp_end, so caller func can restart
 662  * counting.
 663  */
 664 static GstFlowReturn
 665 gst_ogg_mux_pad_queue_page (GstOggMux * mux, GstOggPadData * pad,
 666     ogg_page * page, gboolean delta)
 667 {
 668   GstFlowReturn ret;
 669   GstBuffer *buffer = gst_ogg_mux_buffer_from_page (mux, page, delta);
 670
 671   /* take the timestamp of the first packet on this page */
 672   GST_BUFFER_TIMESTAMP (buffer) = pad->timestamp;
 673   GST_BUFFER_DURATION (buffer) = pad->timestamp_end - pad->timestamp;
 674   /* take the gp time of the last completed packet on this page */
 675   GST_BUFFER_OFFSET (buffer) = pad->gp_time;
 676
 677   /* the next page will start where the current page's end time leaves off */
 678   pad->timestamp = pad->timestamp_end;
 679
 680   g_queue_push_tail (pad->pagebuffers, buffer);
 681   GST_LOG_OBJECT (pad->collect.pad, GST_GP_FORMAT
 682       " queued buffer page %p (gp time %"
 683       GST_TIME_FORMAT ", timestamp %" GST_TIME_FORMAT
 684       "), %d page buffers queued", GST_GP_CAST (ogg_page_granulepos (page)),
 685       buffer, GST_TIME_ARGS (GST_BUFFER_OFFSET (buffer)),
 686       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
 687       g_queue_get_length (pad->pagebuffers));
 688
 689   while (gst_ogg_mux_dequeue_page (mux, &ret)) {
 690     if (ret != GST_FLOW_OK)
 691       break;
 692   }
 693
 694   return ret;
 695 }
 696
 697 /*
 698  * Given two pads, compare the buffers queued on it.
 699  * Returns:
 700  *  0 if they have an equal priority
 701  * -1 if the first is better
 702  *  1 if the second is better
 703  * Priority decided by: a) validity, b) older timestamp, c) smaller number
 704  * of muxed pages
 705  */
 706 static gint
 707 gst_ogg_mux_compare_pads (GstOggMux * ogg_mux, GstOggPadData * first,
 708     GstOggPadData * second)
 709 {
 710   guint64 firsttime, secondtime;
 711
 712   /* if the first pad doesn't contain anything or is even NULL, return
 713    * the second pad as best candidate and vice versa */
 714   if (first == NULL)
 715     return 1;
 716   if (second == NULL)
 717     return -1;
 718
 719   /* no timestamp on first buffer, it must go first */
 720   firsttime = GST_BUFFER_TIMESTAMP (first->buffer);
 721   if (firsttime == GST_CLOCK_TIME_NONE)
 722     return -1;
 723
 724   /* no timestamp on second buffer, it must go first */
 725   secondtime = GST_BUFFER_TIMESTAMP (second->buffer);
 726   if (secondtime == GST_CLOCK_TIME_NONE)
 727     return 1;
 728
 729   /* first buffer has higher timestamp, second one should go first */
 730   if (secondtime < firsttime)
 731     return 1;
 732   /* second buffer has higher timestamp, first one should go first */
 733   else if (secondtime > firsttime)
 734     return -1;
 735   else {
 736     /* buffers with equal timestamps, prefer the pad that has the
 737      * least number of pages muxed */
 738     if (second->pageno < first->pageno)
 739       return 1;
 740     else if (second->pageno > first->pageno)
 741       return -1;
 742   }
 743
 744   /* same priority if all of the above failed */
 745   return 0;
 746 }
 747
 748 static GstBuffer *
 749 gst_ogg_mux_decorate_buffer (GstOggMux * ogg_mux, GstOggPadData * pad,
 750     GstBuffer * buf)
 751 {
 752   GstClockTime time;
 753   gint64 duration, granule, limit;
 754   GstClockTime next_time;
 755   GstClockTimeDiff diff;
 756   ogg_packet packet;
 757   gsize size;
 758
 759   /* ensure messing with metadata is ok */
 760   buf = gst_buffer_make_writable (buf);
 761
 762   /* convert time to running time, so we need no longer bother about that */
 763   time = GST_BUFFER_TIMESTAMP (buf);
 764   if (G_LIKELY (GST_CLOCK_TIME_IS_VALID (time))) {
 765     time = gst_segment_to_running_time (&pad->segment, GST_FORMAT_TIME, time);
 766     if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (time))) {
 767       gst_buffer_unref (buf);
 768       return NULL;
 769     } else {
 770       GST_BUFFER_TIMESTAMP (buf) = time;
 771     }
 772   }
 773
 774   /* now come up with granulepos stuff corresponding to time */
 775   if (!pad->have_type ||
 776       pad->map.granulerate_n <= 0 || pad->map.granulerate_d <= 0)
 777     goto no_granule;
 778
 779   packet.packet = gst_buffer_map (buf, &size, NULL, GST_MAP_READ);
 780   packet.bytes = size;
 781   duration = gst_ogg_stream_get_packet_duration (&pad->map, &packet);
 782   gst_buffer_unmap (buf, packet.packet, size);
 783
 784   /* give up if no duration can be determined, relying on upstream */
 785   if (G_UNLIKELY (duration < 0)) {
 786     /* well, if some day we really could handle sparse input ... */
 787     if (pad->map.is_sparse) {
 788       limit = 1;
 789       diff = 2;
 790       goto resync;
 791     }
 792     GST_WARNING_OBJECT (pad->collect.pad,
 793         "failed to determine packet duration");
 794     goto no_granule;
 795   }
 796
 797   GST_LOG_OBJECT (pad->collect.pad, "buffer ts %" GST_TIME_FORMAT
 798       ", duration %" GST_TIME_FORMAT ", granule duration %" G_GINT64_FORMAT,
 799       GST_TIME_ARGS (time), GST_TIME_ARGS (GST_BUFFER_DURATION (buf)),
 800       duration);
 801
 802   /* determine granule corresponding to time,
 803    * using the inverse of oggdemux' granule -> time */
 804
 805   /* see if interpolated granule matches good enough */
 806   granule = pad->next_granule;
 807   next_time = gst_ogg_stream_granule_to_time (&pad->map, pad->next_granule);
 808   diff = GST_CLOCK_DIFF (next_time, time);
 809
 810   /* we tolerate deviation up to configured or within granule granularity */
 811   limit = gst_ogg_stream_granule_to_time (&pad->map, 1) / 2;
 812   limit = MAX (limit, ogg_mux->max_tolerance);
 813
 814   GST_LOG_OBJECT (pad->collect.pad, "expected granule %" G_GINT64_FORMAT " == "
 815       "time %" GST_TIME_FORMAT " --> ts diff %" GST_TIME_FORMAT
 816       " < tolerance %" GST_TIME_FORMAT " (?)",
 817       granule, GST_TIME_ARGS (next_time), GST_TIME_ARGS (ABS (diff)),
 818       GST_TIME_ARGS (limit));
 819
 820 resync:
 821   /* if not good enough, determine granule based on time */
 822   if (diff > limit || diff < -limit) {
 823     granule = gst_util_uint64_scale_round (time, pad->map.granulerate_n,
 824         GST_SECOND * pad->map.granulerate_d);
 825     GST_DEBUG_OBJECT (pad->collect.pad,
 826         "resyncing to determined granule %" G_GINT64_FORMAT, granule);
 827   }
 828
 829   if (pad->map.is_ogm || pad->map.is_sparse) {
 830     pad->next_granule = granule;
 831   } else {
 832     granule += duration;
 833     pad->next_granule = granule;
 834   }
 835
 836   /* track previous keyframe */
 837   if (!GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT))
 838     pad->keyframe_granule = granule;
 839
 840   /* determine corresponding time and granulepos */
 841   GST_BUFFER_OFFSET (buf) = gst_ogg_stream_granule_to_time (&pad->map, granule);
 842   GST_BUFFER_OFFSET_END (buf) =
 843       gst_ogg_stream_granule_to_granulepos (&pad->map, granule,
 844       pad->keyframe_granule);
 845
 846   return buf;
 847
 848   /* ERRORS */
 849 no_granule:
 850   {
 851     GST_DEBUG_OBJECT (pad->collect.pad, "could not determine granulepos, "
 852         "falling back to upstream provided metadata");
 853     return buf;
 854   }
 855 }
 856
 857
 858 /* make sure at least one buffer is queued on all pads, two if possible
 859  *
 860  * if pad->buffer == NULL, pad->next_buffer !=  NULL, then
 861  *   we do not know if the buffer is the last or not
 862  * if pad->buffer != NULL, pad->next_buffer != NULL, then
 863  *   pad->buffer is not the last buffer for the pad
 864  * if pad->buffer != NULL, pad->next_buffer == NULL, then
 865  *   pad->buffer if the last buffer for the pad
 866  *
 867  * returns a pointer to an oggpad that holds the best buffer, or
 868  * NULL when no pad was usable. "best" means the buffer marked
 869  * with the lowest timestamp. If best->buffer == NULL then either
 870  * we're at EOS (popped = FALSE), or a buffer got dropped, so retry. */
 871 static GstOggPadData *
 872 gst_ogg_mux_queue_pads (GstOggMux * ogg_mux, gboolean * popped)
 873 {
 874   GstOggPadData *bestpad = NULL;
 875   GSList *walk;
 876
 877   *popped = FALSE;
 878
 879   /* try to make sure we have a buffer from each usable pad first */
 880   walk = ogg_mux->collect->data;
 881   while (walk) {
 882     GstOggPadData *pad;
 883     GstCollectData2 *data;
 884
 885     data = (GstCollectData2 *) walk->data;
 886     pad = (GstOggPadData *) data;
 887
 888     walk = g_slist_next (walk);
 889
 890     GST_LOG_OBJECT (data->pad, "looking at pad for buffer");
 891
 892     /* try to get a new buffer for this pad if needed and possible */
 893     if (pad->buffer == NULL) {
 894       GstBuffer *buf;
 895
 896       buf = gst_collect_pads2_pop (ogg_mux->collect, data);
 897       GST_LOG_OBJECT (data->pad, "popped buffer %" GST_PTR_FORMAT, buf);
 898
 899       /* On EOS we get a NULL buffer */
 900       if (buf != NULL) {
 901         *popped = TRUE;
 902
 903         if (ogg_mux->delta_pad == NULL &&
 904             GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT))
 905           ogg_mux->delta_pad = pad;
 906
 907         /* if we need headers */
 908         if (pad->state == GST_OGG_PAD_STATE_CONTROL) {
 909           /* and we have one */
 910           ogg_packet packet;
 911           gboolean is_header;
 912           gsize size;
 913
 914           packet.packet = gst_buffer_map (buf, &size, NULL, GST_MAP_READ);
 915           packet.bytes = size;
 916
 917           /* if we're not yet in data mode, ensure we're setup on the first packet */
 918           if (!pad->have_type) {
 919             GstCaps *caps;
 920
 921             /* Use headers in caps, if any; this will allow us to be resilient
 922              * to starting streams on the fly, and some streams (like VP8
 923              * at least) do not send headers packets, as other muxers don't
 924              * expect/need them. */
 925             caps = gst_pad_get_current_caps (GST_PAD_CAST (data->pad));
 926             pad->have_type =
 927                 gst_ogg_stream_setup_map_from_caps_headers (&pad->map, caps);
 928
 929             if (!pad->have_type) {
 930               /* fallback on the packet */
 931               pad->have_type = gst_ogg_stream_setup_map (&pad->map, &packet);
 932             }
 933             if (!pad->have_type) {
 934               GST_ERROR_OBJECT (pad, "mapper didn't recognise input stream "
 935                   "(pad caps: %" GST_PTR_FORMAT ")", caps);
 936             } else {
 937               GST_DEBUG_OBJECT (pad, "caps detected: %" GST_PTR_FORMAT,
 938                   pad->map.caps);
 939             }
 940             if (caps)
 941               gst_caps_unref (caps);
 942           }
 943
 944           if (pad->have_type)
 945             is_header = gst_ogg_stream_packet_is_header (&pad->map, &packet);
 946           else                  /* fallback (FIXME 0.11: remove IN_CAPS hack) */
 947             is_header = GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_IN_CAPS);
 948
 949           gst_buffer_unmap (buf, packet.packet, size);
 950
 951           if (is_header) {
 952             GST_DEBUG_OBJECT (ogg_mux,
 953                 "got header buffer in control state, ignoring");
 954             /* just ignore */
 955             pad->map.n_header_packets_seen++;
 956             gst_buffer_unref (buf);
 957             buf = NULL;
 958           } else {
 959             GST_DEBUG_OBJECT (ogg_mux,
 960                 "got data buffer in control state, switching to data mode");
 961             /* this is a data buffer so switch to data state */
 962             pad->state = GST_OGG_PAD_STATE_DATA;
 963
 964             /* check if this type of stream allows generating granulepos
 965              * metadata here, if not, upstream will have to provide */
 966             if (gst_ogg_stream_granule_to_granulepos (&pad->map, 1, 1) < 0) {
 967               GST_WARNING_OBJECT (data->pad, "can not generate metadata; "
 968                   "relying on upstream");
 969               /* disable metadata code path, otherwise not used anyway */
 970               pad->map.granulerate_n = 0;
 971             }
 972           }
 973         }
 974
 975         /* so now we should have a real data packet;
 976          * see that it is properly decorated */
 977         if (G_LIKELY (buf)) {
 978           buf = gst_ogg_mux_decorate_buffer (ogg_mux, pad, buf);
 979           if (G_UNLIKELY (!buf))
 980             GST_DEBUG_OBJECT (data->pad, "buffer clipped");
 981         }
 982       }
 983
 984       pad->buffer = buf;
 985     }
 986
 987     /* we should have a buffer now, see if it is the best pad to
 988      * pull on */
 989     if (pad->buffer) {
 990       if (gst_ogg_mux_compare_pads (ogg_mux, bestpad, pad) > 0) {
 991         GST_LOG_OBJECT (data->pad,
 992             "new best pad, with buffer %" GST_PTR_FORMAT, pad->buffer);
 993
 994         bestpad = pad;
 995       }
 996     }
 997   }
 998
 999   return bestpad;
1000 }
1001
1002 static GList *
1003 gst_ogg_mux_get_headers (GstOggPadData * pad)
1004 {
1005   GList *res = NULL;
1006   GstStructure *structure;
1007   GstCaps *caps;
1008   GstPad *thepad;
1009
1010   thepad = pad->collect.pad;
1011
1012   GST_LOG_OBJECT (thepad, "getting headers");
1013
1014   caps = gst_pad_get_current_caps (thepad);
1015   if (caps != NULL) {
1016     const GValue *streamheader;
1017
1018     structure = gst_caps_get_structure (caps, 0);
1019     streamheader = gst_structure_get_value (structure, "streamheader");
1020     if (streamheader != NULL) {
1021       GST_LOG_OBJECT (thepad, "got header");
1022       if (G_VALUE_TYPE (streamheader) == GST_TYPE_ARRAY) {
1023         GArray *bufarr = g_value_peek_pointer (streamheader);
1024         gint i;
1025
1026         GST_LOG_OBJECT (thepad, "got fixed list");
1027
1028         for (i = 0; i < bufarr->len; i++) {
1029           GValue *bufval = &g_array_index (bufarr, GValue, i);
1030
1031           GST_LOG_OBJECT (thepad, "item %d", i);
1032           if (G_VALUE_TYPE (bufval) == GST_TYPE_BUFFER) {
1033             GstBuffer *buf = g_value_peek_pointer (bufval);
1034
1035             GST_LOG_OBJECT (thepad, "adding item %d to header list", i);
1036
1037             gst_buffer_ref (buf);
1038             res = g_list_append (res, buf);
1039           }
1040         }
1041       } else {
1042         GST_LOG_OBJECT (thepad, "streamheader is not fixed list");
1043       }
1044
1045     } else if (gst_structure_has_name (structure, "video/x-dirac")) {
1046       res = g_list_append (res, pad->buffer);
1047       pad->buffer = NULL;
1048     } else {
1049       GST_LOG_OBJECT (thepad, "caps don't have streamheader");
1050     }
1051     gst_caps_unref (caps);
1052   } else {
1053     GST_LOG_OBJECT (thepad, "got empty caps as negotiated format");
1054   }
1055   return res;
1056 }
1057
1058 static GstCaps *
1059 gst_ogg_mux_set_header_on_caps (GstCaps * caps, GList * buffers)
1060 {
1061   GstStructure *structure;
1062   GValue array = { 0 };
1063   GList *walk = buffers;
1064
1065   caps = gst_caps_make_writable (caps);
1066
1067   structure = gst_caps_get_structure (caps, 0);
1068
1069   /* put buffers in a fixed list */
1070   g_value_init (&array, GST_TYPE_ARRAY);
1071
1072   while (walk) {
1073     GstBuffer *buf = GST_BUFFER (walk->data);
1074     GstBuffer *copy;
1075     GValue value = { 0 };
1076
1077     walk = walk->next;
1078
1079     /* mark buffer */
1080     GST_LOG ("Setting IN_CAPS on buffer of length %" G_GSIZE_FORMAT,
1081         gst_buffer_get_size (buf));
1082     GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_IN_CAPS);
1083
1084     g_value_init (&value, GST_TYPE_BUFFER);
1085     copy = gst_buffer_copy (buf);
1086     gst_value_set_buffer (&value, copy);
1087     gst_buffer_unref (copy);
1088     gst_value_array_append_value (&array, &value);
1089     g_value_unset (&value);
1090   }
1091   gst_structure_set_value (structure, "streamheader", &array);
1092   g_value_unset (&array);
1093
1094   return caps;
1095 }
1096
1097 static void
1098 gst_ogg_mux_create_header_packet_with_flags (ogg_packet * packet,
1099     gboolean bos, gboolean eos)
1100 {
1101   packet->granulepos = 0;
1102   /* mark BOS and packet number */
1103   packet->b_o_s = bos;
1104   /* mark EOS */
1105   packet->e_o_s = eos;
1106 }
1107
1108 static void
1109 gst_ogg_mux_create_header_packet (ogg_packet * packet, GstOggPadData * pad)
1110 {
1111   gst_ogg_mux_create_header_packet_with_flags (packet, pad->packetno == 0, 0);
1112   packet->packetno = pad->packetno++;
1113 }
1114
1115 static void
1116 gst_ogg_mux_submit_skeleton_header_packet (GstOggMux * mux,
1117     ogg_stream_state * os, GstBuffer * buf, gboolean bos, gboolean eos)
1118 {
1119   ogg_packet packet;
1120   gsize size;
1121
1122   packet.packet = gst_buffer_map (buf, &size, NULL, GST_MAP_READ);
1123   packet.bytes = size;
1124   gst_ogg_mux_create_header_packet_with_flags (&packet, bos, eos);
1125   ogg_stream_packetin (os, &packet);
1126   gst_buffer_unref (buf);
1127 }
1128
1129 static void
1130 gst_ogg_mux_make_fishead (GstOggMux * mux, ogg_stream_state * os)
1131 {
1132   GstByteWriter bw;
1133   GstBuffer *fishead;
1134
1135   GST_DEBUG_OBJECT (mux, "Creating fishead");
1136
1137   gst_byte_writer_init_with_size (&bw, 64, TRUE);
1138   gst_byte_writer_put_string_utf8 (&bw, "fishead");
1139   gst_byte_writer_put_int16_le (&bw, 3);        /* version major */
1140   gst_byte_writer_put_int16_le (&bw, 0);        /* version minor */
1141   gst_byte_writer_put_int64_le (&bw, 0);        /* presentation time numerator */
1142   gst_byte_writer_put_int64_le (&bw, 1000);     /* ...and denominator */
1143   gst_byte_writer_put_int64_le (&bw, 0);        /* base time numerator */
1144   gst_byte_writer_put_int64_le (&bw, 1000);     /* ...and denominator */
1145   gst_byte_writer_fill (&bw, ' ', 20);  /* UTC time */
1146   g_assert (gst_byte_writer_get_pos (&bw) == 64);
1147   fishead = gst_byte_writer_reset_and_get_buffer (&bw);
1148   gst_ogg_mux_submit_skeleton_header_packet (mux, os, fishead, 1, 0);
1149 }
1150
1151 static void
1152 gst_ogg_mux_byte_writer_put_string_utf8 (GstByteWriter * bw, const char *s)
1153 {
1154   gst_byte_writer_put_data (bw, (const guint8 *) s, strlen (s));
1155 }
1156
1157 static void
1158 gst_ogg_mux_add_fisbone_message_header (GstOggMux * mux, GstByteWriter * bw,
1159     const char *tag, const char *value)
1160 {
1161   /* It is valid to pass NULL as the value to omit the tag */
1162   if (!value)
1163     return;
1164   GST_DEBUG_OBJECT (mux, "Adding fisbone message header %s: %s", tag, value);
1165   gst_ogg_mux_byte_writer_put_string_utf8 (bw, tag);
1166   gst_ogg_mux_byte_writer_put_string_utf8 (bw, ": ");
1167   gst_ogg_mux_byte_writer_put_string_utf8 (bw, value);
1168   gst_ogg_mux_byte_writer_put_string_utf8 (bw, "\r\n");
1169 }
1170
1171 static void
1172 gst_ogg_mux_add_fisbone_message_header_from_tags (GstOggMux * mux,
1173     GstByteWriter * bw, const char *header, const char *tag,
1174     const GstTagList * tags)
1175 {
1176   GString *s;
1177   guint size = gst_tag_list_get_tag_size (tags, tag), n;
1178   GST_DEBUG_OBJECT (mux, "Found %u tags for name %s", size, tag);
1179   if (size == 0)
1180     return;
1181   s = g_string_new ("");
1182   for (n = 0; n < size; ++n) {
1183     gchar *tmp;
1184     if (n)
1185       g_string_append (s, ", ");
1186     gst_tag_list_get_string_index (tags, tag, n, &tmp);
1187     g_string_append (s, tmp);
1188     g_free (tmp);
1189   }
1190   gst_ogg_mux_add_fisbone_message_header (mux, bw, header, s->str);
1191   g_string_free (s, TRUE);
1192 }
1193
1194 /* This is a basic placeholder to generate roles for the tracks.
1195    For tracks with more than one video, both video tracks will get
1196    tagged with a "video/main" role, but we have no way of knowing
1197    which one is the main one, if any. We could just pick one. For
1198    audio, it's more complicated as we don't know which is music,
1199    which is dubbing, etc. For kate, we could take a pretty good
1200    guess based on the category, as role essentially is category.
1201    For now, leave this as is. */
1202 static const char *
1203 gst_ogg_mux_get_default_role (GstOggPadData * pad)
1204 {
1205   const char *type = gst_ogg_stream_get_media_type (&pad->map);
1206   if (type) {
1207     if (!strncmp (type, "video/", strlen ("video/")))
1208       return "video/main";
1209     if (!strncmp (type, "audio/", strlen ("audio/")))
1210       return "audio/main";
1211     if (!strcmp (type + strlen (type) - strlen ("kate"), "kate"))
1212       return "text/caption";
1213   }
1214   return NULL;
1215 }
1216
1217 static void
1218 gst_ogg_mux_make_fisbone (GstOggMux * mux, ogg_stream_state * os,
1219     GstOggPadData * pad)
1220 {
1221   GstByteWriter bw;
1222
1223   GST_DEBUG_OBJECT (mux,
1224       "Creating %s fisbone for serial %08x",
1225       gst_ogg_stream_get_media_type (&pad->map), pad->map.serialno);
1226
1227   gst_byte_writer_init (&bw);
1228   gst_byte_writer_put_string_utf8 (&bw, "fisbone");
1229   gst_byte_writer_put_int32_le (&bw, 44);       /* offset to message headers */
1230   gst_byte_writer_put_uint32_le (&bw, pad->map.serialno);
1231   gst_byte_writer_put_uint32_le (&bw, pad->map.n_header_packets);
1232   gst_byte_writer_put_uint64_le (&bw, pad->map.granulerate_n);
1233   gst_byte_writer_put_uint64_le (&bw, pad->map.granulerate_d);
1234   gst_byte_writer_put_uint64_le (&bw, 0);       /* base granule */
1235   gst_byte_writer_put_uint32_le (&bw, pad->map.preroll);
1236   gst_byte_writer_put_uint8 (&bw, pad->map.granuleshift);
1237   gst_byte_writer_fill (&bw, 0, 3);     /* padding */
1238   /* message header fields - MIME type for now */
1239   gst_ogg_mux_add_fisbone_message_header (mux, &bw, "Content-Type",
1240       gst_ogg_stream_get_media_type (&pad->map));
1241   gst_ogg_mux_add_fisbone_message_header (mux, &bw, "Role",
1242       gst_ogg_mux_get_default_role (pad));
1243   gst_ogg_mux_add_fisbone_message_header_from_tags (mux, &bw, "Language",
1244       GST_TAG_LANGUAGE_CODE, pad->tags);
1245   gst_ogg_mux_add_fisbone_message_header_from_tags (mux, &bw, "Title",
1246       GST_TAG_TITLE, pad->tags);
1247
1248   gst_ogg_mux_submit_skeleton_header_packet (mux, os,
1249       gst_byte_writer_reset_and_get_buffer (&bw), 0, 0);
1250 }
1251
1252 static void
1253 gst_ogg_mux_make_fistail (GstOggMux * mux, ogg_stream_state * os)
1254 {
1255   GST_DEBUG_OBJECT (mux, "Creating fistail");
1256
1257   gst_ogg_mux_submit_skeleton_header_packet (mux, os,
1258       gst_buffer_new_and_alloc (0), 0, 1);
1259 }
1260
1261 /*
1262  * For each pad we need to write out one (small) header in one
1263  * page that allows decoders to identify the type of the stream.
1264  * After that we need to write out all extra info for the decoders.
1265  * In the case of a codec that also needs data as configuration, we can
1266  * find that info in the streamcaps.
1267  * After writing the headers we must start a new page for the data.
1268  */
1269 static GstFlowReturn
1270 gst_ogg_mux_send_headers (GstOggMux * mux)
1271 {
1272   GSList *walk;
1273   GList *hbufs, *hwalk;
1274   GstCaps *caps;
1275   GstFlowReturn ret;
1276   ogg_page page;
1277   ogg_stream_state skeleton_stream;
1278
1279   hbufs = NULL;
1280   ret = GST_FLOW_OK;
1281
1282   GST_LOG_OBJECT (mux, "collecting headers");
1283
1284   walk = mux->collect->data;
1285   while (walk) {
1286     GstOggPadData *pad;
1287     GstPad *thepad;
1288
1289     pad = (GstOggPadData *) walk->data;
1290     thepad = pad->collect.pad;
1291
1292     walk = g_slist_next (walk);
1293
1294     GST_LOG_OBJECT (mux, "looking at pad %s:%s", GST_DEBUG_PAD_NAME (thepad));
1295
1296     /* if the pad has no buffer, we don't care */
1297     if (pad->buffer == NULL)
1298       continue;
1299
1300     /* now figure out the headers */
1301     pad->map.headers = gst_ogg_mux_get_headers (pad);
1302   }
1303
1304   GST_LOG_OBJECT (mux, "creating BOS pages");
1305   walk = mux->collect->data;
1306   while (walk) {
1307     GstOggPadData *pad;
1308     GstBuffer *buf;
1309     ogg_packet packet;
1310     GstPad *thepad;
1311     GstCaps *caps;
1312     GstStructure *structure;
1313     GstBuffer *hbuf;
1314     gsize size;
1315
1316     pad = (GstOggPadData *) walk->data;
1317     thepad = pad->collect.pad;
1318     caps = gst_pad_get_current_caps (thepad);
1319     structure = gst_caps_get_structure (caps, 0);
1320
1321     walk = walk->next;
1322
1323     pad->packetno = 0;
1324
1325     GST_LOG_OBJECT (thepad, "looping over headers");
1326
1327     if (pad->map.headers) {
1328       buf = GST_BUFFER (pad->map.headers->data);
1329       pad->map.headers = g_list_remove (pad->map.headers, buf);
1330     } else if (pad->buffer) {
1331       buf = pad->buffer;
1332       gst_buffer_ref (buf);
1333     } else {
1334       /* fixme -- should be caught in the previous list traversal. */
1335       GST_OBJECT_LOCK (thepad);
1336       g_critical ("No headers or buffers on pad %s:%s",
1337           GST_DEBUG_PAD_NAME (thepad));
1338       GST_OBJECT_UNLOCK (thepad);
1339       continue;
1340     }
1341
1342     /* create a packet from the buffer */
1343     packet.packet = gst_buffer_map (buf, &size, NULL, GST_MAP_READ);
1344     packet.bytes = size;
1345
1346     gst_ogg_mux_create_header_packet (&packet, pad);
1347
1348     /* swap the packet in */
1349     ogg_stream_packetin (&pad->map.stream, &packet);
1350
1351     gst_buffer_unmap (buf, packet.packet, size);
1352     gst_buffer_unref (buf);
1353
1354     GST_LOG_OBJECT (thepad, "flushing out BOS page");
1355     if (!ogg_stream_flush (&pad->map.stream, &page))
1356       g_critical ("Could not flush BOS page");
1357
1358     hbuf = gst_ogg_mux_buffer_from_page (mux, &page, FALSE);
1359
1360     GST_LOG_OBJECT (mux, "swapped out page with mime type %s",
1361         gst_structure_get_name (structure));
1362
1363     /* quick hack: put video pages at the front.
1364      * Ideally, we would have a settable enum for which Ogg
1365      * profile we work with, and order based on that.
1366      * (FIXME: if there is more than one video stream, shouldn't we only put
1367      * one's BOS into the first page, followed by an audio stream's BOS, and
1368      * only then followed by the remaining video and audio streams?) */
1369     if (pad->map.is_video) {
1370       GST_DEBUG_OBJECT (thepad, "putting %s page at the front",
1371           gst_structure_get_name (structure));
1372       hbufs = g_list_prepend (hbufs, hbuf);
1373     } else {
1374       hbufs = g_list_append (hbufs, hbuf);
1375     }
1376
1377     gst_caps_unref (caps);
1378   }
1379
1380   /* The Skeleton BOS goes first - even before the video that went first before */
1381   if (mux->use_skeleton) {
1382     ogg_stream_init (&skeleton_stream, gst_ogg_mux_generate_serialno (mux));
1383     gst_ogg_mux_make_fishead (mux, &skeleton_stream);
1384     while (ogg_stream_flush (&skeleton_stream, &page) > 0) {
1385       GstBuffer *hbuf = gst_ogg_mux_buffer_from_page (mux, &page, FALSE);
1386       hbufs = g_list_append (hbufs, hbuf);
1387     }
1388   }
1389
1390   GST_LOG_OBJECT (mux, "creating next headers");
1391   walk = mux->collect->data;
1392   while (walk) {
1393     GstOggPadData *pad;
1394     GstPad *thepad;
1395
1396     pad = (GstOggPadData *) walk->data;
1397     thepad = pad->collect.pad;
1398
1399     walk = walk->next;
1400
1401     if (mux->use_skeleton)
1402       gst_ogg_mux_make_fisbone (mux, &skeleton_stream, pad);
1403
1404     GST_LOG_OBJECT (mux, "looping over headers for pad %s:%s",
1405         GST_DEBUG_PAD_NAME (thepad));
1406
1407     hwalk = pad->map.headers;
1408     while (hwalk) {
1409       GstBuffer *buf = GST_BUFFER (hwalk->data);
1410       ogg_packet packet;
1411       ogg_page page;
1412       gsize size;
1413
1414       hwalk = hwalk->next;
1415
1416       /* create a packet from the buffer */
1417       packet.packet = gst_buffer_map (buf, &size, NULL, GST_MAP_READ);
1418       packet.bytes = size;
1419
1420       gst_ogg_mux_create_header_packet (&packet, pad);
1421
1422       /* swap the packet in */
1423       ogg_stream_packetin (&pad->map.stream, &packet);
1424       gst_buffer_unmap (buf, packet.packet, size);
1425       gst_buffer_unref (buf);
1426
1427       /* if last header, flush page */
1428       if (hwalk == NULL) {
1429         GST_LOG_OBJECT (mux,
1430             "flushing page as packet %" G_GUINT64_FORMAT " is first or "
1431             "last packet", (guint64) packet.packetno);
1432         while (ogg_stream_flush (&pad->map.stream, &page)) {
1433           GstBuffer *hbuf = gst_ogg_mux_buffer_from_page (mux, &page, FALSE);
1434
1435           GST_LOG_OBJECT (mux, "swapped out page");
1436           hbufs = g_list_append (hbufs, hbuf);
1437         }
1438       } else {
1439         GST_LOG_OBJECT (mux, "try to swap out page");
1440         /* just try to swap out a page then */
1441         while (ogg_stream_pageout (&pad->map.stream, &page) > 0) {
1442           GstBuffer *hbuf = gst_ogg_mux_buffer_from_page (mux, &page, FALSE);
1443
1444           GST_LOG_OBJECT (mux, "swapped out page");
1445           hbufs = g_list_append (hbufs, hbuf);
1446         }
1447       }
1448     }
1449     g_list_free (pad->map.headers);
1450     pad->map.headers = NULL;
1451   }
1452
1453   if (mux->use_skeleton) {
1454     /* flush accumulated fisbones, the fistail must be on a separate page */
1455     while (ogg_stream_flush (&skeleton_stream, &page) > 0) {
1456       GstBuffer *hbuf = gst_ogg_mux_buffer_from_page (mux, &page, FALSE);
1457       hbufs = g_list_append (hbufs, hbuf);
1458     }
1459     gst_ogg_mux_make_fistail (mux, &skeleton_stream);
1460     while (ogg_stream_flush (&skeleton_stream, &page) > 0) {
1461       GstBuffer *hbuf = gst_ogg_mux_buffer_from_page (mux, &page, FALSE);
1462       hbufs = g_list_append (hbufs, hbuf);
1463     }
1464     ogg_stream_clear (&skeleton_stream);
1465   }
1466
1467   /* hbufs holds all buffers for the headers now */
1468
1469   /* create caps with the buffers */
1470   caps = gst_pad_get_caps (mux->srcpad, NULL);
1471   if (caps) {
1472     caps = gst_ogg_mux_set_header_on_caps (caps, hbufs);
1473     gst_pad_set_caps (mux->srcpad, caps);
1474     gst_caps_unref (caps);
1475   }
1476   /* and send the buffers */
1477   while (hbufs != NULL) {
1478     GstBuffer *buf = GST_BUFFER (hbufs->data);
1479
1480     hbufs = g_list_delete_link (hbufs, hbufs);
1481
1482     if ((ret = gst_ogg_mux_push_buffer (mux, buf, NULL)) != GST_FLOW_OK)
1483       break;
1484   }
1485   /* free any remaining nodes/buffers in case we couldn't push them */
1486   g_list_foreach (hbufs, (GFunc) gst_mini_object_unref, NULL);
1487   g_list_free (hbufs);
1488
1489   return ret;
1490 }
1491
1492 /* this function is called to process data on the best pending pad.
1493  *
1494  * basic idea:
1495  *
1496  * 1) store the selected pad and keep on pulling until we fill a
1497  *    complete ogg page or the ogg page is filled above the max-delay
1498  *    threshold. This is needed because the ogg spec says that
1499  *    you should fill a complete page with data from the same logical
1500  *    stream. When the page is filled, go back to 1).
1501  * 2) before filling a page, read ahead one more buffer to see if this
1502  *    packet is the last of the stream. We need to do this because the ogg
1503  *    spec mandates that the last packet should have the EOS flag set before
1504  *    sending it to ogg. if pad->buffer is NULL we need to wait to find out
1505  *    whether there are any more buffers.
1506  * 3) pages get queued on a per-pad queue. Every time a page is queued, a
1507  *    dequeue is called, which will dequeue the oldest page on any pad, provided
1508  *    that ALL pads have at least one marked page in the queue (or remaining
1509  *    pads are at EOS)
1510  */
1511 static GstFlowReturn
1512 gst_ogg_mux_process_best_pad (GstOggMux * ogg_mux, GstOggPadData * best)
1513 {
1514   GstFlowReturn ret = GST_FLOW_OK;
1515   gboolean delta_unit;
1516   gint64 granulepos = 0;
1517   GstClockTime timestamp, gp_time;
1518   GstBuffer *next_buf;
1519
1520   GST_LOG_OBJECT (ogg_mux, "best pad %" GST_PTR_FORMAT
1521       ", currently pulling from %" GST_PTR_FORMAT, best->collect.pad,
1522       ogg_mux->pulling ? ogg_mux->pulling->collect.pad : NULL);
1523
1524   if (ogg_mux->pulling) {
1525     next_buf = gst_collect_pads2_peek (ogg_mux->collect,
1526         &ogg_mux->pulling->collect);
1527     if (next_buf) {
1528       ogg_mux->pulling->eos = FALSE;
1529       gst_buffer_unref (next_buf);
1530     } else {
1531       GST_DEBUG_OBJECT (ogg_mux->pulling->collect.pad, "setting eos to true");
1532       ogg_mux->pulling->eos = TRUE;
1533     }
1534   }
1535
1536   /* We could end up pushing from the best pad instead, so check that
1537    * as well */
1538   if (best && best != ogg_mux->pulling) {
1539     next_buf = gst_collect_pads2_peek (ogg_mux->collect, &best->collect);
1540     if (next_buf) {
1541       best->eos = FALSE;
1542       gst_buffer_unref (next_buf);
1543     } else {
1544       GST_DEBUG_OBJECT (best->collect.pad, "setting eos to true");
1545       best->eos = TRUE;
1546     }
1547   }
1548
1549   /* if we were already pulling from one pad, but the new "best" buffer is
1550    * from another pad, we need to check if we have reason to flush a page
1551    * for the pad we were pulling from before */
1552   if (ogg_mux->pulling && best &&
1553       ogg_mux->pulling != best && ogg_mux->pulling->buffer) {
1554     GstOggPadData *pad = ogg_mux->pulling;
1555     GstClockTime last_ts = GST_BUFFER_END_TIME (pad->buffer);
1556
1557     /* if the next packet in the current page is going to make the page
1558      * too long, we need to flush */
1559     if (last_ts > ogg_mux->next_ts + ogg_mux->max_delay) {
1560       ogg_page page;
1561
1562       GST_LOG_OBJECT (pad->collect.pad,
1563           GST_GP_FORMAT " stored packet %" G_GINT64_FORMAT
1564           " will make page too long, flushing",
1565           GST_BUFFER_OFFSET_END (pad->buffer),
1566           (gint64) pad->map.stream.packetno);
1567
1568       while (ogg_stream_flush (&pad->map.stream, &page)) {
1569         /* end time of this page is the timestamp of the next buffer */
1570         ogg_mux->pulling->timestamp_end = GST_BUFFER_TIMESTAMP (pad->buffer);
1571         /* Place page into the per-pad queue */
1572         ret = gst_ogg_mux_pad_queue_page (ogg_mux, pad, &page,
1573             pad->first_delta);
1574         /* increment the page number counter */
1575         pad->pageno++;
1576         /* mark other pages as delta */
1577         pad->first_delta = TRUE;
1578       }
1579       pad->new_page = TRUE;
1580       ogg_mux->pulling = NULL;
1581     }
1582   }
1583
1584   /* if we don't know which pad to pull on, use the best one */
1585   if (ogg_mux->pulling == NULL) {
1586     ogg_mux->pulling = best;
1587     GST_LOG_OBJECT (ogg_mux->pulling->collect.pad, "pulling from best pad");
1588
1589     /* remember timestamp and gp time of first buffer for this new pad */
1590     if (ogg_mux->pulling != NULL) {
1591       ogg_mux->next_ts = GST_BUFFER_TIMESTAMP (ogg_mux->pulling->buffer);
1592       GST_LOG_OBJECT (ogg_mux->pulling->collect.pad, "updated times, next ts %"
1593           GST_TIME_FORMAT, GST_TIME_ARGS (ogg_mux->next_ts));
1594     } else {
1595       /* no pad to pull on, send EOS */
1596       gst_pad_push_event (ogg_mux->srcpad, gst_event_new_eos ());
1597       return GST_FLOW_WRONG_STATE;
1598     }
1599   }
1600
1601   if (ogg_mux->need_headers) {
1602     ret = gst_ogg_mux_send_headers (ogg_mux);
1603     ogg_mux->need_headers = FALSE;
1604   }
1605
1606   /* we are pulling from a pad, continue to do so until a page
1607    * has been filled and queued */
1608   if (ogg_mux->pulling != NULL) {
1609     ogg_packet packet;
1610     ogg_page page;
1611     GstBuffer *buf, *tmpbuf;
1612     GstOggPadData *pad = ogg_mux->pulling;
1613     gint64 duration;
1614     gboolean force_flush;
1615     gsize size;
1616
1617     GST_LOG_OBJECT (ogg_mux->pulling->collect.pad, "pulling from pad");
1618
1619     /* now see if we have a buffer */
1620     buf = pad->buffer;
1621     if (buf == NULL) {
1622       GST_DEBUG_OBJECT (ogg_mux, "pad was EOS");
1623       ogg_mux->pulling = NULL;
1624       return GST_FLOW_OK;
1625     }
1626
1627     delta_unit = GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT);
1628     duration = GST_BUFFER_DURATION (buf);
1629
1630     /* if the current "next timestamp" on the pad is unset, then this is the
1631      * first packet on the new page.  Update our pad's page timestamp */
1632     if (ogg_mux->pulling->timestamp == GST_CLOCK_TIME_NONE) {
1633       ogg_mux->pulling->timestamp = GST_BUFFER_TIMESTAMP (buf);
1634       GST_LOG_OBJECT (ogg_mux->pulling->collect.pad,
1635           "updated pad timestamp to %" GST_TIME_FORMAT,
1636           GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
1637     }
1638     /* create a packet from the buffer */
1639     packet.packet = gst_buffer_map (buf, &size, NULL, GST_MAP_READ);
1640     packet.bytes = size;
1641     packet.granulepos = GST_BUFFER_OFFSET_END (buf);
1642     if (packet.granulepos == -1)
1643       packet.granulepos = 0;
1644     /* mark BOS and packet number */
1645     packet.b_o_s = (pad->packetno == 0);
1646     packet.packetno = pad->packetno++;
1647     GST_LOG_OBJECT (pad->collect.pad, GST_GP_FORMAT
1648         " packet %" G_GINT64_FORMAT " (%ld bytes) created from buffer",
1649         GST_GP_CAST (packet.granulepos), (gint64) packet.packetno,
1650         packet.bytes);
1651
1652     packet.e_o_s = ogg_mux->pulling->eos ? 1 : 0;
1653     tmpbuf = NULL;
1654
1655     /* we flush when we see a new keyframe */
1656     force_flush = (pad->prev_delta && !delta_unit)
1657         || pad->map.always_flush_page;
1658     if (duration != -1) {
1659       pad->duration += duration;
1660       /* if page duration exceeds max, flush page */
1661       if (pad->duration > ogg_mux->max_page_delay) {
1662         force_flush = TRUE;
1663         pad->duration = 0;
1664       }
1665     }
1666
1667     if (GST_BUFFER_IS_DISCONT (buf)) {
1668       if (pad->data_pushed) {
1669         GST_LOG_OBJECT (pad->collect.pad, "got discont");
1670         packet.packetno++;
1671         /* No public API for this; hack things in */
1672         pad->map.stream.pageno++;
1673         force_flush = TRUE;
1674       } else {
1675         GST_LOG_OBJECT (pad->collect.pad, "discont at stream start");
1676       }
1677     }
1678
1679     /* flush the currently built page if necessary */
1680     if (force_flush) {
1681       GST_LOG_OBJECT (pad->collect.pad,
1682           GST_GP_FORMAT " forced flush of page before this packet",
1683           GST_BUFFER_OFFSET_END (pad->buffer));
1684       while (ogg_stream_flush (&pad->map.stream, &page)) {
1685         /* end time of this page is the timestamp of the next buffer */
1686         ogg_mux->pulling->timestamp_end = GST_BUFFER_TIMESTAMP (pad->buffer);
1687         ret = gst_ogg_mux_pad_queue_page (ogg_mux, pad, &page,
1688             pad->first_delta);
1689
1690         /* increment the page number counter */
1691         pad->pageno++;
1692         /* mark other pages as delta */
1693         pad->first_delta = TRUE;
1694       }
1695       pad->new_page = TRUE;
1696     }
1697
1698     /* if this is the first packet of a new page figure out the delta flag */
1699     if (pad->new_page) {
1700       if (delta_unit) {
1701         /* mark the page as delta */
1702         pad->first_delta = TRUE;
1703       } else {
1704         /* got a keyframe */
1705         if (ogg_mux->delta_pad == pad) {
1706           /* if we get it on the pad with deltaunits,
1707            * we mark the page as non delta */
1708           pad->first_delta = FALSE;
1709         } else if (ogg_mux->delta_pad != NULL) {
1710           /* if there are pads with delta frames, we
1711            * must mark this one as delta */
1712           pad->first_delta = TRUE;
1713         } else {
1714           pad->first_delta = FALSE;
1715         }
1716       }
1717       pad->new_page = FALSE;
1718     }
1719
1720     /* save key unit to track delta->key unit transitions */
1721     pad->prev_delta = delta_unit;
1722
1723     /* swap the packet in */
1724     if (packet.e_o_s == 1)
1725       GST_DEBUG_OBJECT (pad->collect.pad, "swapping in EOS packet");
1726     if (packet.b_o_s == 1)
1727       GST_DEBUG_OBJECT (pad->collect.pad, "swapping in BOS packet");
1728
1729     ogg_stream_packetin (&pad->map.stream, &packet);
1730     gst_buffer_unmap (buf, packet.packet, size);
1731     pad->data_pushed = TRUE;
1732
1733     gp_time = GST_BUFFER_OFFSET (pad->buffer);
1734     granulepos = GST_BUFFER_OFFSET_END (pad->buffer);
1735     timestamp = GST_BUFFER_TIMESTAMP (pad->buffer);
1736
1737     GST_LOG_OBJECT (pad->collect.pad,
1738         GST_GP_FORMAT " packet %" G_GINT64_FORMAT ", gp time %"
1739         GST_TIME_FORMAT ", timestamp %" GST_TIME_FORMAT " packetin'd",
1740         granulepos, (gint64) packet.packetno, GST_TIME_ARGS (gp_time),
1741         GST_TIME_ARGS (timestamp));
1742     /* don't need the old buffer anymore */
1743     gst_buffer_unref (pad->buffer);
1744     /* store new readahead buffer */
1745     pad->buffer = tmpbuf;
1746
1747     /* let ogg write out the pages now. The packet we got could end
1748      * up in more than one page so we need to write them all */
1749     if (ogg_stream_pageout (&pad->map.stream, &page) > 0) {
1750       /* we have a new page, so we need to timestamp it correctly.
1751        * if this fresh packet ends on this page, then the page's granulepos
1752        * comes from that packet, and we should set this buffer's timestamp */
1753
1754       GST_LOG_OBJECT (pad->collect.pad,
1755           GST_GP_FORMAT " packet %" G_GINT64_FORMAT ", time %"
1756           GST_TIME_FORMAT ") caused new page",
1757           granulepos, (gint64) packet.packetno, GST_TIME_ARGS (timestamp));
1758       GST_LOG_OBJECT (pad->collect.pad,
1759           GST_GP_FORMAT " new page %ld",
1760           GST_GP_CAST (ogg_page_granulepos (&page)), pad->map.stream.pageno);
1761
1762       if (ogg_page_granulepos (&page) == granulepos) {
1763         /* the packet we streamed in finishes on the current page,
1764          * because the page's granulepos is the granulepos of the last
1765          * packet completed on that page,
1766          * so update the timestamp that we will give to the page */
1767         GST_LOG_OBJECT (pad->collect.pad,
1768             GST_GP_FORMAT
1769             " packet finishes on current page, updating gp time to %"
1770             GST_TIME_FORMAT, granulepos, GST_TIME_ARGS (gp_time));
1771         pad->gp_time = gp_time;
1772       } else {
1773         GST_LOG_OBJECT (pad->collect.pad,
1774             GST_GP_FORMAT
1775             " packet spans beyond current page, keeping old gp time %"
1776             GST_TIME_FORMAT, granulepos, GST_TIME_ARGS (pad->gp_time));
1777       }
1778
1779       /* push the page */
1780       /* end time of this page is the timestamp of the next buffer */
1781       pad->timestamp_end = timestamp;
1782       ret = gst_ogg_mux_pad_queue_page (ogg_mux, pad, &page, pad->first_delta);
1783       pad->pageno++;
1784       /* mark next pages as delta */
1785       pad->first_delta = TRUE;
1786
1787       /* use an inner loop here to flush the remaining pages and
1788        * mark them as delta frames as well */
1789       while (ogg_stream_pageout (&pad->map.stream, &page) > 0) {
1790         if (ogg_page_granulepos (&page) == granulepos) {
1791           /* the page has taken up the new packet completely, which means
1792            * the packet ends the page and we can update the gp time
1793            * before pushing out */
1794           pad->gp_time = gp_time;
1795         }
1796
1797         /* we have a complete page now, we can push the page
1798          * and make sure to pull on a new pad the next time around */
1799         ret = gst_ogg_mux_pad_queue_page (ogg_mux, pad, &page,
1800             pad->first_delta);
1801         /* increment the page number counter */
1802         pad->pageno++;
1803       }
1804       /* need a new page as well */
1805       pad->new_page = TRUE;
1806       pad->duration = 0;
1807       /* we're done pulling on this pad, make sure to choose a new
1808        * pad for pulling in the next iteration */
1809       ogg_mux->pulling = NULL;
1810     }
1811
1812     /* Update the gp time, if necessary, since any future page will have at
1813      * least this gp time.
1814      */
1815     if (pad->gp_time < gp_time) {
1816       pad->gp_time = gp_time;
1817       GST_LOG_OBJECT (pad->collect.pad,
1818           "Updated running gp time of pad %" GST_PTR_FORMAT
1819           " to %" GST_TIME_FORMAT, pad->collect.pad, GST_TIME_ARGS (gp_time));
1820     }
1821   }
1822
1823   return ret;
1824 }
1825
1826 /* all_pads_eos:
1827  *
1828  * Checks if all pads are EOS'd by peeking.
1829  *
1830  * Returns TRUE if all pads are EOS.
1831  */
1832 static gboolean
1833 all_pads_eos (GstCollectPads2 * pads)
1834 {
1835   GSList *walk;
1836
1837   walk = pads->data;
1838   while (walk) {
1839     GstOggPadData *oggpad = (GstOggPadData *) walk->data;
1840
1841     GST_DEBUG_OBJECT (oggpad->collect.pad,
1842         "oggpad %p eos %d", oggpad, oggpad->eos);
1843
1844     if (oggpad->eos == FALSE)
1845       return FALSE;
1846
1847     walk = g_slist_next (walk);
1848   }
1849
1850   return TRUE;
1851 }
1852
1853 /* This function is called when there is data on all pads.
1854  *
1855  * It finds a pad to pull on, this is done by looking at the buffers
1856  * to decide which one to use, and using the 'oldest' one first. It then calls
1857  * gst_ogg_mux_process_best_pad() to process as much data as possible.
1858  *
1859  * If all the pads have received EOS, it flushes out all data by continually
1860  * getting the best pad and calling gst_ogg_mux_process_best_pad() until they
1861  * are all empty, and then sends EOS.
1862  */
1863 static GstFlowReturn
1864 gst_ogg_mux_collected (GstCollectPads2 * pads, GstOggMux * ogg_mux)
1865 {
1866   GstOggPadData *best;
1867   GstFlowReturn ret;
1868   gboolean popped;
1869
1870   GST_LOG_OBJECT (ogg_mux, "collected");
1871
1872   /* queue buffers on all pads; find a buffer with the lowest timestamp */
1873   best = gst_ogg_mux_queue_pads (ogg_mux, &popped);
1874
1875   if (popped)
1876     return GST_FLOW_OK;
1877
1878   if (best == NULL || best->buffer == NULL) {
1879     /* This is not supposed to happen */
1880     return GST_FLOW_ERROR;
1881   }
1882
1883   ret = gst_ogg_mux_process_best_pad (ogg_mux, best);
1884
1885   if (best->eos && all_pads_eos (pads)) {
1886     gst_pad_push_event (ogg_mux->srcpad, gst_event_new_eos ());
1887     return GST_FLOW_EOS;
1888   }
1889
1890   return ret;
1891 }
1892
1893 static void
1894 gst_ogg_mux_get_property (GObject * object,
1895     guint prop_id, GValue * value, GParamSpec * pspec)
1896 {
1897   GstOggMux *ogg_mux;
1898
1899   ogg_mux = GST_OGG_MUX (object);
1900
1901   switch (prop_id) {
1902     case ARG_MAX_DELAY:
1903       g_value_set_uint64 (value, ogg_mux->max_delay);
1904       break;
1905     case ARG_MAX_PAGE_DELAY:
1906       g_value_set_uint64 (value, ogg_mux->max_page_delay);
1907       break;
1908     case ARG_MAX_TOLERANCE:
1909       g_value_set_uint64 (value, ogg_mux->max_tolerance);
1910       break;
1911     case ARG_SKELETON:
1912       g_value_set_boolean (value, ogg_mux->use_skeleton);
1913       break;
1914     default:
1915       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1916       break;
1917   }
1918 }
1919
1920 static void
1921 gst_ogg_mux_set_property (GObject * object,
1922     guint prop_id, const GValue * value, GParamSpec * pspec)
1923 {
1924   GstOggMux *ogg_mux;
1925
1926   ogg_mux = GST_OGG_MUX (object);
1927
1928   switch (prop_id) {
1929     case ARG_MAX_DELAY:
1930       ogg_mux->max_delay = g_value_get_uint64 (value);
1931       break;
1932     case ARG_MAX_PAGE_DELAY:
1933       ogg_mux->max_page_delay = g_value_get_uint64 (value);
1934       break;
1935     case ARG_MAX_TOLERANCE:
1936       ogg_mux->max_tolerance = g_value_get_uint64 (value);
1937       break;
1938     case ARG_SKELETON:
1939       ogg_mux->use_skeleton = g_value_get_boolean (value);
1940       break;
1941     default:
1942       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1943       break;
1944   }
1945 }
1946
1947 /* reset all variables in the ogg pads. */
1948 static void
1949 gst_ogg_mux_init_collectpads (GstCollectPads2 * collect)
1950 {
1951   GSList *walk;
1952
1953   walk = collect->data;
1954   while (walk) {
1955     GstOggPadData *oggpad = (GstOggPadData *) walk->data;
1956
1957     ogg_stream_init (&oggpad->map.stream, oggpad->map.serialno);
1958     oggpad->packetno = 0;
1959     oggpad->pageno = 0;
1960     oggpad->eos = FALSE;
1961     /* we assume there will be some control data first for this pad */
1962     oggpad->state = GST_OGG_PAD_STATE_CONTROL;
1963     oggpad->new_page = TRUE;
1964     oggpad->first_delta = FALSE;
1965     oggpad->prev_delta = FALSE;
1966     oggpad->data_pushed = FALSE;
1967     oggpad->pagebuffers = g_queue_new ();
1968
1969     gst_segment_init (&oggpad->segment, GST_FORMAT_TIME);
1970
1971     walk = g_slist_next (walk);
1972   }
1973 }
1974
1975 /* Clear all buffers from the collectpads object */
1976 static void
1977 gst_ogg_mux_clear_collectpads (GstCollectPads2 * collect)
1978 {
1979   GSList *walk;
1980
1981   for (walk = collect->data; walk; walk = g_slist_next (walk)) {
1982     GstOggPadData *oggpad = (GstOggPadData *) walk->data;
1983     GstBuffer *buf;
1984
1985     ogg_stream_clear (&oggpad->map.stream);
1986
1987     while ((buf = g_queue_pop_head (oggpad->pagebuffers)) != NULL) {
1988       gst_buffer_unref (buf);
1989     }
1990     g_queue_free (oggpad->pagebuffers);
1991     oggpad->pagebuffers = NULL;
1992
1993     if (oggpad->buffer) {
1994       gst_buffer_unref (oggpad->buffer);
1995       oggpad->buffer = NULL;
1996     }
1997
1998     if (oggpad->tags) {
1999       gst_tag_list_free (oggpad->tags);
2000       oggpad->tags = NULL;
2001     }
2002
2003     gst_segment_init (&oggpad->segment, GST_FORMAT_TIME);
2004   }
2005 }
2006
2007 static GstStateChangeReturn
2008 gst_ogg_mux_change_state (GstElement * element, GstStateChange transition)
2009 {
2010   GstOggMux *ogg_mux;
2011   GstStateChangeReturn ret;
2012
2013   ogg_mux = GST_OGG_MUX (element);
2014
2015   switch (transition) {
2016     case GST_STATE_CHANGE_NULL_TO_READY:
2017       break;
2018     case GST_STATE_CHANGE_READY_TO_PAUSED:
2019       gst_ogg_mux_clear (ogg_mux);
2020       gst_ogg_mux_init_collectpads (ogg_mux->collect);
2021       gst_collect_pads2_start (ogg_mux->collect);
2022       break;
2023     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
2024       break;
2025     case GST_STATE_CHANGE_PAUSED_TO_READY:
2026       gst_collect_pads2_stop (ogg_mux->collect);
2027       break;
2028     default:
2029       break;
2030   }
2031
2032   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
2033
2034   switch (transition) {
2035     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
2036       break;
2037     case GST_STATE_CHANGE_PAUSED_TO_READY:
2038       gst_ogg_mux_clear_collectpads (ogg_mux->collect);
2039       break;
2040     case GST_STATE_CHANGE_READY_TO_NULL:
2041       break;
2042     default:
2043       break;
2044   }
2045
2046   return ret;
2047 }
2048
2049 gboolean
2050 gst_ogg_mux_plugin_init (GstPlugin * plugin)
2051 {
2052   GST_DEBUG_CATEGORY_INIT (gst_ogg_mux_debug, "oggmux", 0, "ogg muxer");
2053
2054   return gst_element_register (plugin, "oggmux", GST_RANK_PRIMARY,
2055       GST_TYPE_OGG_MUX);
2056 }