gst/multifile/gstsplitmuxsink.c

   1 /* GStreamer Muxer bin that splits output stream by size/time
   2  * Copyright (C) <2014> Jan Schmidt <jan@centricular.com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19
  20 /**
  21  * SECTION:element-splitmuxsink
  22  * @short_description: Muxer wrapper for splitting output stream by size or time
  23  *
  24  * This element wraps a muxer and a sink, and starts a new file when the mux
  25  * contents are about to cross a threshold of maximum size of maximum time,
  26  * splitting at video keyframe boundaries. Exactly one input video stream
  27  * is required, with as many accompanying audio and subtitle streams as
  28  * desired.
  29  *
  30  * By default, it uses mp4mux and filesink, but they can be changed via
  31  * the 'muxer' and 'sink' properties.
  32  *
  33  * The minimum file size is 1 GOP, however - so limits may be overrun if the
  34  * distance between any 2 keyframes is larger than the limits.
  35  *
  36  * The splitting process is driven by the video stream contents, and
  37  * the video stream must contain closed GOPs for the output file parts
  38  * to be played individually correctly.
  39  *
  40  * <refsect2>
  41  * <title>Example pipelines</title>
  42  * |[
  43  * gst-launch-1.0 -e v4l2src num-buffers=500 ! video/x-raw,width=320,height=240 ! videoconvert ! queue ! timeoverlay ! x264enc max-key-int=10 ! h264parse ! splitmuxsink location=video%02d.mov max-size-time=10000000000 max-size-bytes=1000000
  44  * ]|
  45  *
  46  * Records a video stream captured from a v4l2 device and muxes it into
  47  * ISO mp4 files, splitting as needed to limit size/duration to 10 seconds
  48  * and 1MB maximum size.
  49  *
  50  * </refsect2>
  51  */
  52
  53 #ifdef HAVE_CONFIG_H
  54 #include "config.h"
  55 #endif
  56
  57 #include <string.h>
  58 #include "gstsplitmuxsink.h"
  59
  60 GST_DEBUG_CATEGORY_STATIC (splitmux_debug);
  61 #define GST_CAT_DEFAULT splitmux_debug
  62
  63 #define GST_SPLITMUX_LOCK(s) g_mutex_lock(&(s)->lock)
  64 #define GST_SPLITMUX_UNLOCK(s) g_mutex_unlock(&(s)->lock)
  65 #define GST_SPLITMUX_WAIT(s) g_cond_wait (&(s)->data_cond, &(s)->lock)
  66 #define GST_SPLITMUX_BROADCAST(s) g_cond_broadcast (&(s)->data_cond)
  67
  68 enum
  69 {
  70   PROP_0,
  71   PROP_LOCATION,
  72   PROP_MAX_SIZE_TIME,
  73   PROP_MAX_SIZE_BYTES,
  74   PROP_MUXER_OVERHEAD,
  75   PROP_MUXER,
  76   PROP_SINK
  77 };
  78
  79 #define DEFAULT_MAX_SIZE_TIME       0
  80 #define DEFAULT_MAX_SIZE_BYTES      0
  81 #define DEFAULT_MUXER_OVERHEAD      0.02
  82 #define DEFAULT_MUXER "mp4mux"
  83 #define DEFAULT_SINK "filesink"
  84
  85 static GstStaticPadTemplate video_sink_template =
  86 GST_STATIC_PAD_TEMPLATE ("video",
  87     GST_PAD_SINK,
  88     GST_PAD_REQUEST,
  89     GST_STATIC_CAPS_ANY);
  90 static GstStaticPadTemplate audio_sink_template =
  91 GST_STATIC_PAD_TEMPLATE ("audio_%u",
  92     GST_PAD_SINK,
  93     GST_PAD_REQUEST,
  94     GST_STATIC_CAPS_ANY);
  95 static GstStaticPadTemplate subtitle_sink_template =
  96 GST_STATIC_PAD_TEMPLATE ("subtitle_%u",
  97     GST_PAD_SINK,
  98     GST_PAD_REQUEST,
  99     GST_STATIC_CAPS_ANY);
 100
 101 static GQuark PAD_CONTEXT;
 102
 103 static void
 104 _do_init (void)
 105 {
 106   PAD_CONTEXT = g_quark_from_static_string ("pad-context");
 107 }
 108
 109 #define gst_splitmux_sink_parent_class parent_class
 110 G_DEFINE_TYPE_EXTENDED (GstSplitMuxSink, gst_splitmux_sink, GST_TYPE_BIN, 0,
 111     _do_init ());
 112
 113 static gboolean create_elements (GstSplitMuxSink * splitmux);
 114 static gboolean create_sink (GstSplitMuxSink * splitmux);
 115 static void gst_splitmux_sink_set_property (GObject * object, guint prop_id,
 116     const GValue * value, GParamSpec * pspec);
 117 static void gst_splitmux_sink_get_property (GObject * object, guint prop_id,
 118     GValue * value, GParamSpec * pspec);
 119 static void gst_splitmux_sink_dispose (GObject * object);
 120 static void gst_splitmux_sink_finalize (GObject * object);
 121
 122 static GstPad *gst_splitmux_sink_request_new_pad (GstElement * element,
 123     GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
 124 static void gst_splitmux_sink_release_pad (GstElement * element, GstPad * pad);
 125
 126 static GstStateChangeReturn gst_splitmux_sink_change_state (GstElement *
 127     element, GstStateChange transition);
 128
 129 static void bus_handler (GstBin * bin, GstMessage * msg);
 130 static void set_next_filename (GstSplitMuxSink * splitmux);
 131 static void start_next_fragment (GstSplitMuxSink * splitmux);
 132 static void check_queue_length (GstSplitMuxSink * splitmux, MqStreamCtx * ctx);
 133 static void mq_stream_ctx_unref (MqStreamCtx * ctx);
 134
 135 static MqStreamBuf *
 136 mq_stream_buf_new (void)
 137 {
 138   return g_slice_new0 (MqStreamBuf);
 139 }
 140
 141 static void
 142 mq_stream_buf_free (MqStreamBuf * data)
 143 {
 144   g_slice_free (MqStreamBuf, data);
 145 }
 146
 147 static void
 148 gst_splitmux_sink_class_init (GstSplitMuxSinkClass * klass)
 149 {
 150   GObjectClass *gobject_class = (GObjectClass *) klass;
 151   GstElementClass *gstelement_class = (GstElementClass *) klass;
 152   GstBinClass *gstbin_class = (GstBinClass *) klass;
 153
 154   gobject_class->set_property = gst_splitmux_sink_set_property;
 155   gobject_class->get_property = gst_splitmux_sink_get_property;
 156   gobject_class->dispose = gst_splitmux_sink_dispose;
 157   gobject_class->finalize = gst_splitmux_sink_finalize;
 158
 159   gst_element_class_set_static_metadata (gstelement_class,
 160       "Split Muxing Bin", "Generic/Bin/Muxer",
 161       "Convenience bin that muxes incoming streams into multiple time/size limited files",
 162       "Jan Schmidt <jan@centricular.com>");
 163
 164   gst_element_class_add_pad_template (gstelement_class,
 165       gst_static_pad_template_get (&video_sink_template));
 166   gst_element_class_add_pad_template (gstelement_class,
 167       gst_static_pad_template_get (&audio_sink_template));
 168   gst_element_class_add_pad_template (gstelement_class,
 169       gst_static_pad_template_get (&subtitle_sink_template));
 170
 171   gstelement_class->change_state =
 172       GST_DEBUG_FUNCPTR (gst_splitmux_sink_change_state);
 173   gstelement_class->request_new_pad =
 174       GST_DEBUG_FUNCPTR (gst_splitmux_sink_request_new_pad);
 175   gstelement_class->release_pad =
 176       GST_DEBUG_FUNCPTR (gst_splitmux_sink_release_pad);
 177
 178   gstbin_class->handle_message = bus_handler;
 179
 180   g_object_class_install_property (gobject_class, PROP_LOCATION,
 181       g_param_spec_string ("location", "File Output Pattern",
 182           "Format string pattern for the location of the files to write (e.g. video%05d.mp4)",
 183           NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 184   g_object_class_install_property (gobject_class, PROP_MUXER_OVERHEAD,
 185       g_param_spec_double ("mux-overhead", "Muxing Overhead",
 186           "Extra size overhead of muxing (0.02 = 2%)", 0.0, 1.0,
 187           DEFAULT_MUXER_OVERHEAD,
 188           G_PARAM_READWRITE | GST_PARAM_CONTROLLABLE | G_PARAM_STATIC_STRINGS));
 189
 190   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
 191       g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
 192           "Max. amount of time per file (in ns, 0=disable)", 0, G_MAXUINT64,
 193           DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 194   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BYTES,
 195       g_param_spec_uint64 ("max-size-bytes", "Max. size bytes",
 196           "Max. amount of data per file (in bytes, 0=disable)", 0, G_MAXUINT64,
 197           DEFAULT_MAX_SIZE_BYTES, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 198
 199   g_object_class_install_property (gobject_class, PROP_MUXER,
 200       g_param_spec_object ("muxer", "Muxer",
 201           "The muxer element to use (NULL = default mp4mux)",
 202           GST_TYPE_ELEMENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 203   g_object_class_install_property (gobject_class, PROP_SINK,
 204       g_param_spec_object ("sink", "Sink",
 205           "The sink element (or element chain) to use (NULL = default filesink)",
 206           GST_TYPE_ELEMENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 207 }
 208
 209 static void
 210 gst_splitmux_sink_init (GstSplitMuxSink * splitmux)
 211 {
 212   g_mutex_init (&splitmux->lock);
 213   g_cond_init (&splitmux->data_cond);
 214
 215   splitmux->mux_overhead = DEFAULT_MUXER_OVERHEAD;
 216   splitmux->threshold_time = DEFAULT_MAX_SIZE_TIME;
 217   splitmux->threshold_bytes = DEFAULT_MAX_SIZE_BYTES;
 218 }
 219
 220 static void
 221 gst_splitmux_reset (GstSplitMuxSink * splitmux)
 222 {
 223   if (splitmux->mq)
 224     gst_bin_remove (GST_BIN (splitmux), splitmux->mq);
 225   if (splitmux->muxer)
 226     gst_bin_remove (GST_BIN (splitmux), splitmux->muxer);
 227   if (splitmux->active_sink)
 228     gst_bin_remove (GST_BIN (splitmux), splitmux->active_sink);
 229
 230   splitmux->sink = splitmux->active_sink = splitmux->muxer = splitmux->mq =
 231       NULL;
 232 }
 233
 234 static void
 235 gst_splitmux_sink_dispose (GObject * object)
 236 {
 237   GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
 238
 239   G_OBJECT_CLASS (parent_class)->dispose (object);
 240
 241   /* Calling parent dispose invalidates all child pointers */
 242   splitmux->sink = splitmux->active_sink = splitmux->muxer = splitmux->mq =
 243       NULL;
 244 }
 245
 246 static void
 247 gst_splitmux_sink_finalize (GObject * object)
 248 {
 249   GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
 250   g_cond_clear (&splitmux->data_cond);
 251   if (splitmux->provided_sink)
 252     gst_object_unref (splitmux->provided_sink);
 253   if (splitmux->provided_muxer)
 254     gst_object_unref (splitmux->provided_muxer);
 255
 256   g_free (splitmux->location);
 257
 258   /* Make sure to free any un-released contexts */
 259   g_list_foreach (splitmux->contexts, (GFunc) mq_stream_ctx_unref, NULL);
 260   g_list_free (splitmux->contexts);
 261
 262   G_OBJECT_CLASS (parent_class)->finalize (object);
 263 }
 264
 265 static void
 266 gst_splitmux_sink_set_property (GObject * object, guint prop_id,
 267     const GValue * value, GParamSpec * pspec)
 268 {
 269   GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
 270
 271   switch (prop_id) {
 272     case PROP_LOCATION:{
 273       GST_OBJECT_LOCK (splitmux);
 274       g_free (splitmux->location);
 275       splitmux->location = g_value_dup_string (value);
 276       GST_OBJECT_UNLOCK (splitmux);
 277       break;
 278     }
 279     case PROP_MAX_SIZE_BYTES:
 280       GST_OBJECT_LOCK (splitmux);
 281       splitmux->threshold_bytes = g_value_get_uint64 (value);
 282       GST_OBJECT_UNLOCK (splitmux);
 283       break;
 284     case PROP_MAX_SIZE_TIME:
 285       GST_OBJECT_LOCK (splitmux);
 286       splitmux->threshold_time = g_value_get_uint64 (value);
 287       GST_OBJECT_UNLOCK (splitmux);
 288       break;
 289     case PROP_MUXER_OVERHEAD:
 290       GST_OBJECT_LOCK (splitmux);
 291       splitmux->mux_overhead = g_value_get_double (value);
 292       GST_OBJECT_UNLOCK (splitmux);
 293       break;
 294     case PROP_SINK:
 295       GST_SPLITMUX_LOCK (splitmux);
 296       if (splitmux->provided_sink)
 297         gst_object_unref (splitmux->provided_sink);
 298       splitmux->provided_sink = g_value_dup_object (value);
 299       GST_SPLITMUX_UNLOCK (splitmux);
 300       break;
 301     case PROP_MUXER:
 302       GST_SPLITMUX_LOCK (splitmux);
 303       if (splitmux->provided_muxer)
 304         gst_object_unref (splitmux->provided_muxer);
 305       splitmux->provided_muxer = g_value_dup_object (value);
 306       GST_SPLITMUX_UNLOCK (splitmux);
 307       break;
 308     default:
 309       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 310       break;
 311   }
 312 }
 313
 314 static void
 315 gst_splitmux_sink_get_property (GObject * object, guint prop_id,
 316     GValue * value, GParamSpec * pspec)
 317 {
 318   GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
 319
 320   switch (prop_id) {
 321     case PROP_LOCATION:
 322       GST_OBJECT_LOCK (splitmux);
 323       g_value_set_string (value, splitmux->location);
 324       GST_OBJECT_UNLOCK (splitmux);
 325       break;
 326     case PROP_MAX_SIZE_BYTES:
 327       GST_OBJECT_LOCK (splitmux);
 328       g_value_set_uint64 (value, splitmux->threshold_bytes);
 329       GST_OBJECT_UNLOCK (splitmux);
 330       break;
 331     case PROP_MAX_SIZE_TIME:
 332       GST_OBJECT_LOCK (splitmux);
 333       g_value_set_uint64 (value, splitmux->threshold_time);
 334       GST_OBJECT_UNLOCK (splitmux);
 335       break;
 336     case PROP_MUXER_OVERHEAD:
 337       GST_OBJECT_LOCK (splitmux);
 338       g_value_set_double (value, splitmux->mux_overhead);
 339       GST_OBJECT_UNLOCK (splitmux);
 340       break;
 341     case PROP_SINK:
 342       GST_SPLITMUX_LOCK (splitmux);
 343       g_value_set_object (value, splitmux->provided_sink);
 344       GST_SPLITMUX_UNLOCK (splitmux);
 345       break;
 346     case PROP_MUXER:
 347       GST_SPLITMUX_LOCK (splitmux);
 348       g_value_set_object (value, splitmux->provided_muxer);
 349       GST_SPLITMUX_UNLOCK (splitmux);
 350       break;
 351     default:
 352       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 353       break;
 354   }
 355 }
 356
 357 static GstPad *
 358 mq_sink_to_src (GstElement * mq, GstPad * sink_pad)
 359 {
 360   gchar *tmp, *sinkname, *srcname;
 361   GstPad *mq_src;
 362
 363   sinkname = gst_pad_get_name (sink_pad);
 364   tmp = sinkname + 5;
 365   srcname = g_strdup_printf ("src_%s", tmp);
 366
 367   mq_src = gst_element_get_static_pad (mq, srcname);
 368
 369   g_free (sinkname);
 370   g_free (srcname);
 371
 372   return mq_src;
 373 }
 374
 375 static gboolean
 376 get_pads_from_mq (GstSplitMuxSink * splitmux, GstPad ** sink_pad,
 377     GstPad ** src_pad)
 378 {
 379   GstPad *mq_sink;
 380   GstPad *mq_src;
 381
 382   /* Request a pad from multiqueue, then connect this one, then
 383    * discover the corresponding output pad and return both */
 384   mq_sink = gst_element_get_request_pad (splitmux->mq, "sink_%u");
 385   if (mq_sink == NULL)
 386     return FALSE;
 387
 388   mq_src = mq_sink_to_src (splitmux->mq, mq_sink);
 389   if (mq_src == NULL)
 390     goto fail;
 391
 392   *sink_pad = mq_sink;
 393   *src_pad = mq_src;
 394
 395   return TRUE;
 396
 397 fail:
 398   gst_element_release_request_pad (splitmux->mq, mq_sink);
 399   return FALSE;
 400 }
 401
 402 static MqStreamCtx *
 403 mq_stream_ctx_new (GstSplitMuxSink * splitmux)
 404 {
 405   MqStreamCtx *ctx;
 406
 407   ctx = g_new0 (MqStreamCtx, 1);
 408   g_atomic_int_set (&ctx->refcount, 1);
 409   ctx->splitmux = splitmux;
 410   gst_segment_init (&ctx->in_segment, GST_FORMAT_UNDEFINED);
 411   gst_segment_init (&ctx->out_segment, GST_FORMAT_UNDEFINED);
 412   ctx->in_running_time = ctx->out_running_time = 0;
 413   g_queue_init (&ctx->queued_bufs);
 414   return ctx;
 415 }
 416
 417 static void
 418 mq_stream_ctx_free (MqStreamCtx * ctx)
 419 {
 420   g_queue_foreach (&ctx->queued_bufs, (GFunc) mq_stream_buf_free, NULL);
 421   g_queue_clear (&ctx->queued_bufs);
 422   g_free (ctx);
 423 }
 424
 425 static void
 426 mq_stream_ctx_unref (MqStreamCtx * ctx)
 427 {
 428   if (g_atomic_int_dec_and_test (&ctx->refcount))
 429     mq_stream_ctx_free (ctx);
 430 }
 431
 432 static void
 433 mq_stream_ctx_ref (MqStreamCtx * ctx)
 434 {
 435   g_atomic_int_inc (&ctx->refcount);
 436 }
 437
 438 static void
 439 _pad_block_destroy_sink_notify (MqStreamCtx * ctx)
 440 {
 441   ctx->sink_pad_block_id = 0;
 442   mq_stream_ctx_unref (ctx);
 443 }
 444
 445 static void
 446 _pad_block_destroy_src_notify (MqStreamCtx * ctx)
 447 {
 448   ctx->src_pad_block_id = 0;
 449   mq_stream_ctx_unref (ctx);
 450 }
 451
 452 /* Called with lock held, drops the lock to send EOS to the
 453  * pad
 454  */
 455 static void
 456 send_eos (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
 457 {
 458   GstEvent *eos;
 459   GstPad *pad;
 460
 461   eos = gst_event_new_eos ();
 462   pad = gst_pad_get_peer (ctx->srcpad);
 463
 464   ctx->out_eos = TRUE;
 465
 466   GST_INFO_OBJECT (splitmux, "Sending EOS on %" GST_PTR_FORMAT, pad);
 467   GST_SPLITMUX_UNLOCK (splitmux);
 468   gst_pad_send_event (pad, eos);
 469   GST_SPLITMUX_LOCK (splitmux);
 470
 471   gst_object_unref (pad);
 472 }
 473
 474 /* Called with splitmux lock held to check if this output
 475  * context needs to sleep to wait for the release of the
 476  * next GOP, or to send EOS to close out the current file
 477  */
 478 static void
 479 complete_or_wait_on_out (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
 480 {
 481   do {
 482
 483     GST_LOG_OBJECT (ctx->srcpad,
 484         "Checking running time %" GST_TIME_FORMAT " against max %"
 485         GST_TIME_FORMAT, GST_TIME_ARGS (ctx->out_running_time),
 486         GST_TIME_ARGS (splitmux->max_out_running_time));
 487
 488     if (splitmux->max_out_running_time == GST_CLOCK_TIME_NONE ||
 489         ctx->out_running_time < splitmux->max_out_running_time)
 490       return;
 491
 492     if (ctx->flushing || splitmux->state == SPLITMUX_STATE_STOPPED)
 493       return;
 494
 495     if (splitmux->state == SPLITMUX_STATE_ENDING_FILE) {
 496       if (ctx->out_eos == FALSE) {
 497         send_eos (splitmux, ctx);
 498         continue;
 499       }
 500     } else if (splitmux->state == SPLITMUX_STATE_START_NEXT_FRAGMENT) {
 501       start_next_fragment (splitmux);
 502       continue;
 503     }
 504
 505     GST_INFO_OBJECT (ctx->srcpad,
 506         "Sleeping for running time %"
 507         GST_TIME_FORMAT " (max %" GST_TIME_FORMAT ")",
 508         GST_TIME_ARGS (ctx->out_running_time),
 509         GST_TIME_ARGS (splitmux->max_out_running_time));
 510     ctx->out_blocked = TRUE;
 511     /* Expand the mq if needed before sleeping */
 512     check_queue_length (splitmux, ctx);
 513     GST_SPLITMUX_WAIT (splitmux);
 514     ctx->out_blocked = FALSE;
 515     GST_INFO_OBJECT (ctx->srcpad,
 516         "Woken for new max running time %" GST_TIME_FORMAT,
 517         GST_TIME_ARGS (splitmux->max_out_running_time));
 518   } while (1);
 519 }
 520
 521 static GstPadProbeReturn
 522 handle_mq_output (GstPad * pad, GstPadProbeInfo * info, MqStreamCtx * ctx)
 523 {
 524   GstSplitMuxSink *splitmux = ctx->splitmux;
 525   MqStreamBuf *buf_info = NULL;
 526
 527   GST_LOG_OBJECT (pad, "Fired probe type 0x%x\n", info->type);
 528
 529   /* FIXME: Handle buffer lists, until then make it clear they won't work */
 530   if (info->type & GST_PAD_PROBE_TYPE_BUFFER_LIST) {
 531     g_warning ("Buffer list handling not implemented");
 532     return GST_PAD_PROBE_DROP;
 533   }
 534   if (info->type & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) {
 535     GstEvent *event = gst_pad_probe_info_get_event (info);
 536
 537     GST_LOG_OBJECT (pad, "Event %" GST_PTR_FORMAT, event);
 538
 539     switch (GST_EVENT_TYPE (event)) {
 540       case GST_EVENT_SEGMENT:
 541         gst_event_copy_segment (event, &ctx->out_segment);
 542         break;
 543       case GST_EVENT_FLUSH_STOP:
 544         GST_SPLITMUX_LOCK (splitmux);
 545         gst_segment_init (&ctx->out_segment, GST_FORMAT_UNDEFINED);
 546         g_queue_foreach (&ctx->queued_bufs, (GFunc) mq_stream_buf_free, NULL);
 547         g_queue_clear (&ctx->queued_bufs);
 548         ctx->flushing = FALSE;
 549         GST_SPLITMUX_UNLOCK (splitmux);
 550         break;
 551       case GST_EVENT_FLUSH_START:
 552         GST_SPLITMUX_LOCK (splitmux);
 553         GST_LOG_OBJECT (pad, "Flush start");
 554         ctx->flushing = TRUE;
 555         GST_SPLITMUX_BROADCAST (splitmux);
 556         GST_SPLITMUX_UNLOCK (splitmux);
 557         break;
 558       case GST_EVENT_EOS:
 559         GST_SPLITMUX_LOCK (splitmux);
 560         if (splitmux->state == SPLITMUX_STATE_STOPPED)
 561           goto beach;
 562         ctx->out_eos = TRUE;
 563         GST_SPLITMUX_UNLOCK (splitmux);
 564         break;
 565       case GST_EVENT_GAP:{
 566         GstClockTime gap_ts;
 567
 568         gst_event_parse_gap (event, &gap_ts, NULL);
 569         if (gap_ts == GST_CLOCK_TIME_NONE)
 570           break;
 571
 572         GST_SPLITMUX_LOCK (splitmux);
 573
 574         gap_ts = gst_segment_to_running_time (&ctx->out_segment,
 575             GST_FORMAT_TIME, gap_ts);
 576
 577         GST_LOG_OBJECT (pad, "Have GAP w/ ts %" GST_TIME_FORMAT,
 578             GST_TIME_ARGS (gap_ts));
 579
 580         if (splitmux->state == SPLITMUX_STATE_STOPPED)
 581           goto beach;
 582         ctx->out_running_time = gap_ts;
 583         complete_or_wait_on_out (splitmux, ctx);
 584         GST_SPLITMUX_UNLOCK (splitmux);
 585         break;
 586       }
 587       default:
 588         break;
 589     }
 590     return GST_PAD_PROBE_PASS;
 591   }
 592
 593   /* Allow everything through until the configured next stopping point */
 594   GST_SPLITMUX_LOCK (splitmux);
 595
 596   buf_info = g_queue_pop_tail (&ctx->queued_bufs);
 597   if (buf_info == NULL)
 598     /* Can only happen due to a poorly timed flush */
 599     goto beach;
 600
 601   /* If we have popped a keyframe, decrement the queued_gop count */
 602   if (buf_info->keyframe && splitmux->queued_gops > 0)
 603     splitmux->queued_gops--;
 604
 605   ctx->out_running_time = buf_info->run_ts;
 606
 607   GST_LOG_OBJECT (splitmux,
 608       "Pad %" GST_PTR_FORMAT " buffer with TS %" GST_TIME_FORMAT
 609       " size %" G_GSIZE_FORMAT,
 610       pad, GST_TIME_ARGS (ctx->out_running_time), buf_info->buf_size);
 611
 612   complete_or_wait_on_out (splitmux, ctx);
 613
 614   if (splitmux->muxed_out_time == GST_CLOCK_TIME_NONE ||
 615       splitmux->muxed_out_time < buf_info->run_ts)
 616     splitmux->muxed_out_time = buf_info->run_ts;
 617
 618   splitmux->muxed_out_bytes += buf_info->buf_size;
 619
 620 #ifndef GST_DISABLE_GST_DEBUG
 621   {
 622     GstBuffer *buf = gst_pad_probe_info_get_buffer (info);
 623     GST_LOG_OBJECT (pad, "Returning to pass buffer %" GST_PTR_FORMAT
 624         " run ts %" GST_TIME_FORMAT, buf,
 625         GST_TIME_ARGS (ctx->out_running_time));
 626   }
 627 #endif
 628
 629   GST_SPLITMUX_UNLOCK (splitmux);
 630
 631   mq_stream_buf_free (buf_info);
 632
 633   return GST_PAD_PROBE_PASS;
 634
 635 beach:
 636   GST_SPLITMUX_UNLOCK (splitmux);
 637   if (buf_info)
 638     mq_stream_buf_free (buf_info);
 639   return GST_PAD_PROBE_DROP;
 640 }
 641
 642 static gboolean
 643 resend_sticky (GstPad * pad, GstEvent ** event, GstPad * peer)
 644 {
 645   return gst_pad_send_event (peer, gst_event_ref (*event));
 646 }
 647
 648 static void
 649 restart_context (MqStreamCtx * ctx, GstSplitMuxSink * splitmux)
 650 {
 651   GstPad *peer = gst_pad_get_peer (ctx->srcpad);
 652
 653   gst_pad_sticky_events_foreach (ctx->srcpad,
 654       (GstPadStickyEventsForeachFunction) (resend_sticky), peer);
 655
 656   /* Clear EOS flag */
 657   ctx->out_eos = FALSE;
 658
 659   gst_object_unref (peer);
 660 }
 661
 662 /* Called with lock held when a fragment
 663  * reaches EOS and it is time to restart
 664  * a new fragment
 665  */
 666 static void
 667 start_next_fragment (GstSplitMuxSink * splitmux)
 668 {
 669   /* 1 change to new file */
 670   gst_element_set_state (splitmux->muxer, GST_STATE_NULL);
 671   gst_element_set_state (splitmux->active_sink, GST_STATE_NULL);
 672
 673   set_next_filename (splitmux);
 674
 675   gst_element_sync_state_with_parent (splitmux->active_sink);
 676   gst_element_sync_state_with_parent (splitmux->muxer);
 677
 678   g_list_foreach (splitmux->contexts, (GFunc) restart_context, splitmux);
 679
 680   /* Switch state and go back to processing */
 681   splitmux->state = SPLITMUX_STATE_COLLECTING_GOP_START;
 682   if (!splitmux->video_ctx->in_eos)
 683     splitmux->max_out_running_time = splitmux->video_ctx->in_running_time;
 684   else
 685     splitmux->max_out_running_time = GST_CLOCK_TIME_NONE;
 686
 687   /* Store the overflow parameters as the basis for the next fragment */
 688   splitmux->mux_start_time = splitmux->muxed_out_time;
 689   splitmux->mux_start_bytes = splitmux->muxed_out_bytes;
 690
 691   GST_DEBUG_OBJECT (splitmux,
 692       "Restarting flow for new fragment. New running time %" GST_TIME_FORMAT,
 693       GST_TIME_ARGS (splitmux->max_out_running_time));
 694
 695   GST_SPLITMUX_BROADCAST (splitmux);
 696 }
 697
 698 static void
 699 bus_handler (GstBin * bin, GstMessage * message)
 700 {
 701   GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (bin);
 702
 703   switch (GST_MESSAGE_TYPE (message)) {
 704     case GST_MESSAGE_EOS:
 705       /* If the state is draining out the current file, drop this EOS */
 706       GST_SPLITMUX_LOCK (splitmux);
 707       if (splitmux->state == SPLITMUX_STATE_ENDING_FILE &&
 708           splitmux->max_out_running_time != GST_CLOCK_TIME_NONE) {
 709         GST_DEBUG_OBJECT (splitmux, "Caught EOS at end of fragment, dropping");
 710         splitmux->state = SPLITMUX_STATE_START_NEXT_FRAGMENT;
 711         GST_SPLITMUX_BROADCAST (splitmux);
 712
 713         gst_message_unref (message);
 714         GST_SPLITMUX_UNLOCK (splitmux);
 715         return;
 716       }
 717       GST_SPLITMUX_UNLOCK (splitmux);
 718       break;
 719     default:
 720       break;
 721   }
 722
 723   GST_BIN_CLASS (parent_class)->handle_message (bin, message);
 724 }
 725
 726 /* Called with splitmux lock held */
 727 /* Called when entering ProcessingCompleteGop state
 728  * Assess if mq contents overflowed the current file
 729  *   -> If yes, need to switch to new file
 730  *   -> if no, set max_out_running_time to let this GOP in and
 731  *      go to COLLECTING_GOP_START state
 732  */
 733 static void
 734 handle_gathered_gop (GstSplitMuxSink * splitmux)
 735 {
 736   GList *cur;
 737   gsize queued_bytes = 0;
 738   GstClockTime queued_time = 0;
 739
 740   /* Assess if the multiqueue contents overflowed the current file */
 741   for (cur = g_list_first (splitmux->contexts);
 742       cur != NULL; cur = g_list_next (cur)) {
 743     MqStreamCtx *tmpctx = (MqStreamCtx *) (cur->data);
 744     if (tmpctx->in_running_time > queued_time)
 745       queued_time = tmpctx->in_running_time;
 746     queued_bytes += tmpctx->in_bytes;
 747   }
 748
 749   g_assert (queued_bytes >= splitmux->mux_start_bytes);
 750   g_assert (queued_time >= splitmux->mux_start_time);
 751
 752   queued_bytes -= splitmux->mux_start_bytes;
 753   queued_time -= splitmux->mux_start_time;
 754
 755   /* Expand queued bytes estimate by muxer overhead */
 756   queued_bytes += (queued_bytes * splitmux->mux_overhead);
 757
 758   GST_LOG_OBJECT (splitmux, "mq at TS %" GST_TIME_FORMAT
 759       " bytes %" G_GSIZE_FORMAT, GST_TIME_ARGS (queued_time), queued_bytes);
 760
 761   /* Check for overrun - have we output at least one byte and overrun
 762    * either threshold? */
 763   if ((splitmux->mux_start_bytes < splitmux->muxed_out_bytes) &&
 764       ((splitmux->threshold_bytes > 0 &&
 765               queued_bytes >= splitmux->threshold_bytes) ||
 766           (splitmux->threshold_time > 0 &&
 767               queued_time >= splitmux->threshold_time))) {
 768
 769     splitmux->state = SPLITMUX_STATE_ENDING_FILE;
 770
 771     GST_INFO_OBJECT (splitmux,
 772         "mq overflowed since last, draining out. max out TS is %"
 773         GST_TIME_FORMAT, GST_TIME_ARGS (splitmux->max_out_running_time));
 774     GST_SPLITMUX_BROADCAST (splitmux);
 775
 776   } else {
 777     /* No overflow */
 778     GST_LOG_OBJECT (splitmux,
 779         "This GOP didn't overflow the fragment. Bytes sent %" G_GSIZE_FORMAT
 780         " queued %" G_GSIZE_FORMAT " time %" GST_TIME_FORMAT " Continuing.",
 781         splitmux->muxed_out_bytes - splitmux->mux_start_bytes,
 782         queued_bytes, GST_TIME_ARGS (queued_time));
 783
 784     /* Wake everyone up to push this one GOP, then sleep */
 785     splitmux->state = SPLITMUX_STATE_COLLECTING_GOP_START;
 786     if (!splitmux->video_ctx->in_eos)
 787       splitmux->max_out_running_time = splitmux->video_ctx->in_running_time;
 788     else
 789       splitmux->max_out_running_time = GST_CLOCK_TIME_NONE;
 790
 791     GST_LOG_OBJECT (splitmux, "Waking output for complete GOP, TS %"
 792         GST_TIME_FORMAT, GST_TIME_ARGS (splitmux->max_out_running_time));
 793     GST_SPLITMUX_BROADCAST (splitmux);
 794   }
 795
 796 }
 797
 798 /* Called with splitmux lock held */
 799 /* Called from each input pad when it is has all the pieces
 800  * for a GOP or EOS, starting with the video pad which has set the
 801  * splitmux->max_in_running_time
 802  */
 803 static void
 804 check_completed_gop (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
 805 {
 806   GList *cur;
 807   gboolean ready = TRUE;
 808
 809   if (splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE) {
 810     /* Iterate each pad, and check that the input running time is at least
 811      * up to the video runnning time, and if so handle the collected GOP */
 812     GST_LOG_OBJECT (splitmux, "Checking GOP collected, ctx %p", ctx);
 813     for (cur = g_list_first (splitmux->contexts);
 814         cur != NULL; cur = g_list_next (cur)) {
 815       MqStreamCtx *tmpctx = (MqStreamCtx *) (cur->data);
 816
 817       GST_LOG_OBJECT (splitmux,
 818           "Context %p (src pad %" GST_PTR_FORMAT ") TS %" GST_TIME_FORMAT
 819           " EOS %d", tmpctx, tmpctx->srcpad,
 820           GST_TIME_ARGS (tmpctx->in_running_time), tmpctx->in_eos);
 821
 822       if (tmpctx->in_running_time < splitmux->max_in_running_time &&
 823           !tmpctx->in_eos) {
 824         GST_LOG_OBJECT (splitmux,
 825             "Context %p (src pad %" GST_PTR_FORMAT ") not ready. We'll sleep",
 826             tmpctx, tmpctx->srcpad);
 827         ready = FALSE;
 828         break;
 829       }
 830     }
 831     if (ready) {
 832       GST_DEBUG_OBJECT (splitmux,
 833           "Collected GOP is complete. Processing (ctx %p)", ctx);
 834       /* All pads have a complete GOP, release it into the multiqueue */
 835       handle_gathered_gop (splitmux);
 836     }
 837   }
 838
 839   /* Some pad is not yet ready, or GOP is being pushed
 840    * either way, sleep and wait to get woken */
 841   while ((splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE ||
 842           splitmux->state == SPLITMUX_STATE_START_NEXT_FRAGMENT) &&
 843       !ctx->flushing) {
 844
 845     GST_LOG_OBJECT (splitmux, "Sleeping for %s (ctx %p)",
 846         splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE ?
 847         "GOP complete" : "EOF draining", ctx);
 848     GST_SPLITMUX_WAIT (splitmux);
 849
 850     GST_LOG_OBJECT (splitmux, "Done waiting for complete GOP (ctx %p)", ctx);
 851   }
 852 }
 853
 854 static void
 855 check_queue_length (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
 856 {
 857   GList *cur;
 858   guint cur_len = g_queue_get_length (&ctx->queued_bufs);
 859
 860   GST_DEBUG_OBJECT (ctx->sinkpad,
 861       "Checking queue length len %u cur_max %u queued gops %u",
 862       cur_len, splitmux->mq_max_buffers, splitmux->queued_gops);
 863
 864   if (cur_len >= splitmux->mq_max_buffers) {
 865     gboolean allow_grow = FALSE;
 866
 867     /* If collecting a GOP and this pad might block,
 868      * and there isn't already a pending GOP in the queue
 869      * then grow
 870      */
 871     if (splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE &&
 872         ctx->in_running_time < splitmux->max_in_running_time &&
 873         splitmux->queued_gops <= 1) {
 874       allow_grow = TRUE;
 875     } else if (splitmux->state == SPLITMUX_STATE_COLLECTING_GOP_START &&
 876         ctx->is_video) {
 877       allow_grow = TRUE;
 878     }
 879
 880     if (!allow_grow) {
 881       for (cur = g_list_first (splitmux->contexts);
 882           cur != NULL; cur = g_list_next (cur)) {
 883         MqStreamCtx *tmpctx = (MqStreamCtx *) (cur->data);
 884         GST_DEBUG_OBJECT (tmpctx->sinkpad,
 885             " len %u out_blocked %d",
 886             g_queue_get_length (&tmpctx->queued_bufs), tmpctx->out_blocked);
 887         /* If another stream is starving, grow */
 888         if (tmpctx != ctx && g_queue_get_length (&tmpctx->queued_bufs) < 1) {
 889           allow_grow = TRUE;
 890         }
 891       }
 892     }
 893
 894     if (allow_grow) {
 895       splitmux->mq_max_buffers = cur_len + 1;
 896
 897       GST_INFO_OBJECT (splitmux,
 898           "Multiqueue overrun - enlarging to %u buffers ctx %p",
 899           splitmux->mq_max_buffers, ctx);
 900
 901       g_object_set (splitmux->mq, "max-size-buffers",
 902           splitmux->mq_max_buffers, NULL);
 903     }
 904   }
 905 }
 906
 907 static GstPadProbeReturn
 908 handle_mq_input (GstPad * pad, GstPadProbeInfo * info, MqStreamCtx * ctx)
 909 {
 910   GstSplitMuxSink *splitmux = ctx->splitmux;
 911   GstBuffer *buf;
 912   MqStreamBuf *buf_info;
 913   GstClockTime ts;
 914   gboolean loop_again;
 915   gboolean keyframe = FALSE;
 916
 917   GST_LOG_OBJECT (pad, "Fired probe type 0x%x\n", info->type);
 918
 919   /* FIXME: Handle buffer lists, until then make it clear they won't work */
 920   if (info->type & GST_PAD_PROBE_TYPE_BUFFER_LIST) {
 921     g_warning ("Buffer list handling not implemented");
 922     return GST_PAD_PROBE_DROP;
 923   }
 924   if (info->type & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) {
 925     GstEvent *event = gst_pad_probe_info_get_event (info);
 926     switch (GST_EVENT_TYPE (event)) {
 927       case GST_EVENT_SEGMENT:
 928         gst_event_copy_segment (event, &ctx->in_segment);
 929         break;
 930       case GST_EVENT_FLUSH_STOP:
 931         GST_SPLITMUX_LOCK (splitmux);
 932         gst_segment_init (&ctx->in_segment, GST_FORMAT_UNDEFINED);
 933         ctx->in_eos = FALSE;
 934         ctx->in_bytes = 0;
 935         ctx->in_running_time = 0;
 936         GST_SPLITMUX_UNLOCK (splitmux);
 937         break;
 938       case GST_EVENT_EOS:
 939         GST_SPLITMUX_LOCK (splitmux);
 940         ctx->in_eos = TRUE;
 941
 942         if (splitmux->state == SPLITMUX_STATE_STOPPED)
 943           goto beach;
 944
 945         if (ctx->is_video) {
 946           GST_INFO_OBJECT (splitmux, "Got Video EOS. Finishing up");
 947           /* Act as if this is a new keyframe with infinite timestamp */
 948           splitmux->max_in_running_time = GST_CLOCK_TIME_NONE;
 949           splitmux->state = SPLITMUX_STATE_WAITING_GOP_COMPLETE;
 950           /* Wake up other input pads to collect this GOP */
 951           GST_SPLITMUX_BROADCAST (splitmux);
 952           check_completed_gop (splitmux, ctx);
 953         } else if (splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE) {
 954           /* If we are waiting for a GOP to be completed (ie, for aux
 955            * pads to catch up), then this pad is complete, so check
 956            * if the whole GOP is.
 957            */
 958           check_completed_gop (splitmux, ctx);
 959         }
 960         GST_SPLITMUX_UNLOCK (splitmux);
 961         break;
 962       default:
 963         break;
 964     }
 965     return GST_PAD_PROBE_PASS;
 966   }
 967
 968   buf = gst_pad_probe_info_get_buffer (info);
 969   ctx->in_running_time = gst_segment_to_running_time (&ctx->in_segment,
 970       GST_FORMAT_TIME, GST_BUFFER_TIMESTAMP (buf));
 971   buf_info = mq_stream_buf_new ();
 972
 973   if (GST_BUFFER_PTS_IS_VALID (buf))
 974     ts = GST_BUFFER_PTS (buf);
 975   else
 976     ts = GST_BUFFER_DTS (buf);
 977
 978   GST_SPLITMUX_LOCK (splitmux);
 979
 980   if (splitmux->state == SPLITMUX_STATE_STOPPED)
 981     goto beach;
 982
 983   /* If this buffer has a timestamp, advance the input timestamp of the
 984    * stream */
 985   if (GST_CLOCK_TIME_IS_VALID (ts)) {
 986     GstClockTime running_time =
 987         gst_segment_to_running_time (&ctx->in_segment, GST_FORMAT_TIME,
 988         GST_BUFFER_TIMESTAMP (buf));
 989
 990     if (GST_CLOCK_TIME_IS_VALID (running_time) &&
 991         (ctx->in_running_time == GST_CLOCK_TIME_NONE
 992             || running_time > ctx->in_running_time))
 993       ctx->in_running_time = running_time;
 994   }
 995
 996   /* Try to make sure we have a valid running time */
 997   if (!GST_CLOCK_TIME_IS_VALID (ctx->in_running_time)) {
 998     ctx->in_running_time =
 999         gst_segment_to_running_time (&ctx->in_segment, GST_FORMAT_TIME,
1000         ctx->in_segment.start);
1001   }
1002
1003   buf_info->run_ts = ctx->in_running_time;
1004   buf_info->buf_size = gst_buffer_get_size (buf);
1005
1006   /* Update total input byte counter for overflow detect */
1007   ctx->in_bytes += buf_info->buf_size;
1008
1009   GST_DEBUG_OBJECT (pad, "Buf TS %" GST_TIME_FORMAT
1010       " total in_bytes %" G_GSIZE_FORMAT,
1011       GST_TIME_ARGS (buf_info->run_ts), ctx->in_bytes);
1012
1013   loop_again = TRUE;
1014   do {
1015     if (ctx->flushing)
1016       break;
1017
1018     switch (splitmux->state) {
1019       case SPLITMUX_STATE_COLLECTING_GOP_START:
1020         if (ctx->is_video) {
1021           /* If a keyframe, we have a complete GOP */
1022           if (GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT) ||
1023               !GST_CLOCK_TIME_IS_VALID (ctx->in_running_time) ||
1024               splitmux->max_in_running_time >= ctx->in_running_time) {
1025             /* Pass this buffer through */
1026             loop_again = FALSE;
1027             break;
1028           }
1029           GST_INFO_OBJECT (pad,
1030               "Have keyframe with running time %" GST_TIME_FORMAT,
1031               GST_TIME_ARGS (ctx->in_running_time));
1032           keyframe = TRUE;
1033           splitmux->state = SPLITMUX_STATE_WAITING_GOP_COMPLETE;
1034           splitmux->max_in_running_time = ctx->in_running_time;
1035           /* Wake up other input pads to collect this GOP */
1036           GST_SPLITMUX_BROADCAST (splitmux);
1037           check_completed_gop (splitmux, ctx);
1038         } else {
1039           /* We're still waiting for a keyframe on the video pad, sleep */
1040           GST_LOG_OBJECT (pad, "Sleeping for GOP start");
1041           GST_SPLITMUX_WAIT (splitmux);
1042           GST_LOG_OBJECT (pad, "Done sleeping for GOP start state now %d",
1043               splitmux->state);
1044         }
1045         break;
1046       case SPLITMUX_STATE_WAITING_GOP_COMPLETE:
1047         /* After a GOP start is found, this buffer might complete the GOP */
1048         /* If we overran the target timestamp, it might be time to process
1049          * the GOP, otherwise bail out for more data
1050          */
1051         GST_LOG_OBJECT (pad,
1052             "Checking TS %" GST_TIME_FORMAT " against max %" GST_TIME_FORMAT,
1053             GST_TIME_ARGS (ctx->in_running_time),
1054             GST_TIME_ARGS (splitmux->max_in_running_time));
1055
1056         if (ctx->in_running_time < splitmux->max_in_running_time) {
1057           loop_again = FALSE;
1058           break;
1059         }
1060
1061         GST_LOG_OBJECT (pad,
1062             "Collected last packet of GOP. Checking other pads");
1063         check_completed_gop (splitmux, ctx);
1064         break;
1065       case SPLITMUX_STATE_ENDING_FILE:
1066       case SPLITMUX_STATE_START_NEXT_FRAGMENT:
1067         /* A fragment is ending, wait until that's done before continuing */
1068         GST_DEBUG_OBJECT (pad, "Sleeping for fragment restart");
1069         GST_SPLITMUX_WAIT (splitmux);
1070         GST_DEBUG_OBJECT (pad,
1071             "Done sleeping for fragment restart state now %d", splitmux->state);
1072         break;
1073       default:
1074         loop_again = FALSE;
1075         break;
1076     }
1077   } while (loop_again);
1078
1079   if (keyframe) {
1080     splitmux->queued_gops++;
1081     buf_info->keyframe = TRUE;
1082   }
1083
1084   /* Now add this buffer to the queue just before returning */
1085   g_queue_push_head (&ctx->queued_bufs, buf_info);
1086
1087   /* Check the buffer will fit in the mq */
1088   check_queue_length (splitmux, ctx);
1089
1090   GST_LOG_OBJECT (pad, "Returning to queue buffer %" GST_PTR_FORMAT
1091       " run ts %" GST_TIME_FORMAT, buf, GST_TIME_ARGS (ctx->in_running_time));
1092
1093 beach:
1094   GST_SPLITMUX_UNLOCK (splitmux);
1095
1096   return GST_PAD_PROBE_PASS;
1097 }
1098
1099 static GstPad *
1100 gst_splitmux_sink_request_new_pad (GstElement * element,
1101     GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
1102 {
1103   GstSplitMuxSink *splitmux = (GstSplitMuxSink *) element;
1104   GstPadTemplate *mux_template = NULL;
1105   GstPad *res = NULL;
1106   GstPad *mq_sink, *mq_src;
1107   gchar *gname;
1108   gboolean is_video = FALSE;
1109   MqStreamCtx *ctx;
1110
1111   GST_DEBUG_OBJECT (element, "templ:%s, name:%s", templ->name_template, name);
1112
1113   GST_SPLITMUX_LOCK (splitmux);
1114   if (!create_elements (splitmux))
1115     goto fail;
1116
1117   if (templ->name_template) {
1118     if (g_str_equal (templ->name_template, "video")) {
1119       /* FIXME: Look for a pad template with matching caps, rather than by name */
1120       mux_template =
1121           gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS
1122           (splitmux->muxer), "video_%u");
1123       is_video = TRUE;
1124       name = NULL;
1125     } else {
1126       mux_template =
1127           gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS
1128           (splitmux->muxer), templ->name_template);
1129     }
1130   }
1131
1132   res = gst_element_request_pad (splitmux->muxer, mux_template, name, caps);
1133   if (res == NULL)
1134     goto fail;
1135
1136   if (is_video)
1137     gname = g_strdup ("video");
1138   else if (name == NULL)
1139     gname = gst_pad_get_name (res);
1140   else
1141     gname = g_strdup (name);
1142
1143   if (!get_pads_from_mq (splitmux, &mq_sink, &mq_src)) {
1144     gst_element_release_request_pad (splitmux->muxer, res);
1145     gst_object_unref (GST_OBJECT (res));
1146     goto fail;
1147   }
1148
1149   if (gst_pad_link (mq_src, res) != GST_PAD_LINK_OK) {
1150     gst_element_release_request_pad (splitmux->muxer, res);
1151     gst_object_unref (GST_OBJECT (res));
1152     gst_element_release_request_pad (splitmux->mq, mq_sink);
1153     gst_object_unref (GST_OBJECT (mq_sink));
1154     goto fail;
1155   }
1156
1157   gst_object_unref (GST_OBJECT (res));
1158
1159   ctx = mq_stream_ctx_new (splitmux);
1160   ctx->is_video = is_video;
1161   ctx->srcpad = mq_src;
1162   ctx->sinkpad = mq_sink;
1163
1164   mq_stream_ctx_ref (ctx);
1165   ctx->src_pad_block_id =
1166       gst_pad_add_probe (mq_src, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM,
1167       (GstPadProbeCallback) handle_mq_output, ctx, (GDestroyNotify)
1168       _pad_block_destroy_src_notify);
1169   if (is_video)
1170     splitmux->video_ctx = ctx;
1171
1172   res = gst_ghost_pad_new (gname, mq_sink);
1173   g_object_set_qdata ((GObject *) (res), PAD_CONTEXT, ctx);
1174
1175   mq_stream_ctx_ref (ctx);
1176   ctx->sink_pad_block_id =
1177       gst_pad_add_probe (res, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM,
1178       (GstPadProbeCallback) handle_mq_input, ctx, (GDestroyNotify)
1179       _pad_block_destroy_sink_notify);
1180
1181   GST_DEBUG_OBJECT (splitmux, "Request pad %" GST_PTR_FORMAT
1182       " is mq pad %" GST_PTR_FORMAT, res, mq_sink);
1183
1184   splitmux->contexts = g_list_prepend (splitmux->contexts, ctx);
1185
1186   g_free (gname);
1187
1188   gst_object_unref (mq_sink);
1189   gst_object_unref (mq_src);
1190
1191   gst_pad_set_active (res, TRUE);
1192   gst_element_add_pad (element, res);
1193   GST_SPLITMUX_UNLOCK (splitmux);
1194
1195   return res;
1196 fail:
1197   GST_SPLITMUX_UNLOCK (splitmux);
1198   return NULL;
1199 }
1200
1201 static void
1202 gst_splitmux_sink_release_pad (GstElement * element, GstPad * pad)
1203 {
1204   GstSplitMuxSink *splitmux = (GstSplitMuxSink *) element;
1205   GstPad *mqsink, *mqsrc, *muxpad;
1206   MqStreamCtx *ctx =
1207       (MqStreamCtx *) (g_object_get_qdata ((GObject *) (pad), PAD_CONTEXT));
1208
1209   GST_SPLITMUX_LOCK (splitmux);
1210
1211   if (splitmux->muxer == NULL || splitmux->mq == NULL)
1212     goto fail;                  /* Elements don't exist yet - nothing to release */
1213
1214   GST_INFO_OBJECT (pad, "releasing request pad");
1215
1216   mqsink = gst_ghost_pad_get_target (GST_GHOST_PAD (pad));
1217   mqsrc = mq_sink_to_src (splitmux->mq, mqsink);
1218   muxpad = gst_pad_get_peer (mqsrc);
1219
1220   /* Remove the context from our consideration */
1221   splitmux->contexts = g_list_remove (splitmux->contexts, ctx);
1222
1223   if (ctx->sink_pad_block_id)
1224     gst_pad_remove_probe (ctx->sinkpad, ctx->sink_pad_block_id);
1225
1226   if (ctx->src_pad_block_id)
1227     gst_pad_remove_probe (ctx->srcpad, ctx->src_pad_block_id);
1228
1229   /* Can release the context now */
1230   mq_stream_ctx_unref (ctx);
1231
1232   /* Release and free the mq input */
1233   gst_element_release_request_pad (splitmux->mq, mqsink);
1234
1235   /* Release and free the muxer input */
1236   gst_element_release_request_pad (splitmux->muxer, muxpad);
1237
1238   gst_object_unref (mqsink);
1239   gst_object_unref (mqsrc);
1240   gst_object_unref (muxpad);
1241
1242   gst_element_remove_pad (element, pad);
1243
1244 fail:
1245   GST_SPLITMUX_UNLOCK (splitmux);
1246 }
1247
1248 static GstElement *
1249 create_element (GstSplitMuxSink * splitmux,
1250     const gchar * factory, const gchar * name)
1251 {
1252   GstElement *ret = gst_element_factory_make (factory, name);
1253   if (ret == NULL) {
1254     g_warning ("Failed to create %s - splitmuxsink will not work", name);
1255     return NULL;
1256   }
1257
1258   if (!gst_bin_add (GST_BIN (splitmux), ret)) {
1259     g_warning ("Could not add %s element - splitmuxsink will not work", name);
1260     gst_object_unref (ret);
1261     return NULL;
1262   }
1263
1264   return ret;
1265 }
1266
1267 static gboolean
1268 create_elements (GstSplitMuxSink * splitmux)
1269 {
1270   /* Create internal elements */
1271   if (splitmux->mq == NULL) {
1272     if ((splitmux->mq =
1273             create_element (splitmux, "multiqueue", "multiqueue")) == NULL)
1274       goto fail;
1275
1276     splitmux->mq_max_buffers = 5;
1277     /* No bytes or time limit, we limit buffers manually */
1278     g_object_set (splitmux->mq, "max-size-bytes", 0, "max-size-time",
1279         (guint64) 0, "max-size-buffers", splitmux->mq_max_buffers, NULL);
1280   }
1281
1282   if (splitmux->muxer == NULL) {
1283     if (splitmux->provided_muxer == NULL) {
1284       if ((splitmux->muxer =
1285               create_element (splitmux, "mp4mux", "muxer")) == NULL)
1286         goto fail;
1287     } else {
1288       splitmux->muxer = splitmux->provided_muxer;
1289       if (!gst_bin_add (GST_BIN (splitmux), splitmux->provided_muxer)) {
1290         g_warning ("Could not add muxer element - splitmuxsink will not work");
1291         goto fail;
1292       }
1293     }
1294   }
1295
1296   return TRUE;
1297 fail:
1298   return FALSE;
1299 }
1300
1301 static GstElement *
1302 find_sink (GstElement * e)
1303 {
1304   GstElement *res = NULL;
1305   GstIterator *iter;
1306   gboolean done = FALSE;
1307   GValue data = { 0, };
1308
1309   if (!GST_IS_BIN (e))
1310     return e;
1311
1312   iter = gst_bin_iterate_sinks (GST_BIN (e));
1313   while (!done) {
1314     switch (gst_iterator_next (iter, &data)) {
1315       case GST_ITERATOR_OK:
1316       {
1317         GstElement *child = g_value_get_object (&data);
1318         if (g_object_class_find_property (G_OBJECT_GET_CLASS (child),
1319                 "location") != NULL) {
1320           res = child;
1321           done = TRUE;
1322         }
1323         g_value_reset (&data);
1324         break;
1325       }
1326       case GST_ITERATOR_RESYNC:
1327         gst_iterator_resync (iter);
1328         break;
1329       case GST_ITERATOR_DONE:
1330         done = TRUE;
1331         break;
1332       case GST_ITERATOR_ERROR:
1333         g_assert_not_reached ();
1334         break;
1335     }
1336   }
1337   g_value_unset (&data);
1338   gst_iterator_free (iter);
1339
1340   return res;
1341 }
1342
1343 static gboolean
1344 create_sink (GstSplitMuxSink * splitmux)
1345 {
1346   g_return_val_if_fail (splitmux->active_sink == NULL, TRUE);
1347
1348   if (splitmux->provided_sink == NULL) {
1349     if ((splitmux->sink =
1350             create_element (splitmux, DEFAULT_SINK, "sink")) == NULL)
1351       goto fail;
1352     splitmux->active_sink = splitmux->sink;
1353   } else {
1354     if (!gst_bin_add (GST_BIN (splitmux), splitmux->provided_sink)) {
1355       g_warning ("Could not add sink elements - splitmuxsink will not work");
1356       goto fail;
1357     }
1358
1359     splitmux->active_sink = splitmux->provided_sink;
1360
1361     /* Find the sink element */
1362     splitmux->sink = find_sink (splitmux->active_sink);
1363     if (splitmux->sink == NULL) {
1364       g_warning
1365           ("Could not locate sink element in provided sink - splitmuxsink will not work");
1366       goto fail;
1367     }
1368   }
1369
1370   if (!gst_element_link (splitmux->muxer, splitmux->active_sink)) {
1371     g_warning ("Failed to link muxer and sink- splitmuxsink will not work");
1372     goto fail;
1373   }
1374
1375   return TRUE;
1376 fail:
1377   return FALSE;
1378 }
1379
1380 #ifdef __GNUC__
1381 #pragma GCC diagnostic ignored "-Wformat-nonliteral"
1382 #endif
1383 static void
1384 set_next_filename (GstSplitMuxSink * splitmux)
1385 {
1386   if (splitmux->location) {
1387     gchar *fname;
1388
1389     fname = g_strdup_printf (splitmux->location, splitmux->fragment_id);
1390     GST_INFO_OBJECT (splitmux, "Setting file to %s", fname);
1391     g_object_set (splitmux->sink, "location", fname, NULL);
1392     g_free (fname);
1393
1394     splitmux->fragment_id++;
1395   }
1396 }
1397
1398 static GstStateChangeReturn
1399 gst_splitmux_sink_change_state (GstElement * element, GstStateChange transition)
1400 {
1401   GstStateChangeReturn ret;
1402   GstSplitMuxSink *splitmux = (GstSplitMuxSink *) element;
1403
1404   switch (transition) {
1405     case GST_STATE_CHANGE_NULL_TO_READY:{
1406       GST_SPLITMUX_LOCK (splitmux);
1407       if (!create_elements (splitmux) || !create_sink (splitmux)) {
1408         ret = GST_STATE_CHANGE_FAILURE;
1409         GST_SPLITMUX_UNLOCK (splitmux);
1410         goto beach;
1411       }
1412       GST_SPLITMUX_UNLOCK (splitmux);
1413       splitmux->fragment_id = 0;
1414       set_next_filename (splitmux);
1415       break;
1416     }
1417     case GST_STATE_CHANGE_READY_TO_PAUSED:{
1418       GST_SPLITMUX_LOCK (splitmux);
1419       /* Start by collecting one input on each pad */
1420       splitmux->state = SPLITMUX_STATE_COLLECTING_GOP_START;
1421       splitmux->max_in_running_time = 0;
1422       splitmux->muxed_out_time = splitmux->mux_start_time = 0;
1423       splitmux->muxed_out_bytes = splitmux->mux_start_bytes = 0;
1424       GST_SPLITMUX_UNLOCK (splitmux);
1425       break;
1426     }
1427     case GST_STATE_CHANGE_PAUSED_TO_READY:
1428     case GST_STATE_CHANGE_READY_TO_NULL:
1429       GST_SPLITMUX_LOCK (splitmux);
1430       splitmux->state = SPLITMUX_STATE_STOPPED;
1431       /* Wake up any blocked threads */
1432       GST_LOG_OBJECT (splitmux,
1433           "State change -> NULL or READY. Waking threads");
1434       GST_SPLITMUX_BROADCAST (splitmux);
1435       GST_SPLITMUX_UNLOCK (splitmux);
1436       break;
1437     default:
1438       break;
1439   }
1440
1441   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1442   if (ret == GST_STATE_CHANGE_FAILURE)
1443     goto beach;
1444
1445   switch (transition) {
1446     case GST_STATE_CHANGE_READY_TO_NULL:
1447       GST_SPLITMUX_LOCK (splitmux);
1448       splitmux->fragment_id = 0;
1449       gst_splitmux_reset (splitmux);
1450       GST_SPLITMUX_UNLOCK (splitmux);
1451       break;
1452     default:
1453       break;
1454   }
1455
1456 beach:
1457
1458   if (transition == GST_STATE_CHANGE_NULL_TO_READY &&
1459       ret == GST_STATE_CHANGE_FAILURE) {
1460     /* Cleanup elements on failed transition out of NULL */
1461     gst_splitmux_reset (splitmux);
1462   }
1463   return ret;
1464 }
1465
1466 gboolean
1467 register_splitmuxsink (GstPlugin * plugin)
1468 {
1469   GST_DEBUG_CATEGORY_INIT (splitmux_debug, "splitmuxsink", 0,
1470       "Split File Muxing Sink");
1471
1472   return gst_element_register (plugin, "splitmuxsink", GST_RANK_NONE,
1473       GST_TYPE_SPLITMUX_SINK);
1474 }