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 #ifdef HAVE_CONFIG_H

    22 #include "config.h"

    23 #endif

    24 

    25 #include <gst/gst.h>

    26 #include <gst/base/gstcollectpads.h>

    27 

    28 #include "gstoggmux.h"

    29 

    30 /* memcpy - if someone knows a way to get rid of it, please speak up

    31  * note: the ogg docs even say you need this... */

    32 #include <string.h>

    33 #include <time.h>

    34 #include <stdlib.h>             /* rand, srand, atoi */

    35 

    36 GST_DEBUG_CATEGORY_STATIC (gst_ogg_mux_debug);

    37 #define GST_CAT_DEFAULT gst_ogg_mux_debug

    38 

    39 /* This isn't generally what you'd want with an end-time macro, because

    40    technically the end time of a buffer with invalid duration is invalid. But

    41    for sorting ogg pages this is what we want. */

    42 #define GST_BUFFER_END_TIME(buf) \

    43     (GST_BUFFER_DURATION_IS_VALID (buf) \

    44     ? GST_BUFFER_TIMESTAMP (buf) + GST_BUFFER_DURATION (buf) \

    45     : GST_BUFFER_TIMESTAMP (buf))

    46 

    47 #define GST_GP_FORMAT "[gp %8" G_GINT64_FORMAT "]"

    48 

    49 typedef enum

    50 {

    51   GST_OGG_FLAG_BOS = GST_ELEMENT_FLAG_LAST,

    52   GST_OGG_FLAG_EOS

    53 }

    54 GstOggFlag;

    55 

    56 /* elementfactory information */

    57 static const GstElementDetails gst_ogg_mux_details =

    58 GST_ELEMENT_DETAILS ("Ogg muxer",

    59     "Codec/Muxer",

    60     "mux ogg streams (info about ogg: http://xiph.org)",

    61     "Wim Taymans <wim@fluendo.com>");

    62 

    63 /* OggMux signals and args */

    64 enum

    65 {

    66   /* FILL ME */

    67   LAST_SIGNAL

    68 };

    69 

    70 /* set to 0.5 seconds by default */

    71 #define DEFAULT_MAX_DELAY       G_GINT64_CONSTANT(500000000)

    72 #define DEFAULT_MAX_PAGE_DELAY  G_GINT64_CONSTANT(500000000)

    73 enum

    74 {

    75   ARG_0,

    76   ARG_MAX_DELAY,

    77   ARG_MAX_PAGE_DELAY,

    78 };

    79 

    80 static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",

    81     GST_PAD_SRC,

    82     GST_PAD_ALWAYS,

    83     GST_STATIC_CAPS ("application/ogg")

    84     );

    85 

    86 static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink_%d",

    87     GST_PAD_SINK,

    88     GST_PAD_REQUEST,

    89     GST_STATIC_CAPS ("video/x-theora; "

    90         "audio/x-vorbis; audio/x-flac; audio/x-speex; "

    91         "application/x-ogm-video; application/x-ogm-audio; video/x-dirac; "

    92         "video/x-smoke; text/x-cmml, encoded = (boolean) TRUE")

    93     );

    94 

    95 static void gst_ogg_mux_base_init (gpointer g_class);

    96 static void gst_ogg_mux_class_init (GstOggMuxClass * klass);

    97 static void gst_ogg_mux_init (GstOggMux * ogg_mux);

    98 static void gst_ogg_mux_finalize (GObject * object);

    99 

   100 static GstFlowReturn

   101 gst_ogg_mux_collected (GstCollectPads * pads, GstOggMux * ogg_mux);

   102 static gboolean gst_ogg_mux_handle_src_event (GstPad * pad, GstEvent * event);

   103 static GstPad *gst_ogg_mux_request_new_pad (GstElement * element,

   104     GstPadTemplate * templ, const gchar * name);

   105 static void gst_ogg_mux_release_pad (GstElement * element, GstPad * pad);

   106 

   107 static void gst_ogg_mux_set_property (GObject * object,

   108     guint prop_id, const GValue * value, GParamSpec * pspec);

   109 static void gst_ogg_mux_get_property (GObject * object,

   110     guint prop_id, GValue * value, GParamSpec * pspec);

   111 static GstStateChangeReturn gst_ogg_mux_change_state (GstElement * element,

   112     GstStateChange transition);

   113 

   114 static GstElementClass *parent_class = NULL;

   115 

   116 /*static guint gst_ogg_mux_signals[LAST_SIGNAL] = { 0 }; */

   117 

   118 GType

   119 gst_ogg_mux_get_type (void)

   120 {

   121   static GType ogg_mux_type = 0;

   122 

   123   if (G_UNLIKELY (ogg_mux_type == 0)) {

   124     static const GTypeInfo ogg_mux_info = {

   125       sizeof (GstOggMuxClass),

   126       gst_ogg_mux_base_init,

   127       NULL,

   128       (GClassInitFunc) gst_ogg_mux_class_init,

   129       NULL,

   130       NULL,

   131       sizeof (GstOggMux),

   132       0,

   133       (GInstanceInitFunc) gst_ogg_mux_init,

   134     };

   135 

   136     ogg_mux_type =

   137         g_type_register_static (GST_TYPE_ELEMENT, "GstOggMux", &ogg_mux_info,

   138         0);

   139   }

   140   return ogg_mux_type;

   141 }

   142 

   143 static void

   144 gst_ogg_mux_base_init (gpointer g_class)

   145 {

   146   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);

   147 

   148   gst_element_class_add_pad_template (element_class,

   149       gst_static_pad_template_get (&src_factory));

   150   gst_element_class_add_pad_template (element_class,

   151       gst_static_pad_template_get (&sink_factory));

   152 

   153   gst_element_class_set_details (element_class, &gst_ogg_mux_details);

   154 }

   155 

   156 static void

   157 gst_ogg_mux_class_init (GstOggMuxClass * klass)

   158 {

   159   GObjectClass *gobject_class;

   160   GstElementClass *gstelement_class;

   161 

   162   gobject_class = (GObjectClass *) klass;

   163   gstelement_class = (GstElementClass *) klass;

   164 

   165   parent_class = g_type_class_peek_parent (klass);

   166 

   167   gobject_class->finalize = gst_ogg_mux_finalize;

   168   gobject_class->get_property = gst_ogg_mux_get_property;

   169   gobject_class->set_property = gst_ogg_mux_set_property;

   170 

   171   gstelement_class->request_new_pad = gst_ogg_mux_request_new_pad;

   172   gstelement_class->release_pad = gst_ogg_mux_release_pad;

   173 

   174   g_object_class_install_property (gobject_class, ARG_MAX_DELAY,

   175       g_param_spec_uint64 ("max-delay", "Max delay",

   176           "Maximum delay in multiplexing streams", 0, G_MAXUINT64,

   177           DEFAULT_MAX_DELAY, (GParamFlags) G_PARAM_READWRITE));

   178   g_object_class_install_property (gobject_class, ARG_MAX_PAGE_DELAY,

   179       g_param_spec_uint64 ("max-page-delay", "Max page delay",

   180           "Maximum delay for sending out a page", 0, G_MAXUINT64,

   181           DEFAULT_MAX_PAGE_DELAY, (GParamFlags) G_PARAM_READWRITE));

   182 

   183   gstelement_class->change_state = gst_ogg_mux_change_state;

   184 

   185 }

   186 

   187 #if 0

   188 static const GstEventMask *

   189 gst_ogg_mux_get_sink_event_masks (GstPad * pad)

   190 {

   191   static const GstEventMask gst_ogg_mux_sink_event_masks[] = {

   192     {GST_EVENT_EOS, 0},

   193     {GST_EVENT_DISCONTINUOUS, 0},

   194     {0,}

   195   };

   196 

   197   return gst_ogg_mux_sink_event_masks;

   198 }

   199 #endif

   200 

   201 static void

   202 gst_ogg_mux_clear (GstOggMux * ogg_mux)

   203 {

   204   ogg_mux->pulling = NULL;

   205   ogg_mux->need_headers = TRUE;

   206   ogg_mux->max_delay = DEFAULT_MAX_DELAY;

   207   ogg_mux->max_page_delay = DEFAULT_MAX_PAGE_DELAY;

   208   ogg_mux->delta_pad = NULL;

   209   ogg_mux->offset = 0;

   210   ogg_mux->next_ts = 0;

   211   ogg_mux->last_ts = GST_CLOCK_TIME_NONE;

   212 }

   213 

   214 static void

   215 gst_ogg_mux_init (GstOggMux * ogg_mux)

   216 {

   217   GstElementClass *klass = GST_ELEMENT_GET_CLASS (ogg_mux);

   218 

   219   ogg_mux->srcpad =

   220       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,

   221           "src"), "src");

   222   gst_pad_set_event_function (ogg_mux->srcpad, gst_ogg_mux_handle_src_event);

   223   gst_element_add_pad (GST_ELEMENT (ogg_mux), ogg_mux->srcpad);

   224 

   225   GST_OBJECT_FLAG_SET (GST_ELEMENT (ogg_mux), GST_OGG_FLAG_BOS);

   226 

   227   /* seed random number generator for creation of serial numbers */

   228   srand (time (NULL));

   229 

   230   ogg_mux->collect = gst_collect_pads_new ();

   231   gst_collect_pads_set_function (ogg_mux->collect,

   232       (GstCollectPadsFunction) GST_DEBUG_FUNCPTR (gst_ogg_mux_collected),

   233       ogg_mux);

   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 (GstCollectData * data)

   255 {

   256   GstOggPad *oggpad = (GstOggPad *) data;

   257   GstBuffer *buf;

   258 

   259   ogg_stream_clear (&oggpad->stream);

   260 

   261   if (oggpad->pagebuffers) {

   262     while ((buf = g_queue_pop_head (oggpad->pagebuffers)) != NULL) {

   263       gst_buffer_unref (buf);

   264     }

   265     g_queue_free (oggpad->pagebuffers);

   266     oggpad->pagebuffers = NULL;

   267   }

   268 }

   269 

   270 static GstPadLinkReturn

   271 gst_ogg_mux_sinkconnect (GstPad * pad, GstPad * peer)

   272 {

   273   GstOggMux *ogg_mux;

   274 

   275   ogg_mux = GST_OGG_MUX (gst_pad_get_parent (pad));

   276 

   277   GST_DEBUG_OBJECT (ogg_mux, "sinkconnect triggered on %s", GST_PAD_NAME (pad));

   278 

   279   gst_object_unref (ogg_mux);

   280 

   281   return GST_PAD_LINK_OK;

   282 }

   283 

   284 static GstPad *

   285 gst_ogg_mux_request_new_pad (GstElement * element,

   286     GstPadTemplate * templ, const gchar * req_name)

   287 {

   288   GstOggMux *ogg_mux;

   289   GstPad *newpad;

   290   GstElementClass *klass;

   291 

   292   g_return_val_if_fail (templ != NULL, NULL);

   293 

   294   if (templ->direction != GST_PAD_SINK)

   295     goto wrong_direction;

   296 

   297   g_return_val_if_fail (GST_IS_OGG_MUX (element), NULL);

   298   ogg_mux = GST_OGG_MUX (element);

   299 

   300   klass = GST_ELEMENT_GET_CLASS (element);

   301 

   302   if (templ != gst_element_class_get_pad_template (klass, "sink_%d"))

   303     goto wrong_template;

   304 

   305   {

   306     gint serial;

   307     gchar *name;

   308 

   309     if (req_name == NULL || strlen (req_name) < 6) {

   310       /* no name given when requesting the pad, use random serial number */

   311       serial = rand ();

   312     } else {

   313       /* parse serial number from requested padname */

   314       serial = atoi (&req_name[5]);

   315     }

   316     /* create new pad with the name */

   317     GST_DEBUG_OBJECT (ogg_mux, "Creating new pad for serial %d", serial);

   318     name = g_strdup_printf ("sink_%d", serial);

   319     newpad = gst_pad_new_from_template (templ, name);

   320     g_free (name);

   321 

   322     /* construct our own wrapper data structure for the pad to

   323      * keep track of its status */

   324     {

   325       GstOggPad *oggpad;

   326 

   327       oggpad = (GstOggPad *)

   328           gst_collect_pads_add_pad_full (ogg_mux->collect, newpad,

   329           sizeof (GstOggPad), gst_ogg_mux_ogg_pad_destroy_notify);

   330       ogg_mux->active_pads++;

   331 

   332       oggpad->serial = serial;

   333       ogg_stream_init (&oggpad->stream, serial);

   334       oggpad->packetno = 0;

   335       oggpad->pageno = 0;

   336       oggpad->eos = FALSE;

   337       /* we assume there will be some control data first for this pad */

   338       oggpad->state = GST_OGG_PAD_STATE_CONTROL;

   339       oggpad->new_page = TRUE;

   340       oggpad->first_delta = FALSE;

   341       oggpad->prev_delta = FALSE;

   342       oggpad->pagebuffers = g_queue_new ();

   343     }

   344   }

   345 

   346   /* setup some pad functions */

   347   gst_pad_set_link_function (newpad, gst_ogg_mux_sinkconnect);

   348   /* dd the pad to the element */

   349   gst_element_add_pad (element, newpad);

   350 

   351   return newpad;

   352 

   353   /* ERRORS */

   354 wrong_direction:

   355   {

   356     g_warning ("ogg_mux: request pad that is not a SINK pad\n");

   357     return NULL;

   358   }

   359 wrong_template:

   360   {

   361     g_warning ("ogg_mux: this is not our template!\n");

   362     return NULL;

   363   }

   364 }

   365 

   366 static void

   367 gst_ogg_mux_release_pad (GstElement * element, GstPad * pad)

   368 {

   369   GstOggMux *ogg_mux;

   370 

   371   ogg_mux = GST_OGG_MUX (gst_pad_get_parent (pad));

   372 

   373   gst_collect_pads_remove_pad (ogg_mux->collect, pad);

   374   gst_element_remove_pad (element, pad);

   375 

   376   gst_object_unref (ogg_mux);

   377 }

   378 

   379 /* handle events */

   380 static gboolean

   381 gst_ogg_mux_handle_src_event (GstPad * pad, GstEvent * event)

   382 {

   383   GstEventType type;

   384 

   385   type = event ? GST_EVENT_TYPE (event) : GST_EVENT_UNKNOWN;

   386 

   387   switch (type) {

   388     case GST_EVENT_SEEK:

   389       /* disable seeking for now */

   390       return FALSE;

   391     default:

   392       break;

   393   }

   394 

   395   return gst_pad_event_default (pad, event);

   396 }

   397 

   398 static GstBuffer *

   399 gst_ogg_mux_buffer_from_page (GstOggMux * mux, ogg_page * page, gboolean delta)

   400 {

   401   GstBuffer *buffer;

   402 

   403   /* allocate space for header and body */

   404   buffer = gst_buffer_new_and_alloc (page->header_len + page->body_len);

   405   memcpy (GST_BUFFER_DATA (buffer), page->header, page->header_len);

   406   memcpy (GST_BUFFER_DATA (buffer) + page->header_len,

   407       page->body, page->body_len);

   408 

   409   /* Here we set granulepos as our OFFSET_END to give easy direct access to

   410    * this value later. Before we push it, we reset this to OFFSET + SIZE

   411    * (see gst_ogg_mux_push_buffer). */

   412   GST_BUFFER_OFFSET_END (buffer) = ogg_page_granulepos (page);

   413   if (delta)

   414     GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);

   415 

   416   GST_LOG_OBJECT (mux, GST_GP_FORMAT

   417       " created buffer %p from ogg page", ogg_page_granulepos (page), buffer);

   418 

   419   return buffer;

   420 }

   421 

   422 static GstFlowReturn

   423 gst_ogg_mux_push_buffer (GstOggMux * mux, GstBuffer * buffer)

   424 {

   425   GstCaps *caps;

   426 

   427   /* fix up OFFSET and OFFSET_END again */

   428   GST_BUFFER_OFFSET (buffer) = mux->offset;

   429   mux->offset += GST_BUFFER_SIZE (buffer);

   430   GST_BUFFER_OFFSET_END (buffer) = mux->offset;

   431 

   432   /* Ensure we have monotonically increasing timestamps in the output. */

   433   if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer)) {

   434     if (mux->last_ts != GST_CLOCK_TIME_NONE &&

   435         GST_BUFFER_TIMESTAMP (buffer) < mux->last_ts)

   436       GST_BUFFER_TIMESTAMP (buffer) = mux->last_ts;

   437     else

   438       mux->last_ts = GST_BUFFER_TIMESTAMP (buffer);

   439   }

   440 

   441   caps = gst_pad_get_negotiated_caps (mux->srcpad);

   442   gst_buffer_set_caps (buffer, caps);

   443   gst_caps_unref (caps);

   444 

   445   return gst_pad_push (mux->srcpad, buffer);

   446 }

   447 

   448 /* if all queues have at least one page, dequeue the page with the lowest

   449  * timestamp */

   450 static gboolean

   451 gst_ogg_mux_dequeue_page (GstOggMux * mux, GstFlowReturn * flowret)

   452 {

   453   GSList *walk;

   454   GstOggPad *opad = NULL;       /* "oldest" pad */

   455   GstClockTime oldest = GST_CLOCK_TIME_NONE;

   456   GstBuffer *buf = NULL;

   457   gboolean ret = FALSE;

   458 

   459   *flowret = GST_FLOW_OK;

   460 

   461   walk = mux->collect->data;

   462   while (walk) {

   463     GstOggPad *pad = (GstOggPad *) walk->data;

   464 

   465     /* We need each queue to either be at EOS, or have one or more pages

   466      * available with a set granulepos (i.e. not -1), otherwise we don't have

   467      * enough data yet to determine which stream needs to go next for correct

   468      * time ordering. */

   469     if (pad->pagebuffers->length == 0) {

   470       if (pad->eos) {

   471         GST_LOG_OBJECT (pad->collect.pad,

   472             "pad is EOS, skipping for dequeue decision");

   473       } else {

   474         GST_LOG_OBJECT (pad->collect.pad,

   475             "no pages in this queue, can't dequeue");

   476         return FALSE;

   477       }

   478     } else {

   479       /* We then need to check for a non-negative granulepos */

   480       int i;

   481       gboolean valid = FALSE;

   482 

   483       for (i = 0; i < pad->pagebuffers->length; i++) {

   484         buf = g_queue_peek_nth (pad->pagebuffers, i);

   485         /* Here we check the OFFSET_END, which is actually temporarily the

   486          * granulepos value for this buffer */

   487         if (GST_BUFFER_OFFSET_END (buf) != -1) {

   488           valid = TRUE;

   489           break;

   490         }

   491       }

   492       if (!valid) {

   493         GST_LOG_OBJECT (pad->collect.pad,

   494             "No page timestamps in queue, can't dequeue");

   495         return FALSE;

   496       }

   497     }

   498 

   499     walk = g_slist_next (walk);

   500   }

   501 

   502   walk = mux->collect->data;

   503   while (walk) {

   504     GstOggPad *pad = (GstOggPad *) walk->data;

   505 

   506     /* any page with a granulepos of -1 can be pushed immediately.

   507      * TODO: it CAN be, but it seems silly to do so? */

   508     buf = g_queue_peek_head (pad->pagebuffers);

   509     while (buf && GST_BUFFER_OFFSET_END (buf) == -1) {

   510       GST_LOG_OBJECT (pad->collect.pad, "[gp        -1] pushing page");

   511       g_queue_pop_head (pad->pagebuffers);

   512       *flowret = gst_ogg_mux_push_buffer (mux, buf);

   513       buf = g_queue_peek_head (pad->pagebuffers);

   514       ret = TRUE;

   515     }

   516 

   517     if (buf) {

   518       /* if no oldest buffer yet, take this one */

   519       if (oldest == GST_CLOCK_TIME_NONE) {

   520         GST_LOG_OBJECT (mux, "no oldest yet, taking buffer %p from pad %"

   521             GST_PTR_FORMAT " with gp time %" GST_TIME_FORMAT,

   522             buf, pad->collect.pad, GST_TIME_ARGS (GST_BUFFER_OFFSET (buf)));

   523         oldest = GST_BUFFER_OFFSET (buf);

   524         opad = pad;

   525       } else {

   526         /* if we have an oldest, compare with this one */

   527         if (GST_BUFFER_OFFSET (buf) < oldest) {

   528           GST_LOG_OBJECT (mux, "older buffer %p, taking from pad %"

   529               GST_PTR_FORMAT " with gp time %" GST_TIME_FORMAT,

   530               buf, pad->collect.pad, GST_TIME_ARGS (GST_BUFFER_OFFSET (buf)));

   531           oldest = GST_BUFFER_OFFSET (buf);

   532           opad = pad;

   533         }

   534       }

   535     }

   536     walk = g_slist_next (walk);

   537   }

   538 

   539   if (oldest != GST_CLOCK_TIME_NONE) {

   540     g_assert (opad);

   541     buf = g_queue_pop_head (opad->pagebuffers);

   542     GST_LOG_OBJECT (opad->collect.pad,

   543         GST_GP_FORMAT " pushing oldest page buffer %p (granulepos time %"

   544         GST_TIME_FORMAT ")", GST_BUFFER_OFFSET_END (buf), buf,

   545         GST_TIME_ARGS (GST_BUFFER_OFFSET (buf)));

   546     *flowret = gst_ogg_mux_push_buffer (mux, buf);

   547     ret = TRUE;

   548   }

   549 

   550   return ret;

   551 }

   552 

   553 /* put the given ogg page on a per-pad queue, timestamping it correctly.

   554  * after that, dequeue and push as many pages as possible.

   555  * Caller should make sure:

   556  * pad->timestamp     was set with the timestamp of the first packet put

   557  *                    on the page

   558  * pad->timestamp_end was set with the timestamp + duration of the last packet

   559  *                    put on the page

   560  * pad->gp_time       was set with the time matching the gp of the last

   561  *                    packet put on the page

   562  *

   563  * will also reset timestamp and timestamp_end, so caller func can restart

   564  * counting.

   565  */

   566 static GstFlowReturn

   567 gst_ogg_mux_pad_queue_page (GstOggMux * mux, GstOggPad * pad, ogg_page * page,

   568     gboolean delta)

   569 {

   570   GstFlowReturn ret;

   571   GstBuffer *buffer = gst_ogg_mux_buffer_from_page (mux, page, delta);

   572 

   573   /* take the timestamp of the first packet on this page */

   574   GST_BUFFER_TIMESTAMP (buffer) = pad->timestamp;

   575   GST_BUFFER_DURATION (buffer) = pad->timestamp_end - pad->timestamp;

   576   /* take the gp time of the last completed packet on this page */

   577   GST_BUFFER_OFFSET (buffer) = pad->gp_time;

   578 

   579   /* the next page will start where the current page's end time leaves off */

   580   pad->timestamp = pad->timestamp_end;

   581 

   582   g_queue_push_tail (pad->pagebuffers, buffer);

   583   GST_LOG_OBJECT (pad->collect.pad, GST_GP_FORMAT

   584       " queued buffer page %p (gp time %"

   585       GST_TIME_FORMAT ", timestamp %" GST_TIME_FORMAT

   586       "), %d page buffers queued", ogg_page_granulepos (page),

   587       buffer, GST_TIME_ARGS (GST_BUFFER_OFFSET (buffer)),

   588       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),

   589       g_queue_get_length (pad->pagebuffers));

   590 

   591   while (gst_ogg_mux_dequeue_page (mux, &ret)) {

   592     if (ret != GST_FLOW_OK)

   593       break;

   594   }

   595 

   596   return ret;

   597 }

   598 

   599 /*

   600  * Given two pads, compare the buffers queued on it.

   601  * Returns:

   602  *  0 if they have an equal priority

   603  * -1 if the first is better

   604  *  1 if the second is better

   605  * Priority decided by: a) validity, b) older timestamp, c) smaller number

   606  * of muxed pages

   607  */

   608 static gint

   609 gst_ogg_mux_compare_pads (GstOggMux * ogg_mux, GstOggPad * first,

   610     GstOggPad * second)

   611 {

   612   guint64 firsttime, secondtime;

   613 

   614   /* if the first pad doesn't contain anything or is even NULL, return

   615    * the second pad as best candidate and vice versa */

   616   if (first == NULL || (first->buffer == NULL && first->next_buffer == NULL))

   617     return 1;

   618   if (second == NULL || (second->buffer == NULL && second->next_buffer == NULL))

   619     return -1;

   620 

   621   /* no timestamp on first buffer, it must go first */

   622   if (first->buffer)

   623     firsttime = GST_BUFFER_TIMESTAMP (first->buffer);

   624   else

   625     firsttime = GST_BUFFER_TIMESTAMP (first->next_buffer);

   626   if (firsttime == GST_CLOCK_TIME_NONE)

   627     return -1;

   628 

   629   /* no timestamp on second buffer, it must go first */

   630   if (second->buffer)

   631     secondtime = GST_BUFFER_TIMESTAMP (second->buffer);

   632   else

   633     secondtime = GST_BUFFER_TIMESTAMP (second->next_buffer);

   634   if (secondtime == GST_CLOCK_TIME_NONE)

   635     return 1;

   636 

   637   /* first buffer has higher timestamp, second one should go first */

   638   if (secondtime < firsttime)

   639     return 1;

   640   /* second buffer has higher timestamp, first one should go first */

   641   else if (secondtime > firsttime)

   642     return -1;

   643   else {

   644     /* buffers with equal timestamps, prefer the pad that has the

   645      * least number of pages muxed */

   646     if (second->pageno < first->pageno)

   647       return 1;

   648     else if (second->pageno > first->pageno)

   649       return -1;

   650   }

   651 

   652   /* same priority if all of the above failed */

   653   return 0;

   654 }

   655 

   656 /* make sure at least one buffer is queued on all pads, two if possible

   657  * 

   658  * if pad->buffer == NULL, pad->next_buffer !=  NULL, then

   659  *   we do not know if the buffer is the last or not

   660  * if pad->buffer != NULL, pad->next_buffer != NULL, then

   661  *   pad->buffer is not the last buffer for the pad

   662  * if pad->buffer != NULL, pad->next_buffer == NULL, then

   663  *   pad->buffer if the last buffer for the pad

   664  * 

   665  * returns a pointer to an oggpad that holds the best buffer, or

   666  * NULL when no pad was usable. "best" means the buffer marked

   667  * with the lowest timestamp. If best->buffer == NULL then nothing

   668  * should be done until more data arrives */

   669 static GstOggPad *

   670 gst_ogg_mux_queue_pads (GstOggMux * ogg_mux)

   671 {

   672   GstOggPad *bestpad = NULL, *still_hungry = NULL;

   673   GSList *walk;

   674 

   675   /* try to make sure we have a buffer from each usable pad first */

   676   walk = ogg_mux->collect->data;

   677   while (walk) {

   678     GstOggPad *pad;

   679     GstCollectData *data;

   680 

   681     data = (GstCollectData *) walk->data;

   682     pad = (GstOggPad *) data;

   683 

   684     walk = g_slist_next (walk);

   685 

   686     GST_LOG_OBJECT (data->pad, "looking at pad for buffer");

   687 

   688     /* try to get a new buffer for this pad if needed and possible */

   689     if (pad->buffer == NULL) {

   690       GstBuffer *buf;

   691       gboolean incaps;

   692 

   693       /* shift the buffer along if needed (it's okay if next_buffer is NULL) */

   694       if (pad->buffer == NULL) {

   695         GST_LOG_OBJECT (data->pad, "shifting buffer %" GST_PTR_FORMAT,

   696             pad->next_buffer);

   697         pad->buffer = pad->next_buffer;

   698         pad->next_buffer = NULL;

   699       }

   700 

   701       buf = gst_collect_pads_pop (ogg_mux->collect, data);

   702       GST_LOG_OBJECT (data->pad, "popped buffer %" GST_PTR_FORMAT, buf);

   703 

   704       /* On EOS we get a NULL buffer */

   705       if (buf != NULL) {

   706         if (ogg_mux->delta_pad == NULL &&

   707             GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT))

   708           ogg_mux->delta_pad = pad;

   709 

   710         incaps = GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_IN_CAPS);

   711         /* if we need headers */

   712         if (pad->state == GST_OGG_PAD_STATE_CONTROL) {

   713           /* and we have one */

   714           if (incaps) {

   715             GST_DEBUG_OBJECT (ogg_mux,

   716                 "got incaps buffer in control state, ignoring");

   717             /* just ignore */

   718             gst_buffer_unref (buf);

   719             buf = NULL;

   720           } else {

   721             GST_DEBUG_OBJECT (ogg_mux,

   722                 "got data buffer in control state, switching " "to data mode");

   723             /* this is a data buffer so switch to data state */

   724             pad->state = GST_OGG_PAD_STATE_DATA;

   725           }

   726         }

   727       } else {

   728         GST_DEBUG_OBJECT (data->pad, "EOS on pad");

   729         if (!pad->eos) {

   730           ogg_page page;

   731           GstFlowReturn ret;

   732 

   733           /* it's no longer active */

   734           ogg_mux->active_pads--;

   735 

   736           /* Just gone to EOS. Flush existing page(s) */

   737           pad->eos = TRUE;

   738 

   739           while (ogg_stream_flush (&pad->stream, &page)) {

   740             /* Place page into the per-pad queue */

   741             ret = gst_ogg_mux_pad_queue_page (ogg_mux, pad, &page,

   742                 pad->first_delta);

   743             /* increment the page number counter */

   744             pad->pageno++;

   745             /* mark other pages as delta */

   746             pad->first_delta = TRUE;

   747           }

   748         }

   749       }

   750 

   751       pad->next_buffer = buf;

   752     }

   753 

   754     /* we should have a buffer now, see if it is the best pad to

   755      * pull on */

   756     if (pad->buffer || pad->next_buffer) {

   757       if (gst_ogg_mux_compare_pads (ogg_mux, bestpad, pad) > 0) {

   758         GST_LOG_OBJECT (data->pad,

   759             "new best pad, with buffers %" GST_PTR_FORMAT

   760             " and %" GST_PTR_FORMAT, pad->buffer, pad->next_buffer);

   761 

   762         bestpad = pad;

   763       }

   764     } else if (!pad->eos) {

   765       GST_LOG_OBJECT (data->pad, "hungry pad");

   766       still_hungry = pad;

   767     }

   768   }

   769 

   770   if (still_hungry)

   771     /* drop back into collectpads... */

   772     return still_hungry;

   773   else

   774     return bestpad;

   775 }

   776 

   777 static GList *

   778 gst_ogg_mux_get_headers (GstOggPad * pad)

   779 {

   780   GList *res = NULL;

   781   GstOggMux *ogg_mux;

   782   GstStructure *structure;

   783   GstCaps *caps;

   784   GstPad *thepad;

   785 

   786   thepad = pad->collect.pad;

   787 

   788   ogg_mux = GST_OGG_MUX (GST_PAD_PARENT (thepad));

   789 

   790   GST_LOG_OBJECT (thepad, "getting headers");

   791 

   792   caps = gst_pad_get_negotiated_caps (thepad);

   793   if (caps != NULL) {

   794     const GValue *streamheader;

   795 

   796     structure = gst_caps_get_structure (caps, 0);

   797     if (strcmp (gst_structure_get_name (structure), "video/x-dirac") == 0) {

   798       GstBuffer *buf = gst_buffer_new_and_alloc (16);

   799       int fps_n = 12;

   800       int fps_d = 1;

   801 

   802       gst_structure_get_fraction (structure, "framerate", &fps_n, &fps_d);

   803 

   804       memcpy (GST_BUFFER_DATA (buf), "KW-DIRAC", 8);

   805       GST_WRITE_UINT32_BE (GST_BUFFER_DATA (buf) + 8, fps_n);

   806       GST_WRITE_UINT32_BE (GST_BUFFER_DATA (buf) + 12, fps_d);

   807 

   808       res = g_list_append (res, buf);

   809 

   810       //res = g_list_append (res, gst_buffer_ref(pad->buffer));

   811     } else {

   812       streamheader = gst_structure_get_value (structure, "streamheader");

   813       if (streamheader != NULL) {

   814         GST_LOG_OBJECT (thepad, "got header");

   815         if (G_VALUE_TYPE (streamheader) == GST_TYPE_ARRAY) {

   816           GArray *bufarr = g_value_peek_pointer (streamheader);

   817           gint i;

   818 

   819           GST_LOG_OBJECT (thepad, "got fixed list");

   820 

   821           for (i = 0; i < bufarr->len; i++) {

   822             GValue *bufval = &g_array_index (bufarr, GValue, i);

   823 

   824             GST_LOG_OBJECT (thepad, "item %d", i);

   825             if (G_VALUE_TYPE (bufval) == GST_TYPE_BUFFER) {

   826               GstBuffer *buf = g_value_peek_pointer (bufval);

   827 

   828               GST_LOG_OBJECT (thepad, "adding item %d to header list", i);

   829 

   830               gst_buffer_ref (buf);

   831               res = g_list_append (res, buf);

   832             }

   833           }

   834         } else {

   835           GST_LOG_OBJECT (thepad, "streamheader is not fixed list");

   836         }

   837       } else {

   838         GST_LOG_OBJECT (thepad, "caps don't have streamheader");

   839       }

   840     }

   841     gst_caps_unref (caps);

   842   } else {

   843     GST_LOG_OBJECT (thepad, "got empty caps as negotiated format");

   844   }

   845   return res;

   846 }

   847 

   848 static GstCaps *

   849 gst_ogg_mux_set_header_on_caps (GstCaps * caps, GList * buffers)

   850 {

   851   GstStructure *structure;

   852   GValue array = { 0 };

   853   GList *walk = buffers;

   854 

   855   caps = gst_caps_make_writable (caps);

   856 

   857   structure = gst_caps_get_structure (caps, 0);

   858 

   859   /* put buffers in a fixed list */

   860   g_value_init (&array, GST_TYPE_ARRAY);

   861 

   862   while (walk) {

   863     GstBuffer *buf = GST_BUFFER (walk->data);

   864     GstBuffer *copy;

   865     GValue value = { 0 };

   866 

   867     walk = walk->next;

   868 

   869     /* mark buffer */

   870     GST_LOG ("Setting IN_CAPS on buffer of length %d", GST_BUFFER_SIZE (buf));

   871     GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_IN_CAPS);

   872 

   873     g_value_init (&value, GST_TYPE_BUFFER);

   874     copy = gst_buffer_copy (buf);

   875     gst_value_set_buffer (&value, copy);

   876     gst_buffer_unref (copy);

   877     gst_value_array_append_value (&array, &value);

   878     g_value_unset (&value);

   879   }

   880   gst_structure_set_value (structure, "streamheader", &array);

   881   g_value_unset (&array);

   882 

   883   return caps;

   884 }

   885 

   886 /*

   887  * For each pad we need to write out one (small) header in one

   888  * page that allows decoders to identify the type of the stream.

   889  * After that we need to write out all extra info for the decoders.

   890  * In the case of a codec that also needs data as configuration, we can

   891  * find that info in the streamcaps. 

   892  * After writing the headers we must start a new page for the data.

   893  */

   894 static GstFlowReturn

   895 gst_ogg_mux_send_headers (GstOggMux * mux)

   896 {

   897   GSList *walk;

   898   GList *hbufs, *hwalk;

   899   GstCaps *caps;

   900   GstFlowReturn ret;

   901 

   902   hbufs = NULL;

   903   ret = GST_FLOW_OK;

   904 

   905   GST_LOG_OBJECT (mux, "collecting headers");

   906 

   907   walk = mux->collect->data;

   908   while (walk) {

   909     GstOggPad *pad;

   910     GstPad *thepad;

   911 

   912     pad = (GstOggPad *) walk->data;

   913     thepad = pad->collect.pad;

   914 

   915     walk = g_slist_next (walk);

   916 

   917     GST_LOG_OBJECT (mux, "looking at pad %s:%s", GST_DEBUG_PAD_NAME (thepad));

   918 

   919     /* if the pad has no buffer, we don't care */

   920     if (pad->buffer == NULL && pad->next_buffer == NULL)

   921       continue;

   922 

   923     /* now figure out the headers */

   924     pad->headers = gst_ogg_mux_get_headers (pad);

   925   }

   926 

   927   GST_LOG_OBJECT (mux, "creating BOS pages");

   928   walk = mux->collect->data;

   929   while (walk) {

   930     GstOggPad *pad;

   931     GstBuffer *buf;

   932     ogg_packet packet;

   933     ogg_page page;

   934     GstPad *thepad;

   935     GstCaps *caps;

   936     GstStructure *structure;

   937     GstBuffer *hbuf;

   938 

   939     pad = (GstOggPad *) walk->data;

   940     thepad = pad->collect.pad;

   941     caps = gst_pad_get_negotiated_caps (thepad);

   942     structure = gst_caps_get_structure (caps, 0);

   943 

   944     walk = walk->next;

   945 

   946     pad->packetno = 0;

   947 

   948     GST_LOG_OBJECT (thepad, "looping over headers");

   949 

   950     if (pad->headers) {

   951       buf = GST_BUFFER (pad->headers->data);

   952       pad->headers = g_list_remove (pad->headers, buf);

   953     } else if (pad->buffer) {

   954       buf = pad->buffer;

   955       gst_buffer_ref (buf);

   956     } else if (pad->next_buffer) {

   957       buf = pad->next_buffer;

   958       gst_buffer_ref (buf);

   959     } else {

   960       /* fixme -- should be caught in the previous list traversal. */

   961       GST_OBJECT_LOCK (pad);

   962       g_critical ("No headers or buffers on pad %s:%s",

   963           GST_DEBUG_PAD_NAME (pad));

   964       GST_OBJECT_UNLOCK (pad);

   965       continue;

   966     }

   967 

   968     /* create a packet from the buffer */

   969     packet.packet = GST_BUFFER_DATA (buf);

   970     packet.bytes = GST_BUFFER_SIZE (buf);

   971     packet.granulepos = GST_BUFFER_OFFSET_END (buf);

   972     if (packet.granulepos == -1)

   973       packet.granulepos = 0;

   974     /* mark BOS and packet number */

   975     packet.b_o_s = (pad->packetno == 0);

   976     packet.packetno = pad->packetno++;

   977     /* mark EOS */

   978     packet.e_o_s = 0;

   979 

   980     /* swap the packet in */

   981     ogg_stream_packetin (&pad->stream, &packet);

   982     gst_buffer_unref (buf);

   983 

   984     GST_LOG_OBJECT (thepad, "flushing out BOS page");

   985     if (!ogg_stream_flush (&pad->stream, &page))

   986       g_critical ("Could not flush BOS page");

   987 

   988     hbuf = gst_ogg_mux_buffer_from_page (mux, &page, FALSE);

   989 

   990     GST_LOG_OBJECT (mux, "swapped out page with mime type %s",

   991         gst_structure_get_name (structure));

   992 

   993     /* quick hack: put Theora and Dirac video pages at the front.

   994      * Ideally, we would have a settable enum for which Ogg

   995      * profile we work with, and order based on that.

   996      * (FIXME: if there is more than one video stream, shouldn't we only put

   997      * one's BOS into the first page, followed by an audio stream's BOS, and

   998      * only then followed by the remaining video and audio streams?) */

   999     if (gst_structure_has_name (structure, "video/x-theora")) {

  1000       GST_DEBUG_OBJECT (thepad, "putting %s page at the front", "Theora");

  1001       hbufs = g_list_prepend (hbufs, hbuf);

  1002     } else if (gst_structure_has_name (structure, "video/x-dirac")) {

  1003       GST_DEBUG_OBJECT (thepad, "putting %s page at the front", "Dirac");

  1004       hbufs = g_list_prepend (hbufs, hbuf);

  1005     } else {

  1006       hbufs = g_list_append (hbufs, hbuf);

  1007     }

  1008     gst_caps_unref (caps);

  1009   }

  1010 

  1011   GST_LOG_OBJECT (mux, "creating next headers");

  1012   walk = mux->collect->data;

  1013   while (walk) {

  1014     GstOggPad *pad;

  1015     GstPad *thepad;

  1016 

  1017     pad = (GstOggPad *) walk->data;

  1018     thepad = pad->collect.pad;

  1019 

  1020     walk = walk->next;

  1021 

  1022     GST_LOG_OBJECT (mux, "looping over headers for pad %s:%s",

  1023         GST_DEBUG_PAD_NAME (thepad));

  1024 

  1025     hwalk = pad->headers;

  1026     while (hwalk) {

  1027       GstBuffer *buf = GST_BUFFER (hwalk->data);

  1028       ogg_packet packet;

  1029       ogg_page page;

  1030 

  1031       hwalk = hwalk->next;

  1032 

  1033       /* create a packet from the buffer */

  1034       packet.packet = GST_BUFFER_DATA (buf);

  1035       packet.bytes = GST_BUFFER_SIZE (buf);

  1036       packet.granulepos = GST_BUFFER_OFFSET_END (buf);

  1037       if (packet.granulepos == -1)

  1038         packet.granulepos = 0;

  1039       /* mark BOS and packet number */

  1040       packet.b_o_s = (pad->packetno == 0);

  1041       packet.packetno = pad->packetno++;

  1042       /* mark EOS */

  1043       packet.e_o_s = 0;

  1044 

  1045       /* swap the packet in */

  1046       ogg_stream_packetin (&pad->stream, &packet);

  1047       gst_buffer_unref (buf);

  1048 

  1049       /* if last header, flush page */

  1050       if (hwalk == NULL) {

  1051         GST_LOG_OBJECT (mux,

  1052             "flushing page as packet %" G_GUINT64_FORMAT " is first or "

  1053             "last packet", pad->packetno);

  1054         while (ogg_stream_flush (&pad->stream, &page)) {

  1055           GstBuffer *hbuf = gst_ogg_mux_buffer_from_page (mux, &page, FALSE);

  1056 

  1057           GST_LOG_OBJECT (mux, "swapped out page");

  1058           hbufs = g_list_append (hbufs, hbuf);

  1059         }

  1060       } else {

  1061         GST_LOG_OBJECT (mux, "try to swap out page");

  1062         /* just try to swap out a page then */

  1063         while (ogg_stream_pageout (&pad->stream, &page) > 0) {

  1064           GstBuffer *hbuf = gst_ogg_mux_buffer_from_page (mux, &page, FALSE);

  1065 

  1066           GST_LOG_OBJECT (mux, "swapped out page");

  1067           hbufs = g_list_append (hbufs, hbuf);

  1068         }

  1069       }

  1070     }

  1071     g_list_free (pad->headers);

  1072     pad->headers = NULL;

  1073   }

  1074   /* hbufs holds all buffers for the headers now */

  1075 

  1076   /* create caps with the buffers */

  1077   caps = gst_pad_get_caps (mux->srcpad);

  1078   if (caps) {

  1079     caps = gst_ogg_mux_set_header_on_caps (caps, hbufs);

  1080     gst_pad_set_caps (mux->srcpad, caps);

  1081     gst_caps_unref (caps);

  1082   }

  1083   /* and send the buffers */

  1084   hwalk = hbufs;

  1085   while (hwalk) {

  1086     GstBuffer *buf = GST_BUFFER (hwalk->data);

  1087 

  1088     hwalk = hwalk->next;

  1089 

  1090     if ((ret = gst_ogg_mux_push_buffer (mux, buf)) != GST_FLOW_OK)

  1091       break;

  1092   }

  1093   g_list_free (hbufs);

  1094 

  1095   return ret;

  1096 }

  1097 

  1098 /* this function is called to process data on the best pending pad.

  1099  *

  1100  * basic idea:

  1101  *

  1102  * 1) store the selected pad and keep on pulling until we fill a

  1103  *    complete ogg page or the ogg page is filled above the max-delay

  1104  *    threshold. This is needed because the ogg spec says that

  1105  *    you should fill a complete page with data from the same logical

  1106  *    stream. When the page is filled, go back to 1).

  1107  * 2) before filling a page, read ahead one more buffer to see if this

  1108  *    packet is the last of the stream. We need to do this because the ogg

  1109  *    spec mandates that the last packet should have the EOS flag set before

  1110  *    sending it to ogg. if pad->buffer is NULL we need to wait to find out

  1111  *    whether there are any more buffers.

  1112  * 3) pages get queued on a per-pad queue. Every time a page is queued, a

  1113  *    dequeue is called, which will dequeue the oldest page on any pad, provided

  1114  *    that ALL pads have at least one marked page in the queue (or remaining

  1115  *    pads are at EOS)

  1116  */

  1117 static GstFlowReturn

  1118 gst_ogg_mux_process_best_pad (GstOggMux * ogg_mux, GstOggPad * best)

  1119 {

  1120   gboolean delta_unit;

  1121   GstFlowReturn ret;

  1122   gint64 granulepos = 0;

  1123   GstClockTime timestamp, gp_time;

  1124 

  1125   GST_LOG_OBJECT (ogg_mux, "best pad %" GST_PTR_FORMAT

  1126       ", currently pulling from %" GST_PTR_FORMAT, best->collect.pad,

  1127       ogg_mux->pulling);

  1128 

  1129   /* best->buffer is non-NULL, either the pad is EOS's or there is a next 

  1130    * buffer */

  1131   if (best->next_buffer == NULL && !best->eos) {

  1132     GST_WARNING_OBJECT (ogg_mux, "no subsequent buffer and EOS not reached");

  1133     return GST_FLOW_WRONG_STATE;

  1134   }