gstreamer/subprojects/gst-plugins-bad/gst-libs/gst/mse/gstappendpipeline.c

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

979 lines
26 KiB
C
Raw Normal View History

/* GStreamer
*
* SPDX-License-Identifier: LGPL-2.1
*
* Copyright (C) 2016, 2017 Metrological Group B.V.
* Copyright (C) 2016, 2017 Igalia S.L
* Copyright (C) 2022, 2023 Collabora Ltd.
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "gstappendpipeline-private.h"
#include "mse.h"
#include "gstmselogging-private.h"
#include "gstmsemediatype-private.h"
#include "gstmediasourcetrack-private.h"
#include <gst/gst.h>
#include <gst/app/gstappsrc.h>
#include <gst/app/gstappsink.h>
typedef enum
{
PARSE_ERROR,
IGNORED,
ADDED,
} AddTrackResult;
typedef struct
{
GstAppendPipeline *pipeline;
GstTask *task;
GRecMutex mutex;
GstBus *bus;
} BackgroundTask;
typedef struct
{
GstAppendPipelineCallbacks callbacks;
gpointer user_data;
} Callbacks;
typedef struct
{
GstAppendPipeline *parent;
GstPad *src_pad;
GstAppSink *sink;
GstMediaSourceTrack *mse_track;
GstStream *stream;
GstClockTime previous_pts;
} Track;
typedef struct
{
GstClockTime duration;
GPtrArray *video_tracks;
GPtrArray *audio_tracks;
GPtrArray *text_tracks;
} InitSegment;
struct _GstAppendPipeline
{
GstObject parent_instance;
GstPipeline *pipeline;
GstAppSrc *src;
GstElement *parsebin;
GstBus *bus;
GstStreamCollection *streams;
GArray *tracks;
gboolean received_init_segment;
gboolean have_outstanding_samples;
InitSegment init_segment;
gboolean encountered_error;
BackgroundTask *task;
Callbacks callbacks;
};
G_DEFINE_TYPE (GstAppendPipeline, gst_append_pipeline, GST_TYPE_OBJECT);
#define END_OF_APPEND "end-of-append"
#define ABORT "abort"
#define SHUTDOWN "shutdown"
static void process_init_segment (GstAppendPipeline *);
static gboolean
send_abort (GstAppendPipeline * self)
{
return gst_bus_post (self->bus, gst_message_new_application (NULL,
gst_structure_new_empty (ABORT)));
}
static gboolean
send_shutdown (GstAppendPipeline * self)
{
return gst_bus_post (self->bus, gst_message_new_application (NULL,
gst_structure_new_empty (SHUTDOWN)));
}
static GstEvent *
new_end_of_append_event (void)
{
GstStructure *structure = gst_structure_new_empty (END_OF_APPEND);
return gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM, structure);
}
static gboolean
is_end_of_append_event (GstEvent * event)
{
return GST_EVENT_TYPE (event) == GST_EVENT_CUSTOM_DOWNSTREAM
&& gst_event_has_name (event, END_OF_APPEND);
}
static inline guint
n_tracks (GstAppendPipeline * self)
{
return self->tracks->len;
}
static inline Track *
index_track (GstAppendPipeline * self, guint i)
{
return &g_array_index (self->tracks, Track, i);
}
static inline void
call_parse_error (GstAppendPipeline * self)
{
GstAppendPipelineCallbacks *callbacks = &self->callbacks.callbacks;
gpointer user_data = self->callbacks.user_data;
if (callbacks->error) {
callbacks->error (self, user_data);
GST_TRACE_OBJECT (self, "done");
} else {
GST_TRACE_OBJECT (self, "dropping");
}
}
static inline void
call_received_init_segment (GstAppendPipeline * self)
{
GstAppendPipelineCallbacks *callbacks = &self->callbacks.callbacks;
gpointer user_data = self->callbacks.user_data;
if (callbacks->received_init_segment) {
callbacks->received_init_segment (self, user_data);
GST_TRACE_OBJECT (self, "done");
} else {
GST_TRACE_OBJECT (self, "dropping");
}
}
static inline void
call_new_sample (GstAppendPipeline * self, GstMediaSourceTrack * track,
GstSample * sample)
{
GstAppendPipelineCallbacks *callbacks = &self->callbacks.callbacks;
gpointer user_data = self->callbacks.user_data;
if (callbacks->new_sample) {
callbacks->new_sample (self, track, sample, user_data);
GST_TRACE_OBJECT (self, "done");
} else {
GST_TRACE_OBJECT (self, "dropping");
}
}
static inline void
call_duration_changed (GstAppendPipeline * self)
{
GstAppendPipelineCallbacks *callbacks = &self->callbacks.callbacks;
gpointer user_data = self->callbacks.user_data;
if (callbacks->duration_changed) {
callbacks->duration_changed (self, user_data);
GST_TRACE_OBJECT (self, "done");
} else {
GST_TRACE_OBJECT (self, "dropping");
}
}
static inline void
call_eos (GstAppendPipeline * self, GstMediaSourceTrack * track)
{
GstAppendPipelineCallbacks *callbacks = &self->callbacks.callbacks;
gpointer user_data = self->callbacks.user_data;
if (callbacks->eos) {
callbacks->eos (self, track, user_data);
GST_TRACE_OBJECT (self, "done");
} else {
GST_TRACE_OBJECT (self, "dropping");
}
}
static inline GstSample *
patch_missing_duration (GstAppendPipeline * self, GstSample * sample)
{
GstBuffer *buffer = gst_sample_get_buffer (sample);
if (!GST_BUFFER_DURATION_IS_VALID (buffer)) {
GST_BUFFER_DURATION (buffer) = GST_SECOND / 60;
GST_TRACE_OBJECT (self, "sample is missing duration, patched to %"
GST_TIMEP_FORMAT, &buffer->duration);
}
return sample;
}
static inline GstSample *
patch_missing_pts (GstAppendPipeline * self, GstSample * sample, GstClockTime
fallback)
{
GstBuffer *buffer = gst_sample_get_buffer (sample);
if (!GST_BUFFER_PTS_IS_VALID (buffer) && GST_CLOCK_TIME_IS_VALID (fallback)) {
GST_TRACE_OBJECT (self, "sample is missing pts, patching with %"
GST_TIMEP_FORMAT, &fallback);
GST_BUFFER_PTS (buffer) = fallback;
}
return sample;
}
static inline GstSample *
patch_missing_dts (GstAppendPipeline * self, GstSample * sample)
{
GstBuffer *buffer = gst_sample_get_buffer (sample);
if (!GST_BUFFER_DTS_IS_VALID (buffer) && GST_BUFFER_PTS_IS_VALID (buffer)) {
GST_TRACE_OBJECT (self, "sample is missing dts, patching with pts %"
GST_TIMEP_FORMAT, &buffer->pts);
GST_BUFFER_DTS (buffer) = GST_BUFFER_PTS (buffer);
}
return sample;
}
static gboolean
consume_sample_from_track (GstAppendPipeline * self, Track * track)
{
GstSample *sample = gst_app_sink_try_pull_sample (track->sink, 0);
if (sample == NULL) {
return FALSE;
}
GstBuffer *buffer = gst_sample_get_buffer (sample);
if (!GST_IS_BUFFER (buffer)) {
GST_WARNING_OBJECT (self, "got null buffer in sample");
goto done;
}
sample = patch_missing_pts (self, sample, track->previous_pts);
sample = patch_missing_duration (self, sample);
sample = patch_missing_dts (self, sample);
track->previous_pts = GST_BUFFER_PTS (buffer);
call_new_sample (self, track->mse_track, sample);
done:
gst_clear_sample (&sample);
return TRUE;
}
static void
consume_all_samples (GstAppendPipeline * self)
{
if (!self->received_init_segment) {
GST_DEBUG_OBJECT (self, "not all tracks are available, delaying");
self->have_outstanding_samples = TRUE;
return;
}
guint track_count = n_tracks (self);
while (TRUE) {
gboolean sample_consumed = FALSE;
for (guint i = 0; i < track_count; i++) {
Track *track = index_track (self, i);
sample_consumed |= consume_sample_from_track (self, track);
}
if (!sample_consumed) {
break;
}
}
call_duration_changed (self);
self->have_outstanding_samples = FALSE;
}
static void
handle_shutdown (BackgroundTask * task)
{
gst_task_stop (task->task);
GstAppendPipeline *self = task->pipeline;
guint track_count = n_tracks (self);
for (guint i = 0; i < track_count; i++) {
Track *track = index_track (self, i);
call_eos (self, track->mse_track);
}
call_eos (self, NULL);
}
static void
handle_abort (BackgroundTask * task)
{
gst_task_stop (task->task);
}
static void
task_function (gpointer user_data)
{
BackgroundTask *task = (BackgroundTask *) user_data;
GstAppendPipeline *self = task->pipeline;
GstMessage *message = gst_bus_timed_pop (task->bus, GST_CLOCK_TIME_NONE);
switch (GST_MESSAGE_TYPE (message)) {
case GST_MESSAGE_APPLICATION:{
if (gst_message_has_name (message, END_OF_APPEND)) {
GST_TRACE_OBJECT (self, "end of append");
consume_all_samples (self);
goto done;
}
if (gst_message_has_name (message, SHUTDOWN)) {
GST_DEBUG_OBJECT (self, "shutdown");
handle_shutdown (task);
goto done;
}
if (gst_message_has_name (message, ABORT)) {
GST_DEBUG_OBJECT (self, "abort");
handle_abort (task);
goto done;
}
g_error ("received unsupported application message");
}
case GST_MESSAGE_STREAM_COLLECTION:{
GST_DEBUG_OBJECT (self, "stream collection");
GstStreamCollection *streams;
gst_message_parse_stream_collection (message, &streams);
gst_clear_object (&self->streams);
self->streams = streams;
process_init_segment (self);
goto done;
}
case GST_MESSAGE_EOS:
GST_DEBUG_OBJECT (self, "end of stream");
if (self->have_outstanding_samples) {
GST_DEBUG_OBJECT (self, "consuming remaining samples before EOS");
consume_all_samples (self);
}
handle_shutdown (task);
goto done;
case GST_MESSAGE_ERROR:
GST_DEBUG_OBJECT (self, "error: %" GST_PTR_FORMAT, message);
self->encountered_error = TRUE;
call_parse_error (self);
handle_shutdown (task);
goto done;
default:
GST_TRACE_OBJECT (self, "ignoring message %" GST_PTR_FORMAT, message);
goto done;
}
done:
gst_message_unref (message);
}
static inline GstAppSink *
new_appsink (GstAppendPipeline * self, GstStreamType type)
{
const gchar *type_name = gst_stream_type_get_name (type);
gchar *name = g_strdup_printf ("%s-%u", type_name, n_tracks (self));
GstAppSink *appsink =
GST_APP_SINK (gst_element_factory_make ("appsink", name));
gst_base_sink_set_sync (GST_BASE_SINK (appsink), FALSE);
gst_base_sink_set_async_enabled (GST_BASE_SINK (appsink), FALSE);
gst_base_sink_set_drop_out_of_segment (GST_BASE_SINK (appsink), FALSE);
gst_base_sink_set_last_sample_enabled (GST_BASE_SINK (appsink), FALSE);
g_free (name);
return appsink;
}
static GstPadProbeReturn
black_hole_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
g_return_val_if_fail (GST_PAD_PROBE_INFO_TYPE (info) &
GST_PAD_PROBE_TYPE_BUFFER, GST_PAD_PROBE_DROP);
return GST_PAD_PROBE_DROP;
}
static GstPadProbeReturn
event_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
GstAppendPipeline *self = GST_APPEND_PIPELINE (user_data);
GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);
if (is_end_of_append_event (event)) {
GST_TRACE_OBJECT (self, "end of append event");
if (gst_bus_post (self->bus, gst_message_new_application (NULL,
gst_structure_new_empty (END_OF_APPEND)))) {
return GST_PAD_PROBE_DROP;
} else {
GST_ERROR_OBJECT (self, "failed to post end of append");
goto error;
}
}
if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
GST_DEBUG_OBJECT (self, "eos event");
if (send_shutdown (self)) {
return GST_PAD_PROBE_OK;
} else {
GST_ERROR_OBJECT (self, "failed to post shutdown");
goto error;
}
}
return GST_PAD_PROBE_OK;
error:
GST_PAD_PROBE_INFO_FLOW_RETURN (info) = GST_FLOW_ERROR;
gst_event_unref (event);
return GST_PAD_PROBE_HANDLED;
}
static AddTrackResult
add_track (GstAppendPipeline * self, GstPad * pad, GstStream * stream,
GstCaps * caps, Track * added_track)
{
GstStreamType type = gst_stream_get_stream_type (stream);
GstMediaSourceTrackType track_type =
gst_media_source_track_type_from_stream_type (type);
switch (type) {
case GST_STREAM_TYPE_AUDIO:
case GST_STREAM_TYPE_TEXT:
case GST_STREAM_TYPE_VIDEO:
break;
default:{
GST_DEBUG_OBJECT (self, "unexpected caps %" GST_PTR_FORMAT
", using black hole probe", caps);
gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_BUFFER, black_hole_probe, self,
NULL);
return IGNORED;
}
}
if (type != GST_STREAM_TYPE_TEXT &&
!gst_media_source_media_type_is_caps_supported (caps)) {
GST_ERROR_OBJECT (self, "unsupported caps: %" GST_PTR_FORMAT, caps);
return PARSE_ERROR;
}
GstAppSink *appsink = new_appsink (self, type);
gst_bin_add (GST_BIN (self->pipeline), GST_ELEMENT (appsink));
gst_element_sync_state_with_parent (GST_ELEMENT (appsink));
GstPad *appsink_pad =
gst_element_get_static_pad (GST_ELEMENT (appsink), "sink");
GstPadLinkReturn link_result = gst_pad_link (pad, appsink_pad);
gst_clear_object (&appsink_pad);
if (GST_PAD_LINK_FAILED (link_result)) {
g_error ("failed to link parser to appsink: %s",
gst_pad_link_get_name (link_result));
}
Track track_template = {
.parent = self,
.sink = gst_object_ref (appsink),
.src_pad = gst_object_ref (pad),
.stream = gst_object_ref (stream),
.mse_track = gst_media_source_track_new_with_initial_caps (track_type,
GST_OBJECT_NAME (appsink), caps),
.previous_pts = GST_CLOCK_TIME_NONE,
};
g_array_append_val (self->tracks, track_template);
GST_TRACE_OBJECT (self, "added appsink %s to pad %s",
GST_OBJECT_NAME (appsink), GST_OBJECT_NAME (pad));
*added_track = track_template;
return ADDED;
}
static void
clear_track (Track * track)
{
gst_clear_object (&track->sink);
gst_clear_object (&track->src_pad);
gst_clear_object (&track->mse_track);
gst_clear_object (&track->stream);
}
static inline GPtrArray *
init_segment_tracks_for (InitSegment * self, GstStreamType type)
{
switch (type) {
case GST_STREAM_TYPE_AUDIO:
return self->audio_tracks;
case GST_STREAM_TYPE_TEXT:
return self->text_tracks;
case GST_STREAM_TYPE_VIDEO:
return self->video_tracks;
default:
g_assert_not_reached ();
return NULL;
}
}
static void
process_init_segment_track (GstPad * pad, GstAppendPipeline * self)
{
GST_OBJECT_LOCK (self);
InitSegment *init_segment = &self->init_segment;
GstStream *stream = gst_pad_get_stream (pad);
GstCaps *caps = gst_stream_get_caps (stream);
GST_DEBUG_OBJECT (self, "%" GST_PTR_FORMAT " got caps %" GST_PTR_FORMAT, pad,
caps);
if (gst_pad_is_linked (pad)) {
GST_TRACE_OBJECT (self, "%" GST_PTR_FORMAT " is already linked, skipping",
pad);
goto done;
}
if (!GST_IS_CAPS (caps)) {
GST_ERROR_OBJECT (self, "no caps on %" GST_PTR_FORMAT
" after stream collection", pad);
call_parse_error (self);
goto done;
}
Track track;
AddTrackResult result = add_track (self, pad, stream, caps, &track);
GstMediaSourceTrack *mse_track = track.mse_track;
switch (result) {
case ADDED:{
GstStreamType type = gst_stream_get_stream_type (stream);
GPtrArray *tracks = init_segment_tracks_for (init_segment, type);
if (tracks->len < 1) {
gst_media_source_track_set_active (mse_track, TRUE);
}
g_ptr_array_add (tracks, gst_object_ref (mse_track));
break;
}
case IGNORED:
break;
case PARSE_ERROR:
call_parse_error (self);
break;
}
done:
gst_clear_object (&stream);
gst_clear_caps (&caps);
GST_OBJECT_UNLOCK (self);
}
static void
on_pad_added (GstElement * parsebin, GstPad * pad, gpointer user_data)
{
GstAppendPipeline *self = GST_APPEND_PIPELINE (user_data);
process_init_segment_track (pad, self);
process_init_segment (self);
}
static gboolean
has_track_for_stream (GstAppendPipeline * self, GstStream * stream)
{
guint track_count = n_tracks (self);
for (guint i = 0; i < track_count; i++) {
Track *track = index_track (self, i);
if (track->stream == stream) {
return TRUE;
}
}
return FALSE;
}
static gboolean
has_all_tracks (GstAppendPipeline * self)
{
if (!GST_IS_STREAM_COLLECTION (self->streams)) {
return FALSE;
}
for (guint i = 0; i < gst_stream_collection_get_size (self->streams); i++) {
GstStream *stream = gst_stream_collection_get_stream (self->streams, i);
switch (gst_stream_get_stream_type (stream)) {
case GST_STREAM_TYPE_AUDIO:
case GST_STREAM_TYPE_VIDEO:
case GST_STREAM_TYPE_TEXT:
break;
default:
continue;
}
if (!has_track_for_stream (self, stream)) {
return FALSE;
}
}
return TRUE;
}
static void
process_init_segment (GstAppendPipeline * self)
{
gint64 duration;
InitSegment *init_segment = &self->init_segment;
GST_OBJECT_LOCK (self);
if (!has_all_tracks (self)) {
goto done;
}
if (gst_element_query_duration (self->parsebin, GST_FORMAT_TIME, &duration)) {
init_segment->duration = MAX (0, duration);
} else {
init_segment->duration = GST_CLOCK_TIME_NONE;
}
GST_DEBUG_OBJECT (self, "init segment says duration=%" GST_TIME_FORMAT,
GST_TIME_ARGS ((GstClockTime) duration));
self->received_init_segment = TRUE;
call_received_init_segment (self);
done:
GST_OBJECT_UNLOCK (self);
}
static inline void
init_segment_init (InitSegment * self)
{
self->audio_tracks = g_ptr_array_new_with_free_func (gst_object_unref);
self->text_tracks = g_ptr_array_new_with_free_func (gst_object_unref);
self->video_tracks = g_ptr_array_new_with_free_func (gst_object_unref);
self->duration = GST_CLOCK_TIME_NONE;
}
static inline void
init_segment_finalize (InitSegment * self)
{
g_ptr_array_free (self->audio_tracks, TRUE);
g_ptr_array_free (self->text_tracks, TRUE);
g_ptr_array_free (self->video_tracks, TRUE);
}
static GArray *
new_tracks_array (void)
{
GArray *tracks = g_array_new (TRUE, TRUE, sizeof (Track));
g_array_set_clear_func (tracks, (GDestroyNotify) clear_track);
return tracks;
}
static BackgroundTask *
background_task_new (GstAppendPipeline * pipeline)
{
BackgroundTask *task = g_new0 (BackgroundTask, 1);
g_rec_mutex_init (&task->mutex);
task->task = gst_task_new (task_function, task, NULL);
task->pipeline = pipeline;
task->bus = gst_object_ref (pipeline->bus);
gst_task_set_lock (task->task, &task->mutex);
return task;
}
static gboolean
background_task_start (BackgroundTask * task)
{
gst_bus_set_flushing (task->bus, FALSE);
return gst_task_start (task->task);
}
static gboolean
background_task_stop (BackgroundTask * task)
{
send_abort (task->pipeline);
gst_task_join (task->task);
gst_bus_set_flushing (task->bus, TRUE);
return TRUE;
}
static void
background_task_cleanup (gpointer ptr)
{
BackgroundTask *task = (BackgroundTask *) ptr;
background_task_stop (task);
task->pipeline = NULL;
gst_clear_object (&task->task);
gst_clear_object (&task->bus);
g_rec_mutex_clear (&task->mutex);
g_free (task);
}
static void
gst_append_pipeline_init (GstAppendPipeline * self)
{
GstElement *appsrc = GST_ELEMENT (gst_element_factory_make ("appsrc", "src"));
GstElement *parsebin =
GST_ELEMENT (gst_element_factory_make ("parsebin", "parse"));
GstElement *pipeline = gst_pipeline_new ("append-pipeline");
GstPad *appsrc_pad = GST_PAD (gst_element_get_static_pad (appsrc, "src"));
gst_pad_add_probe (appsrc_pad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM,
event_probe, self, NULL);
gst_object_unref (appsrc_pad);
gst_bin_add_many (GST_BIN (pipeline), appsrc, parsebin, NULL);
if (!gst_element_link (appsrc, parsebin)) {
g_error ("failed to link appsrc to parsebin");
}
self->bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
self->pipeline = gst_object_ref_sink (GST_PIPELINE (pipeline));
self->src = GST_APP_SRC (appsrc);
self->parsebin = parsebin;
self->task = background_task_new (self);
g_signal_connect_object (parsebin, "pad-added", G_CALLBACK (on_pad_added),
self, 0);
self->received_init_segment = FALSE;
self->encountered_error = FALSE;
self->tracks = new_tracks_array ();
init_segment_init (&self->init_segment);
}
static void
gst_append_pipeline_dispose (GObject * object)
{
GstAppendPipeline *self = (GstAppendPipeline *) object;
send_shutdown (self);
g_clear_pointer (&self->task, background_task_cleanup);
gst_element_set_state (GST_ELEMENT (self->pipeline), GST_STATE_NULL);
G_OBJECT_CLASS (gst_append_pipeline_parent_class)->dispose (object);
}
static void
gst_append_pipeline_finalize (GObject * object)
{
GstAppendPipeline *self = (GstAppendPipeline *) object;
gst_clear_object (&self->pipeline);
gst_clear_object (&self->bus);
init_segment_finalize (&self->init_segment);
g_array_free (self->tracks, TRUE);
gst_clear_object (&self->streams);
G_OBJECT_CLASS (gst_append_pipeline_parent_class)->finalize (object);
}
static void
gst_append_pipeline_class_init (GstAppendPipelineClass * klass)
{
GObjectClass *oclass = G_OBJECT_CLASS (klass);
oclass->dispose = GST_DEBUG_FUNCPTR (gst_append_pipeline_dispose);
oclass->finalize = GST_DEBUG_FUNCPTR (gst_append_pipeline_finalize);
}
GstAppendPipeline *
gst_append_pipeline_new (GstAppendPipelineCallbacks * callbacks,
gpointer user_data, GError ** error)
{
gst_mse_init_logging ();
GstAppendPipeline *self = g_object_new (GST_TYPE_APPEND_PIPELINE, NULL);
GstStateChangeReturn started =
gst_element_set_state (GST_ELEMENT (self->pipeline), GST_STATE_PLAYING);
if (started != GST_STATE_CHANGE_SUCCESS) {
GST_ERROR_OBJECT (self, "failed to start: %s",
gst_element_state_change_return_get_name (started));
g_set_error (error,
GST_MEDIA_SOURCE_ERROR, GST_MEDIA_SOURCE_ERROR_INVALID_STATE,
"failed to start append pipeline");
goto error;
}
if (callbacks) {
self->callbacks.callbacks = *callbacks;
self->callbacks.user_data = user_data;
}
if (!background_task_start (self->task)) {
GST_ERROR_OBJECT (self, "failed to start background task");
g_set_error (error,
GST_MEDIA_SOURCE_ERROR, GST_MEDIA_SOURCE_ERROR_INVALID_STATE,
"failed to start append pipeline's background task");
goto error;
}
return gst_object_ref_sink (self);
error:
gst_clear_object (&self);
return NULL;
}
GstFlowReturn
gst_append_pipeline_append (GstAppendPipeline * self, GstBuffer * buffer)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), GST_FLOW_ERROR);
GstFlowReturn push_result = gst_app_src_push_buffer (self->src, buffer);
if (push_result != GST_FLOW_OK)
return push_result;
if (!gst_element_send_event (GST_ELEMENT_CAST (self->src),
new_end_of_append_event ())) {
GST_ERROR_OBJECT (self, "failed to push end-of-append event");
return GST_FLOW_ERROR;
}
return GST_FLOW_OK;
}
GstFlowReturn
gst_append_pipeline_eos (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), GST_FLOW_ERROR);
return gst_app_src_end_of_stream (self->src);
}
gboolean
gst_append_pipeline_stop (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), FALSE);
GstElement *pipeline = GST_ELEMENT (self->pipeline);
GstStateChangeReturn stopped =
gst_element_set_state (pipeline, GST_STATE_NULL);
if (stopped != GST_STATE_CHANGE_SUCCESS) {
GST_ERROR_OBJECT (self, "failed to stop: %s",
gst_element_state_change_return_get_name (stopped));
return FALSE;
}
self->received_init_segment = FALSE;
self->encountered_error = FALSE;
return TRUE;
}
gboolean
gst_append_pipeline_reset (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), FALSE);
GstElement *pipeline = GST_ELEMENT (self->pipeline);
GstStateChangeReturn stopped =
gst_element_set_state (pipeline, GST_STATE_READY);
if (stopped != GST_STATE_CHANGE_SUCCESS) {
GST_ERROR_OBJECT (self, "failed to stop: %s",
gst_element_state_change_return_get_name (stopped));
return FALSE;
}
background_task_stop (self->task);
init_segment_finalize (&self->init_segment);
gst_clear_object (&self->streams);
g_array_free (self->tracks, TRUE);
self->received_init_segment = FALSE;
self->have_outstanding_samples = FALSE;
self->encountered_error = FALSE;
self->tracks = new_tracks_array ();
init_segment_init (&self->init_segment);
if (!background_task_start (self->task)) {
GST_ERROR_OBJECT (self, "failed to start background task");
return FALSE;
}
GstStateChangeReturn started =
gst_element_set_state (pipeline, GST_STATE_PLAYING);
if (started == GST_STATE_CHANGE_SUCCESS) {
return TRUE;
} else {
GST_ERROR_OBJECT (self, "failed to start: %s",
gst_element_state_change_return_get_name (started));
return FALSE;
}
}
gsize
gst_append_pipeline_n_tracks (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), 0);
return n_tracks (self);
}
gboolean
gst_append_pipeline_has_init_segment (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), FALSE);
return self->received_init_segment;
}
GstClockTime
gst_append_pipeline_get_duration (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), GST_CLOCK_TIME_NONE);
if (self->received_init_segment) {
return self->init_segment.duration;
} else {
return GST_CLOCK_TIME_NONE;
}
}
GPtrArray *
gst_append_pipeline_get_audio_tracks (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), NULL);
if (self->received_init_segment) {
return self->init_segment.audio_tracks;
} else {
return NULL;
}
}
GPtrArray *
gst_append_pipeline_get_text_tracks (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), NULL);
if (self->received_init_segment) {
return self->init_segment.text_tracks;
} else {
return NULL;
}
}
GPtrArray *
gst_append_pipeline_get_video_tracks (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), NULL);
if (self->received_init_segment) {
return self->init_segment.video_tracks;
} else {
return NULL;
}
}
gboolean
gst_append_pipeline_get_eos (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), FALSE);
return gst_task_get_state (self->task->task) != GST_TASK_STARTED;
}
void
gst_append_pipeline_fail (GstAppendPipeline * self)
{
g_return_if_fail (GST_IS_APPEND_PIPELINE (self));
gst_bus_post (self->bus, gst_message_new_error (NULL, NULL, NULL));
}
gboolean
gst_append_pipeline_get_failed (GstAppendPipeline * self)
{
g_return_val_if_fail (GST_IS_APPEND_PIPELINE (self), FALSE);
return self->encountered_error;
}