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 * is required, 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 * 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.
41 * <title>Example pipelines</title>
43 * 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
45 * Records a video stream captured from a v4l2 device and muxes it into
46 * ISO mp4 files, splitting as needed to limit size/duration to 10 seconds
47 * and 1MB maximum size.
56 #include "gstsplitmuxsink.h"
58 GST_DEBUG_CATEGORY_STATIC (splitmux_debug);
59 #define GST_CAT_DEFAULT splitmux_debug
61 #define GST_SPLITMUX_LOCK(s) g_mutex_lock(&(s)->lock)
62 #define GST_SPLITMUX_UNLOCK(s) g_mutex_unlock(&(s)->lock)
63 #define GST_SPLITMUX_WAIT(s) g_cond_wait (&(s)->data_cond, &(s)->lock)
64 #define GST_SPLITMUX_BROADCAST(s) g_cond_broadcast (&(s)->data_cond)
77 #define DEFAULT_MAX_SIZE_TIME 0
78 #define DEFAULT_MAX_SIZE_BYTES 0
79 #define DEFAULT_MUXER_OVERHEAD 0.02
80 #define DEFAULT_MUXER "mp4mux"
81 #define DEFAULT_SINK "filesink"
85 SIGNAL_FORMAT_LOCATION,
89 static guint signals[SIGNAL_LAST];
91 static GstStaticPadTemplate video_sink_template =
92 GST_STATIC_PAD_TEMPLATE ("video",
96 static GstStaticPadTemplate audio_sink_template =
97 GST_STATIC_PAD_TEMPLATE ("audio_%u",
100 GST_STATIC_CAPS_ANY);
101 static GstStaticPadTemplate subtitle_sink_template =
102 GST_STATIC_PAD_TEMPLATE ("subtitle_%u",
105 GST_STATIC_CAPS_ANY);
107 static GQuark PAD_CONTEXT;
112 PAD_CONTEXT = g_quark_from_static_string ("pad-context");
115 #define gst_splitmux_sink_parent_class parent_class
116 G_DEFINE_TYPE_EXTENDED (GstSplitMuxSink, gst_splitmux_sink, GST_TYPE_BIN, 0,
119 static gboolean create_elements (GstSplitMuxSink * splitmux);
120 static gboolean create_sink (GstSplitMuxSink * splitmux);
121 static void gst_splitmux_sink_set_property (GObject * object, guint prop_id,
122 const GValue * value, GParamSpec * pspec);
123 static void gst_splitmux_sink_get_property (GObject * object, guint prop_id,
124 GValue * value, GParamSpec * pspec);
125 static void gst_splitmux_sink_dispose (GObject * object);
126 static void gst_splitmux_sink_finalize (GObject * object);
128 static GstPad *gst_splitmux_sink_request_new_pad (GstElement * element,
129 GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
130 static void gst_splitmux_sink_release_pad (GstElement * element, GstPad * pad);
132 static GstStateChangeReturn gst_splitmux_sink_change_state (GstElement *
133 element, GstStateChange transition);
135 static void bus_handler (GstBin * bin, GstMessage * msg);
136 static void set_next_filename (GstSplitMuxSink * splitmux);
137 static void start_next_fragment (GstSplitMuxSink * splitmux);
138 static void check_queue_length (GstSplitMuxSink * splitmux, MqStreamCtx * ctx);
139 static void mq_stream_ctx_unref (MqStreamCtx * ctx);
142 mq_stream_buf_new (void)
144 return g_slice_new0 (MqStreamBuf);
148 mq_stream_buf_free (MqStreamBuf * data)
150 g_slice_free (MqStreamBuf, data);
154 gst_splitmux_sink_class_init (GstSplitMuxSinkClass * klass)
156 GObjectClass *gobject_class = (GObjectClass *) klass;
157 GstElementClass *gstelement_class = (GstElementClass *) klass;
158 GstBinClass *gstbin_class = (GstBinClass *) klass;
160 gobject_class->set_property = gst_splitmux_sink_set_property;
161 gobject_class->get_property = gst_splitmux_sink_get_property;
162 gobject_class->dispose = gst_splitmux_sink_dispose;
163 gobject_class->finalize = gst_splitmux_sink_finalize;
165 gst_element_class_set_static_metadata (gstelement_class,
166 "Split Muxing Bin", "Generic/Bin/Muxer",
167 "Convenience bin that muxes incoming streams into multiple time/size limited files",
168 "Jan Schmidt <jan@centricular.com>");
170 gst_element_class_add_pad_template (gstelement_class,
171 gst_static_pad_template_get (&video_sink_template));
172 gst_element_class_add_pad_template (gstelement_class,
173 gst_static_pad_template_get (&audio_sink_template));
174 gst_element_class_add_pad_template (gstelement_class,
175 gst_static_pad_template_get (&subtitle_sink_template));
177 gstelement_class->change_state =
178 GST_DEBUG_FUNCPTR (gst_splitmux_sink_change_state);
179 gstelement_class->request_new_pad =
180 GST_DEBUG_FUNCPTR (gst_splitmux_sink_request_new_pad);
181 gstelement_class->release_pad =
182 GST_DEBUG_FUNCPTR (gst_splitmux_sink_release_pad);
184 gstbin_class->handle_message = bus_handler;
186 g_object_class_install_property (gobject_class, PROP_LOCATION,
187 g_param_spec_string ("location", "File Output Pattern",
188 "Format string pattern for the location of the files to write (e.g. video%05d.mp4)",
189 NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
190 g_object_class_install_property (gobject_class, PROP_MUXER_OVERHEAD,
191 g_param_spec_double ("mux-overhead", "Muxing Overhead",
192 "Extra size overhead of muxing (0.02 = 2%)", 0.0, 1.0,
193 DEFAULT_MUXER_OVERHEAD,
194 G_PARAM_READWRITE | GST_PARAM_CONTROLLABLE | G_PARAM_STATIC_STRINGS));
196 g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
197 g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
198 "Max. amount of time per file (in ns, 0=disable)", 0, G_MAXUINT64,
199 DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
200 g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BYTES,
201 g_param_spec_uint64 ("max-size-bytes", "Max. size bytes",
202 "Max. amount of data per file (in bytes, 0=disable)", 0, G_MAXUINT64,
203 DEFAULT_MAX_SIZE_BYTES, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
205 g_object_class_install_property (gobject_class, PROP_MUXER,
206 g_param_spec_object ("muxer", "Muxer",
207 "The muxer element to use (NULL = default mp4mux)",
208 GST_TYPE_ELEMENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
209 g_object_class_install_property (gobject_class, PROP_SINK,
210 g_param_spec_object ("sink", "Sink",
211 "The sink element (or element chain) to use (NULL = default filesink)",
212 GST_TYPE_ELEMENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
215 * GstSplitMuxSink::format-location:
216 * @splitmux: the #GstSplitMuxSink
217 * @fragment_id: the sequence number of the file to be created
219 * Returns: the location to be used for the next output file
221 signals[SIGNAL_FORMAT_LOCATION] =
222 g_signal_new ("format-location", G_TYPE_FROM_CLASS (klass),
223 G_SIGNAL_RUN_LAST, 0, NULL, NULL, NULL, G_TYPE_STRING, 1, G_TYPE_UINT);
227 gst_splitmux_sink_init (GstSplitMuxSink * splitmux)
229 g_mutex_init (&splitmux->lock);
230 g_cond_init (&splitmux->data_cond);
232 splitmux->mux_overhead = DEFAULT_MUXER_OVERHEAD;
233 splitmux->threshold_time = DEFAULT_MAX_SIZE_TIME;
234 splitmux->threshold_bytes = DEFAULT_MAX_SIZE_BYTES;
236 GST_OBJECT_FLAG_SET (splitmux, GST_ELEMENT_FLAG_SINK);
240 gst_splitmux_reset (GstSplitMuxSink * splitmux)
243 gst_bin_remove (GST_BIN (splitmux), splitmux->mq);
245 gst_bin_remove (GST_BIN (splitmux), splitmux->muxer);
246 if (splitmux->active_sink)
247 gst_bin_remove (GST_BIN (splitmux), splitmux->active_sink);
249 splitmux->sink = splitmux->active_sink = splitmux->muxer = splitmux->mq =
254 gst_splitmux_sink_dispose (GObject * object)
256 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
258 G_OBJECT_CLASS (parent_class)->dispose (object);
260 /* Calling parent dispose invalidates all child pointers */
261 splitmux->sink = splitmux->active_sink = splitmux->muxer = splitmux->mq =
266 gst_splitmux_sink_finalize (GObject * object)
268 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
269 g_cond_clear (&splitmux->data_cond);
270 g_mutex_clear (&splitmux->lock);
271 if (splitmux->provided_sink)
272 gst_object_unref (splitmux->provided_sink);
273 if (splitmux->provided_muxer)
274 gst_object_unref (splitmux->provided_muxer);
276 g_free (splitmux->location);
278 /* Make sure to free any un-released contexts */
279 g_list_foreach (splitmux->contexts, (GFunc) mq_stream_ctx_unref, NULL);
280 g_list_free (splitmux->contexts);
282 G_OBJECT_CLASS (parent_class)->finalize (object);
286 gst_splitmux_sink_set_property (GObject * object, guint prop_id,
287 const GValue * value, GParamSpec * pspec)
289 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
293 GST_OBJECT_LOCK (splitmux);
294 g_free (splitmux->location);
295 splitmux->location = g_value_dup_string (value);
296 GST_OBJECT_UNLOCK (splitmux);
299 case PROP_MAX_SIZE_BYTES:
300 GST_OBJECT_LOCK (splitmux);
301 splitmux->threshold_bytes = g_value_get_uint64 (value);
302 GST_OBJECT_UNLOCK (splitmux);
304 case PROP_MAX_SIZE_TIME:
305 GST_OBJECT_LOCK (splitmux);
306 splitmux->threshold_time = g_value_get_uint64 (value);
307 GST_OBJECT_UNLOCK (splitmux);
309 case PROP_MUXER_OVERHEAD:
310 GST_OBJECT_LOCK (splitmux);
311 splitmux->mux_overhead = g_value_get_double (value);
312 GST_OBJECT_UNLOCK (splitmux);
315 GST_OBJECT_LOCK (splitmux);
316 if (splitmux->provided_sink)
317 gst_object_unref (splitmux->provided_sink);
318 splitmux->provided_sink = g_value_dup_object (value);
319 GST_OBJECT_UNLOCK (splitmux);
322 GST_OBJECT_LOCK (splitmux);
323 if (splitmux->provided_muxer)
324 gst_object_unref (splitmux->provided_muxer);
325 splitmux->provided_muxer = g_value_dup_object (value);
326 GST_OBJECT_UNLOCK (splitmux);
329 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
335 gst_splitmux_sink_get_property (GObject * object, guint prop_id,
336 GValue * value, GParamSpec * pspec)
338 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (object);
342 GST_OBJECT_LOCK (splitmux);
343 g_value_set_string (value, splitmux->location);
344 GST_OBJECT_UNLOCK (splitmux);
346 case PROP_MAX_SIZE_BYTES:
347 GST_OBJECT_LOCK (splitmux);
348 g_value_set_uint64 (value, splitmux->threshold_bytes);
349 GST_OBJECT_UNLOCK (splitmux);
351 case PROP_MAX_SIZE_TIME:
352 GST_OBJECT_LOCK (splitmux);
353 g_value_set_uint64 (value, splitmux->threshold_time);
354 GST_OBJECT_UNLOCK (splitmux);
356 case PROP_MUXER_OVERHEAD:
357 GST_OBJECT_LOCK (splitmux);
358 g_value_set_double (value, splitmux->mux_overhead);
359 GST_OBJECT_UNLOCK (splitmux);
362 GST_OBJECT_LOCK (splitmux);
363 g_value_set_object (value, splitmux->provided_sink);
364 GST_OBJECT_UNLOCK (splitmux);
367 GST_OBJECT_LOCK (splitmux);
368 g_value_set_object (value, splitmux->provided_muxer);
369 GST_OBJECT_UNLOCK (splitmux);
372 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
378 mq_sink_to_src (GstElement * mq, GstPad * sink_pad)
380 gchar *tmp, *sinkname, *srcname;
383 sinkname = gst_pad_get_name (sink_pad);
385 srcname = g_strdup_printf ("src_%s", tmp);
387 mq_src = gst_element_get_static_pad (mq, srcname);
396 get_pads_from_mq (GstSplitMuxSink * splitmux, GstPad ** sink_pad,
402 /* Request a pad from multiqueue, then connect this one, then
403 * discover the corresponding output pad and return both */
404 mq_sink = gst_element_get_request_pad (splitmux->mq, "sink_%u");
408 mq_src = mq_sink_to_src (splitmux->mq, mq_sink);
418 gst_element_release_request_pad (splitmux->mq, mq_sink);
423 mq_stream_ctx_new (GstSplitMuxSink * splitmux)
427 ctx = g_new0 (MqStreamCtx, 1);
428 g_atomic_int_set (&ctx->refcount, 1);
429 ctx->splitmux = splitmux;
430 gst_segment_init (&ctx->in_segment, GST_FORMAT_UNDEFINED);
431 gst_segment_init (&ctx->out_segment, GST_FORMAT_UNDEFINED);
432 ctx->in_running_time = ctx->out_running_time = 0;
433 g_queue_init (&ctx->queued_bufs);
438 mq_stream_ctx_free (MqStreamCtx * ctx)
440 g_queue_foreach (&ctx->queued_bufs, (GFunc) mq_stream_buf_free, NULL);
441 g_queue_clear (&ctx->queued_bufs);
446 mq_stream_ctx_unref (MqStreamCtx * ctx)
448 if (g_atomic_int_dec_and_test (&ctx->refcount))
449 mq_stream_ctx_free (ctx);
453 mq_stream_ctx_ref (MqStreamCtx * ctx)
455 g_atomic_int_inc (&ctx->refcount);
459 _pad_block_destroy_sink_notify (MqStreamCtx * ctx)
461 ctx->sink_pad_block_id = 0;
462 mq_stream_ctx_unref (ctx);
466 _pad_block_destroy_src_notify (MqStreamCtx * ctx)
468 ctx->src_pad_block_id = 0;
469 mq_stream_ctx_unref (ctx);
472 /* Called with lock held, drops the lock to send EOS to the
476 send_eos (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
481 eos = gst_event_new_eos ();
482 pad = gst_pad_get_peer (ctx->srcpad);
486 GST_INFO_OBJECT (splitmux, "Sending EOS on %" GST_PTR_FORMAT, pad);
487 GST_SPLITMUX_UNLOCK (splitmux);
488 gst_pad_send_event (pad, eos);
489 GST_SPLITMUX_LOCK (splitmux);
491 gst_object_unref (pad);
494 /* Called with splitmux lock held to check if this output
495 * context needs to sleep to wait for the release of the
496 * next GOP, or to send EOS to close out the current file
499 complete_or_wait_on_out (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
503 GST_LOG_OBJECT (ctx->srcpad,
504 "Checking running time %" GST_TIME_FORMAT " against max %"
505 GST_TIME_FORMAT, GST_TIME_ARGS (ctx->out_running_time),
506 GST_TIME_ARGS (splitmux->max_out_running_time));
508 if (splitmux->max_out_running_time == GST_CLOCK_TIME_NONE ||
509 ctx->out_running_time < splitmux->max_out_running_time) {
510 splitmux->have_muxed_something = TRUE;
514 if (ctx->flushing || splitmux->state == SPLITMUX_STATE_STOPPED)
517 if (splitmux->state == SPLITMUX_STATE_ENDING_FILE) {
518 if (ctx->out_eos == FALSE) {
519 send_eos (splitmux, ctx);
522 } else if (splitmux->state == SPLITMUX_STATE_START_NEXT_FRAGMENT) {
523 start_next_fragment (splitmux);
527 GST_INFO_OBJECT (ctx->srcpad,
528 "Sleeping for running time %"
529 GST_TIME_FORMAT " (max %" GST_TIME_FORMAT ")",
530 GST_TIME_ARGS (ctx->out_running_time),
531 GST_TIME_ARGS (splitmux->max_out_running_time));
532 ctx->out_blocked = TRUE;
533 /* Expand the mq if needed before sleeping */
534 check_queue_length (splitmux, ctx);
535 GST_SPLITMUX_WAIT (splitmux);
536 ctx->out_blocked = FALSE;
537 GST_INFO_OBJECT (ctx->srcpad,
538 "Woken for new max running time %" GST_TIME_FORMAT,
539 GST_TIME_ARGS (splitmux->max_out_running_time));
543 static GstPadProbeReturn
544 handle_mq_output (GstPad * pad, GstPadProbeInfo * info, MqStreamCtx * ctx)
546 GstSplitMuxSink *splitmux = ctx->splitmux;
547 MqStreamBuf *buf_info = NULL;
549 GST_LOG_OBJECT (pad, "Fired probe type 0x%x\n", info->type);
551 /* FIXME: Handle buffer lists, until then make it clear they won't work */
552 if (info->type & GST_PAD_PROBE_TYPE_BUFFER_LIST) {
553 g_warning ("Buffer list handling not implemented");
554 return GST_PAD_PROBE_DROP;
556 if (info->type & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) {
557 GstEvent *event = gst_pad_probe_info_get_event (info);
559 GST_LOG_OBJECT (pad, "Event %" GST_PTR_FORMAT, event);
561 switch (GST_EVENT_TYPE (event)) {
562 case GST_EVENT_SEGMENT:
563 gst_event_copy_segment (event, &ctx->out_segment);
565 case GST_EVENT_FLUSH_STOP:
566 GST_SPLITMUX_LOCK (splitmux);
567 gst_segment_init (&ctx->out_segment, GST_FORMAT_UNDEFINED);
568 g_queue_foreach (&ctx->queued_bufs, (GFunc) mq_stream_buf_free, NULL);
569 g_queue_clear (&ctx->queued_bufs);
570 ctx->flushing = FALSE;
571 GST_SPLITMUX_UNLOCK (splitmux);
573 case GST_EVENT_FLUSH_START:
574 GST_SPLITMUX_LOCK (splitmux);
575 GST_LOG_OBJECT (pad, "Flush start");
576 ctx->flushing = TRUE;
577 GST_SPLITMUX_BROADCAST (splitmux);
578 GST_SPLITMUX_UNLOCK (splitmux);
581 GST_SPLITMUX_LOCK (splitmux);
582 if (splitmux->state == SPLITMUX_STATE_STOPPED)
585 GST_SPLITMUX_UNLOCK (splitmux);
590 gst_event_parse_gap (event, &gap_ts, NULL);
591 if (gap_ts == GST_CLOCK_TIME_NONE)
594 GST_SPLITMUX_LOCK (splitmux);
596 gap_ts = gst_segment_to_running_time (&ctx->out_segment,
597 GST_FORMAT_TIME, gap_ts);
599 GST_LOG_OBJECT (pad, "Have GAP w/ ts %" GST_TIME_FORMAT,
600 GST_TIME_ARGS (gap_ts));
602 if (splitmux->state == SPLITMUX_STATE_STOPPED)
604 ctx->out_running_time = gap_ts;
605 complete_or_wait_on_out (splitmux, ctx);
606 GST_SPLITMUX_UNLOCK (splitmux);
612 return GST_PAD_PROBE_PASS;
615 /* Allow everything through until the configured next stopping point */
616 GST_SPLITMUX_LOCK (splitmux);
618 buf_info = g_queue_pop_tail (&ctx->queued_bufs);
619 if (buf_info == NULL)
620 /* Can only happen due to a poorly timed flush */
623 /* If we have popped a keyframe, decrement the queued_gop count */
624 if (buf_info->keyframe && splitmux->queued_gops > 0)
625 splitmux->queued_gops--;
627 ctx->out_running_time = buf_info->run_ts;
629 GST_LOG_OBJECT (splitmux,
630 "Pad %" GST_PTR_FORMAT " buffer with TS %" GST_TIME_FORMAT
631 " size %" G_GSIZE_FORMAT,
632 pad, GST_TIME_ARGS (ctx->out_running_time), buf_info->buf_size);
634 complete_or_wait_on_out (splitmux, ctx);
636 if (splitmux->muxed_out_time == GST_CLOCK_TIME_NONE ||
637 splitmux->muxed_out_time < buf_info->run_ts)
638 splitmux->muxed_out_time = buf_info->run_ts;
640 splitmux->muxed_out_bytes += buf_info->buf_size;
642 #ifndef GST_DISABLE_GST_DEBUG
644 GstBuffer *buf = gst_pad_probe_info_get_buffer (info);
645 GST_LOG_OBJECT (pad, "Returning to pass buffer %" GST_PTR_FORMAT
646 " run ts %" GST_TIME_FORMAT, buf,
647 GST_TIME_ARGS (ctx->out_running_time));
651 GST_SPLITMUX_UNLOCK (splitmux);
653 mq_stream_buf_free (buf_info);
655 return GST_PAD_PROBE_PASS;
658 GST_SPLITMUX_UNLOCK (splitmux);
659 return GST_PAD_PROBE_DROP;
663 resend_sticky (GstPad * pad, GstEvent ** event, GstPad * peer)
665 return gst_pad_send_event (peer, gst_event_ref (*event));
669 restart_context (MqStreamCtx * ctx, GstSplitMuxSink * splitmux)
671 GstPad *peer = gst_pad_get_peer (ctx->srcpad);
673 gst_pad_sticky_events_foreach (ctx->srcpad,
674 (GstPadStickyEventsForeachFunction) (resend_sticky), peer);
677 ctx->out_eos = FALSE;
679 gst_object_unref (peer);
682 /* Called with lock held when a fragment
683 * reaches EOS and it is time to restart
687 start_next_fragment (GstSplitMuxSink * splitmux)
689 /* 1 change to new file */
690 gst_element_set_state (splitmux->muxer, GST_STATE_NULL);
691 gst_element_set_state (splitmux->active_sink, GST_STATE_NULL);
693 set_next_filename (splitmux);
695 gst_element_sync_state_with_parent (splitmux->active_sink);
696 gst_element_sync_state_with_parent (splitmux->muxer);
698 g_list_foreach (splitmux->contexts, (GFunc) restart_context, splitmux);
700 /* Switch state and go back to processing */
701 splitmux->state = SPLITMUX_STATE_COLLECTING_GOP_START;
703 if (!splitmux->video_ctx->in_eos) {
704 splitmux->max_out_running_time = splitmux->video_ctx->in_running_time;
706 splitmux->max_out_running_time = GST_CLOCK_TIME_NONE;
707 splitmux->have_muxed_something = FALSE;
709 splitmux->have_muxed_something =
710 (splitmux->video_ctx->in_running_time > splitmux->muxed_out_time);
712 /* Store the overflow parameters as the basis for the next fragment */
713 splitmux->mux_start_time = splitmux->muxed_out_time;
714 splitmux->mux_start_bytes = splitmux->muxed_out_bytes;
716 GST_DEBUG_OBJECT (splitmux,
717 "Restarting flow for new fragment. New running time %" GST_TIME_FORMAT,
718 GST_TIME_ARGS (splitmux->max_out_running_time));
720 GST_SPLITMUX_BROADCAST (splitmux);
724 bus_handler (GstBin * bin, GstMessage * message)
726 GstSplitMuxSink *splitmux = GST_SPLITMUX_SINK (bin);
728 switch (GST_MESSAGE_TYPE (message)) {
729 case GST_MESSAGE_EOS:
730 /* If the state is draining out the current file, drop this EOS */
731 GST_SPLITMUX_LOCK (splitmux);
732 if (splitmux->state == SPLITMUX_STATE_ENDING_FILE &&
733 splitmux->max_out_running_time != GST_CLOCK_TIME_NONE) {
734 GST_DEBUG_OBJECT (splitmux, "Caught EOS at end of fragment, dropping");
735 splitmux->state = SPLITMUX_STATE_START_NEXT_FRAGMENT;
736 GST_SPLITMUX_BROADCAST (splitmux);
738 gst_message_unref (message);
739 GST_SPLITMUX_UNLOCK (splitmux);
742 GST_SPLITMUX_UNLOCK (splitmux);
748 GST_BIN_CLASS (parent_class)->handle_message (bin, message);
751 /* Called with splitmux lock held */
752 /* Called when entering ProcessingCompleteGop state
753 * Assess if mq contents overflowed the current file
754 * -> If yes, need to switch to new file
755 * -> if no, set max_out_running_time to let this GOP in and
756 * go to COLLECTING_GOP_START state
759 handle_gathered_gop (GstSplitMuxSink * splitmux)
762 gsize queued_bytes = 0;
763 GstClockTime queued_time = 0;
765 /* Assess if the multiqueue contents overflowed the current file */
766 for (cur = g_list_first (splitmux->contexts);
767 cur != NULL; cur = g_list_next (cur)) {
768 MqStreamCtx *tmpctx = (MqStreamCtx *) (cur->data);
769 if (tmpctx->in_running_time > queued_time)
770 queued_time = tmpctx->in_running_time;
771 queued_bytes += tmpctx->in_bytes;
774 g_assert (queued_bytes >= splitmux->mux_start_bytes);
775 g_assert (queued_time >= splitmux->mux_start_time);
777 queued_bytes -= splitmux->mux_start_bytes;
778 queued_time -= splitmux->mux_start_time;
780 /* Expand queued bytes estimate by muxer overhead */
781 queued_bytes += (queued_bytes * splitmux->mux_overhead);
783 GST_LOG_OBJECT (splitmux, "mq at TS %" GST_TIME_FORMAT
784 " bytes %" G_GSIZE_FORMAT, GST_TIME_ARGS (queued_time), queued_bytes);
786 /* Check for overrun - have we output at least one byte and overrun
787 * either threshold? */
788 if (splitmux->have_muxed_something &&
789 ((splitmux->threshold_bytes > 0 &&
790 queued_bytes >= splitmux->threshold_bytes) ||
791 (splitmux->threshold_time > 0 &&
792 queued_time >= splitmux->threshold_time))) {
794 splitmux->state = SPLITMUX_STATE_ENDING_FILE;
796 GST_INFO_OBJECT (splitmux,
797 "mq overflowed since last, draining out. max out TS is %"
798 GST_TIME_FORMAT, GST_TIME_ARGS (splitmux->max_out_running_time));
799 GST_SPLITMUX_BROADCAST (splitmux);
803 GST_LOG_OBJECT (splitmux,
804 "This GOP didn't overflow the fragment. Bytes sent %" G_GSIZE_FORMAT
805 " queued %" G_GSIZE_FORMAT " time %" GST_TIME_FORMAT " Continuing.",
806 splitmux->muxed_out_bytes - splitmux->mux_start_bytes,
807 queued_bytes, GST_TIME_ARGS (queued_time));
809 /* Wake everyone up to push this one GOP, then sleep */
810 splitmux->state = SPLITMUX_STATE_COLLECTING_GOP_START;
811 splitmux->have_muxed_something = TRUE;
813 if (!splitmux->video_ctx->in_eos)
814 splitmux->max_out_running_time = splitmux->video_ctx->in_running_time;
816 splitmux->max_out_running_time = GST_CLOCK_TIME_NONE;
818 GST_LOG_OBJECT (splitmux, "Waking output for complete GOP, TS %"
819 GST_TIME_FORMAT, GST_TIME_ARGS (splitmux->max_out_running_time));
820 GST_SPLITMUX_BROADCAST (splitmux);
825 /* Called with splitmux lock held */
826 /* Called from each input pad when it is has all the pieces
827 * for a GOP or EOS, starting with the video pad which has set the
828 * splitmux->max_in_running_time
831 check_completed_gop (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
834 gboolean ready = TRUE;
835 GstClockTime current_max_in_running_time;
837 if (splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE) {
838 /* Iterate each pad, and check that the input running time is at least
839 * up to the video runnning time, and if so handle the collected GOP */
840 GST_LOG_OBJECT (splitmux, "Checking GOP collected, ctx %p", ctx);
841 for (cur = g_list_first (splitmux->contexts);
842 cur != NULL; cur = g_list_next (cur)) {
843 MqStreamCtx *tmpctx = (MqStreamCtx *) (cur->data);
845 GST_LOG_OBJECT (splitmux,
846 "Context %p (src pad %" GST_PTR_FORMAT ") TS %" GST_TIME_FORMAT
847 " EOS %d", tmpctx, tmpctx->srcpad,
848 GST_TIME_ARGS (tmpctx->in_running_time), tmpctx->in_eos);
850 if (tmpctx->in_running_time < splitmux->max_in_running_time &&
852 GST_LOG_OBJECT (splitmux,
853 "Context %p (src pad %" GST_PTR_FORMAT ") not ready. We'll sleep",
854 tmpctx, tmpctx->srcpad);
860 GST_DEBUG_OBJECT (splitmux,
861 "Collected GOP is complete. Processing (ctx %p)", ctx);
862 /* All pads have a complete GOP, release it into the multiqueue */
863 handle_gathered_gop (splitmux);
867 /* Some pad is not yet ready, or GOP is being pushed
868 * either way, sleep and wait to get woken */
869 current_max_in_running_time = splitmux->max_in_running_time;
870 while ((splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE ||
871 splitmux->state == SPLITMUX_STATE_START_NEXT_FRAGMENT) &&
873 (current_max_in_running_time == splitmux->max_in_running_time)) {
875 GST_LOG_OBJECT (splitmux, "Sleeping for %s (ctx %p)",
876 splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE ?
877 "GOP complete" : "EOF draining", ctx);
878 GST_SPLITMUX_WAIT (splitmux);
880 GST_LOG_OBJECT (splitmux, "Done waiting for complete GOP (ctx %p)", ctx);
885 check_queue_length (GstSplitMuxSink * splitmux, MqStreamCtx * ctx)
888 guint cur_len = g_queue_get_length (&ctx->queued_bufs);
890 GST_DEBUG_OBJECT (ctx->sinkpad,
891 "Checking queue length len %u cur_max %u queued gops %u",
892 cur_len, splitmux->mq_max_buffers, splitmux->queued_gops);
894 if (cur_len >= splitmux->mq_max_buffers) {
895 gboolean allow_grow = FALSE;
897 /* If collecting a GOP and this pad might block,
898 * and there isn't already a pending GOP in the queue
901 if (splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE &&
902 ctx->in_running_time < splitmux->max_in_running_time &&
903 splitmux->queued_gops <= 1) {
905 } else if (splitmux->state == SPLITMUX_STATE_COLLECTING_GOP_START &&
906 ctx->is_video && splitmux->queued_gops <= 1) {
911 for (cur = g_list_first (splitmux->contexts);
912 cur != NULL; cur = g_list_next (cur)) {
913 MqStreamCtx *tmpctx = (MqStreamCtx *) (cur->data);
914 GST_DEBUG_OBJECT (tmpctx->sinkpad,
915 " len %u out_blocked %d",
916 g_queue_get_length (&tmpctx->queued_bufs), tmpctx->out_blocked);
917 /* If another stream is starving, grow */
918 if (tmpctx != ctx && g_queue_get_length (&tmpctx->queued_bufs) < 1) {
925 splitmux->mq_max_buffers = cur_len + 1;
927 GST_INFO_OBJECT (splitmux,
928 "Multiqueue overrun - enlarging to %u buffers ctx %p",
929 splitmux->mq_max_buffers, ctx);
931 g_object_set (splitmux->mq, "max-size-buffers",
932 splitmux->mq_max_buffers, NULL);
937 static GstPadProbeReturn
938 handle_mq_input (GstPad * pad, GstPadProbeInfo * info, MqStreamCtx * ctx)
940 GstSplitMuxSink *splitmux = ctx->splitmux;
942 MqStreamBuf *buf_info = NULL;
945 gboolean keyframe = FALSE;
947 GST_LOG_OBJECT (pad, "Fired probe type 0x%x", info->type);
949 /* FIXME: Handle buffer lists, until then make it clear they won't work */
950 if (info->type & GST_PAD_PROBE_TYPE_BUFFER_LIST) {
951 g_warning ("Buffer list handling not implemented");
952 return GST_PAD_PROBE_DROP;
954 if (info->type & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) {
955 GstEvent *event = gst_pad_probe_info_get_event (info);
956 switch (GST_EVENT_TYPE (event)) {
957 case GST_EVENT_SEGMENT:
958 gst_event_copy_segment (event, &ctx->in_segment);
960 case GST_EVENT_FLUSH_STOP:
961 GST_SPLITMUX_LOCK (splitmux);
962 gst_segment_init (&ctx->in_segment, GST_FORMAT_UNDEFINED);
965 ctx->in_running_time = 0;
966 GST_SPLITMUX_UNLOCK (splitmux);
969 GST_SPLITMUX_LOCK (splitmux);
972 if (splitmux->state == SPLITMUX_STATE_STOPPED)
976 GST_INFO_OBJECT (splitmux, "Got Video EOS. Finishing up");
977 /* Act as if this is a new keyframe with infinite timestamp */
978 splitmux->max_in_running_time = GST_CLOCK_TIME_NONE;
979 splitmux->state = SPLITMUX_STATE_WAITING_GOP_COMPLETE;
980 /* Wake up other input pads to collect this GOP */
981 GST_SPLITMUX_BROADCAST (splitmux);
982 check_completed_gop (splitmux, ctx);
983 } else if (splitmux->state == SPLITMUX_STATE_WAITING_GOP_COMPLETE) {
984 /* If we are waiting for a GOP to be completed (ie, for aux
985 * pads to catch up), then this pad is complete, so check
986 * if the whole GOP is.
988 check_completed_gop (splitmux, ctx);
990 GST_SPLITMUX_UNLOCK (splitmux);
995 return GST_PAD_PROBE_PASS;
998 buf = gst_pad_probe_info_get_buffer (info);
999 ctx->in_running_time = gst_segment_to_running_time (&ctx->in_segment,
1000 GST_FORMAT_TIME, GST_BUFFER_TIMESTAMP (buf));
1001 buf_info = mq_stream_buf_new ();
1003 if (GST_BUFFER_PTS_IS_VALID (buf))
1004 ts = GST_BUFFER_PTS (buf);
1006 ts = GST_BUFFER_DTS (buf);
1008 GST_SPLITMUX_LOCK (splitmux);
1010 if (splitmux->state == SPLITMUX_STATE_STOPPED)
1013 /* If this buffer has a timestamp, advance the input timestamp of the
1015 if (GST_CLOCK_TIME_IS_VALID (ts)) {
1016 GstClockTime running_time =
1017 gst_segment_to_running_time (&ctx->in_segment, GST_FORMAT_TIME,
1018 GST_BUFFER_TIMESTAMP (buf));
1020 if (GST_CLOCK_TIME_IS_VALID (running_time) &&
1021 (ctx->in_running_time == GST_CLOCK_TIME_NONE
1022 || running_time > ctx->in_running_time))
1023 ctx->in_running_time = running_time;
1026 /* Try to make sure we have a valid running time */
1027 if (!GST_CLOCK_TIME_IS_VALID (ctx->in_running_time)) {
1028 ctx->in_running_time =
1029 gst_segment_to_running_time (&ctx->in_segment, GST_FORMAT_TIME,
1030 ctx->in_segment.start);
1033 buf_info->run_ts = ctx->in_running_time;
1034 buf_info->buf_size = gst_buffer_get_size (buf);
1036 /* Update total input byte counter for overflow detect */
1037 ctx->in_bytes += buf_info->buf_size;
1039 /* initialize mux_start_time */
1040 if (ctx->is_reference && splitmux->mux_start_time == 0)
1041 splitmux->mux_start_time = buf_info->run_ts;
1043 GST_DEBUG_OBJECT (pad, "Buf TS %" GST_TIME_FORMAT
1044 " total in_bytes %" G_GSIZE_FORMAT,
1045 GST_TIME_ARGS (buf_info->run_ts), ctx->in_bytes);
1052 switch (splitmux->state) {
1053 case SPLITMUX_STATE_COLLECTING_GOP_START:
1054 if (ctx->is_video) {
1055 /* If a keyframe, we have a complete GOP */
1056 if (GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT) ||
1057 !GST_CLOCK_TIME_IS_VALID (ctx->in_running_time) ||
1058 splitmux->max_in_running_time >= ctx->in_running_time) {
1059 /* Pass this buffer through */
1063 GST_INFO_OBJECT (pad,
1064 "Have keyframe with running time %" GST_TIME_FORMAT,
1065 GST_TIME_ARGS (ctx->in_running_time));
1067 splitmux->state = SPLITMUX_STATE_WAITING_GOP_COMPLETE;
1068 splitmux->max_in_running_time = ctx->in_running_time;
1069 /* Wake up other input pads to collect this GOP */
1070 GST_SPLITMUX_BROADCAST (splitmux);
1071 check_completed_gop (splitmux, ctx);
1073 /* We're still waiting for a keyframe on the video pad, sleep */
1074 GST_LOG_OBJECT (pad, "Sleeping for GOP start");
1075 GST_SPLITMUX_WAIT (splitmux);
1076 GST_LOG_OBJECT (pad, "Done sleeping for GOP start state now %d",
1080 case SPLITMUX_STATE_WAITING_GOP_COMPLETE:
1081 /* After a GOP start is found, this buffer might complete the GOP */
1082 /* If we overran the target timestamp, it might be time to process
1083 * the GOP, otherwise bail out for more data
1085 GST_LOG_OBJECT (pad,
1086 "Checking TS %" GST_TIME_FORMAT " against max %" GST_TIME_FORMAT,
1087 GST_TIME_ARGS (ctx->in_running_time),
1088 GST_TIME_ARGS (splitmux->max_in_running_time));
1090 if (ctx->in_running_time < splitmux->max_in_running_time) {
1095 GST_LOG_OBJECT (pad,
1096 "Collected last packet of GOP. Checking other pads");
1097 check_completed_gop (splitmux, ctx);
1099 case SPLITMUX_STATE_ENDING_FILE:
1100 case SPLITMUX_STATE_START_NEXT_FRAGMENT:
1101 /* A fragment is ending, wait until that's done before continuing */
1102 GST_DEBUG_OBJECT (pad, "Sleeping for fragment restart");
1103 GST_SPLITMUX_WAIT (splitmux);
1104 GST_DEBUG_OBJECT (pad,
1105 "Done sleeping for fragment restart state now %d", splitmux->state);
1111 } while (loop_again);
1114 splitmux->queued_gops++;
1115 buf_info->keyframe = TRUE;
1118 /* Now add this buffer to the queue just before returning */
1119 g_queue_push_head (&ctx->queued_bufs, buf_info);
1121 /* Check the buffer will fit in the mq */
1122 check_queue_length (splitmux, ctx);
1124 GST_LOG_OBJECT (pad, "Returning to queue buffer %" GST_PTR_FORMAT
1125 " run ts %" GST_TIME_FORMAT, buf, GST_TIME_ARGS (ctx->in_running_time));
1127 GST_SPLITMUX_UNLOCK (splitmux);
1128 return GST_PAD_PROBE_PASS;
1131 GST_SPLITMUX_UNLOCK (splitmux);
1133 mq_stream_buf_free (buf_info);
1134 return GST_PAD_PROBE_PASS;
1138 gst_splitmux_sink_request_new_pad (GstElement * element,
1139 GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
1141 GstSplitMuxSink *splitmux = (GstSplitMuxSink *) element;
1142 GstPadTemplate *mux_template = NULL;
1144 GstPad *mq_sink, *mq_src;
1146 gboolean is_video = FALSE;
1149 GST_DEBUG_OBJECT (element, "templ:%s, name:%s", templ->name_template, name);
1151 GST_SPLITMUX_LOCK (splitmux);
1152 if (!create_elements (splitmux))
1155 if (templ->name_template) {
1156 if (g_str_equal (templ->name_template, "video")) {
1157 /* FIXME: Look for a pad template with matching caps, rather than by name */
1159 gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS
1160 (splitmux->muxer), "video_%u");
1165 gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS
1166 (splitmux->muxer), templ->name_template);
1168 if (mux_template == NULL) {
1169 /* Fallback to find sink pad templates named 'sink_%d' (mpegtsmux) */
1171 gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS
1172 (splitmux->muxer), "sink_%d");
1176 res = gst_element_request_pad (splitmux->muxer, mux_template, name, caps);
1181 gname = g_strdup ("video");
1182 else if (name == NULL)
1183 gname = gst_pad_get_name (res);
1185 gname = g_strdup (name);
1187 if (!get_pads_from_mq (splitmux, &mq_sink, &mq_src)) {
1188 gst_element_release_request_pad (splitmux->muxer, res);
1189 gst_object_unref (GST_OBJECT (res));
1193 if (gst_pad_link (mq_src, res) != GST_PAD_LINK_OK) {
1194 gst_element_release_request_pad (splitmux->muxer, res);
1195 gst_object_unref (GST_OBJECT (res));
1196 gst_element_release_request_pad (splitmux->mq, mq_sink);
1197 gst_object_unref (GST_OBJECT (mq_sink));
1201 gst_object_unref (GST_OBJECT (res));
1203 ctx = mq_stream_ctx_new (splitmux);
1204 ctx->is_video = is_video;
1205 ctx->srcpad = mq_src;
1206 ctx->sinkpad = mq_sink;
1208 mq_stream_ctx_ref (ctx);
1209 ctx->src_pad_block_id =
1210 gst_pad_add_probe (mq_src, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM,
1211 (GstPadProbeCallback) handle_mq_output, ctx, (GDestroyNotify)
1212 _pad_block_destroy_src_notify);
1214 splitmux->video_ctx = ctx;
1216 res = gst_ghost_pad_new (gname, mq_sink);
1217 g_object_set_qdata ((GObject *) (res), PAD_CONTEXT, ctx);
1219 mq_stream_ctx_ref (ctx);
1220 ctx->sink_pad_block_id =
1221 gst_pad_add_probe (mq_sink, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM,
1222 (GstPadProbeCallback) handle_mq_input, ctx, (GDestroyNotify)
1223 _pad_block_destroy_sink_notify);
1225 GST_DEBUG_OBJECT (splitmux, "Request pad %" GST_PTR_FORMAT
1226 " is mq pad %" GST_PTR_FORMAT, res, mq_sink);
1228 splitmux->contexts = g_list_prepend (splitmux->contexts, ctx);
1232 gst_object_unref (mq_sink);
1233 gst_object_unref (mq_src);
1235 gst_pad_set_active (res, TRUE);
1236 gst_element_add_pad (element, res);
1237 GST_SPLITMUX_UNLOCK (splitmux);
1241 GST_SPLITMUX_UNLOCK (splitmux);
1246 gst_splitmux_sink_release_pad (GstElement * element, GstPad * pad)
1248 GstSplitMuxSink *splitmux = (GstSplitMuxSink *) element;
1249 GstPad *mqsink, *mqsrc, *muxpad;
1251 (MqStreamCtx *) (g_object_get_qdata ((GObject *) (pad), PAD_CONTEXT));
1253 GST_SPLITMUX_LOCK (splitmux);
1255 if (splitmux->muxer == NULL || splitmux->mq == NULL)
1256 goto fail; /* Elements don't exist yet - nothing to release */
1258 GST_INFO_OBJECT (pad, "releasing request pad");
1260 mqsink = gst_ghost_pad_get_target (GST_GHOST_PAD (pad));
1261 mqsrc = mq_sink_to_src (splitmux->mq, mqsink);
1262 muxpad = gst_pad_get_peer (mqsrc);
1264 /* Remove the context from our consideration */
1265 splitmux->contexts = g_list_remove (splitmux->contexts, ctx);
1267 if (ctx->sink_pad_block_id)
1268 gst_pad_remove_probe (ctx->sinkpad, ctx->sink_pad_block_id);
1270 if (ctx->src_pad_block_id)
1271 gst_pad_remove_probe (ctx->srcpad, ctx->src_pad_block_id);
1273 /* Can release the context now */
1274 mq_stream_ctx_unref (ctx);
1276 /* Release and free the mq input */
1277 gst_element_release_request_pad (splitmux->mq, mqsink);
1279 /* Release and free the muxer input */
1280 gst_element_release_request_pad (splitmux->muxer, muxpad);
1282 gst_object_unref (mqsink);
1283 gst_object_unref (mqsrc);
1284 gst_object_unref (muxpad);
1286 gst_element_remove_pad (element, pad);
1289 GST_SPLITMUX_UNLOCK (splitmux);
1293 create_element (GstSplitMuxSink * splitmux,
1294 const gchar * factory, const gchar * name)
1296 GstElement *ret = gst_element_factory_make (factory, name);
1298 g_warning ("Failed to create %s - splitmuxsink will not work", name);
1302 if (!gst_bin_add (GST_BIN (splitmux), ret)) {
1303 g_warning ("Could not add %s element - splitmuxsink will not work", name);
1304 gst_object_unref (ret);
1312 create_elements (GstSplitMuxSink * splitmux)
1314 /* Create internal elements */
1315 if (splitmux->mq == NULL) {
1317 create_element (splitmux, "multiqueue", "multiqueue")) == NULL)
1320 splitmux->mq_max_buffers = 5;
1321 /* No bytes or time limit, we limit buffers manually */
1322 g_object_set (splitmux->mq, "max-size-bytes", 0, "max-size-time",
1323 (guint64) 0, "max-size-buffers", splitmux->mq_max_buffers, NULL);
1326 if (splitmux->muxer == NULL) {
1327 GstElement *provided_muxer = NULL;
1329 GST_OBJECT_LOCK (splitmux);
1330 if (splitmux->provided_muxer != NULL)
1331 provided_muxer = gst_object_ref (splitmux->provided_muxer);
1332 GST_OBJECT_UNLOCK (splitmux);
1334 if (provided_muxer == NULL) {
1335 if ((splitmux->muxer =
1336 create_element (splitmux, "mp4mux", "muxer")) == NULL)
1339 if (!gst_bin_add (GST_BIN (splitmux), provided_muxer)) {
1340 g_warning ("Could not add muxer element - splitmuxsink will not work");
1341 gst_object_unref (provided_muxer);
1345 splitmux->muxer = provided_muxer;
1346 gst_object_unref (provided_muxer);
1356 find_sink (GstElement * e)
1358 GstElement *res = NULL;
1360 gboolean done = FALSE;
1361 GValue data = { 0, };
1363 if (!GST_IS_BIN (e))
1366 iter = gst_bin_iterate_sinks (GST_BIN (e));
1368 switch (gst_iterator_next (iter, &data)) {
1369 case GST_ITERATOR_OK:
1371 GstElement *child = g_value_get_object (&data);
1372 if (g_object_class_find_property (G_OBJECT_GET_CLASS (child),
1373 "location") != NULL) {
1377 g_value_reset (&data);
1380 case GST_ITERATOR_RESYNC:
1381 gst_iterator_resync (iter);
1383 case GST_ITERATOR_DONE:
1386 case GST_ITERATOR_ERROR:
1387 g_assert_not_reached ();
1391 g_value_unset (&data);
1392 gst_iterator_free (iter);
1398 create_sink (GstSplitMuxSink * splitmux)
1400 GstElement *provided_sink = NULL;
1402 g_return_val_if_fail (splitmux->active_sink == NULL, TRUE);
1404 GST_OBJECT_LOCK (splitmux);
1405 if (splitmux->provided_sink != NULL)
1406 provided_sink = gst_object_ref (splitmux->provided_sink);
1407 GST_OBJECT_UNLOCK (splitmux);
1409 if (provided_sink == NULL) {
1410 if ((splitmux->sink =
1411 create_element (splitmux, DEFAULT_SINK, "sink")) == NULL)
1413 splitmux->active_sink = splitmux->sink;
1415 if (!gst_bin_add (GST_BIN (splitmux), provided_sink)) {
1416 g_warning ("Could not add sink elements - splitmuxsink will not work");
1417 gst_object_unref (provided_sink);
1421 splitmux->active_sink = provided_sink;
1423 /* The bin holds a ref now, we can drop our tmp ref */
1424 gst_object_unref (provided_sink);
1426 /* Find the sink element */
1427 splitmux->sink = find_sink (splitmux->active_sink);
1428 if (splitmux->sink == NULL) {
1430 ("Could not locate sink element in provided sink - splitmuxsink will not work");
1435 if (!gst_element_link (splitmux->muxer, splitmux->active_sink)) {
1436 g_warning ("Failed to link muxer and sink- splitmuxsink will not work");
1446 #pragma GCC diagnostic ignored "-Wformat-nonliteral"
1449 set_next_filename (GstSplitMuxSink * splitmux)
1451 gchar *fname = NULL;
1453 g_signal_emit (splitmux, signals[SIGNAL_FORMAT_LOCATION], 0,
1454 splitmux->fragment_id, &fname);
1457 fname = splitmux->location ?
1458 g_strdup_printf (splitmux->location, splitmux->fragment_id) : NULL;
1461 GST_INFO_OBJECT (splitmux, "Setting file to %s", fname);
1462 g_object_set (splitmux->sink, "location", fname, NULL);
1465 splitmux->fragment_id++;
1469 static GstStateChangeReturn
1470 gst_splitmux_sink_change_state (GstElement * element, GstStateChange transition)
1472 GstStateChangeReturn ret;
1473 GstSplitMuxSink *splitmux = (GstSplitMuxSink *) element;
1475 switch (transition) {
1476 case GST_STATE_CHANGE_NULL_TO_READY:{
1477 GST_SPLITMUX_LOCK (splitmux);
1478 if (!create_elements (splitmux) || !create_sink (splitmux)) {
1479 ret = GST_STATE_CHANGE_FAILURE;
1480 GST_SPLITMUX_UNLOCK (splitmux);
1483 GST_SPLITMUX_UNLOCK (splitmux);
1484 splitmux->fragment_id = 0;
1485 set_next_filename (splitmux);
1488 case GST_STATE_CHANGE_READY_TO_PAUSED:{
1489 GST_SPLITMUX_LOCK (splitmux);
1490 /* Start by collecting one input on each pad */
1491 splitmux->state = SPLITMUX_STATE_COLLECTING_GOP_START;
1492 splitmux->max_in_running_time = 0;
1493 splitmux->muxed_out_time = splitmux->mux_start_time = 0;
1494 splitmux->muxed_out_bytes = splitmux->mux_start_bytes = 0;
1495 GST_SPLITMUX_UNLOCK (splitmux);
1498 case GST_STATE_CHANGE_PAUSED_TO_READY:
1499 case GST_STATE_CHANGE_READY_TO_NULL:
1500 GST_SPLITMUX_LOCK (splitmux);
1501 splitmux->state = SPLITMUX_STATE_STOPPED;
1502 /* Wake up any blocked threads */
1503 GST_LOG_OBJECT (splitmux,
1504 "State change -> NULL or READY. Waking threads");
1505 GST_SPLITMUX_BROADCAST (splitmux);
1506 GST_SPLITMUX_UNLOCK (splitmux);
1512 ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1513 if (ret == GST_STATE_CHANGE_FAILURE)
1516 switch (transition) {
1517 case GST_STATE_CHANGE_READY_TO_NULL:
1518 GST_SPLITMUX_LOCK (splitmux);
1519 splitmux->fragment_id = 0;
1520 gst_splitmux_reset (splitmux);
1521 GST_SPLITMUX_UNLOCK (splitmux);
1529 if (transition == GST_STATE_CHANGE_NULL_TO_READY &&
1530 ret == GST_STATE_CHANGE_FAILURE) {
1531 /* Cleanup elements on failed transition out of NULL */
1532 gst_splitmux_reset (splitmux);
1538 register_splitmuxsink (GstPlugin * plugin)
1540 GST_DEBUG_CATEGORY_INIT (splitmux_debug, "splitmuxsink", 0,
1541 "Split File Muxing Sink");
1543 return gst_element_register (plugin, "splitmuxsink", GST_RANK_NONE,
1544 GST_TYPE_SPLITMUX_SINK);