1 /* GStreamer Muxer bin that splits output stream by size/time
2 * Copyright (C) <2014> Jan Schmidt <jan@centricular.com>
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.
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.
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.
21 * SECTION:element-splitmuxsink
22 * @short_description: Muxer wrapper for splitting output stream by size or time
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 * can be muxed, with as many accompanying audio and subtitle streams as
30 * By default, it uses mp4mux and filesink, but they can be changed via
31 * the 'muxer' and 'sink' properties.
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.
36 * If a video stream is available, the splitting process is driven by the video
37 * stream contents, and the video stream must contain closed GOPs for the output
38 * file parts to be played individually correctly. In the absence of a video
39 * stream, the first available stream is used as reference for synchronization.
42 * <title>Example pipelines</title>
44 * gst-launch-1.0 -e v4l2src num-buffers=500 ! video/x-raw,width=320,height=240 ! videoconvert ! queue ! timeoverlay ! x264enc key-int-max=10 ! h264parse ! splitmuxsink location=video%02d.mov max-size-time=10000000000 max-size-bytes=1000000
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.
57 #include "gstsplitmuxsink.h"
59 GST_DEBUG_CATEGORY_STATIC (splitmux_debug);
60 #define GST_CAT_DEFAULT splitmux_debug
62 #define GST_SPLITMUX_LOCK(s) g_mutex_lock(&(s)->lock)
63 #define GST_SPLITMUX_UNLOCK(s) g_mutex_unlock(&(s)->lock)
64 #define GST_SPLITMUX_WAIT(s) g_cond_wait (&(s)->data_cond, &(s)->lock)
65 #define GST_SPLITMUX_BROADCAST(s) g_cond_broadcast (&(s)->data_cond)
78 #define DEFAULT_MAX_SIZE_TIME 0
79 #define DEFAULT_MAX_SIZE_BYTES 0
80 #define DEFAULT_MUXER_OVERHEAD 0.02
81 #define DEFAULT_MUXER "mp4mux"
82 #define DEFAULT_SINK "filesink"
86 SIGNAL_FORMAT_LOCATION,
90 static guint signals[SIGNAL_LAST];
92 static GstStaticPadTemplate video_sink_template =
93 GST_STATIC_PAD_TEMPLATE ("video",
97 static GstStaticPadTemplate audio_sink_template =
98 GST_STATIC_PAD_TEMPLATE ("audio_%u",
101 GST_STATIC_CAPS_ANY);
102 static GstStaticPadTemplate subtitle_sink_template =
103 GST_STATIC_PAD_TEMPLATE ("subtitle_%u",
106 GST_STATIC_CAPS_ANY);
108 static GQuark PAD_CONTEXT;
113 PAD_CONTEXT = g_quark_from_static_string ("pad-context");
116 #define gst_splitmux_sink_parent_class parent_class
117 G_DEFINE_TYPE_EXTENDED (GstSplitMuxSink, gst_splitmux_sink, GST_TYPE_BIN, 0,
120 static gboolean create_elements (GstSplitMuxSink * splitmux);
121 static gboolean create_sink (GstSplitMuxSink * splitmux);
122 static void gst_splitmux_sink_set_property (GObject * object, guint prop_id,
123 const GValue * value, GParamSpec * pspec);
124 static void gst_splitmux_sink_get_property (GObject * object, guint prop_id,
125 GValue * value, GParamSpec * pspec);
126 static void gst_splitmux_sink_dispose (GObject * object);
127 static void gst_splitmux_sink_finalize (GObject * object);
129 static GstPad *gst_splitmux_sink_request_new_pad (GstElement * element,
130 GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
131 static void gst_splitmux_sink_release_pad (GstElement * element, GstPad * pad);
133 static GstStateChangeReturn gst_splitmux_sink_change_state (GstElement *
134 element, GstStateChange transition);
136 static void bus_handler (GstBin * bin, GstMessage * msg);
137 static void set_next_filename (GstSplitMuxSink * splitmux);
138 static void start_next_fragment (GstSplitMuxSink * splitmux);
139 static void check_queue_length (GstSplitMuxSink * splitmux, MqStreamCtx * ctx);
140 static void mq_stream_ctx_unref (MqStreamCtx * ctx);
143 mq_stream_buf_new (void)
145 return g_slice_new0 (MqStreamBuf);
149 mq_stream_buf_free (MqStreamBuf * data)
151 g_slice_free (MqStreamBuf, data);
155 gst_splitmux_sink_class_init (GstSplitMuxSinkClass * klass)
157 GObjectClass *gobject_class = (GObjectClass *) klass;
158 GstElementClass *gstelement_class = (GstElementClass *) klass;
159 GstBinClass *gstbin_class = (GstBinClass *) klass;
161 gobject_class->set_property = gst_splitmux_sink_set_property;
162 gobject_class->get_property = gst_splitmux_sink_get_property;
163 gobject_class->dispose = gst_splitmux_sink_dispose;
164 gobject_class->finalize = gst_splitmux_sink_finalize;
166 gst_element_class_set_static_metadata (gstelement_class,
167 "Split Muxing Bin", "Generic/Bin/Muxer",
168 "Convenience bin that muxes incoming streams into multiple time/size limited files",
169 "Jan Schmidt <jan@centricular.com>");
171 gst_element_class_add_static_pad_template (gstelement_class,
172 &video_sink_template);
173 gst_element_class_add_static_pad_template (gstelement_class,
174 &audio_sink_template);
175 gst_element_class_add_static_pad_template (gstelement_class,
176 &subtitle_sink_template);
178 gstelement_class->change_state =
179 GST_DEBUG_FUNCPTR (gst_splitmux_sink_change_state);
180 gstelement_class->request_new_pad =
181 GST_DEBUG_FUNCPTR (gst_splitmux_sink_request_new_pad);
182 gstelement_class->release_pad =
183 GST_DEBUG_FUNCPTR (gst_splitmux_sink_release_pad);
185 gstbin_class->handle_message = bus_handler;
187 g_object_class_install_property (gobject_class, PROP_LOCATION,
188 g_param_spec_string ("location", "File Output Pattern",
189 "Format string pattern for the location of the files to write (e.g. video%05d.mp4)",
190 NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
191 g_object_class_install_property (gobject_class, PROP_MUXER_OVERHEAD,
192 g_param_spec_double ("mux-overhead", "Muxing Overhead",
193 "Extra size overhead of muxing (0.02 = 2%)", 0.0, 1.0,
194 DEFAULT_MUXER_OVERHEAD,
195 G_PARAM_READWRITE | GST_PARAM_CONTROLLABLE | G_PARAM_STATIC_STRINGS));
197 g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
198 g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
199 "Max. amount of time per file (in ns, 0=disable)", 0, G_MAXUINT64,
200 DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
201 g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BYTES,
202 g_param_spec_uint64 ("max-size-bytes", "Max. size bytes",
203 "Max. amount of data per file (in bytes, 0=disable)", 0, G_MAXUINT64,
204 DEFAULT_MAX_SIZE_BYTES, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
206 g_object_class_install_property (gobject_class, PROP_MUXER,
207 g_param_spec_object ("muxer", "Muxer",
208 "The muxer element to use (NULL = default mp4mux)",
209 GST_TYPE_ELEMENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
210 g_object_class_install_property (gobject_class, PROP_SINK,
211 g_param_spec_object ("sink", "Sink",
212 "The sink element (or element chain) to use (NULL = default filesink)",
213 GST_TYPE_ELEMENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
216 * GstSplitMuxSink::format-location:
217 * @splitmux: the #GstSplitMuxSink
218 * @fragment_id: the sequence number of the file to be created
220 * Returns: the location to be used for the next output file
222 signals[SIGNAL_FORMAT_LOCATION] =
223 g_signal_new ("format-location", G_TYPE_FROM_CLASS (klass),
224 G_SIGNAL_RUN_LAST, 0, NULL, NULL, NULL, G_TYPE_STRING, 1, G_TYPE_UINT);
228 gst_splitmux_sink_init (GstSplitMuxSink * splitmux)
230 g_mutex_init (&splitmux->lock);
231 g_cond_init (&splitmux->data_cond);
233 splitmux->mux_overhead = DEFAULT_MUXER_OVERHEAD;
234 splitmux->threshold_time = DEFAULT_MAX_SIZE_TIME;
235 splitmux->threshold_bytes = DEFAULT_MAX_SIZE_BYTES;
237 GST_OBJECT_FLAG_SET (splitmux, GST_ELEMENT_FLAG_SINK);
241 gst_splitmux_reset (GstSplitMuxSink * splitmux)
244 gst_bin_remove (GST_BIN (splitmux), splitmux->mq);
246 gst_bin_remove (GST_BIN (splitmux), splitmux->muxer);
247 if (splitmux->active_sink)
248 gst_bin_remove (GST_BIN (splitmux), splitmux->active_sink);
250 splitmux->sink = splitmux->active_sink = splitmux->muxer = splitmux->mq =
255 gst_splitmux_sink_dispose (GObject * object)
257 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
259 G_OBJECT_CLASS (parent_class)->dispose (object);
261 /* Calling parent dispose invalidates all child pointers */
262 splitmux->sink = splitmux->active_sink = splitmux->muxer = splitmux->mq =
267 gst_splitmux_sink_finalize (GObject * object)
269 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
270 g_cond_clear (&splitmux->data_cond);
271 g_mutex_clear (&splitmux->lock);
272 if (splitmux->provided_sink)
273 gst_object_unref (splitmux->provided_sink);
274 if (splitmux->provided_muxer)
275 gst_object_unref (splitmux->provided_muxer);
277 g_free (splitmux->location);
279 /* Make sure to free any un-released contexts */
280 g_list_foreach (splitmux->contexts, (GFunc) mq_stream_ctx_unref, NULL);
281 g_list_free (splitmux->contexts);
283 G_OBJECT_CLASS (parent_class)->finalize (object);
287 gst_splitmux_sink_set_property (GObject * object, guint prop_id,
288 const GValue * value, GParamSpec * pspec)
290 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
294 GST_OBJECT_LOCK (splitmux);
295 g_free (splitmux->location);
296 splitmux->location = g_value_dup_string (value);
297 GST_OBJECT_UNLOCK (splitmux);
300 case PROP_MAX_SIZE_BYTES:
301 GST_OBJECT_LOCK (splitmux);
302 splitmux->threshold_bytes = g_value_get_uint64 (value);
303 GST_OBJECT_UNLOCK (splitmux);
305 case PROP_MAX_SIZE_TIME:
306 GST_OBJECT_LOCK (splitmux);
307 splitmux->threshold_time = g_value_get_uint64 (value);
308 GST_OBJECT_UNLOCK (splitmux);
310 case PROP_MUXER_OVERHEAD:
311 GST_OBJECT_LOCK (splitmux);
312 splitmux->mux_overhead = g_value_get_double (value);
313 GST_OBJECT_UNLOCK (splitmux);
316 GST_OBJECT_LOCK (splitmux);
317 if (splitmux->provided_sink)
318 gst_object_unref (splitmux->provided_sink);
319 splitmux->provided_sink = g_value_dup_object (value);
320 GST_OBJECT_UNLOCK (splitmux);
323 GST_OBJECT_LOCK (splitmux);
324 if (splitmux->provided_muxer)
325 gst_object_unref (splitmux->provided_muxer);
326 splitmux->provided_muxer = g_value_dup_object (value);
327 GST_OBJECT_UNLOCK (splitmux);
330 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
336 gst_splitmux_sink_get_property (GObject * object, guint prop_id,
337 GValue * value, GParamSpec * pspec)
339 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
343 GST_OBJECT_LOCK (splitmux);
344 g_value_set_string (value, splitmux->location);
345 GST_OBJECT_UNLOCK (splitmux);
347 case PROP_MAX_SIZE_BYTES:
348 GST_OBJECT_LOCK (splitmux);
349 g_value_set_uint64 (value, splitmux->threshold_bytes);
350 GST_OBJECT_UNLOCK (splitmux);
352 case PROP_MAX_SIZE_TIME:
353 GST_OBJECT_LOCK (splitmux);
354 g_value_set_uint64 (value, splitmux->threshold_time);
355 GST_OBJECT_UNLOCK (splitmux);
357 case PROP_MUXER_OVERHEAD:
358 GST_OBJECT_LOCK (splitmux);
359 g_value_set_double (value, splitmux->mux_overhead);
360 GST_OBJECT_UNLOCK (splitmux);
363 GST_OBJECT_LOCK (splitmux);
364 g_value_set_object (value, splitmux->provided_sink);
365 GST_OBJECT_UNLOCK (splitmux);
368 GST_OBJECT_LOCK (splitmux);
369 g_value_set_object (value, splitmux->provided_muxer);
370 GST_OBJECT_UNLOCK (splitmux);
373 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
379 mq_sink_to_src (GstElement * mq, GstPad * sink_pad)
381 gchar *tmp, *sinkname, *srcname;
384 sinkname = gst_pad_get_name (sink_pad);
386 srcname = g_strdup_printf ("src_%s", tmp);
388 mq_src = gst_element_get_static_pad (mq, srcname);
397 get_pads_from_mq (GstSplitMuxSink * splitmux, GstPad ** sink_pad,
403 /* Request a pad from multiqueue, then connect this one, then
404 * discover the corresponding output pad and return both */
405 mq_sink = gst_element_get_request_pad (splitmux->mq, "sink_%u");
409 mq_src = mq_sink_to_src (splitmux->mq, mq_sink);
419 gst_element_release_request_pad (splitmux->mq, mq_sink);
424 mq_stream_ctx_new (GstSplitMuxSink * splitmux)
428 ctx = g_new0 (MqStreamCtx, 1);
429 g_atomic_int_set (&ctx->refcount, 1);
430 ctx->splitmux = splitmux;
431 gst_segment_init (&ctx->in_segment, GST_FORMAT_UNDEFINED);
432 gst_segment_init (&ctx->out_segment, GST_FORMAT_UNDEFINED);
433 ctx->in_running_time = ctx->out_running_time = 0;
434 g_queue_init (&ctx->queued_bufs);
439 mq_stream_ctx_free (MqStreamCtx * ctx)
441 g_queue_foreach (&ctx->queued_bufs, (GFunc) mq_stream_buf_free, NULL);
442 g_queue_clear (&ctx->queued_bufs);
447 mq_stream_ctx_unref (MqStreamCtx * ctx)
449 if (g_atomic_int_dec_and_test (&ctx->refcount))
450 mq_stream_ctx_free (ctx);
454 mq_stream_ctx_ref (MqStreamCtx * ctx)
456 g_atomic_int_inc (&ctx->refcount);
460 _pad_block_destroy_sink_notify (MqStreamCtx * ctx)
462 ctx->sink_pad_block_id = 0;
463 mq_stream_ctx_unref (ctx);
467 _pad_block_destroy_src_notify (MqStreamCtx * ctx)
469 ctx->src_pad_block_id = 0;
470 mq_stream_ctx_unref (ctx);
474 send_fragment_opened_closed_msg (GstSplitMuxSink * splitmux, gboolean opened)
476 gchar *location = NULL;
478 const gchar *msg_name = opened ?
479 "splitmuxsink-fragment-opened" : "splitmuxsink-fragment-closed";
481 g_object_get (splitmux->sink, "location", &location, NULL);
483 msg = gst_message_new_element (GST_OBJECT (splitmux),
484 gst_structure_new (msg_name,
485 "location", G_TYPE_STRING, location,
486 "running-time", GST_TYPE_CLOCK_TIME,
487 splitmux->reference_ctx->out_running_time, NULL));
488 gst_element_post_message (GST_ELEMENT_CAST (splitmux), msg);
493 /* Called with lock held, drops the lock to send EOS to the
497 send_eos (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
502 eos = gst_event_new_eos ();
503 pad = gst_pad_get_peer (ctx->srcpad);
507 GST_INFO_OBJECT (splitmux, "Sending EOS on %" GST_PTR_FORMAT, pad);
508 GST_SPLITMUX_UNLOCK (splitmux);
509 gst_pad_send_event (pad, eos);
510 GST_SPLITMUX_LOCK (splitmux);
512 gst_object_unref (pad);
515 /* Called with splitmux lock held to check if this output
516 * context needs to sleep to wait for the release of the
517 * next GOP, or to send EOS to close out the current file
520 complete_or_wait_on_out (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
524 GST_LOG_OBJECT (ctx->srcpad,
525 "Checking running time %" GST_TIME_FORMAT " against max %"
526 GST_TIME_FORMAT, GST_TIME_ARGS (ctx->out_running_time),
527 GST_TIME_ARGS (splitmux->max_out_running_time));
529 if (splitmux->max_out_running_time == GST_CLOCK_TIME_NONE ||
530 ctx->out_running_time < splitmux->max_out_running_time) {
531 splitmux->have_muxed_something = TRUE;
535 if (ctx->flushing || splitmux->state == SPLITMUX_STATE_STOPPED)
538 if (splitmux->state == SPLITMUX_STATE_ENDING_FILE) {
539 if (ctx->out_eos == FALSE) {
540 send_eos (splitmux, ctx);
543 } else if (splitmux->state == SPLITMUX_STATE_START_NEXT_FRAGMENT) {
544 start_next_fragment (splitmux);
548 GST_INFO_OBJECT (ctx->srcpad,
549 "Sleeping for running time %"
550 GST_TIME_FORMAT " (max %" GST_TIME_FORMAT ")",
551 GST_TIME_ARGS (ctx->out_running_time),
552 GST_TIME_ARGS (splitmux->max_out_running_time));
553 ctx->out_blocked = TRUE;
554 /* Expand the mq if needed before sleeping */
555 check_queue_length (splitmux, ctx);
556 GST_SPLITMUX_WAIT (splitmux);
557 ctx->out_blocked = FALSE;
558 GST_INFO_OBJECT (ctx->srcpad,
559 "Woken for new max running time %" GST_TIME_FORMAT,
560 GST_TIME_ARGS (splitmux->max_out_running_time));
564 static GstPadProbeReturn
565 handle_mq_output (GstPad * pad, GstPadProbeInfo * info, MqStreamCtx * ctx)
567 GstSplitMuxSink *splitmux = ctx->splitmux;
568 MqStreamBuf *buf_info = NULL;
570 GST_LOG_OBJECT (pad, "Fired probe type 0x%x\n", info->type);
572 /* FIXME: Handle buffer lists, until then make it clear they won't work */
573 if (info->type & GST_PAD_PROBE_TYPE_BUFFER_LIST) {
574 g_warning ("Buffer list handling not implemented");
575 return GST_PAD_PROBE_DROP;
577 if (info->type & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) {
578 GstEvent *event = gst_pad_probe_info_get_event (info);
580 GST_LOG_OBJECT (pad, "Event %" GST_PTR_FORMAT, event);
582 switch (GST_EVENT_TYPE (event)) {
583 case GST_EVENT_SEGMENT:
584 gst_event_copy_segment (event, &ctx->out_segment);
586 case GST_EVENT_FLUSH_STOP:
587 GST_SPLITMUX_LOCK (splitmux);
588 gst_segment_init (&ctx->out_segment, GST_FORMAT_UNDEFINED);
589 g_queue_foreach (&ctx->queued_bufs, (GFunc) mq_stream_buf_free, NULL);
590 g_queue_clear (&ctx->queued_bufs);
591 ctx->flushing = FALSE;
592 GST_SPLITMUX_UNLOCK (splitmux);
594 case GST_EVENT_FLUSH_START:
595 GST_SPLITMUX_LOCK (splitmux);
596 GST_LOG_OBJECT (pad, "Flush start");
597 ctx->flushing = TRUE;
598 GST_SPLITMUX_BROADCAST (splitmux);
599 GST_SPLITMUX_UNLOCK (splitmux);
602 GST_SPLITMUX_LOCK (splitmux);
603 if (splitmux->state == SPLITMUX_STATE_STOPPED)
606 GST_SPLITMUX_UNLOCK (splitmux);
611 gst_event_parse_gap (event, &gap_ts, NULL);
612 if (gap_ts == GST_CLOCK_TIME_NONE)
615 GST_SPLITMUX_LOCK (splitmux);
617 gap_ts = gst_segment_to_running_time (&ctx->out_segment,
618 GST_FORMAT_TIME, gap_ts);
620 GST_LOG_OBJECT (pad, "Have GAP w/ ts %" GST_TIME_FORMAT,
621 GST_TIME_ARGS (gap_ts));
623 if (splitmux->state == SPLITMUX_STATE_STOPPED)
625 ctx->out_running_time = gap_ts;
626 complete_or_wait_on_out (splitmux, ctx);
627 GST_SPLITMUX_UNLOCK (splitmux);
630 case GST_EVENT_CUSTOM_DOWNSTREAM:{
631 const GstStructure *s;
634 s = gst_event_get_structure (event);
635 if (!gst_structure_has_name (s, "splitmuxsink-unblock"))
638 gst_structure_get_uint64 (s, "timestamp", &ts);
640 GST_SPLITMUX_LOCK (splitmux);
642 if (splitmux->state == SPLITMUX_STATE_STOPPED)
644 ctx->out_running_time = ts;
645 complete_or_wait_on_out (splitmux, ctx);
646 GST_SPLITMUX_UNLOCK (splitmux);
647 return GST_PAD_PROBE_DROP;
652 return GST_PAD_PROBE_PASS;
655 /* Allow everything through until the configured next stopping point */
656 GST_SPLITMUX_LOCK (splitmux);
658 buf_info = g_queue_pop_tail (&ctx->queued_bufs);
659 if (buf_info == NULL)
660 /* Can only happen due to a poorly timed flush */
663 /* If we have popped a keyframe, decrement the queued_gop count */
664 if (buf_info->keyframe && splitmux->queued_gops > 0)
665 splitmux->queued_gops--;
667 ctx->out_running_time = buf_info->run_ts;
669 GST_LOG_OBJECT (splitmux,
670 "Pad %" GST_PTR_FORMAT " buffer with TS %" GST_TIME_FORMAT
671 " size %" G_GSIZE_FORMAT,
672 pad, GST_TIME_ARGS (ctx->out_running_time), buf_info->buf_size);
674 if (splitmux->opening_first_fragment) {
675 send_fragment_opened_closed_msg (splitmux, TRUE);
676 splitmux->opening_first_fragment = FALSE;
679 complete_or_wait_on_out (splitmux, ctx);
681 if (splitmux->muxed_out_time == GST_CLOCK_TIME_NONE ||
682 splitmux->muxed_out_time < buf_info->run_ts)
683 splitmux->muxed_out_time = buf_info->run_ts;
685 splitmux->muxed_out_bytes += buf_info->buf_size;
687 #ifndef GST_DISABLE_GST_DEBUG
689 GstBuffer *buf = gst_pad_probe_info_get_buffer (info);
690 GST_LOG_OBJECT (pad, "Returning to pass buffer %" GST_PTR_FORMAT
691 " run ts %" GST_TIME_FORMAT, buf,
692 GST_TIME_ARGS (ctx->out_running_time));
696 GST_SPLITMUX_UNLOCK (splitmux);
698 mq_stream_buf_free (buf_info);
700 return GST_PAD_PROBE_PASS;
703 GST_SPLITMUX_UNLOCK (splitmux);
704 return GST_PAD_PROBE_DROP;
708 resend_sticky (GstPad * pad, GstEvent ** event, GstPad * peer)
710 return gst_pad_send_event (peer, gst_event_ref (*event));
714 restart_context (MqStreamCtx * ctx, GstSplitMuxSink * splitmux)
716 GstPad *peer = gst_pad_get_peer (ctx->srcpad);
718 gst_pad_sticky_events_foreach (ctx->srcpad,
719 (GstPadStickyEventsForeachFunction) (resend_sticky), peer);
721 /* Clear EOS flag if not actually EOS */
722 ctx->out_eos = GST_PAD_IS_EOS (ctx->srcpad);
724 gst_object_unref (peer);
727 /* Called with lock held when a fragment
728 * reaches EOS and it is time to restart
732 start_next_fragment (GstSplitMuxSink * splitmux)
734 /* 1 change to new file */
735 gst_element_set_state (splitmux->muxer, GST_STATE_NULL);
736 gst_element_set_state (splitmux->active_sink, GST_STATE_NULL);
738 set_next_filename (splitmux);
740 gst_element_sync_state_with_parent (splitmux->active_sink);
741 gst_element_sync_state_with_parent (splitmux->muxer);
743 g_list_foreach (splitmux->contexts, (GFunc) restart_context, splitmux);
745 /* Switch state and go back to processing */
746 if (!splitmux->reference_ctx->in_eos) {
747 splitmux->state = SPLITMUX_STATE_COLLECTING_GOP_START;
748 splitmux->max_out_running_time = splitmux->reference_ctx->in_running_time;
750 splitmux->state = SPLITMUX_STATE_WAITING_GOP_COMPLETE;
751 splitmux->max_out_running_time = GST_CLOCK_TIME_NONE;
752 splitmux->have_muxed_something = FALSE;
754 splitmux->have_muxed_something =
755 (splitmux->reference_ctx->in_running_time > splitmux->muxed_out_time);
757 /* Store the overflow parameters as the basis for the next fragment */
758 splitmux->mux_start_time = splitmux->muxed_out_time;
759 splitmux->mux_start_bytes = splitmux->muxed_out_bytes;
761 GST_DEBUG_OBJECT (splitmux,
762 "Restarting flow for new fragment. New running time %" GST_TIME_FORMAT,
763 GST_TIME_ARGS (splitmux->max_out_running_time));
765 send_fragment_opened_closed_msg (splitmux, TRUE);
767 GST_SPLITMUX_BROADCAST (splitmux);
771 bus_handler (GstBin * bin, GstMessage * message)
773 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (bin);
775 switch (GST_MESSAGE_TYPE (message)) {
776 case GST_MESSAGE_EOS:
777 /* If the state is draining out the current file, drop this EOS */
778 GST_SPLITMUX_LOCK (splitmux);
780 send_fragment_opened_closed_msg (splitmux, FALSE);
782 if (splitmux->state == SPLITMUX_STATE_ENDING_FILE &&
783 splitmux->max_out_running_time != GST_CLOCK_TIME_NONE) {
784 GST_DEBUG_OBJECT (splitmux, "Caught EOS at end of fragment, dropping");
785 splitmux->state = SPLITMUX_STATE_START_NEXT_FRAGMENT;
786 GST_SPLITMUX_BROADCAST (splitmux);
788 gst_message_unref (message);
789 GST_SPLITMUX_UNLOCK (splitmux);
792 GST_SPLITMUX_UNLOCK (splitmux);
798 GST_BIN_CLASS (parent_class)->handle_message (bin, message);
801 /* Called with splitmux lock held */
802 /* Called when entering ProcessingCompleteGop state
803 * Assess if mq contents overflowed the current file
804 * -> If yes, need to switch to new file
805 * -> if no, set max_out_running_time to let this GOP in and
806 * go to COLLECTING_GOP_START state
809 handle_gathered_gop (GstSplitMuxSink * splitmux)
812 gsize queued_bytes = 0;
813 GstClockTime queued_time = 0;
814 gboolean at_eos = TRUE;
816 /* Assess if the multiqueue contents overflowed the current file */
817 for (cur = g_list_first (splitmux->contexts);
818 cur != NULL; cur = g_list_next (cur)) {
819 MqStreamCtx *tmpctx = (MqStreamCtx *) (cur->data);
820 if (tmpctx->in_running_time > queued_time)
821 queued_time = tmpctx->in_running_time;
822 queued_bytes += tmpctx->in_bytes;
823 at_eos &= tmpctx->in_eos;
826 g_assert (queued_bytes >= splitmux->mux_start_bytes);
827 g_assert (queued_time >= splitmux->mux_start_time);
829 queued_bytes -= splitmux->mux_start_bytes;
830 queued_time -= splitmux->mux_start_time;
832 /* Expand queued bytes estimate by muxer overhead */
833 queued_bytes += (queued_bytes * splitmux->mux_overhead);
835 GST_LOG_OBJECT (splitmux, "mq at TS %" GST_TIME_FORMAT
836 " bytes %" G_GSIZE_FORMAT, GST_TIME_ARGS (queued_time), queued_bytes);
838 /* Check for overrun - have we output at least one byte and overrun
839 * either threshold? */
840 if ((splitmux->have_muxed_something &&
841 ((splitmux->threshold_bytes > 0 &&
842 queued_bytes >= splitmux->threshold_bytes) ||
843 (splitmux->threshold_time > 0 &&
844 queued_time >= splitmux->threshold_time))) || at_eos) {
846 splitmux->state = SPLITMUX_STATE_ENDING_FILE;
848 GST_INFO_OBJECT (splitmux,
849 "mq overflowed since last, draining out. max out TS is %"
850 GST_TIME_FORMAT, GST_TIME_ARGS (splitmux->max_out_running_time));
851 GST_SPLITMUX_BROADCAST (splitmux);
855 GST_LOG_OBJECT (splitmux,
856 "This GOP didn't overflow the fragment. Bytes sent %" G_GSIZE_FORMAT
857 " queued %" G_GSIZE_FORMAT " time %" GST_TIME_FORMAT " Continuing.",
858 splitmux->muxed_out_bytes - splitmux->mux_start_bytes,
859 queued_bytes, GST_TIME_ARGS (queued_time));
861 /* Wake everyone up to push this one GOP, then sleep */
862 splitmux->have_muxed_something = TRUE;
864 if (!splitmux->reference_ctx->in_eos) {
865 splitmux->state = SPLITMUX_STATE_COLLECTING_GOP_START;
866 splitmux->max_out_running_time = splitmux->reference_ctx->in_running_time;
868 splitmux->state = SPLITMUX_STATE_WAITING_GOP_COMPLETE;
869 splitmux->max_out_running_time = GST_CLOCK_TIME_NONE;
872 GST_LOG_OBJECT (splitmux, "Waking output for complete GOP, TS %"
873 GST_TIME_FORMAT, GST_TIME_ARGS (splitmux->max_out_running_time));
874 GST_SPLITMUX_BROADCAST (splitmux);
879 /* Called with splitmux lock held */
880 /* Called from each input pad when it is has all the pieces
881 * for a GOP or EOS, starting with the reference pad which has set the
882 * splitmux->max_in_running_time
885 check_completed_gop (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
888 gboolean ready = TRUE;
889 GstClockTime current_max_in_running_time;
891 if (splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE) {
892 /* Iterate each pad, and check that the input running time is at least
893 * up to the reference running time, and if so handle the collected GOP */
894 GST_LOG_OBJECT (splitmux, "Checking GOP collected, Max in running time %"
895 GST_TIME_FORMAT " ctx %p",
896 GST_TIME_ARGS (splitmux->max_in_running_time), ctx);
897 for (cur = g_list_first (splitmux->contexts); cur != NULL;
898 cur = g_list_next (cur)) {
899 MqStreamCtx *tmpctx = (MqStreamCtx *) (cur->data);
901 GST_LOG_OBJECT (splitmux,
902 "Context %p (src pad %" GST_PTR_FORMAT ") TS %" GST_TIME_FORMAT
903 " EOS %d", tmpctx, tmpctx->srcpad,
904 GST_TIME_ARGS (tmpctx->in_running_time), tmpctx->in_eos);
906 if (splitmux->max_in_running_time != GST_CLOCK_TIME_NONE &&
907 tmpctx->in_running_time < splitmux->max_in_running_time &&
909 GST_LOG_OBJECT (splitmux,
910 "Context %p (src pad %" GST_PTR_FORMAT ") not ready. We'll sleep",
911 tmpctx, tmpctx->srcpad);
917 GST_DEBUG_OBJECT (splitmux,
918 "Collected GOP is complete. Processing (ctx %p)", ctx);
919 /* All pads have a complete GOP, release it into the multiqueue */
920 handle_gathered_gop (splitmux);
924 /* Some pad is not yet ready, or GOP is being pushed
925 * either way, sleep and wait to get woken */
926 current_max_in_running_time = splitmux->max_in_running_time;
927 while ((splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE ||
928 splitmux->state == SPLITMUX_STATE_START_NEXT_FRAGMENT) &&
930 (current_max_in_running_time == splitmux->max_in_running_time)) {
932 GST_LOG_OBJECT (splitmux, "Sleeping for %s (ctx %p)",
933 splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE ?
934 "GOP complete" : "EOF draining", ctx);
935 GST_SPLITMUX_WAIT (splitmux);
937 GST_LOG_OBJECT (splitmux, "Done waiting for complete GOP (ctx %p)", ctx);
942 check_queue_length (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
945 guint cur_len = g_queue_get_length (&ctx->queued_bufs);
947 GST_DEBUG_OBJECT (ctx->sinkpad,
948 "Checking queue length len %u cur_max %u queued gops %u",
949 cur_len, splitmux->mq_max_buffers, splitmux->queued_gops);
951 if (cur_len >= splitmux->mq_max_buffers) {
952 gboolean allow_grow = FALSE;
954 /* If collecting a GOP and this pad might block,
955 * and there isn't already a pending GOP in the queue
958 if (splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE &&
959 ctx->in_running_time < splitmux->max_in_running_time &&
960 splitmux->queued_gops <= 1) {
962 } else if (splitmux->state == SPLITMUX_STATE_COLLECTING_GOP_START &&
963 ctx->is_reference && splitmux->queued_gops <= 1) {
968 for (cur = g_list_first (splitmux->contexts);
969 cur != NULL; cur = g_list_next (cur)) {
970 MqStreamCtx *tmpctx = (MqStreamCtx *) (cur->data);
971 GST_DEBUG_OBJECT (tmpctx->sinkpad,
972 " len %u out_blocked %d",
973 g_queue_get_length (&tmpctx->queued_bufs), tmpctx->out_blocked);
974 /* If another stream is starving, grow */
975 if (tmpctx != ctx && g_queue_get_length (&tmpctx->queued_bufs) < 1) {
982 splitmux->mq_max_buffers = cur_len + 1;
984 GST_INFO_OBJECT (splitmux,
985 "Multiqueue overrun - enlarging to %u buffers ctx %p",
986 splitmux->mq_max_buffers, ctx);
988 g_object_set (splitmux->mq, "max-size-buffers",
989 splitmux->mq_max_buffers, NULL);
994 static GstPadProbeReturn
995 handle_mq_input (GstPad * pad, GstPadProbeInfo * info, MqStreamCtx * ctx)
997 GstSplitMuxSink *splitmux = ctx->splitmux;
999 MqStreamBuf *buf_info = NULL;
1001 gboolean loop_again;
1002 gboolean keyframe = FALSE;
1004 GST_LOG_OBJECT (pad, "Fired probe type 0x%x", info->type);
1006 /* FIXME: Handle buffer lists, until then make it clear they won't work */
1007 if (info->type & GST_PAD_PROBE_TYPE_BUFFER_LIST) {
1008 g_warning ("Buffer list handling not implemented");
1009 return GST_PAD_PROBE_DROP;
1011 if (info->type & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) {
1012 GstEvent *event = gst_pad_probe_info_get_event (info);
1013 switch (GST_EVENT_TYPE (event)) {
1014 case GST_EVENT_SEGMENT:
1015 gst_event_copy_segment (event, &ctx->in_segment);
1017 case GST_EVENT_FLUSH_STOP:
1018 GST_SPLITMUX_LOCK (splitmux);
1019 gst_segment_init (&ctx->in_segment, GST_FORMAT_UNDEFINED);
1020 ctx->in_eos = FALSE;
1022 ctx->in_running_time = 0;
1023 GST_SPLITMUX_UNLOCK (splitmux);
1026 GST_SPLITMUX_LOCK (splitmux);
1029 if (splitmux->state == SPLITMUX_STATE_STOPPED)
1032 if (ctx->is_reference) {
1033 GST_INFO_OBJECT (splitmux, "Got Reference EOS. Finishing up");
1034 /* Act as if this is a new keyframe with infinite timestamp */
1035 splitmux->max_in_running_time = GST_CLOCK_TIME_NONE;
1036 splitmux->state = SPLITMUX_STATE_WAITING_GOP_COMPLETE;
1037 /* Wake up other input pads to collect this GOP */
1038 GST_SPLITMUX_BROADCAST (splitmux);
1039 check_completed_gop (splitmux, ctx);
1040 } else if (splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE) {
1041 /* If we are waiting for a GOP to be completed (ie, for aux
1042 * pads to catch up), then this pad is complete, so check
1043 * if the whole GOP is.
1045 check_completed_gop (splitmux, ctx);
1047 GST_SPLITMUX_UNLOCK (splitmux);
1052 return GST_PAD_PROBE_PASS;
1055 buf = gst_pad_probe_info_get_buffer (info);
1056 buf_info = mq_stream_buf_new ();
1058 if (GST_BUFFER_PTS_IS_VALID (buf))
1059 ts = GST_BUFFER_PTS (buf);
1061 ts = GST_BUFFER_DTS (buf);
1063 GST_SPLITMUX_LOCK (splitmux);
1065 if (splitmux->state == SPLITMUX_STATE_STOPPED)
1068 /* If this buffer has a timestamp, advance the input timestamp of the
1070 if (GST_CLOCK_TIME_IS_VALID (ts)) {
1071 GstClockTime running_time =
1072 gst_segment_to_running_time (&ctx->in_segment, GST_FORMAT_TIME,
1073 GST_BUFFER_TIMESTAMP (buf));
1075 if (GST_CLOCK_TIME_IS_VALID (running_time) &&
1076 (ctx->in_running_time == GST_CLOCK_TIME_NONE
1077 || running_time > ctx->in_running_time))
1078 ctx->in_running_time = running_time;
1081 /* Try to make sure we have a valid running time */
1082 if (!GST_CLOCK_TIME_IS_VALID (ctx->in_running_time)) {
1083 ctx->in_running_time =
1084 gst_segment_to_running_time (&ctx->in_segment, GST_FORMAT_TIME,
1085 ctx->in_segment.start);
1088 buf_info->run_ts = ctx->in_running_time;
1089 buf_info->buf_size = gst_buffer_get_size (buf);
1091 /* Update total input byte counter for overflow detect */
1092 ctx->in_bytes += buf_info->buf_size;
1094 /* initialize mux_start_time */
1095 if (ctx->is_reference && splitmux->mux_start_time == 0)
1096 splitmux->mux_start_time = buf_info->run_ts;
1098 GST_DEBUG_OBJECT (pad, "Buf TS %" GST_TIME_FORMAT
1099 " total in_bytes %" G_GSIZE_FORMAT,
1100 GST_TIME_ARGS (buf_info->run_ts), ctx->in_bytes);
1107 switch (splitmux->state) {
1108 case SPLITMUX_STATE_COLLECTING_GOP_START:
1109 if (ctx->is_reference) {
1110 /* If a keyframe, we have a complete GOP */
1111 if (GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT) ||
1112 !GST_CLOCK_TIME_IS_VALID (ctx->in_running_time) ||
1113 splitmux->max_in_running_time >= ctx->in_running_time) {
1114 /* Pass this buffer through */
1118 GST_INFO_OBJECT (pad,
1119 "Have keyframe with running time %" GST_TIME_FORMAT,
1120 GST_TIME_ARGS (ctx->in_running_time));
1122 splitmux->state = SPLITMUX_STATE_WAITING_GOP_COMPLETE;
1123 splitmux->max_in_running_time = ctx->in_running_time;
1124 /* Wake up other input pads to collect this GOP */
1125 GST_SPLITMUX_BROADCAST (splitmux);
1126 check_completed_gop (splitmux, ctx);
1128 /* We're still waiting for a keyframe on the reference pad, sleep */
1129 GST_LOG_OBJECT (pad, "Sleeping for GOP start");
1130 GST_SPLITMUX_WAIT (splitmux);
1131 GST_LOG_OBJECT (pad, "Done sleeping for GOP start state now %d",
1135 case SPLITMUX_STATE_WAITING_GOP_COMPLETE:
1136 /* After a GOP start is found, this buffer might complete the GOP */
1137 /* If we overran the target timestamp, it might be time to process
1138 * the GOP, otherwise bail out for more data
1140 GST_LOG_OBJECT (pad,
1141 "Checking TS %" GST_TIME_FORMAT " against max %" GST_TIME_FORMAT,
1142 GST_TIME_ARGS (ctx->in_running_time),
1143 GST_TIME_ARGS (splitmux->max_in_running_time));
1145 if (ctx->in_running_time < splitmux->max_in_running_time) {
1150 GST_LOG_OBJECT (pad,
1151 "Collected last packet of GOP. Checking other pads");
1152 check_completed_gop (splitmux, ctx);
1154 case SPLITMUX_STATE_ENDING_FILE:{
1157 /* If somes streams received no buffer during the last GOP that overran,
1158 * because its next buffer has a timestamp bigger than
1159 * ctx->max_in_running_time, its queue is empty. In that case the only
1160 * way to wakeup the output thread is by injecting an event in the
1161 * queue. This usually happen with subtitle streams.
1162 * See https://bugzilla.gnome.org/show_bug.cgi?id=763711. */
1163 GST_LOG_OBJECT (pad, "Sending splitmuxsink-unblock event");
1164 event = gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM |
1165 GST_EVENT_TYPE_SERIALIZED,
1166 gst_structure_new ("splitmuxsink-unblock", "timestamp",
1167 G_TYPE_UINT64, splitmux->max_in_running_time, NULL));
1168 gst_pad_send_event (ctx->sinkpad, event);
1171 case SPLITMUX_STATE_START_NEXT_FRAGMENT:
1172 /* A fragment is ending, wait until that's done before continuing */
1173 GST_DEBUG_OBJECT (pad, "Sleeping for fragment restart");
1174 GST_SPLITMUX_WAIT (splitmux);
1175 GST_DEBUG_OBJECT (pad,
1176 "Done sleeping for fragment restart state now %d", splitmux->state);
1182 } while (loop_again);
1185 splitmux->queued_gops++;
1186 buf_info->keyframe = TRUE;
1189 /* Now add this buffer to the queue just before returning */
1190 g_queue_push_head (&ctx->queued_bufs, buf_info);
1192 /* Check the buffer will fit in the mq */
1193 check_queue_length (splitmux, ctx);
1195 GST_LOG_OBJECT (pad, "Returning to queue buffer %" GST_PTR_FORMAT
1196 " run ts %" GST_TIME_FORMAT, buf, GST_TIME_ARGS (ctx->in_running_time));
1198 GST_SPLITMUX_UNLOCK (splitmux);
1199 return GST_PAD_PROBE_PASS;
1202 GST_SPLITMUX_UNLOCK (splitmux);
1204 mq_stream_buf_free (buf_info);
1205 return GST_PAD_PROBE_PASS;
1209 gst_splitmux_sink_request_new_pad (GstElement * element,
1210 GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
1212 GstSplitMuxSink *splitmux = (GstSplitMuxSink *) element;
1213 GstPadTemplate *mux_template = NULL;
1215 GstPad *mq_sink, *mq_src;
1217 gboolean is_video = FALSE;
1220 GST_DEBUG_OBJECT (element, "templ:%s, name:%s", templ->name_template, name);
1222 GST_SPLITMUX_LOCK (splitmux);
1223 if (!create_elements (splitmux))
1226 if (templ->name_template) {
1227 if (g_str_equal (templ->name_template, "video")) {
1228 /* FIXME: Look for a pad template with matching caps, rather than by name */
1230 gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS
1231 (splitmux->muxer), "video_%u");
1236 gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS
1237 (splitmux->muxer), templ->name_template);
1239 if (mux_template == NULL) {
1240 /* Fallback to find sink pad templates named 'sink_%d' (mpegtsmux) */
1242 gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS
1243 (splitmux->muxer), "sink_%d");
1247 res = gst_element_request_pad (splitmux->muxer, mux_template, name, caps);
1252 gname = g_strdup ("video");
1253 else if (name == NULL)
1254 gname = gst_pad_get_name (res);
1256 gname = g_strdup (name);
1258 if (!get_pads_from_mq (splitmux, &mq_sink, &mq_src)) {
1259 gst_element_release_request_pad (splitmux->muxer, res);
1260 gst_object_unref (GST_OBJECT (res));
1264 if (gst_pad_link (mq_src, res) != GST_PAD_LINK_OK) {
1265 gst_element_release_request_pad (splitmux->muxer, res);
1266 gst_object_unref (GST_OBJECT (res));
1267 gst_element_release_request_pad (splitmux->mq, mq_sink);
1268 gst_object_unref (GST_OBJECT (mq_sink));
1272 gst_object_unref (GST_OBJECT (res));
1274 ctx = mq_stream_ctx_new (splitmux);
1275 ctx->srcpad = mq_src;
1276 ctx->sinkpad = mq_sink;
1278 mq_stream_ctx_ref (ctx);
1279 ctx->src_pad_block_id =
1280 gst_pad_add_probe (mq_src, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM,
1281 (GstPadProbeCallback) handle_mq_output, ctx, (GDestroyNotify)
1282 _pad_block_destroy_src_notify);
1283 if (is_video && splitmux->reference_ctx != NULL) {
1284 splitmux->reference_ctx->is_reference = FALSE;
1285 splitmux->reference_ctx = NULL;
1287 if (splitmux->reference_ctx == NULL) {
1288 splitmux->reference_ctx = ctx;
1289 ctx->is_reference = TRUE;
1292 res = gst_ghost_pad_new (gname, mq_sink);
1293 g_object_set_qdata ((GObject *) (res), PAD_CONTEXT, ctx);
1295 mq_stream_ctx_ref (ctx);
1296 ctx->sink_pad_block_id =
1297 gst_pad_add_probe (mq_sink, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM,
1298 (GstPadProbeCallback) handle_mq_input, ctx, (GDestroyNotify)
1299 _pad_block_destroy_sink_notify);
1301 GST_DEBUG_OBJECT (splitmux, "Request pad %" GST_PTR_FORMAT
1302 " is mq pad %" GST_PTR_FORMAT, res, mq_sink);
1304 splitmux->contexts = g_list_prepend (splitmux->contexts, ctx);
1308 gst_object_unref (mq_sink);
1309 gst_object_unref (mq_src);
1311 gst_pad_set_active (res, TRUE);
1312 gst_element_add_pad (element, res);
1313 GST_SPLITMUX_UNLOCK (splitmux);
1317 GST_SPLITMUX_UNLOCK (splitmux);
1322 gst_splitmux_sink_release_pad (GstElement * element, GstPad * pad)
1324 GstSplitMuxSink *splitmux = (GstSplitMuxSink *) element;
1325 GstPad *mqsink, *mqsrc, *muxpad;
1327 (MqStreamCtx *) (g_object_get_qdata ((GObject *) (pad), PAD_CONTEXT));
1329 GST_SPLITMUX_LOCK (splitmux);
1331 if (splitmux->muxer == NULL || splitmux->mq == NULL)
1332 goto fail; /* Elements don't exist yet - nothing to release */
1334 GST_INFO_OBJECT (pad, "releasing request pad");
1336 mqsink = gst_ghost_pad_get_target (GST_GHOST_PAD (pad));
1337 mqsrc = mq_sink_to_src (splitmux->mq, mqsink);
1338 muxpad = gst_pad_get_peer (mqsrc);
1340 /* Remove the context from our consideration */
1341 splitmux->contexts = g_list_remove (splitmux->contexts, ctx);
1343 if (ctx->sink_pad_block_id)
1344 gst_pad_remove_probe (ctx->sinkpad, ctx->sink_pad_block_id);
1346 if (ctx->src_pad_block_id)
1347 gst_pad_remove_probe (ctx->srcpad, ctx->src_pad_block_id);
1349 /* Can release the context now */
1350 mq_stream_ctx_unref (ctx);
1352 /* Release and free the mq input */
1353 gst_element_release_request_pad (splitmux->mq, mqsink);
1355 /* Release and free the muxer input */
1356 gst_element_release_request_pad (splitmux->muxer, muxpad);
1358 gst_object_unref (mqsink);
1359 gst_object_unref (mqsrc);
1360 gst_object_unref (muxpad);
1362 gst_element_remove_pad (element, pad);
1364 /* Reset the internal elements only after all request pads are released */
1365 if (splitmux->contexts == NULL)
1366 gst_splitmux_reset (splitmux);
1369 GST_SPLITMUX_UNLOCK (splitmux);
1373 create_element (GstSplitMuxSink * splitmux,
1374 const gchar * factory, const gchar * name)
1376 GstElement *ret = gst_element_factory_make (factory, name);
1378 g_warning ("Failed to create %s - splitmuxsink will not work", name);
1382 if (!gst_bin_add (GST_BIN (splitmux), ret)) {
1383 g_warning ("Could not add %s element - splitmuxsink will not work", name);
1384 gst_object_unref (ret);
1392 create_elements (GstSplitMuxSink * splitmux)
1394 /* Create internal elements */
1395 if (splitmux->mq == NULL) {
1397 create_element (splitmux, "multiqueue", "multiqueue")) == NULL)
1400 splitmux->mq_max_buffers = 5;
1401 /* No bytes or time limit, we limit buffers manually */
1402 g_object_set (splitmux->mq, "max-size-bytes", 0, "max-size-time",
1403 (guint64) 0, "max-size-buffers", splitmux->mq_max_buffers, NULL);
1406 if (splitmux->muxer == NULL) {
1407 GstElement *provided_muxer = NULL;
1409 GST_OBJECT_LOCK (splitmux);
1410 if (splitmux->provided_muxer != NULL)
1411 provided_muxer = gst_object_ref (splitmux->provided_muxer);
1412 GST_OBJECT_UNLOCK (splitmux);
1414 if (provided_muxer == NULL) {
1415 if ((splitmux->muxer =
1416 create_element (splitmux, "mp4mux", "muxer")) == NULL)
1419 if (!gst_bin_add (GST_BIN (splitmux), provided_muxer)) {
1420 g_warning ("Could not add muxer element - splitmuxsink will not work");
1421 gst_object_unref (provided_muxer);
1425 splitmux->muxer = provided_muxer;
1426 gst_object_unref (provided_muxer);
1436 find_sink (GstElement * e)
1438 GstElement *res = NULL;
1440 gboolean done = FALSE;
1441 GValue data = { 0, };
1443 if (!GST_IS_BIN (e))
1446 iter = gst_bin_iterate_sinks (GST_BIN (e));
1448 switch (gst_iterator_next (iter, &data)) {
1449 case GST_ITERATOR_OK:
1451 GstElement *child = g_value_get_object (&data);
1452 if (g_object_class_find_property (G_OBJECT_GET_CLASS (child),
1453 "location") != NULL) {
1457 g_value_reset (&data);
1460 case GST_ITERATOR_RESYNC:
1461 gst_iterator_resync (iter);
1463 case GST_ITERATOR_DONE:
1466 case GST_ITERATOR_ERROR:
1467 g_assert_not_reached ();
1471 g_value_unset (&data);
1472 gst_iterator_free (iter);
1478 create_sink (GstSplitMuxSink * splitmux)
1480 GstElement *provided_sink = NULL;
1482 if (splitmux->active_sink == NULL) {
1484 GST_OBJECT_LOCK (splitmux);
1485 if (splitmux->provided_sink != NULL)
1486 provided_sink = gst_object_ref (splitmux->provided_sink);
1487 GST_OBJECT_UNLOCK (splitmux);
1489 if (provided_sink == NULL) {
1490 if ((splitmux->sink =
1491 create_element (splitmux, DEFAULT_SINK, "sink")) == NULL)
1493 splitmux->active_sink = splitmux->sink;
1495 if (!gst_bin_add (GST_BIN (splitmux), provided_sink)) {
1496 g_warning ("Could not add sink elements - splitmuxsink will not work");
1497 gst_object_unref (provided_sink);
1501 splitmux->active_sink = provided_sink;
1503 /* The bin holds a ref now, we can drop our tmp ref */
1504 gst_object_unref (provided_sink);
1506 /* Find the sink element */
1507 splitmux->sink = find_sink (splitmux->active_sink);
1508 if (splitmux->sink == NULL) {
1510 ("Could not locate sink element in provided sink - splitmuxsink will not work");
1515 if (!gst_element_link (splitmux->muxer, splitmux->active_sink)) {
1516 g_warning ("Failed to link muxer and sink- splitmuxsink will not work");
1527 #pragma GCC diagnostic ignored "-Wformat-nonliteral"
1530 set_next_filename (GstSplitMuxSink * splitmux)
1532 gchar *fname = NULL;
1534 g_signal_emit (splitmux, signals[SIGNAL_FORMAT_LOCATION], 0,
1535 splitmux->fragment_id, &fname);
1538 fname = splitmux->location ?
1539 g_strdup_printf (splitmux->location, splitmux->fragment_id) : NULL;
1542 GST_INFO_OBJECT (splitmux, "Setting file to %s", fname);
1543 g_object_set (splitmux->sink, "location", fname, NULL);
1546 splitmux->fragment_id++;
1550 static GstStateChangeReturn
1551 gst_splitmux_sink_change_state (GstElement * element, GstStateChange transition)
1553 GstStateChangeReturn ret;
1554 GstSplitMuxSink *splitmux = (GstSplitMuxSink *) element;
1556 switch (transition) {
1557 case GST_STATE_CHANGE_NULL_TO_READY:{
1558 GST_SPLITMUX_LOCK (splitmux);
1559 if (!create_elements (splitmux) || !create_sink (splitmux)) {
1560 ret = GST_STATE_CHANGE_FAILURE;
1561 GST_SPLITMUX_UNLOCK (splitmux);
1564 GST_SPLITMUX_UNLOCK (splitmux);
1565 splitmux->fragment_id = 0;
1566 set_next_filename (splitmux);
1569 case GST_STATE_CHANGE_READY_TO_PAUSED:{
1570 GST_SPLITMUX_LOCK (splitmux);
1571 /* Start by collecting one input on each pad */
1572 splitmux->state = SPLITMUX_STATE_COLLECTING_GOP_START;
1573 splitmux->max_in_running_time = 0;
1574 splitmux->muxed_out_time = splitmux->mux_start_time = 0;
1575 splitmux->muxed_out_bytes = splitmux->mux_start_bytes = 0;
1576 splitmux->opening_first_fragment = TRUE;
1577 GST_SPLITMUX_UNLOCK (splitmux);
1580 case GST_STATE_CHANGE_PAUSED_TO_READY:
1581 case GST_STATE_CHANGE_READY_TO_NULL:
1582 GST_SPLITMUX_LOCK (splitmux);
1583 splitmux->state = SPLITMUX_STATE_STOPPED;
1584 /* Wake up any blocked threads */
1585 GST_LOG_OBJECT (splitmux,
1586 "State change -> NULL or READY. Waking threads");
1587 GST_SPLITMUX_BROADCAST (splitmux);
1588 GST_SPLITMUX_UNLOCK (splitmux);
1594 ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1595 if (ret == GST_STATE_CHANGE_FAILURE)
1598 switch (transition) {
1599 case GST_STATE_CHANGE_READY_TO_NULL:
1600 GST_SPLITMUX_LOCK (splitmux);
1601 splitmux->fragment_id = 0;
1602 /* Reset internal elements only if no pad contexts are using them */
1603 if (splitmux->contexts == NULL)
1604 gst_splitmux_reset (splitmux);
1605 GST_SPLITMUX_UNLOCK (splitmux);
1613 if (transition == GST_STATE_CHANGE_NULL_TO_READY &&
1614 ret == GST_STATE_CHANGE_FAILURE) {
1615 /* Cleanup elements on failed transition out of NULL */
1616 gst_splitmux_reset (splitmux);
1622 register_splitmuxsink (GstPlugin * plugin)
1624 GST_DEBUG_CATEGORY_INIT (splitmux_debug, "splitmuxsink", 0,
1625 "Split File Muxing Sink");
1627 return gst_element_register (plugin, "splitmuxsink", GST_RANK_NONE,
1628 GST_TYPE_SPLITMUX_SINK);