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., 59 Temple Place - Suite 330,
  17  * Boston, MA 02111-1307, 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 line</title>
  34  * |[
  35  * gst-launch splitfilesrc location="/path/to/part-*.mpg" ! decodebin ! ... \
  36  * ]| Plays the different parts as if they were one single MPEG file.
  37  * </refsect2>
  38  *
  39  * Since: 0.10.31
  40  */
  41
  42 /* TODO:
  43  *  - implement splitfile:// URI handler?
  44  */
  45
  46 #ifdef HAVE_CONFIG_H
  47 #  include "config.h"
  48 #endif
  49
  50 #include "gstsplitfilesrc.h"
  51 #include "patternspec.h"
  52
  53 #include <string.h>
  54
  55 #ifdef G_OS_WIN32
  56 #define DEFAULT_PATTERN_MATCH_MODE MATCH_MODE_UTF8
  57 #else
  58 #define DEFAULT_PATTERN_MATCH_MODE MATCH_MODE_AUTO
  59 #endif
  60
  61 enum
  62 {
  63   PROP_LOCATION = 1
  64 };
  65
  66 #define DEFAULT_LOCATION NULL
  67
  68 static void gst_split_file_src_set_property (GObject * object, guint prop_id,
  69     const GValue * value, GParamSpec * pspec);
  70 static void gst_split_file_src_get_property (GObject * object, guint prop_id,
  71     GValue * value, GParamSpec * pspec);
  72 static void gst_split_file_src_finalize (GObject * obj);
  73
  74 static gboolean gst_split_file_src_start (GstBaseSrc * basesrc);
  75 static gboolean gst_split_file_src_stop (GstBaseSrc * basesrc);
  76 static gboolean gst_split_file_src_can_seek (GstBaseSrc * basesrc);
  77 static gboolean gst_split_file_src_check_get_range (GstBaseSrc * basesrc);
  78 static gboolean gst_split_file_src_get_size (GstBaseSrc * basesrc, guint64 * s);
  79 static gboolean gst_split_file_src_unlock (GstBaseSrc * basesrc);
  80 static GstFlowReturn gst_split_file_src_create (GstBaseSrc * basesrc,
  81     guint64 offset, guint size, GstBuffer ** buffer);
  82
  83 static GstStaticPadTemplate gst_split_file_src_pad_template =
  84 GST_STATIC_PAD_TEMPLATE ("src",
  85     GST_PAD_SRC,
  86     GST_PAD_ALWAYS,
  87     GST_STATIC_CAPS_ANY);
  88
  89 GST_DEBUG_CATEGORY_STATIC (splitfilesrc_debug);
  90 #define GST_CAT_DEFAULT splitfilesrc_debug
  91
  92 GST_BOILERPLATE (GstSplitFileSrc, gst_split_file_src, GstBaseSrc,
  93     GST_TYPE_BASE_SRC);
  94
  95 static void
  96 gst_split_file_src_base_init (gpointer g_class)
  97 {
  98   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
  99
 100   GST_DEBUG_CATEGORY_INIT (splitfilesrc_debug, "splitfilesrc", 0,
 101       "splitfilesrc element");
 102
 103   gst_element_class_add_pad_template (gstelement_class,
 104       gst_static_pad_template_get (&gst_split_file_src_pad_template));
 105
 106   gst_element_class_set_details_simple (gstelement_class, "Split-File Source",
 107       "Source/File",
 108       "Read a sequentially named set of files as if it was one large file",
 109       "Tim-Philipp Müller <tim.muller@collabora.co.uk>");
 110 }
 111
 112 #ifdef G_OS_WIN32
 113 #define WIN32_BLURB " Location string must be in UTF-8 encoding (on Windows)."
 114 #else
 115 #define WIN32_BLURB             /* nothing */
 116 #endif
 117
 118 static void
 119 gst_split_file_src_class_init (GstSplitFileSrcClass * klass)
 120 {
 121   GstBaseSrcClass *gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
 122   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 123
 124   gobject_class->set_property = gst_split_file_src_set_property;
 125   gobject_class->get_property = gst_split_file_src_get_property;
 126   gobject_class->finalize = gst_split_file_src_finalize;
 127
 128   g_object_class_install_property (gobject_class, PROP_LOCATION,
 129       g_param_spec_string ("location", "File Location",
 130           "Wildcard pattern to match file names of the input files. If "
 131           "the location is an absolute path or contains directory components, "
 132           "only the base file name part will be considered for pattern "
 133           "matching. The results will be sorted." WIN32_BLURB,
 134           DEFAULT_LOCATION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 135
 136   gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_split_file_src_start);
 137   gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_split_file_src_stop);
 138   gstbasesrc_class->create = GST_DEBUG_FUNCPTR (gst_split_file_src_create);
 139   gstbasesrc_class->get_size = GST_DEBUG_FUNCPTR (gst_split_file_src_get_size);
 140   gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_split_file_src_unlock);
 141   gstbasesrc_class->is_seekable =
 142       GST_DEBUG_FUNCPTR (gst_split_file_src_can_seek);
 143   gstbasesrc_class->check_get_range =
 144       GST_DEBUG_FUNCPTR (gst_split_file_src_check_get_range);
 145 }
 146
 147 static void
 148 gst_split_file_src_init (GstSplitFileSrc * splitfilesrc,
 149     GstSplitFileSrcClass * g_class)
 150 {
 151 }
 152
 153 static void
 154 gst_split_file_src_finalize (GObject * obj)
 155 {
 156   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (obj);
 157
 158   g_free (src->location);
 159   src->location = NULL;
 160
 161   G_OBJECT_CLASS (parent_class)->finalize (obj);
 162 }
 163
 164 static gboolean
 165 gst_split_file_src_can_seek (GstBaseSrc * basesrc)
 166 {
 167   return TRUE;
 168 }
 169
 170 static gboolean
 171 gst_split_file_src_check_get_range (GstBaseSrc * basesrc)
 172 {
 173   return TRUE;
 174 }
 175
 176 static gboolean
 177 gst_split_file_src_unlock (GstBaseSrc * basesrc)
 178 {
 179   /* This is not actually that useful, since all normal file
 180    * operations are fully blocking anyway */
 181 #if 0
 182   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 183
 184   GST_DEBUG_OBJECT (src, "cancelling pending I/O operation if there is one");
 185   /* g_cancellable_cancel (src->cancellable); */
 186   GST_DEBUG_OBJECT (src, "done");
 187 #endif
 188
 189   return TRUE;
 190 }
 191
 192 static gboolean
 193 gst_split_file_src_get_size (GstBaseSrc * basesrc, guint64 * size)
 194 {
 195   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 196
 197   *size = src->parts[src->num_parts - 1].stop + 1;
 198   return TRUE;
 199 }
 200
 201 static void
 202 gst_split_file_src_set_property (GObject * object, guint prop_id,
 203     const GValue * value, GParamSpec * pspec)
 204 {
 205   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (object);
 206
 207   switch (prop_id) {
 208     case PROP_LOCATION:
 209       GST_OBJECT_LOCK (src);
 210       g_free (src->location);
 211       src->location = g_value_dup_string (value);
 212 #ifdef G_OS_WIN32
 213       if (!g_utf8_validate (src->location, -1, NULL)) {
 214         g_warning ("splitfilesrc 'location' property must be in UTF-8 "
 215             "encoding on Windows");
 216       }
 217 #endif
 218       GST_OBJECT_UNLOCK (src);
 219       break;
 220     default:
 221       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 222       break;
 223   }
 224 }
 225
 226 static void
 227 gst_split_file_src_get_property (GObject * object, guint prop_id,
 228     GValue * value, GParamSpec * pspec)
 229 {
 230   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (object);
 231
 232   switch (prop_id) {
 233     case PROP_LOCATION:
 234       GST_OBJECT_LOCK (src);
 235       g_value_set_string (value, src->location);
 236       GST_OBJECT_UNLOCK (src);
 237       break;
 238     default:
 239       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 240       break;
 241   }
 242 }
 243
 244 static int
 245 gst_split_file_src_array_sortfunc (gchar ** a, gchar ** b)
 246 {
 247   return strcmp (*a, *b);
 248 }
 249
 250 static gchar **
 251 gst_split_file_src_find_files (GstSplitFileSrc * src, const gchar * dirname,
 252     const gchar * basename, GError ** err)
 253 {
 254   PatternSpec *pspec;
 255   GPtrArray *files;
 256   const gchar *name;
 257   GDir *dir;
 258
 259   if (dirname == NULL || basename == NULL)
 260     goto invalid_location;
 261
 262   GST_INFO_OBJECT (src, "checking in directory '%s' for pattern '%s'",
 263       dirname, basename);
 264
 265   dir = g_dir_open (dirname, 0, err);
 266   if (dir == NULL)
 267     return NULL;
 268
 269   if (DEFAULT_PATTERN_MATCH_MODE == MATCH_MODE_UTF8 &&
 270       !g_utf8_validate (basename, -1, NULL)) {
 271     goto not_utf8;
 272   }
 273
 274   /* mode will be AUTO on linux/unix and UTF8 on win32 */
 275   pspec = pattern_spec_new (basename, DEFAULT_PATTERN_MATCH_MODE);
 276
 277   files = g_ptr_array_new ();
 278
 279   while ((name = g_dir_read_name (dir))) {
 280     GST_TRACE_OBJECT (src, "check: %s", name);
 281     if (pattern_match_string (pspec, name)) {
 282       GST_DEBUG_OBJECT (src, "match: %s", name);
 283       g_ptr_array_add (files, g_build_filename (dirname, name, NULL));
 284     }
 285   }
 286
 287   if (files->len == 0)
 288     goto no_matches;
 289
 290   g_ptr_array_sort (files, (GCompareFunc) gst_split_file_src_array_sortfunc);
 291   g_ptr_array_add (files, NULL);
 292
 293   pattern_spec_free (pspec);
 294   g_dir_close (dir);
 295
 296   return (gchar **) g_ptr_array_free (files, FALSE);
 297
 298 /* ERRORS */
 299 invalid_location:
 300   {
 301     g_set_error_literal (err, G_FILE_ERROR, G_FILE_ERROR_INVAL,
 302         "No filename specified.");
 303     return NULL;
 304   }
 305 not_utf8:
 306   {
 307     g_dir_close (dir);
 308     g_set_error_literal (err, G_FILE_ERROR, G_FILE_ERROR_INVAL,
 309         "Filename pattern must be UTF-8 on Windows.");
 310     return NULL;
 311   }
 312 no_matches:
 313   {
 314     pattern_spec_free (pspec);
 315     g_dir_close (dir);
 316     g_set_error_literal (err, G_FILE_ERROR, G_FILE_ERROR_NOENT,
 317         "Found no files matching the pattern.");
 318     return NULL;
 319   }
 320 }
 321
 322 static gboolean
 323 gst_split_file_src_start (GstBaseSrc * basesrc)
 324 {
 325   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 326   GCancellable *cancel;
 327   gboolean ret = FALSE;
 328   guint64 offset;
 329   GError *err = NULL;
 330   gchar *basename = NULL;
 331   gchar *dirname = NULL;
 332   gchar **files;
 333   guint i;
 334
 335   GST_OBJECT_LOCK (src);
 336   if (src->location != NULL && src->location[0] != '\0') {
 337     basename = g_path_get_basename (src->location);
 338     dirname = g_path_get_dirname (src->location);
 339   }
 340   GST_OBJECT_UNLOCK (src);
 341
 342   files = gst_split_file_src_find_files (src, dirname, basename, &err);
 343
 344   if (files == NULL || *files == NULL)
 345     goto no_files;
 346
 347   src->num_parts = g_strv_length (files);
 348   src->parts = g_new0 (GstFilePart, src->num_parts);
 349
 350   cancel = src->cancellable;
 351
 352   offset = 0;
 353   for (i = 0; i < src->num_parts; ++i) {
 354     GFileInputStream *stream;
 355     GFileInfo *info;
 356     goffset size;
 357     GFile *file;
 358
 359     file = g_file_new_for_path (files[i]);
 360     stream = g_file_read (file, cancel, &err);
 361     g_object_unref (file);
 362
 363     if (err != NULL)
 364       goto open_read_error;
 365
 366     info = g_file_input_stream_query_info (stream, "standard::*", NULL, &err);
 367     if (err != NULL) {
 368       g_object_unref (stream);
 369       goto query_info_error;
 370     }
 371
 372     size = g_file_info_get_size (info);
 373     g_object_unref (info);
 374
 375     src->parts[i].stream = stream;
 376     src->parts[i].path = g_strdup (files[i]);
 377     src->parts[i].start = offset;
 378     src->parts[i].stop = offset + size - 1;
 379
 380     GST_DEBUG ("[%010" G_GUINT64_FORMAT "-%010" G_GUINT64_FORMAT "] %s",
 381         src->parts[i].start, src->parts[i].stop, src->parts[i].path);
 382
 383     offset += size;
 384   }
 385
 386   GST_INFO ("Successfully opened %u file parts for reading", src->num_parts);
 387
 388   src->cur_part = 0;
 389
 390   src->cancellable = g_cancellable_new ();
 391
 392   ret = TRUE;
 393
 394 done:
 395   if (err != NULL)
 396     g_error_free (err);
 397   g_strfreev (files);
 398   g_free (basename);
 399   g_free (dirname);
 400   return ret;
 401
 402 /* ERRORS */
 403 no_files:
 404   {
 405     if (err->code == G_IO_ERROR_CANCELLED)
 406       goto cancelled;
 407
 408     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, ("%s", err->message),
 409         ("Failed to find files in '%s' for pattern '%s'",
 410             GST_STR_NULL (dirname), GST_STR_NULL (basename)));
 411     goto done;
 412   }
 413 open_read_error:
 414   {
 415     if (err->code == G_IO_ERROR_CANCELLED)
 416       goto cancelled;
 417
 418     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, ("%s", err->message),
 419         ("Failed to open file '%s' for reading", files[i]));
 420     goto done;
 421   }
 422 query_info_error:
 423   {
 424     if (err->code == G_IO_ERROR_CANCELLED)
 425       goto cancelled;
 426
 427     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, ("%s", err->message),
 428         ("Failed to query info for file '%s'", files[i]));
 429     goto done;
 430   }
 431 cancelled:
 432   {
 433     GST_DEBUG_OBJECT (src, "I/O operation cancelled from another thread");
 434     goto done;
 435   }
 436 }
 437
 438 static gboolean
 439 gst_split_file_src_stop (GstBaseSrc * basesrc)
 440 {
 441   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 442   guint i;
 443
 444   for (i = 0; i < src->num_parts; ++i) {
 445     if (src->parts[i].stream != NULL)
 446       g_object_unref (src->parts[i].stream);
 447     g_free (src->parts[i].path);
 448   }
 449   g_free (src->parts);
 450   src->parts = NULL;
 451   src->num_parts = 0;
 452
 453   g_object_unref (src->cancellable);
 454   src->cancellable = NULL;
 455
 456   return TRUE;
 457 }
 458
 459 static gboolean
 460 gst_split_file_src_find_part_for_offset (GstSplitFileSrc * src, guint64 offset,
 461     guint * part_number)
 462 {
 463   GstFilePart *part;
 464   guint i;
 465
 466   /* TODO: could use gst_util_array_binary_search() here */
 467   part = src->parts;
 468   for (i = 0; i < src->num_parts; ++i) {
 469     if (offset >= part->start && offset <= part->stop) {
 470       *part_number = i;
 471       return TRUE;
 472     }
 473     ++part;
 474   }
 475
 476   return FALSE;
 477 }
 478
 479 static GstFlowReturn
 480 gst_split_file_src_create (GstBaseSrc * basesrc, guint64 offset, guint size,
 481     GstBuffer ** buffer)
 482 {
 483   GstSplitFileSrc *src = GST_SPLIT_FILE_SRC (basesrc);
 484   GstFilePart cur_part;
 485   GInputStream *stream;
 486   GCancellable *cancel;
 487   GSeekable *seekable;
 488   GstBuffer *buf;
 489   GError *err = NULL;
 490   guint64 read_offset;
 491   guint8 *data;
 492   guint to_read;
 493
 494   cur_part = src->parts[src->cur_part];
 495   if (offset < cur_part.start || offset > cur_part.stop) {
 496     if (!gst_split_file_src_find_part_for_offset (src, offset, &src->cur_part))
 497       return GST_FLOW_UNEXPECTED;
 498     cur_part = src->parts[src->cur_part];
 499   }
 500
 501   GST_LOG_OBJECT (src, "current part: %u (%" G_GUINT64_FORMAT " - "
 502       "%" G_GUINT64_FORMAT ", %s)", src->cur_part, cur_part.start,
 503       cur_part.stop, cur_part.path);
 504
 505   buf = gst_buffer_new_and_alloc (size);
 506
 507   GST_BUFFER_OFFSET (buf) = offset;
 508
 509   data = GST_BUFFER_DATA (buf);
 510
 511   cancel = src->cancellable;
 512
 513   while (size > 0) {
 514     guint64 bytes_to_end_of_part;
 515     gsize read = 0;
 516
 517     /* we want the offset into the file part */
 518     read_offset = offset - cur_part.start;
 519
 520     GST_LOG ("Reading part %03u from offset %" G_GUINT64_FORMAT " (%s)",
 521         src->cur_part, read_offset, cur_part.path);
 522
 523     /* FIXME: only seek when needed (hopefully gio is smart) */
 524     seekable = G_SEEKABLE (cur_part.stream);
 525     if (!g_seekable_seek (seekable, read_offset, G_SEEK_SET, cancel, &err))
 526       goto seek_failed;
 527
 528     GST_LOG_OBJECT (src, "now: %" G_GUINT64_FORMAT, g_seekable_tell (seekable));
 529
 530     bytes_to_end_of_part = (cur_part.stop - cur_part.start) + 1 - read_offset;
 531     to_read = MIN (size, bytes_to_end_of_part);
 532
 533     GST_LOG_OBJECT (src, "reading %u bytes from part %u (bytes to end of "
 534         "part: %u)", to_read, src->cur_part, (guint) bytes_to_end_of_part);
 535
 536     stream = G_INPUT_STREAM (cur_part.stream);
 537
 538     /* NB: we won't try to read beyond EOF */
 539     if (!g_input_stream_read_all (stream, data, to_read, &read, cancel, &err))
 540       goto read_failed;
 541
 542     GST_LOG_OBJECT (src, "read %u bytes", (guint) read);
 543
 544     data += read;
 545     size -= read;
 546     offset += read;
 547
 548     /* are we done? */
 549     if (size == 0)
 550       break;
 551
 552     GST_LOG_OBJECT (src, "%u bytes left to read for this chunk", size);
 553
 554     /* corner case, this should never really happen (assuming basesrc clips
 555      * requests beyond the file size) */
 556     if (read < to_read) {
 557       if (src->cur_part == src->num_parts - 1) {
 558         /* last file part, stop reading and truncate buffer */
 559         GST_BUFFER_SIZE (buf) = offset - GST_BUFFER_OFFSET (buf);
 560         break;
 561       } else {
 562         goto file_part_changed;
 563       }
 564     }
 565
 566     ++src->cur_part;
 567     cur_part = src->parts[src->cur_part];
 568   }
 569
 570   GST_BUFFER_OFFSET_END (buf) = offset;
 571
 572   *buffer = buf;
 573   GST_LOG_OBJECT (src, "read %u bytes into buf %p", GST_BUFFER_SIZE (buf), buf);
 574   return GST_FLOW_OK;
 575
 576 /* ERRORS */
 577 seek_failed:
 578   {
 579     if (err->code == G_IO_ERROR_CANCELLED)
 580       goto cancelled;
 581
 582     GST_ELEMENT_ERROR (src, RESOURCE, SEEK, (NULL),
 583         ("Seek to %" G_GUINT64_FORMAT " in %s failed", read_offset,
 584             cur_part.path));
 585     g_error_free (err);
 586     gst_buffer_unref (buf);
 587     return GST_FLOW_ERROR;
 588   }
 589 read_failed:
 590   {
 591     if (err->code == G_IO_ERROR_CANCELLED)
 592       goto cancelled;
 593
 594     GST_ELEMENT_ERROR (src, RESOURCE, READ, ("%s", err->message),
 595         ("Read from %" G_GUINT64_FORMAT " in %s failed", read_offset,
 596             cur_part.path));
 597     g_error_free (err);
 598     gst_buffer_unref (buf);
 599     return GST_FLOW_ERROR;
 600   }
 601 file_part_changed:
 602   {
 603     GST_ELEMENT_ERROR (src, RESOURCE, READ,
 604         ("Read error while reading file part %s", cur_part.path),
 605         ("Short read in file part, file may have been modified since start"));
 606     gst_buffer_unref (buf);
 607     return GST_FLOW_ERROR;
 608   }
 609 cancelled:
 610   {
 611     GST_DEBUG_OBJECT (src, "I/O operation cancelled from another thread");
 612     g_error_free (err);
 613     gst_buffer_unref (buf);
 614     return GST_FLOW_WRONG_STATE;
 615   }
 616 }