gst/multifile/gstsplitfilesrc.c

   1 /* GStreamer Split File Source
   2  * Copyright (C) 2011 Collabora Ltd. <tim.muller@collabora.co.uk>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19 /**
  20  * SECTION:element-splitfilesrc
  21  * @see_also: #GstFileSrc, #GstMultiFileSrc
  22  *
  23  * Reads data from multiple files, presenting those files as one continuous
  24  * file to downstream elements. This is useful for reading a large file that
  25  * had to be split into multiple parts due to filesystem file size limitations,
  26  * for example.
  27  *
  28  * The files to select are chosen via the location property, which supports
  29  * (and expects) shell-style wildcards (but only for the filename, not for
  30  * directories). The results will be sorted.
  31  *
  32  * <refsect2>
  33  * <title>Example launch lines</title>
  34  * |[
  35  * gst-launch-1.0 splitfilesrc location="/path/to/part-*.mpg" ! decodebin ! ...
  36  * ]| Plays the different parts as if they were one single MPEG file.
  37  * |[
  38  * gst-launch-1.0 playbin uri="splitfile://path/to/foo.avi.*"
  39  * ]| Plays the different parts as if they were one single AVI file.
  40  * </refsect2>
  41  */
  42
  43 #ifdef HAVE_CONFIG_H
  44 #  include "config.h"
  45 #endif
  46
  47 #include "gstsplitfilesrc.h"
  48 #include "gstsplitutils.h"
  49
  50 #include <string.h>
  51
  52 enum
  53 {
  54   PROP_LOCATION = 1
  55 };
  56
  57 #define DEFAULT_LOCATION NULL
  58
  59 static void gst_split_file_src_uri_handler_init (gpointer g_iface,
  60     gpointer iface_data);
  61 static void gst_split_file_src_set_property (GObject * object, guint prop_id,
  62     const GValue * value, GParamSpec * pspec);
  63 static void gst_split_file_src_get_property (GObject * object, guint prop_id,
  64     GValue * value, GParamSpec * pspec);
  65 static void gst_split_file_src_finalize (GObject * obj);
  66
  67 static gboolean gst_split_file_src_start (GstBaseSrc * basesrc);
  68 static gboolean gst_split_file_src_stop (GstBaseSrc * basesrc);
  69 static gboolean gst_split_file_src_can_seek (GstBaseSrc * basesrc);
  70 static gboolean gst_split_file_src_get_size (GstBaseSrc * basesrc, guint64 * s);
  71 static gboolean gst_split_file_src_unlock (GstBaseSrc * basesrc);
  72 static GstFlowReturn gst_split_file_src_create (GstBaseSrc * basesrc,
  73     guint64 offset, guint size, GstBuffer ** buffer);
  74
  75 static GstStaticPadTemplate gst_split_file_src_pad_template =
  76 GST_STATIC_PAD_TEMPLATE ("src",
  77     GST_PAD_SRC,
  78     GST_PAD_ALWAYS,
  79     GST_STATIC_CAPS_ANY);
  80
  81 GST_DEBUG_CATEGORY_STATIC (splitfilesrc_debug);
  82 #define GST_CAT_DEFAULT splitfilesrc_debug
  83
  84
  85 G_DEFINE_TYPE_WITH_CODE (GstSplitFileSrc, gst_split_file_src, GST_TYPE_BASE_SRC,
  86     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER,
  87         gst_split_file_src_uri_handler_init));
  88
  89 #ifdef G_OS_WIN32
  90 #define WIN32_BLURB " Location string must be in UTF-8 encoding (on Windows)."
  91 #else
  92 #define WIN32_BLURB             /* nothing */
  93 #endif
  94
  95 static void
  96 gst_split_file_src_class_init (GstSplitFileSrcClass * klass)
  97 {
  98   GstBaseSrcClass *gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
  99   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 100   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 101
 102   gobject_class->set_property = gst_split_file_src_set_property;
 103   gobject_class->get_property = gst_split_file_src_get_property;
 104   gobject_class->finalize = gst_split_file_src_finalize;
 105
 106   g_object_class_install_property (gobject_class, PROP_LOCATION,
 107       g_param_spec_string ("location", "File Location",
 108           "Wildcard pattern to match file names of the input files. If "
 109           "the location is an absolute path or contains directory components, "
 110           "only the base file name part will be considered for pattern "
 111           "matching. The results will be sorted." WIN32_BLURB,
 112           DEFAULT_LOCATION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 113
 114   gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_split_file_src_start);
 115   gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_split_file_src_stop);
 116   gstbasesrc_class->create = GST_DEBUG_FUNCPTR (gst_split_file_src_create);
 117   gstbasesrc_class->get_size = GST_DEBUG_FUNCPTR (gst_split_file_src_get_size);
 118   gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_split_file_src_unlock);
 119   gstbasesrc_class->is_seekable =
 120       GST_DEBUG_FUNCPTR (gst_split_file_src_can_seek);
 121
 122   GST_DEBUG_CATEGORY_INIT (splitfilesrc_debug, "splitfilesrc", 0,
 123       "splitfilesrc element");
 124
 125   gst_element_class_add_static_pad_template (gstelement_class,
 126       &gst_split_file_src_pad_template);
 127
 128   gst_element_class_set_static_metadata (gstelement_class, "Split-File Source",
 129       "Source/File",
 130       "Read a sequentially named set of files as if it was one large file",
 131       "Tim-Philipp Müller <tim.muller@collabora.co.uk>");
 132 }
 133
 134 static void
 135 gst_split_file_src_init (GstSplitFileSrc * splitfilesrc)
 136 {
 137 }
 138
 139 static void
 140 gst_split_file_src_finalize (GObject * obj)
 141 {
 142   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (obj);
 143
 144   g_free (src->location);
 145   src->location = NULL;
 146
 147   G_OBJECT_CLASS (gst_split_file_src_parent_class)->finalize (obj);
 148 }
 149
 150 static gboolean
 151 gst_split_file_src_can_seek (GstBaseSrc * basesrc)
 152 {
 153   return TRUE;
 154 }
 155
 156 static gboolean
 157 gst_split_file_src_unlock (GstBaseSrc * basesrc)
 158 {
 159   /* This is not actually that useful, since all normal file
 160    * operations are fully blocking anyway */
 161 #if 0
 162   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 163
 164   GST_DEBUG_OBJECT (src, "cancelling pending I/O operation if there is one");
 165   /* g_cancellable_cancel (src->cancellable); */
 166   GST_DEBUG_OBJECT (src, "done");
 167 #endif
 168
 169   return TRUE;
 170 }
 171
 172 static gboolean
 173 gst_split_file_src_get_size (GstBaseSrc * basesrc, guint64 * size)
 174 {
 175   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 176
 177   *size = src->parts[src->num_parts - 1].stop + 1;
 178   return TRUE;
 179 }
 180
 181 static void
 182 gst_split_file_src_set_location (GstSplitFileSrc * src, const char *location)
 183 {
 184   GST_OBJECT_LOCK (src);
 185   g_free (src->location);
 186
 187   if (location != NULL && g_str_has_prefix (location, "splitfile://"))
 188     src->location = gst_uri_get_location (location);
 189   else
 190     src->location = g_strdup (location);
 191 #ifdef G_OS_WIN32
 192   if (!g_utf8_validate (src->location, -1, NULL)) {
 193     g_warning ("splitfilesrc 'location' property must be in UTF-8 "
 194         "encoding on Windows");
 195   }
 196 #endif
 197   GST_OBJECT_UNLOCK (src);
 198 }
 199
 200 static void
 201 gst_split_file_src_set_property (GObject * object, guint prop_id,
 202     const GValue * value, GParamSpec * pspec)
 203 {
 204   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (object);
 205
 206   switch (prop_id) {
 207     case PROP_LOCATION:
 208       gst_split_file_src_set_location (src, g_value_get_string (value));
 209       break;
 210     default:
 211       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 212       break;
 213   }
 214 }
 215
 216 static void
 217 gst_split_file_src_get_property (GObject * object, guint prop_id,
 218     GValue * value, GParamSpec * pspec)
 219 {
 220   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (object);
 221
 222   switch (prop_id) {
 223     case PROP_LOCATION:
 224       GST_OBJECT_LOCK (src);
 225       g_value_set_string (value, src->location);
 226       GST_OBJECT_UNLOCK (src);
 227       break;
 228     default:
 229       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 230       break;
 231   }
 232 }
 233
 234 static gboolean
 235 gst_split_file_src_start (GstBaseSrc * basesrc)
 236 {
 237   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 238   GCancellable *cancel;
 239   gboolean ret = FALSE;
 240   guint64 offset;
 241   GError *err = NULL;
 242   gchar *basename = NULL;
 243   gchar *dirname = NULL;
 244   gchar **files;
 245   guint i;
 246
 247   GST_OBJECT_LOCK (src);
 248   if (src->location != NULL && src->location[0] != '\0') {
 249     basename = g_path_get_basename (src->location);
 250     dirname = g_path_get_dirname (src->location);
 251   }
 252   GST_OBJECT_UNLOCK (src);
 253
 254   files = gst_split_util_find_files (dirname, basename, &err);
 255
 256   if (files == NULL || *files == NULL)
 257     goto no_files;
 258
 259   src->num_parts = g_strv_length (files);
 260   src->parts = g_new0 (GstFilePart, src->num_parts);
 261
 262   cancel = src->cancellable;
 263
 264   offset = 0;
 265   for (i = 0; i < src->num_parts; ++i) {
 266     GFileInputStream *stream;
 267     GFileInfo *info;
 268     goffset size;
 269     GFile *file;
 270
 271     file = g_file_new_for_path (files[i]);
 272     stream = g_file_read (file, cancel, &err);
 273     g_object_unref (file);
 274
 275     if (err != NULL)
 276       goto open_read_error;
 277
 278     info = g_file_input_stream_query_info (stream, "standard::*", NULL, &err);
 279     if (err != NULL) {
 280       g_object_unref (stream);
 281       goto query_info_error;
 282     }
 283
 284     size = g_file_info_get_size (info);
 285     g_object_unref (info);
 286
 287     src->parts[i].stream = stream;
 288     src->parts[i].path = g_strdup (files[i]);
 289     src->parts[i].start = offset;
 290     src->parts[i].stop = offset + size - 1;
 291
 292     GST_DEBUG ("[%010" G_GUINT64_FORMAT "-%010" G_GUINT64_FORMAT "] %s",
 293         src->parts[i].start, src->parts[i].stop, src->parts[i].path);
 294
 295     offset += size;
 296   }
 297
 298   GST_INFO ("Successfully opened %u file parts for reading", src->num_parts);
 299
 300   src->cur_part = 0;
 301
 302   src->cancellable = g_cancellable_new ();
 303
 304   ret = TRUE;
 305
 306 done:
 307   if (err != NULL)
 308     g_error_free (err);
 309   g_strfreev (files);
 310   g_free (basename);
 311   g_free (dirname);
 312   return ret;
 313
 314 /* ERRORS */
 315 no_files:
 316   {
 317     if (err->code == G_IO_ERROR_CANCELLED)
 318       goto cancelled;
 319
 320     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, ("%s", err->message),
 321         ("Failed to find files in '%s' for pattern '%s'",
 322             GST_STR_NULL (dirname), GST_STR_NULL (basename)));
 323     goto done;
 324   }
 325 open_read_error:
 326   {
 327     if (err->code == G_IO_ERROR_CANCELLED)
 328       goto cancelled;
 329
 330     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, ("%s", err->message),
 331         ("Failed to open file '%s' for reading", files[i]));
 332     goto done;
 333   }
 334 query_info_error:
 335   {
 336     if (err->code == G_IO_ERROR_CANCELLED)
 337       goto cancelled;
 338
 339     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, ("%s", err->message),
 340         ("Failed to query info for file '%s'", files[i]));
 341     goto done;
 342   }
 343 cancelled:
 344   {
 345     GST_DEBUG_OBJECT (src, "I/O operation cancelled from another thread");
 346     goto done;
 347   }
 348 }
 349
 350 static gboolean
 351 gst_split_file_src_stop (GstBaseSrc * basesrc)
 352 {
 353   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 354   guint i;
 355
 356   for (i = 0; i < src->num_parts; ++i) {
 357     if (src->parts[i].stream != NULL)
 358       g_object_unref (src->parts[i].stream);
 359     g_free (src->parts[i].path);
 360   }
 361   g_free (src->parts);
 362   src->parts = NULL;
 363   src->num_parts = 0;
 364
 365   g_object_unref (src->cancellable);
 366   src->cancellable = NULL;
 367
 368   return TRUE;
 369 }
 370
 371 static gint
 372 gst_split_file_src_part_search (GstFilePart * part, guint64 * offset,
 373     gpointer user_data)
 374 {
 375   if (*offset > part->stop)
 376     return -1;                  /* The target is after this part */
 377   else if (*offset < part->start)
 378     return 1;                   /* The target is before this part */
 379   else
 380     return 0;                   /* This is the target part */
 381 }
 382
 383 static gboolean
 384 gst_split_file_src_find_part_for_offset (GstSplitFileSrc * src, guint64 offset,
 385     guint * part_number)
 386 {
 387   gboolean res = TRUE;
 388   GstFilePart *part;
 389
 390   part =
 391       gst_util_array_binary_search (src->parts, src->num_parts,
 392       sizeof (GstFilePart),
 393       (GCompareDataFunc) gst_split_file_src_part_search,
 394       GST_SEARCH_MODE_AFTER, &offset, NULL);
 395
 396   if (part)
 397     *part_number = part - src->parts;
 398   else
 399     res = FALSE;
 400
 401   return res;
 402 }
 403
 404 static GstFlowReturn
 405 gst_split_file_src_create (GstBaseSrc * basesrc, guint64 offset, guint size,
 406     GstBuffer ** buffer)
 407 {
 408   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 409   GstFilePart cur_part;
 410   GInputStream *stream;
 411   GCancellable *cancel;
 412   GSeekable *seekable;
 413   GstBuffer *buf;
 414   GError *err = NULL;
 415   guint64 read_offset;
 416   GstMapInfo map;
 417   guint8 *data;
 418   guint to_read;
 419
 420   cur_part = src->parts[src->cur_part];
 421   if (offset < cur_part.start || offset > cur_part.stop) {
 422     if (!gst_split_file_src_find_part_for_offset (src, offset, &src->cur_part))
 423       return GST_FLOW_EOS;
 424     cur_part = src->parts[src->cur_part];
 425   }
 426
 427   GST_LOG_OBJECT (src, "current part: %u (%" G_GUINT64_FORMAT " - "
 428       "%" G_GUINT64_FORMAT ", %s)", src->cur_part, cur_part.start,
 429       cur_part.stop, cur_part.path);
 430
 431   buf = gst_buffer_new_allocate (NULL, size, NULL);
 432
 433   GST_BUFFER_OFFSET (buf) = offset;
 434
 435   gst_buffer_map (buf, &map, GST_MAP_WRITE);
 436   data = map.data;
 437
 438   cancel = src->cancellable;
 439
 440   while (size > 0) {
 441     guint64 bytes_to_end_of_part;
 442     gsize read = 0;
 443
 444     /* we want the offset into the file part */
 445     read_offset = offset - cur_part.start;
 446
 447     GST_LOG ("Reading part %03u from offset %" G_GUINT64_FORMAT " (%s)",
 448         src->cur_part, read_offset, cur_part.path);
 449
 450     /* FIXME: only seek when needed (hopefully gio is smart) */
 451     seekable = G_SEEKABLE (cur_part.stream);
 452     if (!g_seekable_seek (seekable, read_offset, G_SEEK_SET, cancel, &err))
 453       goto seek_failed;
 454
 455     GST_LOG_OBJECT (src, "now: %" G_GUINT64_FORMAT, g_seekable_tell (seekable));
 456
 457     bytes_to_end_of_part = (cur_part.stop - cur_part.start) + 1 - read_offset;
 458     to_read = MIN (size, bytes_to_end_of_part);
 459
 460     GST_LOG_OBJECT (src, "reading %u bytes from part %u (bytes to end of "
 461         "part: %u)", to_read, src->cur_part, (guint) bytes_to_end_of_part);
 462
 463     stream = G_INPUT_STREAM (cur_part.stream);
 464
 465     /* NB: we won't try to read beyond EOF */
 466     if (!g_input_stream_read_all (stream, data, to_read, &read, cancel, &err))
 467       goto read_failed;
 468
 469     GST_LOG_OBJECT (src, "read %u bytes", (guint) read);
 470
 471     data += read;
 472     size -= read;
 473     offset += read;
 474
 475     /* are we done? */
 476     if (size == 0)
 477       break;
 478
 479     GST_LOG_OBJECT (src, "%u bytes left to read for this chunk", size);
 480
 481     /* corner case, this should never really happen (assuming basesrc clips
 482      * requests beyond the file size) */
 483     if (read < to_read) {
 484       if (src->cur_part == src->num_parts - 1) {
 485         /* last file part, stop reading and truncate buffer */
 486         gst_buffer_set_size (buf, offset - GST_BUFFER_OFFSET (buf));
 487         break;
 488       } else {
 489         goto file_part_changed;
 490       }
 491     }
 492
 493     ++src->cur_part;
 494     cur_part = src->parts[src->cur_part];
 495   }
 496
 497   GST_BUFFER_OFFSET_END (buf) = offset;
 498
 499   gst_buffer_unmap (buf, &map);
 500
 501   *buffer = buf;
 502   GST_LOG_OBJECT (src, "read %" G_GSIZE_FORMAT " bytes into buf %p",
 503       gst_buffer_get_size (buf), buf);
 504   return GST_FLOW_OK;
 505
 506 /* ERRORS */
 507 seek_failed:
 508   {
 509     if (err->code == G_IO_ERROR_CANCELLED)
 510       goto cancelled;
 511
 512     GST_ELEMENT_ERROR (src, RESOURCE, SEEK, (NULL),
 513         ("Seek to %" G_GUINT64_FORMAT " in %s failed", read_offset,
 514             cur_part.path));
 515     g_error_free (err);
 516     gst_buffer_unref (buf);
 517     return GST_FLOW_ERROR;
 518   }
 519 read_failed:
 520   {
 521     if (err->code == G_IO_ERROR_CANCELLED)
 522       goto cancelled;
 523
 524     GST_ELEMENT_ERROR (src, RESOURCE, READ, ("%s", err->message),
 525         ("Read from %" G_GUINT64_FORMAT " in %s failed", read_offset,
 526             cur_part.path));
 527     g_error_free (err);
 528     gst_buffer_unref (buf);
 529     return GST_FLOW_ERROR;
 530   }
 531 file_part_changed:
 532   {
 533     GST_ELEMENT_ERROR (src, RESOURCE, READ,
 534         ("Read error while reading file part %s", cur_part.path),
 535         ("Short read in file part, file may have been modified since start"));
 536     gst_buffer_unref (buf);
 537     return GST_FLOW_ERROR;
 538   }
 539 cancelled:
 540   {
 541     GST_DEBUG_OBJECT (src, "I/O operation cancelled from another thread");
 542     g_error_free (err);
 543     gst_buffer_unref (buf);
 544     return GST_FLOW_FLUSHING;
 545   }
 546 }
 547
 548 static guint
 549 gst_split_file_src_uri_get_type (GType type)
 550 {
 551   return GST_URI_SRC;
 552 }
 553
 554 static const gchar *const *
 555 gst_split_file_src_uri_get_protocols (GType type)
 556 {
 557   static const gchar *protocols[] = { "splitfile", NULL };
 558
 559   return (const gchar * const *) protocols;
 560 }
 561
 562 static gchar *
 563 gst_split_file_src_uri_get_uri (GstURIHandler * handler)
 564 {
 565   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (handler);
 566   gchar *ret;
 567
 568   GST_OBJECT_LOCK (src);
 569   if (src->location != NULL)
 570     ret = g_strdup_printf ("splitfile://%s", src->location);
 571   else
 572     ret = NULL;
 573   GST_OBJECT_UNLOCK (src);
 574
 575   return ret;
 576 }
 577
 578 static gboolean
 579 gst_split_file_src_uri_set_uri (GstURIHandler * handler, const gchar * uri,
 580     GError ** error)
 581 {
 582   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (handler);
 583
 584   gst_split_file_src_set_location (src, uri);
 585
 586   return TRUE;
 587 }
 588
 589 static void
 590 gst_split_file_src_uri_handler_init (gpointer g_iface, gpointer iface_data)
 591 {
 592   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
 593
 594   iface->get_type = gst_split_file_src_uri_get_type;
 595   iface->get_protocols = gst_split_file_src_uri_get_protocols;
 596   iface->get_uri = gst_split_file_src_uri_get_uri;
 597   iface->set_uri = gst_split_file_src_uri_set_uri;
 598 }