git.maemo.org Git - mafwsubrenderer/blob - gst-plugins-base-subtitles0.10/ext/gio/gstgiobasesrc.c

   1 /* GStreamer
   2  *
   3  * Copyright (C) 2007 Rene Stadler <mail@renestadler.de>
   4  * Copyright (C) 2007-2009 Sebastian Dröge <sebastian.droege@collabora.co.uk>
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  */
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include <config.h>
  24 #endif
  25
  26 #include "gstgiobasesrc.h"
  27
  28 #include <gst/base/gsttypefindhelper.h>
  29
  30 GST_DEBUG_CATEGORY_STATIC (gst_gio_base_src_debug);
  31 #define GST_CAT_DEFAULT gst_gio_base_src_debug
  32
  33 static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",
  34     GST_PAD_SRC,
  35     GST_PAD_ALWAYS,
  36     GST_STATIC_CAPS_ANY);
  37
  38 GST_BOILERPLATE (GstGioBaseSrc, gst_gio_base_src, GstBaseSrc,
  39     GST_TYPE_BASE_SRC);
  40
  41 static void gst_gio_base_src_finalize (GObject * object);
  42
  43 static gboolean gst_gio_base_src_start (GstBaseSrc * base_src);
  44 static gboolean gst_gio_base_src_stop (GstBaseSrc * base_src);
  45 static gboolean gst_gio_base_src_get_size (GstBaseSrc * base_src,
  46     guint64 * size);
  47 static gboolean gst_gio_base_src_is_seekable (GstBaseSrc * base_src);
  48 static gboolean gst_gio_base_src_unlock (GstBaseSrc * base_src);
  49 static gboolean gst_gio_base_src_unlock_stop (GstBaseSrc * base_src);
  50 static gboolean gst_gio_base_src_check_get_range (GstBaseSrc * base_src);
  51 static GstFlowReturn gst_gio_base_src_create (GstBaseSrc * base_src,
  52     guint64 offset, guint size, GstBuffer ** buf);
  53 static gboolean gst_gio_base_src_query (GstBaseSrc * base_src,
  54     GstQuery * query);
  55
  56 static void
  57 gst_gio_base_src_base_init (gpointer gclass)
  58 {
  59   GstElementClass *element_class = GST_ELEMENT_CLASS (gclass);
  60
  61   GST_DEBUG_CATEGORY_INIT (gst_gio_base_src_debug, "gio_base_src", 0,
  62       "GIO base source");
  63
  64   gst_element_class_add_pad_template (element_class,
  65       gst_static_pad_template_get (&src_factory));
  66 }
  67
  68 static void
  69 gst_gio_base_src_class_init (GstGioBaseSrcClass * klass)
  70 {
  71   GObjectClass *gobject_class = (GObjectClass *) klass;
  72   GstBaseSrcClass *gstbasesrc_class = (GstBaseSrcClass *) klass;
  73
  74   gobject_class->finalize = gst_gio_base_src_finalize;
  75
  76   gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_gio_base_src_start);
  77   gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_gio_base_src_stop);
  78   gstbasesrc_class->get_size = GST_DEBUG_FUNCPTR (gst_gio_base_src_get_size);
  79   gstbasesrc_class->is_seekable =
  80       GST_DEBUG_FUNCPTR (gst_gio_base_src_is_seekable);
  81   gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_gio_base_src_unlock);
  82   gstbasesrc_class->unlock_stop =
  83       GST_DEBUG_FUNCPTR (gst_gio_base_src_unlock_stop);
  84   gstbasesrc_class->check_get_range =
  85       GST_DEBUG_FUNCPTR (gst_gio_base_src_check_get_range);
  86   gstbasesrc_class->create = GST_DEBUG_FUNCPTR (gst_gio_base_src_create);
  87   gstbasesrc_class->query = GST_DEBUG_FUNCPTR (gst_gio_base_src_query);
  88 }
  89
  90 static void
  91 gst_gio_base_src_init (GstGioBaseSrc * src, GstGioBaseSrcClass * gclass)
  92 {
  93   src->cancel = g_cancellable_new ();
  94 }
  95
  96 static void
  97 gst_gio_base_src_finalize (GObject * object)
  98 {
  99   GstGioBaseSrc *src = GST_GIO_BASE_SRC (object);
 100
 101   if (src->cancel) {
 102     g_object_unref (src->cancel);
 103     src->cancel = NULL;
 104   }
 105
 106   if (src->stream) {
 107     g_object_unref (src->stream);
 108     src->stream = NULL;
 109   }
 110
 111   if (src->cache) {
 112     gst_buffer_unref (src->cache);
 113     src->cache = NULL;
 114   }
 115
 116   GST_CALL_PARENT (G_OBJECT_CLASS, finalize, (object));
 117 }
 118
 119 static gboolean
 120 gst_gio_base_src_start (GstBaseSrc * base_src)
 121 {
 122   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
 123   GstGioBaseSrcClass *gbsrc_class = GST_GIO_BASE_SRC_GET_CLASS (src);
 124
 125   src->position = 0;
 126
 127   /* FIXME: This will likely block */
 128   src->stream = gbsrc_class->get_stream (src);
 129   if (G_UNLIKELY (!G_IS_INPUT_STREAM (src->stream))) {
 130     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 131         ("No input stream provided by subclass"));
 132     return FALSE;
 133   } else if (G_UNLIKELY (g_input_stream_is_closed (src->stream))) {
 134     GST_ELEMENT_ERROR (src, LIBRARY, FAILED, (NULL),
 135         ("Input stream is already closed"));
 136     return FALSE;
 137   }
 138
 139   if (G_IS_SEEKABLE (src->stream))
 140     src->position = g_seekable_tell (G_SEEKABLE (src->stream));
 141
 142   GST_DEBUG_OBJECT (src, "started source");
 143
 144   return TRUE;
 145 }
 146
 147 static gboolean
 148 gst_gio_base_src_stop (GstBaseSrc * base_src)
 149 {
 150   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
 151   GstGioBaseSrcClass *klass = GST_GIO_BASE_SRC_GET_CLASS (src);
 152   gboolean success;
 153   GError *err = NULL;
 154
 155   if (klass->close_on_stop && G_IS_INPUT_STREAM (src->stream)) {
 156     GST_DEBUG_OBJECT (src, "closing stream");
 157
 158     /* FIXME: can block but unfortunately we can't use async operations
 159      * here because they require a running main loop */
 160     success = g_input_stream_close (src->stream, src->cancel, &err);
 161
 162     if (!success && !gst_gio_error (src, "g_input_stream_close", &err, NULL)) {
 163       GST_ELEMENT_WARNING (src, RESOURCE, CLOSE, (NULL),
 164           ("g_input_stream_close failed: %s", err->message));
 165       g_clear_error (&err);
 166     } else if (!success) {
 167       GST_ELEMENT_WARNING (src, RESOURCE, CLOSE, (NULL),
 168           ("g_input_stream_close failed"));
 169     } else {
 170       GST_DEBUG_OBJECT (src, "g_input_stream_close succeeded");
 171     }
 172
 173     g_object_unref (src->stream);
 174     src->stream = NULL;
 175   } else {
 176     g_object_unref (src->stream);
 177     src->stream = NULL;
 178   }
 179
 180   return TRUE;
 181 }
 182
 183 static gboolean
 184 gst_gio_base_src_get_size (GstBaseSrc * base_src, guint64 * size)
 185 {
 186   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
 187
 188   if (G_IS_FILE_INPUT_STREAM (src->stream)) {
 189     GFileInfo *info;
 190     GError *err = NULL;
 191
 192     info = g_file_input_stream_query_info (G_FILE_INPUT_STREAM (src->stream),
 193         G_FILE_ATTRIBUTE_STANDARD_SIZE, src->cancel, &err);
 194
 195     if (info != NULL) {
 196       *size = g_file_info_get_size (info);
 197       g_object_unref (info);
 198       GST_DEBUG_OBJECT (src, "found size: %" G_GUINT64_FORMAT, *size);
 199       return TRUE;
 200     }
 201
 202     if (!gst_gio_error (src, "g_file_input_stream_query_info", &err, NULL)) {
 203
 204       if (GST_GIO_ERROR_MATCHES (err, NOT_SUPPORTED))
 205         GST_DEBUG_OBJECT (src, "size information not available");
 206       else
 207         GST_WARNING_OBJECT (src, "size information retrieval failed: %s",
 208             err->message);
 209
 210       g_clear_error (&err);
 211     }
 212   }
 213
 214   if (GST_GIO_STREAM_IS_SEEKABLE (src->stream)) {
 215     goffset old;
 216     goffset stream_size;
 217     gboolean ret;
 218     GSeekable *seekable = G_SEEKABLE (src->stream);
 219     GError *err = NULL;
 220
 221     old = g_seekable_tell (seekable);
 222
 223     ret = g_seekable_seek (seekable, 0, G_SEEK_END, src->cancel, &err);
 224     if (!ret) {
 225       if (!gst_gio_error (src, "g_seekable_seek", &err, NULL)) {
 226         if (GST_GIO_ERROR_MATCHES (err, NOT_SUPPORTED))
 227           GST_DEBUG_OBJECT (src,
 228               "Seeking to the end of stream is not supported");
 229         else
 230           GST_WARNING_OBJECT (src, "Seeking to end of stream failed: %s",
 231               err->message);
 232         g_clear_error (&err);
 233       } else {
 234         GST_WARNING_OBJECT (src, "Seeking to end of stream failed");
 235       }
 236       return FALSE;
 237     }
 238
 239     stream_size = g_seekable_tell (seekable);
 240
 241     ret = g_seekable_seek (seekable, old, G_SEEK_SET, src->cancel, &err);
 242     if (!ret) {
 243       if (!gst_gio_error (src, "g_seekable_seek", &err, NULL)) {
 244         if (GST_GIO_ERROR_MATCHES (err, NOT_SUPPORTED))
 245           GST_ERROR_OBJECT (src, "Seeking to the old position not supported");
 246         else
 247           GST_ERROR_OBJECT (src, "Seeking to the old position failed: %s",
 248               err->message);
 249         g_clear_error (&err);
 250       } else {
 251         GST_ERROR_OBJECT (src, "Seeking to the old position faile");
 252       }
 253       return FALSE;
 254     }
 255
 256     if (stream_size >= 0) {
 257       *size = stream_size;
 258       return TRUE;
 259     }
 260   }
 261
 262   return FALSE;
 263 }
 264
 265 static gboolean
 266 gst_gio_base_src_is_seekable (GstBaseSrc * base_src)
 267 {
 268   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
 269   gboolean seekable;
 270
 271   seekable = GST_GIO_STREAM_IS_SEEKABLE (src->stream);
 272
 273   GST_DEBUG_OBJECT (src, "can seek: %d", seekable);
 274
 275   return seekable;
 276 }
 277
 278 static gboolean
 279 gst_gio_base_src_unlock (GstBaseSrc * base_src)
 280 {
 281   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
 282
 283   GST_LOG_OBJECT (src, "triggering cancellation");
 284
 285   g_cancellable_cancel (src->cancel);
 286
 287   return TRUE;
 288 }
 289
 290 static gboolean
 291 gst_gio_base_src_unlock_stop (GstBaseSrc * base_src)
 292 {
 293   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
 294
 295   GST_LOG_OBJECT (src, "resetting cancellable");
 296
 297   g_cancellable_reset (src->cancel);
 298
 299   return TRUE;
 300 }
 301
 302 static gboolean
 303 gst_gio_base_src_check_get_range (GstBaseSrc * base_src)
 304 {
 305   return GST_CALL_PARENT_WITH_DEFAULT (GST_BASE_SRC_CLASS,
 306       check_get_range, (base_src), FALSE);
 307 }
 308
 309 static GstFlowReturn
 310 gst_gio_base_src_create (GstBaseSrc * base_src, guint64 offset, guint size,
 311     GstBuffer ** buf_return)
 312 {
 313   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
 314   GstBuffer *buf;
 315   GstFlowReturn ret = GST_FLOW_OK;
 316
 317   g_return_val_if_fail (G_IS_INPUT_STREAM (src->stream), GST_FLOW_ERROR);
 318
 319   /* If we have the requested part in our cache take a subbuffer of that,
 320    * otherwise fill the cache again with at least 4096 bytes from the
 321    * requested offset and return a subbuffer of that.
 322    *
 323    * We need caching because every read/seek operation will need to go
 324    * over DBus if our backend is GVfs and this is painfully slow. */
 325   if (src->cache && offset >= GST_BUFFER_OFFSET (src->cache) &&
 326       offset + size <= GST_BUFFER_OFFSET_END (src->cache)) {
 327     GST_DEBUG_OBJECT (src, "Creating subbuffer from cached buffer: offset %"
 328         G_GUINT64_FORMAT " length %u", offset, size);
 329
 330     buf = gst_buffer_create_sub (src->cache,
 331         offset - GST_BUFFER_OFFSET (src->cache), size);
 332
 333     GST_BUFFER_OFFSET (buf) = offset;
 334     GST_BUFFER_OFFSET_END (buf) = offset + size;
 335     GST_BUFFER_SIZE (buf) = size;
 336   } else {
 337     guint cachesize = MAX (4096, size);
 338     gssize read, res;
 339     gboolean success, eos;
 340     GError *err = NULL;
 341
 342     if (src->cache) {
 343       gst_buffer_unref (src->cache);
 344       src->cache = NULL;
 345     }
 346
 347     if (G_UNLIKELY (offset != src->position)) {
 348       if (!GST_GIO_STREAM_IS_SEEKABLE (src->stream))
 349         return GST_FLOW_NOT_SUPPORTED;
 350
 351       GST_DEBUG_OBJECT (src, "Seeking to position %" G_GUINT64_FORMAT, offset);
 352       ret = gst_gio_seek (src, G_SEEKABLE (src->stream), offset, src->cancel);
 353
 354       if (ret == GST_FLOW_OK)
 355         src->position = offset;
 356       else
 357         return ret;
 358     }
 359
 360     src->cache = gst_buffer_try_new_and_alloc (cachesize);
 361     if (G_UNLIKELY (src->cache == NULL)) {
 362       GST_ERROR_OBJECT (src, "Failed to allocate %u bytes", cachesize);
 363       return GST_FLOW_ERROR;
 364     }
 365
 366     GST_LOG_OBJECT (src, "Reading %u bytes from offset %" G_GUINT64_FORMAT,
 367         cachesize, offset);
 368
 369     /* GIO sometimes gives less bytes than requested although
 370      * it's not at the end of file. SMB for example only
 371      * supports reads up to 64k. So we loop here until we get at
 372      * at least the requested amount of bytes or a read returns
 373      * nothing. */
 374     read = 0;
 375     while (size - read > 0 && (res =
 376             g_input_stream_read (G_INPUT_STREAM (src->stream),
 377                 GST_BUFFER_DATA (src->cache) + read, cachesize - read,
 378                 src->cancel, &err)) > 0) {
 379       read += res;
 380     }
 381
 382     success = (read >= 0);
 383     eos = (cachesize > 0 && read == 0);
 384
 385     if (!success && !gst_gio_error (src, "g_input_stream_read", &err, &ret)) {
 386       GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 387           ("Could not read from stream: %s", err->message));
 388       g_clear_error (&err);
 389     }
 390
 391     if (success && !eos) {
 392       src->position += read;
 393       GST_BUFFER_SIZE (src->cache) = read;
 394
 395       GST_BUFFER_OFFSET (src->cache) = offset;
 396       GST_BUFFER_OFFSET_END (src->cache) = offset + read;
 397
 398       GST_DEBUG_OBJECT (src, "Read successful");
 399       GST_DEBUG_OBJECT (src, "Creating subbuffer from new "
 400           "cached buffer: offset %" G_GUINT64_FORMAT " length %u", offset,
 401           size);
 402
 403       buf = gst_buffer_create_sub (src->cache, 0, MIN (size, read));
 404
 405       GST_BUFFER_OFFSET (buf) = offset;
 406       GST_BUFFER_OFFSET_END (buf) = offset + MIN (size, read);
 407       GST_BUFFER_SIZE (buf) = MIN (size, read);
 408     } else {
 409       GST_DEBUG_OBJECT (src, "Read not successful");
 410       gst_buffer_unref (src->cache);
 411       src->cache = NULL;
 412       buf = NULL;
 413     }
 414
 415     if (eos)
 416       ret = GST_FLOW_UNEXPECTED;
 417   }
 418
 419   *buf_return = buf;
 420
 421   return ret;
 422 }
 423
 424 static gboolean
 425 gst_gio_base_src_query (GstBaseSrc * base_src, GstQuery * query)
 426 {
 427   gboolean ret = FALSE;
 428   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
 429
 430   switch (GST_QUERY_TYPE (query)) {
 431     case GST_QUERY_URI:
 432       if (GST_IS_URI_HANDLER (src)) {
 433         const gchar *uri = gst_uri_handler_get_uri (GST_URI_HANDLER (src));
 434         gst_query_set_uri (query, uri);
 435         ret = TRUE;
 436       }
 437       break;
 438     default:
 439       ret = FALSE;
 440       break;
 441   }
 442
 443   if (!ret)
 444     ret = GST_BASE_SRC_CLASS (parent_class)->query (base_src, query);
 445
 446   return ret;
 447 }