/* GStreamer Split File Source * Copyright (C) 2011 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. */ /** * SECTION:element-splitfilesrc * @see_also: #GstFileSrc, #GstMultiFileSrc * * Reads data from multiple files, presenting those files as one continuous * file to downstream elements. This is useful for reading a large file that * had to be split into multiple parts due to filesystem file size limitations, * for example. * * The files to select are chosen via the location property, which supports * (and expects) shell-style wildcards (but only for the filename, not for * directories). The results will be sorted. * * * Example launch lines * |[ * gst-launch-1.0 splitfilesrc location="/path/to/part-*.mpg" ! decodebin ! ... * ]| Plays the different parts as if they were one single MPEG file. * |[ * gst-launch-1.0 playbin uri="splitfile://path/to/foo.avi.*" * ]| Plays the different parts as if they were one single AVI file. * */ #ifdef HAVE_CONFIG_H # include "config.h" #endif #include "gstsplitfilesrc.h" #include "gstsplitutils.h" #include enum { PROP_LOCATION = 1 }; #define DEFAULT_LOCATION NULL static void gst_split_file_src_uri_handler_init (gpointer g_iface, gpointer iface_data); static void gst_split_file_src_set_property (GObject * object, guint prop_id, const GValue * value, GParamSpec * pspec); static void gst_split_file_src_get_property (GObject * object, guint prop_id, GValue * value, GParamSpec * pspec); static void gst_split_file_src_finalize (GObject * obj); static gboolean gst_split_file_src_start (GstBaseSrc * basesrc); static gboolean gst_split_file_src_stop (GstBaseSrc * basesrc); static gboolean gst_split_file_src_can_seek (GstBaseSrc * basesrc); static gboolean gst_split_file_src_get_size (GstBaseSrc * basesrc, guint64 * s); static gboolean gst_split_file_src_unlock (GstBaseSrc * basesrc); static GstFlowReturn gst_split_file_src_create (GstBaseSrc * basesrc, guint64 offset, guint size, GstBuffer ** buffer); static GstStaticPadTemplate gst_split_file_src_pad_template = GST_STATIC_PAD_TEMPLATE ("src", GST_PAD_SRC, GST_PAD_ALWAYS, GST_STATIC_CAPS_ANY); GST_DEBUG_CATEGORY_STATIC (splitfilesrc_debug); #define GST_CAT_DEFAULT splitfilesrc_debug G_DEFINE_TYPE_WITH_CODE (GstSplitFileSrc, gst_split_file_src, GST_TYPE_BASE_SRC, G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_split_file_src_uri_handler_init)); #ifdef G_OS_WIN32 #define WIN32_BLURB " Location string must be in UTF-8 encoding (on Windows)." #else #define WIN32_BLURB /* nothing */ #endif static void gst_split_file_src_class_init (GstSplitFileSrcClass * klass) { GstBaseSrcClass *gstbasesrc_class = GST_BASE_SRC_CLASS (klass); GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass); GObjectClass *gobject_class = G_OBJECT_CLASS (klass); gobject_class->set_property = gst_split_file_src_set_property; gobject_class->get_property = gst_split_file_src_get_property; gobject_class->finalize = gst_split_file_src_finalize; g_object_class_install_property (gobject_class, PROP_LOCATION, g_param_spec_string ("location", "File Location", "Wildcard pattern to match file names of the input files. If " "the location is an absolute path or contains directory components, " "only the base file name part will be considered for pattern " "matching. The results will be sorted." WIN32_BLURB, DEFAULT_LOCATION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_split_file_src_start); gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_split_file_src_stop); gstbasesrc_class->create = GST_DEBUG_FUNCPTR (gst_split_file_src_create); gstbasesrc_class->get_size = GST_DEBUG_FUNCPTR (gst_split_file_src_get_size); gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_split_file_src_unlock); gstbasesrc_class->is_seekable = GST_DEBUG_FUNCPTR (gst_split_file_src_can_seek); GST_DEBUG_CATEGORY_INIT (splitfilesrc_debug, "splitfilesrc", 0, "splitfilesrc element"); gst_element_class_add_static_pad_template (gstelement_class, &gst_split_file_src_pad_template); gst_element_class_set_static_metadata (gstelement_class, "Split-File Source", "Source/File", "Read a sequentially named set of files as if it was one large file", "Tim-Philipp Müller "); } static void gst_split_file_src_init (GstSplitFileSrc * splitfilesrc) { } static void gst_split_file_src_finalize (GObject * obj) { GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (obj); g_free (src->location); src->location = NULL; G_OBJECT_CLASS (gst_split_file_src_parent_class)->finalize (obj); } static gboolean gst_split_file_src_can_seek (GstBaseSrc * basesrc) { return TRUE; } static gboolean gst_split_file_src_unlock (GstBaseSrc * basesrc) { /* This is not actually that useful, since all normal file * operations are fully blocking anyway */ #if 0 GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc); GST_DEBUG_OBJECT (src, "cancelling pending I/O operation if there is one"); /* g_cancellable_cancel (src->cancellable); */ GST_DEBUG_OBJECT (src, "done"); #endif return TRUE; } static gboolean gst_split_file_src_get_size (GstBaseSrc * basesrc, guint64 * size) { GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc); *size = src->parts[src->num_parts - 1].stop + 1; return TRUE; } static void gst_split_file_src_set_location (GstSplitFileSrc * src, const char *location) { GST_OBJECT_LOCK (src); g_free (src->location); if (location != NULL && g_str_has_prefix (location, "splitfile://")) src->location = gst_uri_get_location (location); else src->location = g_strdup (location); #ifdef G_OS_WIN32 if (!g_utf8_validate (src->location, -1, NULL)) { g_warning ("splitfilesrc 'location' property must be in UTF-8 " "encoding on Windows"); } #endif GST_OBJECT_UNLOCK (src); } static void gst_split_file_src_set_property (GObject * object, guint prop_id, const GValue * value, GParamSpec * pspec) { GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (object); switch (prop_id) { case PROP_LOCATION: gst_split_file_src_set_location (src, g_value_get_string (value)); break; default: G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); break; } } static void gst_split_file_src_get_property (GObject * object, guint prop_id, GValue * value, GParamSpec * pspec) { GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (object); switch (prop_id) { case PROP_LOCATION: GST_OBJECT_LOCK (src); g_value_set_string (value, src->location); GST_OBJECT_UNLOCK (src); break; default: G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); break; } } static gboolean gst_split_file_src_start (GstBaseSrc * basesrc) { GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc); GCancellable *cancel; gboolean ret = FALSE; guint64 offset; GError *err = NULL; gchar *basename = NULL; gchar *dirname = NULL; gchar **files; guint i; GST_OBJECT_LOCK (src); if (src->location != NULL && src->location[0] != '\0') { basename = g_path_get_basename (src->location); dirname = g_path_get_dirname (src->location); } GST_OBJECT_UNLOCK (src); files = gst_split_util_find_files (dirname, basename, &err); if (files == NULL || *files == NULL) goto no_files; src->num_parts = g_strv_length (files); src->parts = g_new0 (GstFilePart, src->num_parts); cancel = src->cancellable; offset = 0; for (i = 0; i < src->num_parts; ++i) { GFileInputStream *stream; GFileInfo *info; goffset size; GFile *file; file = g_file_new_for_path (files[i]); stream = g_file_read (file, cancel, &err); g_object_unref (file); if (err != NULL) goto open_read_error; info = g_file_input_stream_query_info (stream, "standard::*", NULL, &err); if (err != NULL) { g_object_unref (stream); goto query_info_error; } size = g_file_info_get_size (info); g_object_unref (info); src->parts[i].stream = stream; src->parts[i].path = g_strdup (files[i]); src->parts[i].start = offset; src->parts[i].stop = offset + size - 1; GST_DEBUG ("[%010" G_GUINT64_FORMAT "-%010" G_GUINT64_FORMAT "] %s", src->parts[i].start, src->parts[i].stop, src->parts[i].path); offset += size; } GST_INFO ("Successfully opened %u file parts for reading", src->num_parts); src->cur_part = 0; src->cancellable = g_cancellable_new (); ret = TRUE; done: if (err != NULL) g_error_free (err); g_strfreev (files); g_free (basename); g_free (dirname); return ret; /* ERRORS */ no_files: { if (err->code == G_IO_ERROR_CANCELLED) goto cancelled; GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, ("%s", err->message), ("Failed to find files in '%s' for pattern '%s'", GST_STR_NULL (dirname), GST_STR_NULL (basename))); goto done; } open_read_error: { if (err->code == G_IO_ERROR_CANCELLED) goto cancelled; GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, ("%s", err->message), ("Failed to open file '%s' for reading", files[i])); goto done; } query_info_error: { if (err->code == G_IO_ERROR_CANCELLED) goto cancelled; GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, ("%s", err->message), ("Failed to query info for file '%s'", files[i])); goto done; } cancelled: { GST_DEBUG_OBJECT (src, "I/O operation cancelled from another thread"); goto done; } } static gboolean gst_split_file_src_stop (GstBaseSrc * basesrc) { GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc); guint i; for (i = 0; i < src->num_parts; ++i) { if (src->parts[i].stream != NULL) g_object_unref (src->parts[i].stream); g_free (src->parts[i].path); } g_free (src->parts); src->parts = NULL; src->num_parts = 0; g_object_unref (src->cancellable); src->cancellable = NULL; return TRUE; } static gint gst_split_file_src_part_search (GstFilePart * part, guint64 * offset, gpointer user_data) { if (*offset > part->stop) return -1; /* The target is after this part */ else if (*offset < part->start) return 1; /* The target is before this part */ else return 0; /* This is the target part */ } static gboolean gst_split_file_src_find_part_for_offset (GstSplitFileSrc * src, guint64 offset, guint * part_number) { gboolean res = TRUE; GstFilePart *part; part = gst_util_array_binary_search (src->parts, src->num_parts, sizeof (GstFilePart), (GCompareDataFunc) gst_split_file_src_part_search, GST_SEARCH_MODE_AFTER, &offset, NULL); if (part) *part_number = part - src->parts; else res = FALSE; return res; } static GstFlowReturn gst_split_file_src_create (GstBaseSrc * basesrc, guint64 offset, guint size, GstBuffer ** buffer) { GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc); GstFilePart cur_part; GInputStream *stream; GCancellable *cancel; GSeekable *seekable; GstBuffer *buf; GError *err = NULL; guint64 read_offset; GstMapInfo map; guint8 *data; guint to_read; cur_part = src->parts[src->cur_part]; if (offset < cur_part.start || offset > cur_part.stop) { if (!gst_split_file_src_find_part_for_offset (src, offset, &src->cur_part)) return GST_FLOW_EOS; cur_part = src->parts[src->cur_part]; } GST_LOG_OBJECT (src, "current part: %u (%" G_GUINT64_FORMAT " - " "%" G_GUINT64_FORMAT ", %s)", src->cur_part, cur_part.start, cur_part.stop, cur_part.path); buf = gst_buffer_new_allocate (NULL, size, NULL); GST_BUFFER_OFFSET (buf) = offset; gst_buffer_map (buf, &map, GST_MAP_WRITE); data = map.data; cancel = src->cancellable; while (size > 0) { guint64 bytes_to_end_of_part; gsize read = 0; /* we want the offset into the file part */ read_offset = offset - cur_part.start; GST_LOG ("Reading part %03u from offset %" G_GUINT64_FORMAT " (%s)", src->cur_part, read_offset, cur_part.path); /* FIXME: only seek when needed (hopefully gio is smart) */ seekable = G_SEEKABLE (cur_part.stream); if (!g_seekable_seek (seekable, read_offset, G_SEEK_SET, cancel, &err)) goto seek_failed; GST_LOG_OBJECT (src, "now: %" G_GUINT64_FORMAT, g_seekable_tell (seekable)); bytes_to_end_of_part = (cur_part.stop - cur_part.start) + 1 - read_offset; to_read = MIN (size, bytes_to_end_of_part); GST_LOG_OBJECT (src, "reading %u bytes from part %u (bytes to end of " "part: %u)", to_read, src->cur_part, (guint) bytes_to_end_of_part); stream = G_INPUT_STREAM (cur_part.stream); /* NB: we won't try to read beyond EOF */ if (!g_input_stream_read_all (stream, data, to_read, &read, cancel, &err)) goto read_failed; GST_LOG_OBJECT (src, "read %u bytes", (guint) read); data += read; size -= read; offset += read; /* are we done? */ if (size == 0) break; GST_LOG_OBJECT (src, "%u bytes left to read for this chunk", size); /* corner case, this should never really happen (assuming basesrc clips * requests beyond the file size) */ if (read < to_read) { if (src->cur_part == src->num_parts - 1) { /* last file part, stop reading and truncate buffer */ gst_buffer_set_size (buf, offset - GST_BUFFER_OFFSET (buf)); break; } else { goto file_part_changed; } } ++src->cur_part; cur_part = src->parts[src->cur_part]; } GST_BUFFER_OFFSET_END (buf) = offset; gst_buffer_unmap (buf, &map); *buffer = buf; GST_LOG_OBJECT (src, "read %" G_GSIZE_FORMAT " bytes into buf %p", gst_buffer_get_size (buf), buf); return GST_FLOW_OK; /* ERRORS */ seek_failed: { if (err->code == G_IO_ERROR_CANCELLED) goto cancelled; GST_ELEMENT_ERROR (src, RESOURCE, SEEK, (NULL), ("Seek to %" G_GUINT64_FORMAT " in %s failed", read_offset, cur_part.path)); g_error_free (err); gst_buffer_unref (buf); return GST_FLOW_ERROR; } read_failed: { if (err->code == G_IO_ERROR_CANCELLED) goto cancelled; GST_ELEMENT_ERROR (src, RESOURCE, READ, ("%s", err->message), ("Read from %" G_GUINT64_FORMAT " in %s failed", read_offset, cur_part.path)); g_error_free (err); gst_buffer_unref (buf); return GST_FLOW_ERROR; } file_part_changed: { GST_ELEMENT_ERROR (src, RESOURCE, READ, ("Read error while reading file part %s", cur_part.path), ("Short read in file part, file may have been modified since start")); gst_buffer_unref (buf); return GST_FLOW_ERROR; } cancelled: { GST_DEBUG_OBJECT (src, "I/O operation cancelled from another thread"); g_error_free (err); gst_buffer_unref (buf); return GST_FLOW_FLUSHING; } } static GstURIType gst_split_file_src_uri_get_type (GType type) { return GST_URI_SRC; } static const gchar *const * gst_split_file_src_uri_get_protocols (GType type) { static const gchar *protocols[] = { "splitfile", NULL }; return (const gchar * const *) protocols; } static gchar * gst_split_file_src_uri_get_uri (GstURIHandler * handler) { GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (handler); gchar *ret; GST_OBJECT_LOCK (src); if (src->location != NULL) ret = g_strdup_printf ("splitfile://%s", src->location); else ret = NULL; GST_OBJECT_UNLOCK (src); return ret; } static gboolean gst_split_file_src_uri_set_uri (GstURIHandler * handler, const gchar * uri, GError ** error) { GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (handler); gst_split_file_src_set_location (src, uri); return TRUE; } static void gst_split_file_src_uri_handler_init (gpointer g_iface, gpointer iface_data) { GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface; iface->get_type = gst_split_file_src_uri_get_type; iface->get_protocols = gst_split_file_src_uri_get_protocols; iface->get_uri = gst_split_file_src_uri_get_uri; iface->set_uri = gst_split_file_src_uri_set_uri; }