ext/rtmp/gstrtmpsrc.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2000 Wim Taymans <wtay@chello.be>
   4  *                    2002 Kristian Rietveld <kris@gtk.org>
   5  *                    2002,2003 Colin Walters <walters@gnu.org>
   6  *                    2001,2010 Bastien Nocera <hadess@hadess.net>
   7  *                    2010 Sebastian Dröge <sebastian.droege@collabora.co.uk>
   8  *
   9  * rtmpsrc.c:
  10  *
  11  * This library is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU Library General Public
  13  * License as published by the Free Software Foundation; either
  14  * version 2 of the License, or (at your option) any later version.
  15  *
  16  * This library is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  19  * Library General Public License for more details.
  20  *
  21  * You should have received a copy of the GNU Library General Public
  22  * License along with this library; if not, write to the
  23  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  24  * Boston, MA 02111-1307, USA.
  25  */
  26
  27 /**
  28  * SECTION:element-rtmpsrc
  29  *
  30  * This plugin reads data from a local or remote location specified
  31  * by an URI. This location can be specified using any protocol supported by
  32  * the RTMP library, i.e. rtmp, rtmpt, rtmps, rtmpe, rtmfp, rtmpte and rtmpts.
  33  *
  34  * <refsect2>
  35  * <title>Example launch lines</title>
  36  * |[
  37  * gst-launch -v rtmpsrc location=rtmp://somehost/someurl ! fakesink
  38  * ]| Open an RTMP location and pass its content to fakesink.
  39  * </refsect2>
  40  */
  41
  42 #ifdef HAVE_CONFIG_H
  43 #include "config.h"
  44 #endif
  45
  46 #include <glib/gi18n-lib.h>
  47
  48 #include "gstrtmpsrc.h"
  49
  50 #include <stdio.h>
  51 #include <stdlib.h>
  52 #include <string.h>
  53
  54 #include <gst/gst.h>
  55
  56 #ifdef G_OS_WIN32
  57 #include <winsock2.h>
  58 #endif
  59
  60 GST_DEBUG_CATEGORY_STATIC (rtmpsrc_debug);
  61 #define GST_CAT_DEFAULT rtmpsrc_debug
  62
  63 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  64     GST_PAD_SRC,
  65     GST_PAD_ALWAYS,
  66     GST_STATIC_CAPS_ANY);
  67
  68 enum
  69 {
  70   PROP_0,
  71   PROP_LOCATION,
  72   PROP_SWF_URL,
  73   PROP_PAGE_URL
  74 };
  75
  76 static void gst_rtmp_src_uri_handler_init (gpointer g_iface,
  77     gpointer iface_data);
  78
  79 static void gst_rtmp_src_set_property (GObject * object, guint prop_id,
  80     const GValue * value, GParamSpec * pspec);
  81 static void gst_rtmp_src_get_property (GObject * object, guint prop_id,
  82     GValue * value, GParamSpec * pspec);
  83 static void gst_rtmp_src_finalize (GObject * object);
  84
  85 static gboolean gst_rtmp_src_stop (GstBaseSrc * src);
  86 static gboolean gst_rtmp_src_start (GstBaseSrc * src);
  87 static gboolean gst_rtmp_src_is_seekable (GstBaseSrc * src);
  88 static gboolean gst_rtmp_src_prepare_seek_segment (GstBaseSrc * src,
  89     GstEvent * event, GstSegment * segment);
  90 static gboolean gst_rtmp_src_do_seek (GstBaseSrc * src, GstSegment * segment);
  91 static GstFlowReturn gst_rtmp_src_create (GstPushSrc * pushsrc,
  92     GstBuffer ** buffer);
  93 static gboolean gst_rtmp_src_query (GstBaseSrc * src, GstQuery * query);
  94
  95 static void
  96 _do_init (GType gtype)
  97 {
  98   static const GInterfaceInfo urihandler_info = {
  99     gst_rtmp_src_uri_handler_init,
 100     NULL,
 101     NULL
 102   };
 103
 104   g_type_add_interface_static (gtype, GST_TYPE_URI_HANDLER, &urihandler_info);
 105
 106   GST_DEBUG_CATEGORY_INIT (rtmpsrc_debug, "rtmpsrc", 0, "RTMP Source");
 107 }
 108
 109 GST_BOILERPLATE_FULL (GstRTMPSrc, gst_rtmp_src, GstPushSrc, GST_TYPE_PUSH_SRC,
 110     _do_init);
 111
 112 static void
 113 gst_rtmp_src_base_init (gpointer g_class)
 114 {
 115   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
 116
 117   gst_element_class_add_static_pad_template (element_class, &srctemplate);
 118
 119   gst_element_class_set_details_simple (element_class,
 120       "RTMP Source",
 121       "Source/File",
 122       "Read RTMP streams",
 123       "Bastien Nocera <hadess@hadess.net>, "
 124       "Sebastian Dröge <sebastian.droege@collabora.co.uk>");
 125 }
 126
 127 static void
 128 gst_rtmp_src_class_init (GstRTMPSrcClass * klass)
 129 {
 130   GObjectClass *gobject_class;
 131   GstBaseSrcClass *gstbasesrc_class;
 132   GstPushSrcClass *gstpushsrc_class;
 133
 134   gobject_class = G_OBJECT_CLASS (klass);
 135   gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
 136   gstpushsrc_class = GST_PUSH_SRC_CLASS (klass);
 137
 138   gobject_class->finalize = gst_rtmp_src_finalize;
 139   gobject_class->set_property = gst_rtmp_src_set_property;
 140   gobject_class->get_property = gst_rtmp_src_get_property;
 141
 142   /* properties */
 143   gst_element_class_install_std_props (GST_ELEMENT_CLASS (klass),
 144       "location", PROP_LOCATION, G_PARAM_READWRITE, NULL);
 145
 146   gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_rtmp_src_start);
 147   gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_rtmp_src_stop);
 148   gstbasesrc_class->is_seekable = GST_DEBUG_FUNCPTR (gst_rtmp_src_is_seekable);
 149   gstbasesrc_class->prepare_seek_segment =
 150       GST_DEBUG_FUNCPTR (gst_rtmp_src_prepare_seek_segment);
 151   gstbasesrc_class->do_seek = GST_DEBUG_FUNCPTR (gst_rtmp_src_do_seek);
 152   gstpushsrc_class->create = GST_DEBUG_FUNCPTR (gst_rtmp_src_create);
 153   gstbasesrc_class->query = GST_DEBUG_FUNCPTR (gst_rtmp_src_query);
 154 }
 155
 156 static void
 157 gst_rtmp_src_init (GstRTMPSrc * rtmpsrc, GstRTMPSrcClass * klass)
 158 {
 159 #ifdef G_OS_WIN32
 160   WSADATA wsa_data;
 161
 162   if (WSAStartup (MAKEWORD (2, 2), &wsa_data) != 0) {
 163     GST_ERROR_OBJECT (rtmpsrc, "WSAStartup failed: 0x%08x", WSAGetLastError ());
 164   }
 165 #endif
 166
 167   rtmpsrc->cur_offset = 0;
 168   rtmpsrc->last_timestamp = 0;
 169
 170   gst_base_src_set_format (GST_BASE_SRC (rtmpsrc), GST_FORMAT_TIME);
 171 }
 172
 173 static void
 174 gst_rtmp_src_finalize (GObject * object)
 175 {
 176   GstRTMPSrc *rtmpsrc = GST_RTMP_SRC (object);
 177
 178   g_free (rtmpsrc->uri);
 179   rtmpsrc->uri = NULL;
 180
 181 #ifdef G_OS_WIN32
 182   WSACleanup ();
 183 #endif
 184
 185   G_OBJECT_CLASS (parent_class)->finalize (object);
 186 }
 187
 188 /*
 189  * URI interface support.
 190  */
 191
 192 static GstURIType
 193 gst_rtmp_src_uri_get_type (void)
 194 {
 195   return GST_URI_SRC;
 196 }
 197
 198 static gchar **
 199 gst_rtmp_src_uri_get_protocols (void)
 200 {
 201   static gchar *protocols[] =
 202       { (char *) "rtmp", (char *) "rtmpt", (char *) "rtmps", (char *) "rtmpe",
 203     (char *) "rtmfp", (char *) "rtmpte", (char *) "rtmpts", NULL
 204   };
 205   return protocols;
 206 }
 207
 208 static const gchar *
 209 gst_rtmp_src_uri_get_uri (GstURIHandler * handler)
 210 {
 211   GstRTMPSrc *src = GST_RTMP_SRC (handler);
 212
 213   return src->uri;
 214 }
 215
 216 static gboolean
 217 gst_rtmp_src_uri_set_uri (GstURIHandler * handler, const gchar * uri)
 218 {
 219   GstRTMPSrc *src = GST_RTMP_SRC (handler);
 220
 221   if (GST_STATE (src) >= GST_STATE_PAUSED)
 222     return FALSE;
 223
 224   g_free (src->uri);
 225   src->uri = NULL;
 226
 227   if (uri != NULL) {
 228     int protocol;
 229     AVal host;
 230     unsigned int port;
 231     AVal playpath, app;
 232
 233     if (!RTMP_ParseURL (uri, &protocol, &host, &port, &playpath, &app) ||
 234         !host.av_len || !playpath.av_len) {
 235       GST_ERROR_OBJECT (src, "Failed to parse URI %s", uri);
 236       return FALSE;
 237     }
 238     src->uri = g_strdup (uri);
 239   }
 240
 241   GST_DEBUG_OBJECT (src, "Changed URI to %s", GST_STR_NULL (uri));
 242
 243   return TRUE;
 244 }
 245
 246 static void
 247 gst_rtmp_src_uri_handler_init (gpointer g_iface, gpointer iface_data)
 248 {
 249   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
 250
 251   iface->get_type = gst_rtmp_src_uri_get_type;
 252   iface->get_protocols = gst_rtmp_src_uri_get_protocols;
 253   iface->get_uri = gst_rtmp_src_uri_get_uri;
 254   iface->set_uri = gst_rtmp_src_uri_set_uri;
 255 }
 256
 257 static void
 258 gst_rtmp_src_set_property (GObject * object, guint prop_id,
 259     const GValue * value, GParamSpec * pspec)
 260 {
 261   GstRTMPSrc *src;
 262
 263   src = GST_RTMP_SRC (object);
 264
 265   switch (prop_id) {
 266     case PROP_LOCATION:{
 267       gst_rtmp_src_uri_set_uri (GST_URI_HANDLER (src),
 268           g_value_get_string (value));
 269       break;
 270     }
 271     default:
 272       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 273       break;
 274   }
 275 }
 276
 277 static void
 278 gst_rtmp_src_get_property (GObject * object, guint prop_id, GValue * value,
 279     GParamSpec * pspec)
 280 {
 281   GstRTMPSrc *src;
 282
 283   src = GST_RTMP_SRC (object);
 284
 285   switch (prop_id) {
 286     case PROP_LOCATION:
 287       g_value_set_string (value, src->uri);
 288       break;
 289     default:
 290       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 291       break;
 292   }
 293 }
 294
 295 /*
 296  * Read a new buffer from src->reqoffset, takes care of events
 297  * and seeking and such.
 298  */
 299 static GstFlowReturn
 300 gst_rtmp_src_create (GstPushSrc * pushsrc, GstBuffer ** buffer)
 301 {
 302   GstRTMPSrc *src;
 303   GstBuffer *buf;
 304   guint8 *data;
 305   guint todo;
 306   int read;
 307   int size;
 308
 309   src = GST_RTMP_SRC (pushsrc);
 310
 311   g_return_val_if_fail (src->rtmp != NULL, GST_FLOW_ERROR);
 312
 313   size = GST_BASE_SRC_CAST (pushsrc)->blocksize;
 314
 315   GST_DEBUG ("reading from %" G_GUINT64_FORMAT
 316       ", size %u", src->cur_offset, size);
 317
 318   buf = gst_buffer_try_new_and_alloc (size);
 319   if (G_UNLIKELY (buf == NULL)) {
 320     GST_ERROR_OBJECT (src, "Failed to allocate %u bytes", size);
 321     return GST_FLOW_ERROR;
 322   }
 323
 324   todo = size;
 325   data = GST_BUFFER_DATA (buf);
 326   read = 0;
 327
 328   while (todo > 0) {
 329     read = RTMP_Read (src->rtmp, (char *) data, todo);
 330
 331     if (G_UNLIKELY (read == 0 && todo == size)) {
 332       goto eos;
 333     } else if (G_UNLIKELY (read == 0)) {
 334       GST_BUFFER_SIZE (buf) -= todo;
 335       todo = 0;
 336       break;
 337     }
 338
 339     if (G_UNLIKELY (read < 0))
 340       goto read_failed;
 341
 342     if (read < todo) {
 343       data = &data[read];
 344       todo -= read;
 345     } else {
 346       todo = 0;
 347     }
 348     GST_LOG ("  got size %d", read);
 349   }
 350
 351   if (src->discont) {
 352     GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT);
 353     src->discont = FALSE;
 354   }
 355
 356   GST_BUFFER_TIMESTAMP (buf) = src->last_timestamp;
 357   GST_BUFFER_OFFSET (buf) = src->cur_offset;
 358
 359   src->cur_offset += size;
 360   if (src->last_timestamp == GST_CLOCK_TIME_NONE)
 361     src->last_timestamp = src->rtmp->m_mediaStamp * GST_MSECOND;
 362   else
 363     src->last_timestamp =
 364         MAX (src->last_timestamp, src->rtmp->m_mediaStamp * GST_MSECOND);
 365
 366   GST_LOG_OBJECT (src, "Created buffer of size %u at %" G_GINT64_FORMAT
 367       " with timestamp %" GST_TIME_FORMAT, size, GST_BUFFER_OFFSET (buf),
 368       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
 369
 370
 371   /* we're done, return the buffer */
 372   *buffer = buf;
 373
 374   return GST_FLOW_OK;
 375
 376 read_failed:
 377   {
 378     gst_buffer_unref (buf);
 379     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), ("Failed to read data"));
 380     return GST_FLOW_ERROR;
 381   }
 382 eos:
 383   {
 384     gst_buffer_unref (buf);
 385     GST_DEBUG_OBJECT (src, "Reading data gave EOS");
 386     return GST_FLOW_UNEXPECTED;
 387   }
 388 }
 389
 390 static gboolean
 391 gst_rtmp_src_query (GstBaseSrc * basesrc, GstQuery * query)
 392 {
 393   gboolean ret = FALSE;
 394   GstRTMPSrc *src = GST_RTMP_SRC (basesrc);
 395
 396   switch (GST_QUERY_TYPE (query)) {
 397     case GST_QUERY_URI:
 398       gst_query_set_uri (query, src->uri);
 399       ret = TRUE;
 400       break;
 401     case GST_QUERY_POSITION:{
 402       GstFormat format;
 403
 404       gst_query_parse_position (query, &format, NULL);
 405       if (format == GST_FORMAT_TIME) {
 406         gst_query_set_duration (query, format, src->last_timestamp);
 407         ret = TRUE;
 408       }
 409       break;
 410     }
 411     case GST_QUERY_DURATION:{
 412       GstFormat format;
 413       gdouble duration;
 414
 415       gst_query_parse_duration (query, &format, NULL);
 416       if (format == GST_FORMAT_TIME && src->rtmp) {
 417         duration = RTMP_GetDuration (src->rtmp);
 418         if (duration != 0.0) {
 419           gst_query_set_duration (query, format, duration * GST_SECOND);
 420           ret = TRUE;
 421         }
 422       }
 423       break;
 424     }
 425     default:
 426       ret = FALSE;
 427       break;
 428   }
 429
 430   if (!ret)
 431     ret = GST_BASE_SRC_CLASS (parent_class)->query (basesrc, query);
 432
 433   return ret;
 434 }
 435
 436 static gboolean
 437 gst_rtmp_src_is_seekable (GstBaseSrc * basesrc)
 438 {
 439   GstRTMPSrc *src;
 440
 441   src = GST_RTMP_SRC (basesrc);
 442
 443   return src->seekable;
 444 }
 445
 446 static gboolean
 447 gst_rtmp_src_prepare_seek_segment (GstBaseSrc * basesrc, GstEvent * event,
 448     GstSegment * segment)
 449 {
 450   GstRTMPSrc *src;
 451   GstSeekType cur_type, stop_type;
 452   gint64 cur, stop;
 453   GstSeekFlags flags;
 454   GstFormat format;
 455   gdouble rate;
 456
 457   src = GST_RTMP_SRC (basesrc);
 458
 459   gst_event_parse_seek (event, &rate, &format, &flags,
 460       &cur_type, &cur, &stop_type, &stop);
 461
 462   if (!src->seekable) {
 463     GST_LOG_OBJECT (src, "Not a seekable stream");
 464     return FALSE;
 465   }
 466
 467   if (!src->rtmp) {
 468     GST_LOG_OBJECT (src, "Not connected yet");
 469     return FALSE;
 470   }
 471
 472   if (format != GST_FORMAT_TIME) {
 473     GST_LOG_OBJECT (src, "Seeking only supported in TIME format");
 474     return FALSE;
 475   }
 476
 477   if (stop_type != GST_SEEK_TYPE_NONE) {
 478     GST_LOG_OBJECT (src, "Setting a stop position is not supported");
 479     return FALSE;
 480   }
 481
 482   gst_segment_init (segment, GST_FORMAT_TIME);
 483   gst_segment_set_seek (segment, rate, format, flags, cur_type, cur, stop_type,
 484       stop, NULL);
 485
 486   return TRUE;
 487 }
 488
 489 static gboolean
 490 gst_rtmp_src_do_seek (GstBaseSrc * basesrc, GstSegment * segment)
 491 {
 492   GstRTMPSrc *src;
 493
 494   src = GST_RTMP_SRC (basesrc);
 495
 496   if (segment->format != GST_FORMAT_TIME) {
 497     GST_LOG_OBJECT (src, "Only time based seeks are supported");
 498     return FALSE;
 499   }
 500
 501   if (!src->seekable) {
 502     GST_LOG_OBJECT (src, "Not a seekable stream");
 503     return FALSE;
 504   }
 505
 506   if (!src->rtmp) {
 507     GST_LOG_OBJECT (src, "Not connected yet");
 508     return FALSE;
 509   }
 510
 511   src->discont = TRUE;
 512
 513   /* Initial seek */
 514   if (src->cur_offset == 0 && segment->start == 0)
 515     return TRUE;
 516
 517   src->last_timestamp = GST_CLOCK_TIME_NONE;
 518   if (!RTMP_SendSeek (src->rtmp, segment->start / GST_MSECOND)) {
 519     GST_ERROR_OBJECT (src, "Seeking failed");
 520     src->seekable = FALSE;
 521     return FALSE;
 522   }
 523
 524   GST_DEBUG_OBJECT (src, "Seek to %" GST_TIME_FORMAT " successfull",
 525       GST_TIME_ARGS (segment->start));
 526
 527   return TRUE;
 528 }
 529
 530 #define STR2AVAL(av,str) G_STMT_START { \
 531   av.av_val = str; \
 532   av.av_len = strlen(av.av_val); \
 533 } G_STMT_END;
 534
 535 /* open the file, do stuff necessary to go to PAUSED state */
 536 static gboolean
 537 gst_rtmp_src_start (GstBaseSrc * basesrc)
 538 {
 539   GstRTMPSrc *src;
 540   gchar *uri_copy;
 541
 542   src = GST_RTMP_SRC (basesrc);
 543
 544   if (!src->uri) {
 545     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL), ("No filename given"));
 546     return FALSE;
 547   }
 548
 549   src->cur_offset = 0;
 550   src->last_timestamp = 0;
 551   src->seekable = TRUE;
 552   src->discont = TRUE;
 553
 554   uri_copy = g_strdup (src->uri);
 555   src->rtmp = RTMP_Alloc ();
 556   RTMP_Init (src->rtmp);
 557   if (!RTMP_SetupURL (src->rtmp, uri_copy)) {
 558     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 559         ("Failed to setup URL '%s'", src->uri));
 560     g_free (uri_copy);
 561     RTMP_Free (src->rtmp);
 562     src->rtmp = NULL;
 563     return FALSE;
 564   }
 565
 566   /* open if required */
 567   if (!RTMP_IsConnected (src->rtmp)) {
 568     if (!RTMP_Connect (src->rtmp, NULL)) {
 569       GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 570           ("Could not connect to RTMP stream \"%s\" for reading", src->uri));
 571       RTMP_Free (src->rtmp);
 572       src->rtmp = NULL;
 573       return FALSE;
 574     }
 575   }
 576
 577   return TRUE;
 578 }
 579
 580 #undef STR2AVAL
 581
 582 static gboolean
 583 gst_rtmp_src_stop (GstBaseSrc * basesrc)
 584 {
 585   GstRTMPSrc *src;
 586
 587   src = GST_RTMP_SRC (basesrc);
 588
 589   if (src->rtmp) {
 590     RTMP_Close (src->rtmp);
 591     RTMP_Free (src->rtmp);
 592     src->rtmp = NULL;
 593   }
 594
 595   src->cur_offset = 0;
 596   src->last_timestamp = 0;
 597   src->discont = TRUE;
 598
 599   return TRUE;
 600 }