ext/gio/gstgiobasesink.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 "gstgiobasesink.h"
  27
  28 GST_DEBUG_CATEGORY_STATIC (gst_gio_base_sink_debug);
  29 #define GST_CAT_DEFAULT gst_gio_base_sink_debug
  30
  31 static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink",
  32     GST_PAD_SINK,
  33     GST_PAD_ALWAYS,
  34     GST_STATIC_CAPS_ANY);
  35
  36 GST_BOILERPLATE (GstGioBaseSink, gst_gio_base_sink, GstBaseSink,
  37     GST_TYPE_BASE_SINK);
  38
  39 static void gst_gio_base_sink_finalize (GObject * object);
  40 static gboolean gst_gio_base_sink_start (GstBaseSink * base_sink);
  41 static gboolean gst_gio_base_sink_stop (GstBaseSink * base_sink);
  42 static gboolean gst_gio_base_sink_unlock (GstBaseSink * base_sink);
  43 static gboolean gst_gio_base_sink_unlock_stop (GstBaseSink * base_sink);
  44 static gboolean gst_gio_base_sink_event (GstBaseSink * base_sink,
  45     GstEvent * event);
  46 static GstFlowReturn gst_gio_base_sink_render (GstBaseSink * base_sink,
  47     GstBuffer * buffer);
  48 static gboolean gst_gio_base_sink_query (GstPad * pad, GstQuery * query);
  49
  50 static void
  51 gst_gio_base_sink_base_init (gpointer gclass)
  52 {
  53   GstElementClass *element_class = GST_ELEMENT_CLASS (gclass);
  54
  55   GST_DEBUG_CATEGORY_INIT (gst_gio_base_sink_debug, "gio_base_sink", 0,
  56       "GIO base sink");
  57
  58   gst_element_class_add_static_pad_template (element_class, &sink_factory);
  59 }
  60
  61 static void
  62 gst_gio_base_sink_class_init (GstGioBaseSinkClass * klass)
  63 {
  64   GObjectClass *gobject_class = (GObjectClass *) klass;
  65   GstBaseSinkClass *gstbasesink_class = (GstBaseSinkClass *) klass;
  66
  67   gobject_class->finalize = gst_gio_base_sink_finalize;
  68
  69   gstbasesink_class->start = GST_DEBUG_FUNCPTR (gst_gio_base_sink_start);
  70   gstbasesink_class->stop = GST_DEBUG_FUNCPTR (gst_gio_base_sink_stop);
  71   gstbasesink_class->unlock = GST_DEBUG_FUNCPTR (gst_gio_base_sink_unlock);
  72   gstbasesink_class->unlock_stop =
  73       GST_DEBUG_FUNCPTR (gst_gio_base_sink_unlock_stop);
  74   gstbasesink_class->event = GST_DEBUG_FUNCPTR (gst_gio_base_sink_event);
  75   gstbasesink_class->render = GST_DEBUG_FUNCPTR (gst_gio_base_sink_render);
  76 }
  77
  78 static void
  79 gst_gio_base_sink_init (GstGioBaseSink * sink, GstGioBaseSinkClass * gclass)
  80 {
  81   gst_pad_set_query_function (GST_BASE_SINK_PAD (sink),
  82       GST_DEBUG_FUNCPTR (gst_gio_base_sink_query));
  83
  84   gst_base_sink_set_sync (GST_BASE_SINK (sink), FALSE);
  85
  86   sink->cancel = g_cancellable_new ();
  87 }
  88
  89 static void
  90 gst_gio_base_sink_finalize (GObject * object)
  91 {
  92   GstGioBaseSink *sink = GST_GIO_BASE_SINK (object);
  93
  94   if (sink->cancel) {
  95     g_object_unref (sink->cancel);
  96     sink->cancel = NULL;
  97   }
  98
  99   if (sink->stream) {
 100     g_object_unref (sink->stream);
 101     sink->stream = NULL;
 102   }
 103
 104   GST_CALL_PARENT (G_OBJECT_CLASS, finalize, (object));
 105 }
 106
 107 static gboolean
 108 gst_gio_base_sink_start (GstBaseSink * base_sink)
 109 {
 110   GstGioBaseSink *sink = GST_GIO_BASE_SINK (base_sink);
 111   GstGioBaseSinkClass *gbsink_class = GST_GIO_BASE_SINK_GET_CLASS (sink);
 112
 113   sink->position = 0;
 114
 115   /* FIXME: This will likely block */
 116   sink->stream = gbsink_class->get_stream (sink);
 117   if (G_UNLIKELY (!G_IS_OUTPUT_STREAM (sink->stream))) {
 118     GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_WRITE, (NULL),
 119         ("No output stream provided by subclass"));
 120     return FALSE;
 121   } else if (G_UNLIKELY (g_output_stream_is_closed (sink->stream))) {
 122     GST_ELEMENT_ERROR (sink, LIBRARY, FAILED, (NULL),
 123         ("Output stream is already closed"));
 124     return FALSE;
 125   }
 126
 127   GST_DEBUG_OBJECT (sink, "started sink");
 128
 129   return TRUE;
 130 }
 131
 132 static gboolean
 133 gst_gio_base_sink_stop (GstBaseSink * base_sink)
 134 {
 135   GstGioBaseSink *sink = GST_GIO_BASE_SINK (base_sink);
 136   GstGioBaseSinkClass *klass = GST_GIO_BASE_SINK_GET_CLASS (sink);
 137   gboolean success;
 138   GError *err = NULL;
 139
 140   if (klass->close_on_stop && G_IS_OUTPUT_STREAM (sink->stream)) {
 141     GST_DEBUG_OBJECT (sink, "closing stream");
 142
 143     /* FIXME: can block but unfortunately we can't use async operations
 144      * here because they require a running main loop */
 145     success = g_output_stream_close (sink->stream, sink->cancel, &err);
 146
 147     if (!success && !gst_gio_error (sink, "g_output_stream_close", &err, NULL)) {
 148       GST_ELEMENT_WARNING (sink, RESOURCE, CLOSE, (NULL),
 149           ("gio_output_stream_close failed: %s", err->message));
 150       g_clear_error (&err);
 151     } else if (!success) {
 152       GST_ELEMENT_WARNING (sink, RESOURCE, CLOSE, (NULL),
 153           ("g_output_stream_close failed"));
 154     } else {
 155       GST_DEBUG_OBJECT (sink, "g_outut_stream_close succeeded");
 156     }
 157
 158     g_object_unref (sink->stream);
 159     sink->stream = NULL;
 160   } else {
 161     success = g_output_stream_flush (sink->stream, sink->cancel, &err);
 162
 163     if (!success && !gst_gio_error (sink, "g_output_stream_flush", &err, NULL)) {
 164       GST_ELEMENT_WARNING (sink, RESOURCE, CLOSE, (NULL),
 165           ("gio_output_stream_flush failed: %s", err->message));
 166       g_clear_error (&err);
 167     } else if (!success) {
 168       GST_ELEMENT_WARNING (sink, RESOURCE, CLOSE, (NULL),
 169           ("g_output_stream_flush failed"));
 170     } else {
 171       GST_DEBUG_OBJECT (sink, "g_outut_stream_flush succeeded");
 172     }
 173
 174     g_object_unref (sink->stream);
 175     sink->stream = NULL;
 176   }
 177
 178   return TRUE;
 179 }
 180
 181 static gboolean
 182 gst_gio_base_sink_unlock (GstBaseSink * base_sink)
 183 {
 184   GstGioBaseSink *sink = GST_GIO_BASE_SINK (base_sink);
 185
 186   GST_LOG_OBJECT (sink, "triggering cancellation");
 187
 188   g_cancellable_cancel (sink->cancel);
 189
 190   return TRUE;
 191 }
 192
 193 static gboolean
 194 gst_gio_base_sink_unlock_stop (GstBaseSink * base_sink)
 195 {
 196   GstGioBaseSink *sink = GST_GIO_BASE_SINK (base_sink);
 197
 198   GST_LOG_OBJECT (sink, "resetting cancellable");
 199
 200   g_cancellable_reset (sink->cancel);
 201
 202   return TRUE;
 203 }
 204
 205 static gboolean
 206 gst_gio_base_sink_event (GstBaseSink * base_sink, GstEvent * event)
 207 {
 208   GstGioBaseSink *sink = GST_GIO_BASE_SINK (base_sink);
 209   GstFlowReturn ret = GST_FLOW_OK;
 210
 211   if (sink->stream == NULL)
 212     return TRUE;
 213
 214   switch (GST_EVENT_TYPE (event)) {
 215     case GST_EVENT_NEWSEGMENT:
 216       if (G_IS_OUTPUT_STREAM (sink->stream)) {
 217         GstFormat format;
 218         gint64 offset;
 219
 220         gst_event_parse_new_segment (event, NULL, NULL, &format, &offset, NULL,
 221             NULL);
 222
 223         if (format != GST_FORMAT_BYTES) {
 224           GST_WARNING_OBJECT (sink, "ignored NEWSEGMENT event in %s format",
 225               gst_format_get_name (format));
 226           break;
 227         }
 228
 229         if (GST_GIO_STREAM_IS_SEEKABLE (sink->stream)) {
 230           ret = gst_gio_seek (sink, G_SEEKABLE (sink->stream), offset,
 231               sink->cancel);
 232           if (ret == GST_FLOW_OK)
 233             sink->position = offset;
 234         } else {
 235           ret = GST_FLOW_NOT_SUPPORTED;
 236         }
 237       }
 238       break;
 239
 240     case GST_EVENT_EOS:
 241     case GST_EVENT_FLUSH_START:
 242       if (G_IS_OUTPUT_STREAM (sink->stream)) {
 243         gboolean success;
 244         GError *err = NULL;
 245
 246         success = g_output_stream_flush (sink->stream, sink->cancel, &err);
 247
 248         if (!success && !gst_gio_error (sink, "g_output_stream_flush", &err,
 249                 &ret)) {
 250           GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
 251               ("flush failed: %s", err->message));
 252           g_clear_error (&err);
 253         }
 254       }
 255       break;
 256
 257     default:
 258       break;
 259   }
 260
 261   return (ret == GST_FLOW_OK);
 262 }
 263
 264 static GstFlowReturn
 265 gst_gio_base_sink_render (GstBaseSink * base_sink, GstBuffer * buffer)
 266 {
 267   GstGioBaseSink *sink = GST_GIO_BASE_SINK (base_sink);
 268   gssize written;
 269   gboolean success;
 270   GError *err = NULL;
 271
 272   g_return_val_if_fail (G_IS_OUTPUT_STREAM (sink->stream), GST_FLOW_ERROR);
 273
 274   GST_LOG_OBJECT (sink, "writing %u bytes to offset %" G_GUINT64_FORMAT,
 275       GST_BUFFER_SIZE (buffer), sink->position);
 276
 277   written = g_output_stream_write (sink->stream,
 278       GST_BUFFER_DATA (buffer), GST_BUFFER_SIZE (buffer), sink->cancel, &err);
 279
 280   success = (written >= 0);
 281
 282   if (G_UNLIKELY (success && written < GST_BUFFER_SIZE (buffer))) {
 283     /* FIXME: Can this happen?  Should we handle it gracefully?  gnomevfssink
 284      * doesn't... */
 285     GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
 286         ("Could not write to stream: (short write, only %"
 287             G_GSSIZE_FORMAT " bytes of %d bytes written)",
 288             written, GST_BUFFER_SIZE (buffer)));
 289     return GST_FLOW_ERROR;
 290   }
 291
 292   if (success) {
 293     sink->position += written;
 294     return GST_FLOW_OK;
 295
 296   } else {
 297     GstFlowReturn ret;
 298
 299     if (!gst_gio_error (sink, "g_output_stream_write", &err, &ret)) {
 300       if (GST_GIO_ERROR_MATCHES (err, NO_SPACE)) {
 301         GST_ELEMENT_ERROR (sink, RESOURCE, NO_SPACE_LEFT, (NULL),
 302             ("Could not write to stream: %s", err->message));
 303       } else {
 304         GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
 305             ("Could not write to stream: %s", err->message));
 306       }
 307       g_clear_error (&err);
 308     }
 309
 310     return ret;
 311   }
 312 }
 313
 314 static gboolean
 315 gst_gio_base_sink_query (GstPad * pad, GstQuery * query)
 316 {
 317   GstGioBaseSink *sink = GST_GIO_BASE_SINK (GST_PAD_PARENT (pad));
 318   GstFormat format;
 319
 320   switch (GST_QUERY_TYPE (query)) {
 321     case GST_QUERY_POSITION:
 322       gst_query_parse_position (query, &format, NULL);
 323       switch (format) {
 324         case GST_FORMAT_BYTES:
 325         case GST_FORMAT_DEFAULT:
 326           gst_query_set_position (query, GST_FORMAT_BYTES, sink->position);
 327           return TRUE;
 328         default:
 329           return FALSE;
 330       }
 331     case GST_QUERY_FORMATS:
 332       gst_query_set_formats (query, 2, GST_FORMAT_DEFAULT, GST_FORMAT_BYTES);
 333       return TRUE;
 334     case GST_QUERY_URI:
 335       if (GST_IS_URI_HANDLER (sink)) {
 336         const gchar *uri;
 337
 338         uri = gst_uri_handler_get_uri (GST_URI_HANDLER (sink));
 339         gst_query_set_uri (query, uri);
 340         return TRUE;
 341       }
 342       return FALSE;
 343     default:
 344       return gst_pad_query_default (pad, query);
 345   }
 346 }