gst-libs/gst/uridownloader/gsturidownloader.c

   1 /* GStreamer
   2  * Copyright (C) 2011 Andoni Morales Alastruey <ylatuya@gmail.com>
   3  *
   4  * gstfragment.c:
   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., 51 Franklin St, Fifth Floor,
  19  * Boston, MA 02110-1301, USA.
  20  */
  21
  22 #include <glib.h>
  23 #include "gstfragment.h"
  24 #include "gsturidownloader.h"
  25 #include "gsturidownloader_debug.h"
  26
  27 #define GST_CAT_DEFAULT uridownloader_debug
  28 GST_DEBUG_CATEGORY (uridownloader_debug);
  29
  30 #define GST_URI_DOWNLOADER_GET_PRIVATE(obj)  \
  31    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), \
  32     GST_TYPE_URI_DOWNLOADER, GstUriDownloaderPrivate))
  33
  34 struct _GstUriDownloaderPrivate
  35 {
  36   /* Fragments fetcher */
  37   GstElement *urisrc;
  38   GstBus *bus;
  39   GstPad *pad;
  40   GTimeVal *timeout;
  41   GstFragment *download;
  42   GMutex lock;
  43   GCond cond;
  44   gboolean cancelled;
  45 };
  46
  47 static void gst_uri_downloader_finalize (GObject * object);
  48 static void gst_uri_downloader_dispose (GObject * object);
  49
  50 static GstFlowReturn gst_uri_downloader_chain (GstPad * pad, GstObject * parent,
  51     GstBuffer * buf);
  52 static gboolean gst_uri_downloader_sink_event (GstPad * pad, GstObject * parent,
  53     GstEvent * event);
  54 static GstBusSyncReply gst_uri_downloader_bus_handler (GstBus * bus,
  55     GstMessage * message, gpointer data);
  56
  57 static GstStaticPadTemplate sinkpadtemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  58     GST_PAD_SINK,
  59     GST_PAD_ALWAYS,
  60     GST_STATIC_CAPS_ANY);
  61
  62 #define _do_init \
  63 { \
  64   GST_DEBUG_CATEGORY_INIT (uridownloader_debug, "uridownloader", 0, "URI downloader"); \
  65 }
  66
  67 G_DEFINE_TYPE_WITH_CODE (GstUriDownloader, gst_uri_downloader, GST_TYPE_OBJECT,
  68     _do_init);
  69
  70 static void
  71 gst_uri_downloader_class_init (GstUriDownloaderClass * klass)
  72 {
  73   GObjectClass *gobject_class;
  74
  75   gobject_class = (GObjectClass *) klass;
  76
  77   g_type_class_add_private (klass, sizeof (GstUriDownloaderPrivate));
  78
  79   gobject_class->dispose = gst_uri_downloader_dispose;
  80   gobject_class->finalize = gst_uri_downloader_finalize;
  81 }
  82
  83 static void
  84 gst_uri_downloader_init (GstUriDownloader * downloader)
  85 {
  86   downloader->priv = GST_URI_DOWNLOADER_GET_PRIVATE (downloader);
  87
  88   /* Initialize the sink pad. This pad will be connected to the src pad of the
  89    * element created with gst_element_make_from_uri and will handle the download */
  90   downloader->priv->pad =
  91       gst_pad_new_from_static_template (&sinkpadtemplate, "sink");
  92   gst_pad_set_chain_function (downloader->priv->pad,
  93       GST_DEBUG_FUNCPTR (gst_uri_downloader_chain));
  94   gst_pad_set_event_function (downloader->priv->pad,
  95       GST_DEBUG_FUNCPTR (gst_uri_downloader_sink_event));
  96   gst_pad_set_element_private (downloader->priv->pad, downloader);
  97   gst_pad_set_active (downloader->priv->pad, TRUE);
  98
  99   /* Create a bus to handle error and warning message from the source element */
 100   downloader->priv->bus = gst_bus_new ();
 101
 102   g_mutex_init (&downloader->priv->lock);
 103   g_cond_init (&downloader->priv->cond);
 104 }
 105
 106 static void
 107 gst_uri_downloader_dispose (GObject * object)
 108 {
 109   GstUriDownloader *downloader = GST_URI_DOWNLOADER (object);
 110
 111   if (downloader->priv->urisrc != NULL) {
 112     gst_object_unref (downloader->priv->urisrc);
 113     downloader->priv->urisrc = NULL;
 114   }
 115
 116   if (downloader->priv->bus != NULL) {
 117     gst_object_unref (downloader->priv->bus);
 118     downloader->priv->bus = NULL;
 119   }
 120
 121   if (downloader->priv->pad) {
 122     gst_object_unref (downloader->priv->pad);
 123     downloader->priv->pad = NULL;
 124   }
 125
 126   if (downloader->priv->download) {
 127     g_object_unref (downloader->priv->download);
 128     downloader->priv->download = NULL;
 129   }
 130
 131   G_OBJECT_CLASS (gst_uri_downloader_parent_class)->dispose (object);
 132 }
 133
 134 static void
 135 gst_uri_downloader_finalize (GObject * object)
 136 {
 137   GstUriDownloader *downloader = GST_URI_DOWNLOADER (object);
 138
 139   g_mutex_clear (&downloader->priv->lock);
 140   g_cond_clear (&downloader->priv->cond);
 141
 142   G_OBJECT_CLASS (gst_uri_downloader_parent_class)->finalize (object);
 143 }
 144
 145 GstUriDownloader *
 146 gst_uri_downloader_new (void)
 147 {
 148   return g_object_new (GST_TYPE_URI_DOWNLOADER, NULL);
 149 }
 150
 151 static gboolean
 152 gst_uri_downloader_sink_event (GstPad * pad, GstObject * parent,
 153     GstEvent * event)
 154 {
 155   gboolean ret = FALSE;
 156   GstUriDownloader *downloader;
 157
 158   downloader = GST_URI_DOWNLOADER (gst_pad_get_element_private (pad));
 159
 160   switch (event->type) {
 161     case GST_EVENT_EOS:{
 162       GST_OBJECT_LOCK (downloader);
 163       GST_DEBUG_OBJECT (downloader, "Got EOS on the fetcher pad");
 164       if (downloader->priv->download != NULL) {
 165         /* signal we have fetched the URI */
 166         downloader->priv->download->completed = TRUE;
 167         downloader->priv->download->download_stop_time =
 168             gst_util_get_timestamp ();
 169         GST_OBJECT_UNLOCK (downloader);
 170         GST_DEBUG_OBJECT (downloader, "Signaling chain funtion");
 171         g_mutex_lock (&downloader->priv->lock);
 172         g_cond_signal (&downloader->priv->cond);
 173         g_mutex_unlock (&downloader->priv->lock);
 174       } else {
 175         GST_OBJECT_UNLOCK (downloader);
 176       }
 177       gst_event_unref (event);
 178       break;
 179     }
 180     default:
 181       ret = gst_pad_event_default (pad, parent, event);
 182       break;
 183   }
 184
 185   return ret;
 186 }
 187
 188 static GstBusSyncReply
 189 gst_uri_downloader_bus_handler (GstBus * bus,
 190     GstMessage * message, gpointer data)
 191 {
 192   GstUriDownloader *downloader = (GstUriDownloader *) (data);
 193
 194   if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_ERROR ||
 195       GST_MESSAGE_TYPE (message) == GST_MESSAGE_WARNING) {
 196     GError *err = NULL;
 197     gchar *dbg_info = NULL;
 198
 199     gst_message_parse_error (message, &err, &dbg_info);
 200     GST_WARNING_OBJECT (downloader,
 201         "Received error: %s from %s, the download will be cancelled",
 202         GST_OBJECT_NAME (message->src), err->message);
 203     GST_DEBUG ("Debugging info: %s\n", (dbg_info) ? dbg_info : "none");
 204     g_error_free (err);
 205     g_free (dbg_info);
 206
 207     /* remove the sync handler to avoid duplicated messages */
 208     gst_bus_set_sync_handler (downloader->priv->bus, NULL, NULL, NULL);
 209     gst_uri_downloader_cancel (downloader);
 210   }
 211
 212   gst_message_unref (message);
 213   return GST_BUS_DROP;
 214 }
 215
 216 static GstFlowReturn
 217 gst_uri_downloader_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
 218 {
 219   GstUriDownloader *downloader;
 220
 221   downloader = GST_URI_DOWNLOADER (gst_pad_get_element_private (pad));
 222
 223   /* HTML errors (404, 500, etc...) are also pushed through this pad as
 224    * response but the source element will also post a warning or error message
 225    * in the bus, which is handled synchronously cancelling the download.
 226    */
 227   GST_OBJECT_LOCK (downloader);
 228   if (downloader->priv->download == NULL) {
 229     /* Download cancelled, quit */
 230     GST_OBJECT_UNLOCK (downloader);
 231     goto done;
 232   }
 233
 234   GST_LOG_OBJECT (downloader, "The uri fetcher received a new buffer "
 235       "of size %" G_GSIZE_FORMAT, gst_buffer_get_size (buf));
 236   if (!gst_fragment_add_buffer (downloader->priv->download, buf))
 237     GST_WARNING_OBJECT (downloader, "Could not add buffer to fragment");
 238   GST_OBJECT_UNLOCK (downloader);
 239
 240 done:
 241   {
 242     return GST_FLOW_OK;
 243   }
 244 }
 245
 246 static void
 247 gst_uri_downloader_stop (GstUriDownloader * downloader)
 248 {
 249   GstPad *pad;
 250
 251   GST_DEBUG_OBJECT (downloader, "Stopping source element");
 252
 253   /* remove the bus' sync handler */
 254   gst_bus_set_sync_handler (downloader->priv->bus, NULL, NULL, NULL);
 255   /* unlink the source element from the internal pad */
 256   pad = gst_pad_get_peer (downloader->priv->pad);
 257   if (pad) {
 258     gst_pad_unlink (pad, downloader->priv->pad);
 259     gst_object_unref (pad);
 260   }
 261   /* set the element state to NULL */
 262   gst_element_set_state (downloader->priv->urisrc, GST_STATE_NULL);
 263   gst_element_get_state (downloader->priv->urisrc, NULL, NULL,
 264       GST_CLOCK_TIME_NONE);
 265 }
 266
 267 void
 268 gst_uri_downloader_reset (GstUriDownloader * downloader)
 269 {
 270   g_return_if_fail (downloader != NULL);
 271
 272   GST_OBJECT_LOCK (downloader);
 273   downloader->priv->cancelled = FALSE;
 274   GST_OBJECT_UNLOCK (downloader);
 275 }
 276
 277 void
 278 gst_uri_downloader_cancel (GstUriDownloader * downloader)
 279 {
 280   GST_OBJECT_LOCK (downloader);
 281   if (downloader->priv->download != NULL) {
 282     GST_DEBUG_OBJECT (downloader, "Cancelling download");
 283     g_object_unref (downloader->priv->download);
 284     downloader->priv->download = NULL;
 285     downloader->priv->cancelled = TRUE;
 286     GST_OBJECT_UNLOCK (downloader);
 287     GST_DEBUG_OBJECT (downloader, "Signaling chain funtion");
 288     g_mutex_lock (&downloader->priv->lock);
 289     g_cond_signal (&downloader->priv->cond);
 290     g_mutex_unlock (&downloader->priv->lock);
 291   } else {
 292     gboolean cancelled;
 293
 294     cancelled = downloader->priv->cancelled;
 295     downloader->priv->cancelled = TRUE;
 296     GST_OBJECT_UNLOCK (downloader);
 297     if (cancelled)
 298       GST_DEBUG_OBJECT (downloader,
 299           "Trying to cancell a download that was alredy cancelled");
 300   }
 301 }
 302
 303 static gboolean
 304 gst_uri_downloader_set_uri (GstUriDownloader * downloader, const gchar * uri)
 305 {
 306   GstPad *pad;
 307
 308   if (!gst_uri_is_valid (uri))
 309     return FALSE;
 310
 311   GST_DEBUG_OBJECT (downloader, "Creating source element for the URI:%s", uri);
 312   downloader->priv->urisrc =
 313       gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL);
 314   if (!downloader->priv->urisrc)
 315     return FALSE;
 316
 317   /* add a sync handler for the bus messages to detect errors in the download */
 318   gst_element_set_bus (GST_ELEMENT (downloader->priv->urisrc),
 319       downloader->priv->bus);
 320   gst_bus_set_sync_handler (downloader->priv->bus,
 321       gst_uri_downloader_bus_handler, downloader, NULL);
 322
 323   pad = gst_element_get_static_pad (downloader->priv->urisrc, "src");
 324   if (!pad)
 325     return FALSE;
 326   gst_pad_link (pad, downloader->priv->pad);
 327   gst_object_unref (pad);
 328   return TRUE;
 329 }
 330
 331 GstFragment *
 332 gst_uri_downloader_fetch_uri (GstUriDownloader * downloader, const gchar * uri)
 333 {
 334   GstStateChangeReturn ret;
 335   GstFragment *download = NULL;
 336
 337   g_mutex_lock (&downloader->priv->lock);
 338
 339   if (downloader->priv->cancelled) {
 340     goto quit;
 341   }
 342
 343   if (!gst_uri_downloader_set_uri (downloader, uri)) {
 344     goto quit;
 345   }
 346
 347   downloader->priv->download = gst_fragment_new ();
 348
 349   ret = gst_element_set_state (downloader->priv->urisrc, GST_STATE_PLAYING);
 350   if (ret == GST_STATE_CHANGE_FAILURE) {
 351     g_object_unref (downloader->priv->download);
 352     downloader->priv->download = NULL;
 353     goto quit;
 354   }
 355
 356   /* wait until:
 357    *   - the download succeed (EOS in the src pad)
 358    *   - the download failed (Error message on the fetcher bus)
 359    *   - the download was canceled
 360    */
 361   GST_DEBUG_OBJECT (downloader, "Waiting to fetch the URI %s", uri);
 362   g_cond_wait (&downloader->priv->cond, &downloader->priv->lock);
 363
 364   if (downloader->priv->cancelled) {
 365     if (downloader->priv->download) {
 366       g_object_unref (downloader->priv->download);
 367       downloader->priv->download = NULL;
 368     }
 369     goto quit;
 370   }
 371
 372   GST_OBJECT_LOCK (downloader);
 373   download = downloader->priv->download;
 374   downloader->priv->download = NULL;
 375   GST_OBJECT_UNLOCK (downloader);
 376
 377   if (download != NULL)
 378     GST_INFO_OBJECT (downloader, "URI fetched successfully");
 379   else
 380     GST_INFO_OBJECT (downloader, "Error fetching URI");
 381
 382 quit:
 383   {
 384     gst_uri_downloader_stop (downloader);
 385     g_mutex_unlock (&downloader->priv->lock);
 386     return download;
 387   }
 388 }