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