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., 51 Franklin St, Fifth Floor,
  24  * Boston, MA 02110-1301, USA.
  25  */
  26
  27 /**
  28  * SECTION:element-rtmpsrc
  29  * @title: rtmpsrc
  30  *
  31  * This plugin reads data from a local or remote location specified
  32  * by an URI. This location can be specified using any protocol supported by
  33  * the RTMP library, i.e. rtmp, rtmpt, rtmps, rtmpe, rtmfp, rtmpte and rtmpts.
  34  *
  35  * ## Example launch lines
  36  * |[
  37  * gst-launch-1.0 -v rtmpsrc location=rtmp://somehost/someurl ! fakesink
  38  * ]| Open an RTMP location and pass its content to fakesink.
  39  *
  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_TIMEOUT
  73 #if 0
  74       PROP_SWF_URL,
  75   PROP_PAGE_URL
  76 #endif
  77 };
  78
  79 #define DEFAULT_LOCATION NULL
  80 #define DEFAULT_TIMEOUT 120
  81
  82 static void gst_rtmp_src_uri_handler_init (gpointer g_iface,
  83     gpointer iface_data);
  84
  85 static void gst_rtmp_src_set_property (GObject * object, guint prop_id,
  86     const GValue * value, GParamSpec * pspec);
  87 static void gst_rtmp_src_get_property (GObject * object, guint prop_id,
  88     GValue * value, GParamSpec * pspec);
  89 static void gst_rtmp_src_finalize (GObject * object);
  90
  91 static gboolean gst_rtmp_src_connect (GstRTMPSrc * src);
  92 static gboolean gst_rtmp_src_unlock (GstBaseSrc * src);
  93 static gboolean gst_rtmp_src_stop (GstBaseSrc * src);
  94 static gboolean gst_rtmp_src_start (GstBaseSrc * src);
  95 static gboolean gst_rtmp_src_is_seekable (GstBaseSrc * src);
  96 static gboolean gst_rtmp_src_prepare_seek_segment (GstBaseSrc * src,
  97     GstEvent * event, GstSegment * segment);
  98 static gboolean gst_rtmp_src_do_seek (GstBaseSrc * src, GstSegment * segment);
  99 static GstFlowReturn gst_rtmp_src_create (GstPushSrc * pushsrc,
 100     GstBuffer ** buffer);
 101 static gboolean gst_rtmp_src_query (GstBaseSrc * src, GstQuery * query);
 102
 103 #define gst_rtmp_src_parent_class parent_class
 104 G_DEFINE_TYPE_WITH_CODE (GstRTMPSrc, gst_rtmp_src, GST_TYPE_PUSH_SRC,
 105     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER,
 106         gst_rtmp_src_uri_handler_init));
 107
 108 static void
 109 gst_rtmp_src_class_init (GstRTMPSrcClass * klass)
 110 {
 111   GObjectClass *gobject_class;
 112   GstElementClass *gstelement_class;
 113   GstBaseSrcClass *gstbasesrc_class;
 114   GstPushSrcClass *gstpushsrc_class;
 115
 116   gobject_class = G_OBJECT_CLASS (klass);
 117   gstelement_class = GST_ELEMENT_CLASS (klass);
 118   gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
 119   gstpushsrc_class = GST_PUSH_SRC_CLASS (klass);
 120
 121   gobject_class->finalize = gst_rtmp_src_finalize;
 122   gobject_class->set_property = gst_rtmp_src_set_property;
 123   gobject_class->get_property = gst_rtmp_src_get_property;
 124
 125   /* properties */
 126   g_object_class_install_property (gobject_class, PROP_LOCATION,
 127       g_param_spec_string ("location", "RTMP Location",
 128           "Location of the RTMP url to read",
 129           DEFAULT_LOCATION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 130
 131   g_object_class_install_property (gobject_class, PROP_TIMEOUT,
 132       g_param_spec_int ("timeout", "RTMP Timeout",
 133           "Time without receiving any data from the server to wait before to timeout the session",
 134           0, G_MAXINT,
 135           DEFAULT_TIMEOUT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 136
 137   gst_element_class_add_static_pad_template (gstelement_class, &srctemplate);
 138
 139   gst_element_class_set_static_metadata (gstelement_class,
 140       "RTMP Source",
 141       "Source/File",
 142       "Read RTMP streams",
 143       "Bastien Nocera <hadess@hadess.net>, "
 144       "Sebastian Dröge <sebastian.droege@collabora.co.uk>");
 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->unlock = GST_DEBUG_FUNCPTR (gst_rtmp_src_unlock);
 149   gstbasesrc_class->is_seekable = GST_DEBUG_FUNCPTR (gst_rtmp_src_is_seekable);
 150   gstbasesrc_class->prepare_seek_segment =
 151       GST_DEBUG_FUNCPTR (gst_rtmp_src_prepare_seek_segment);
 152   gstbasesrc_class->do_seek = GST_DEBUG_FUNCPTR (gst_rtmp_src_do_seek);
 153   gstpushsrc_class->create = GST_DEBUG_FUNCPTR (gst_rtmp_src_create);
 154   gstbasesrc_class->query = GST_DEBUG_FUNCPTR (gst_rtmp_src_query);
 155
 156   GST_DEBUG_CATEGORY_INIT (rtmpsrc_debug, "rtmpsrc", 0, "RTMP Source");
 157 }
 158
 159 static void
 160 gst_rtmp_src_init (GstRTMPSrc * rtmpsrc)
 161 {
 162 #ifdef G_OS_WIN32
 163   WSADATA wsa_data;
 164
 165   if (WSAStartup (MAKEWORD (2, 2), &wsa_data) != 0) {
 166     GST_ERROR_OBJECT (rtmpsrc, "WSAStartup failed: 0x%08x", WSAGetLastError ());
 167   }
 168 #endif
 169
 170   rtmpsrc->cur_offset = 0;
 171   rtmpsrc->last_timestamp = 0;
 172   rtmpsrc->timeout = DEFAULT_TIMEOUT;
 173
 174   gst_base_src_set_format (GST_BASE_SRC (rtmpsrc), GST_FORMAT_TIME);
 175 }
 176
 177 static void
 178 gst_rtmp_src_finalize (GObject * object)
 179 {
 180   GstRTMPSrc *rtmpsrc = GST_RTMP_SRC (object);
 181
 182   g_free (rtmpsrc->uri);
 183   rtmpsrc->uri = NULL;
 184
 185 #ifdef G_OS_WIN32
 186   WSACleanup ();
 187 #endif
 188
 189   G_OBJECT_CLASS (parent_class)->finalize (object);
 190 }
 191
 192 /*
 193  * URI interface support.
 194  */
 195
 196 static GstURIType
 197 gst_rtmp_src_uri_get_type (GType type)
 198 {
 199   return GST_URI_SRC;
 200 }
 201
 202 static const gchar *const *
 203 gst_rtmp_src_uri_get_protocols (GType type)
 204 {
 205   static const gchar *protocols[] =
 206       { "rtmp", "rtmpt", "rtmps", "rtmpe", "rtmfp", "rtmpte", "rtmpts", NULL };
 207
 208   return protocols;
 209 }
 210
 211 static gchar *
 212 gst_rtmp_src_uri_get_uri (GstURIHandler * handler)
 213 {
 214   GstRTMPSrc *src = GST_RTMP_SRC (handler);
 215
 216   /* FIXME: make thread-safe */
 217   return g_strdup (src->uri);
 218 }
 219
 220 static gboolean
 221 gst_rtmp_src_uri_set_uri (GstURIHandler * handler, const gchar * uri,
 222     GError ** error)
 223 {
 224   GstRTMPSrc *src = GST_RTMP_SRC (handler);
 225
 226   if (GST_STATE (src) >= GST_STATE_PAUSED) {
 227     g_set_error (error, GST_URI_ERROR, GST_URI_ERROR_BAD_STATE,
 228         "Changing the URI on rtmpsrc when it is running is not supported");
 229     return FALSE;
 230   }
 231
 232   g_free (src->uri);
 233   src->uri = NULL;
 234
 235   if (uri != NULL) {
 236     int protocol;
 237     AVal host;
 238     unsigned int port;
 239     AVal playpath, app;
 240
 241     if (!RTMP_ParseURL (uri, &protocol, &host, &port, &playpath, &app) ||
 242         !host.av_len || !playpath.av_len) {
 243       GST_ERROR_OBJECT (src, "Failed to parse URI %s", uri);
 244       g_set_error (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
 245           "Could not parse RTMP URI");
 246       /* FIXME: we should not be freeing RTMP internals to avoid leaking */
 247       free (playpath.av_val);
 248       return FALSE;
 249     }
 250     free (playpath.av_val);
 251     src->uri = g_strdup (uri);
 252   }
 253
 254   GST_DEBUG_OBJECT (src, "Changed URI to %s", GST_STR_NULL (uri));
 255
 256   return TRUE;
 257 }
 258
 259 static void
 260 gst_rtmp_src_uri_handler_init (gpointer g_iface, gpointer iface_data)
 261 {
 262   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
 263
 264   iface->get_type = gst_rtmp_src_uri_get_type;
 265   iface->get_protocols = gst_rtmp_src_uri_get_protocols;
 266   iface->get_uri = gst_rtmp_src_uri_get_uri;
 267   iface->set_uri = gst_rtmp_src_uri_set_uri;
 268 }
 269
 270 static void
 271 gst_rtmp_src_set_property (GObject * object, guint prop_id,
 272     const GValue * value, GParamSpec * pspec)
 273 {
 274   GstRTMPSrc *src;
 275
 276   src = GST_RTMP_SRC (object);
 277
 278   switch (prop_id) {
 279     case PROP_LOCATION:{
 280       gst_rtmp_src_uri_set_uri (GST_URI_HANDLER (src),
 281           g_value_get_string (value), NULL);
 282       break;
 283     }
 284     case PROP_TIMEOUT:{
 285       src->timeout = g_value_get_int (value);
 286       break;
 287     }
 288     default:
 289       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 290       break;
 291   }
 292 }
 293
 294 static void
 295 gst_rtmp_src_get_property (GObject * object, guint prop_id, GValue * value,
 296     GParamSpec * pspec)
 297 {
 298   GstRTMPSrc *src;
 299
 300   src = GST_RTMP_SRC (object);
 301
 302   switch (prop_id) {
 303     case PROP_LOCATION:
 304       g_value_set_string (value, src->uri);
 305       break;
 306     case PROP_TIMEOUT:
 307       g_value_set_int (value, src->timeout);
 308       break;
 309     default:
 310       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 311       break;
 312   }
 313 }
 314
 315 /*
 316  * Read a new buffer from src->reqoffset, takes care of events
 317  * and seeking and such.
 318  */
 319 static GstFlowReturn
 320 gst_rtmp_src_create (GstPushSrc * pushsrc, GstBuffer ** buffer)
 321 {
 322   GstRTMPSrc *src;
 323   GstBuffer *buf;
 324   GstMapInfo map;
 325   guint8 *data;
 326   guint todo;
 327   gsize bsize;
 328   int size;
 329
 330   src = GST_RTMP_SRC (pushsrc);
 331
 332   g_return_val_if_fail (src->rtmp != NULL, GST_FLOW_ERROR);
 333
 334   if (!RTMP_IsConnected (src->rtmp)) {
 335     GST_DEBUG_OBJECT (src, "reconnecting");
 336     if (!gst_rtmp_src_connect (src))
 337       return GST_FLOW_ERROR;
 338   }
 339
 340   size = GST_BASE_SRC_CAST (pushsrc)->blocksize;
 341
 342   GST_DEBUG ("reading from %" G_GUINT64_FORMAT
 343       ", size %u", src->cur_offset, size);
 344
 345   buf = gst_buffer_new_allocate (NULL, size, NULL);
 346   if (G_UNLIKELY (buf == NULL)) {
 347     GST_ERROR_OBJECT (src, "Failed to allocate %u bytes", size);
 348     return GST_FLOW_ERROR;
 349   }
 350
 351   todo = size;
 352   gst_buffer_map (buf, &map, GST_MAP_WRITE);
 353   data = map.data;
 354   bsize = 0;
 355
 356   while (todo > 0) {
 357     int read = RTMP_Read (src->rtmp, (char *) data, todo);
 358
 359     if (G_UNLIKELY (read == 0 && todo == size))
 360       goto eos;
 361
 362     if (G_UNLIKELY (read == 0))
 363       break;
 364
 365     if (G_UNLIKELY (read < 0))
 366       goto read_failed;
 367
 368     if (read < todo) {
 369       data += read;
 370       todo -= read;
 371       bsize += read;
 372     } else {
 373       bsize += todo;
 374       todo = 0;
 375     }
 376     GST_LOG ("  got size %d", read);
 377   }
 378   gst_buffer_unmap (buf, &map);
 379   gst_buffer_resize (buf, 0, bsize);
 380
 381   if (src->discont) {
 382     GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT);
 383     src->discont = FALSE;
 384   }
 385
 386   GST_BUFFER_TIMESTAMP (buf) = src->last_timestamp;
 387   GST_BUFFER_OFFSET (buf) = src->cur_offset;
 388
 389   src->cur_offset += size;
 390   if (src->last_timestamp == GST_CLOCK_TIME_NONE)
 391     src->last_timestamp = src->rtmp->m_mediaStamp * GST_MSECOND;
 392   else
 393     src->last_timestamp =
 394         MAX (src->last_timestamp, src->rtmp->m_mediaStamp * GST_MSECOND);
 395
 396   GST_LOG_OBJECT (src, "Created buffer of size %u at %" G_GINT64_FORMAT
 397       " with timestamp %" GST_TIME_FORMAT, size, GST_BUFFER_OFFSET (buf),
 398       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
 399
 400
 401   /* we're done, return the buffer */
 402   *buffer = buf;
 403
 404   return GST_FLOW_OK;
 405
 406 read_failed:
 407   {
 408     gst_buffer_unmap (buf, &map);
 409     gst_buffer_unref (buf);
 410     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), ("Failed to read data"));
 411     return GST_FLOW_ERROR;
 412   }
 413 eos:
 414   {
 415     gst_buffer_unmap (buf, &map);
 416     gst_buffer_unref (buf);
 417     if (src->cur_offset == 0) {
 418       GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 419           ("Failed to read any data from stream, check your URL"));
 420       return GST_FLOW_ERROR;
 421     } else {
 422       GST_DEBUG_OBJECT (src, "Reading data gave EOS");
 423       return GST_FLOW_EOS;
 424     }
 425   }
 426 }
 427
 428 static gboolean
 429 gst_rtmp_src_query (GstBaseSrc * basesrc, GstQuery * query)
 430 {
 431   gboolean ret = FALSE;
 432   GstRTMPSrc *src = GST_RTMP_SRC (basesrc);
 433
 434   switch (GST_QUERY_TYPE (query)) {
 435     case GST_QUERY_URI:
 436       gst_query_set_uri (query, src->uri);
 437       ret = TRUE;
 438       break;
 439     case GST_QUERY_POSITION:{
 440       GstFormat format;
 441
 442       gst_query_parse_position (query, &format, NULL);
 443       if (format == GST_FORMAT_TIME) {
 444         gst_query_set_position (query, format, src->last_timestamp);
 445         ret = TRUE;
 446       }
 447       break;
 448     }
 449     case GST_QUERY_DURATION:{
 450       GstFormat format;
 451       gdouble duration;
 452
 453       gst_query_parse_duration (query, &format, NULL);
 454       if (format == GST_FORMAT_TIME && src->rtmp) {
 455         duration = RTMP_GetDuration (src->rtmp);
 456         if (duration != 0.0) {
 457           gst_query_set_duration (query, format, duration * GST_SECOND);
 458           ret = TRUE;
 459         }
 460       }
 461       break;
 462     }
 463     case GST_QUERY_SCHEDULING:{
 464       gst_query_set_scheduling (query,
 465           GST_SCHEDULING_FLAG_SEQUENTIAL |
 466           GST_SCHEDULING_FLAG_BANDWIDTH_LIMITED, 1, -1, 0);
 467       gst_query_add_scheduling_mode (query, GST_PAD_MODE_PUSH);
 468
 469       ret = TRUE;
 470       break;
 471     }
 472     default:
 473       ret = FALSE;
 474       break;
 475   }
 476
 477   if (!ret)
 478     ret = GST_BASE_SRC_CLASS (parent_class)->query (basesrc, query);
 479
 480   return ret;
 481 }
 482
 483 static gboolean
 484 gst_rtmp_src_is_seekable (GstBaseSrc * basesrc)
 485 {
 486   GstRTMPSrc *src;
 487
 488   src = GST_RTMP_SRC (basesrc);
 489
 490   return src->seekable;
 491 }
 492
 493 static gboolean
 494 gst_rtmp_src_prepare_seek_segment (GstBaseSrc * basesrc, GstEvent * event,
 495     GstSegment * segment)
 496 {
 497   GstRTMPSrc *src;
 498   GstSeekType cur_type, stop_type;
 499   gint64 cur, stop;
 500   GstSeekFlags flags;
 501   GstFormat format;
 502   gdouble rate;
 503
 504   src = GST_RTMP_SRC (basesrc);
 505
 506   gst_event_parse_seek (event, &rate, &format, &flags,
 507       &cur_type, &cur, &stop_type, &stop);
 508
 509   if (!src->seekable) {
 510     GST_LOG_OBJECT (src, "Not a seekable stream");
 511     return FALSE;
 512   }
 513
 514   if (!src->rtmp) {
 515     GST_LOG_OBJECT (src, "Not connected yet");
 516     return FALSE;
 517   }
 518
 519   if (format != GST_FORMAT_TIME) {
 520     GST_LOG_OBJECT (src, "Seeking only supported in TIME format");
 521     return FALSE;
 522   }
 523
 524   if (stop_type != GST_SEEK_TYPE_NONE) {
 525     GST_LOG_OBJECT (src, "Setting a stop position is not supported");
 526     return FALSE;
 527   }
 528
 529   gst_segment_init (segment, GST_FORMAT_TIME);
 530   gst_segment_do_seek (segment, rate, format, flags, cur_type, cur, stop_type,
 531       stop, NULL);
 532
 533   return TRUE;
 534 }
 535
 536 static gboolean
 537 gst_rtmp_src_do_seek (GstBaseSrc * basesrc, GstSegment * segment)
 538 {
 539   GstRTMPSrc *src;
 540
 541   src = GST_RTMP_SRC (basesrc);
 542
 543   if (segment->format != GST_FORMAT_TIME) {
 544     GST_LOG_OBJECT (src, "Only time based seeks are supported");
 545     return FALSE;
 546   }
 547
 548   if (!src->rtmp) {
 549     GST_LOG_OBJECT (src, "Not connected yet");
 550     return FALSE;
 551   }
 552
 553   /* Initial seek */
 554   if (src->cur_offset == 0 && segment->start == 0)
 555     goto success;
 556
 557   if (!src->seekable) {
 558     GST_LOG_OBJECT (src, "Not a seekable stream");
 559     return FALSE;
 560   }
 561
 562   /* If we have just disconnected in unlock(), we need to re-connect
 563    * and also let librtmp read some data before sending a seek,
 564    * otherwise it will stall. Calling create() does both. */
 565   if (!RTMP_IsConnected (src->rtmp)) {
 566     GstBuffer *buffer = NULL;
 567     gst_rtmp_src_create (GST_PUSH_SRC (basesrc), &buffer);
 568     gst_buffer_replace (&buffer, NULL);
 569   }
 570
 571   src->last_timestamp = GST_CLOCK_TIME_NONE;
 572   if (!RTMP_SendSeek (src->rtmp, segment->start / GST_MSECOND)) {
 573     GST_ERROR_OBJECT (src, "Seeking failed");
 574     src->seekable = FALSE;
 575     return FALSE;
 576   }
 577
 578 success:
 579   /* This is set here so that the call to create() above doesn't clear it */
 580   src->discont = TRUE;
 581
 582   GST_DEBUG_OBJECT (src, "Seek to %" GST_TIME_FORMAT " successful",
 583       GST_TIME_ARGS (segment->start));
 584
 585   return TRUE;
 586 }
 587
 588 static gboolean
 589 gst_rtmp_src_connect (GstRTMPSrc * src)
 590 {
 591   RTMP_Init (src->rtmp);
 592   src->rtmp->Link.timeout = src->timeout;
 593   if (!RTMP_SetupURL (src->rtmp, src->uri)) {
 594     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 595         ("Failed to setup URL '%s'", src->uri));
 596     return FALSE;
 597   }
 598   src->seekable = !(src->rtmp->Link.lFlags & RTMP_LF_LIVE);
 599   GST_INFO_OBJECT (src, "seekable %d", src->seekable);
 600
 601   /* open if required */
 602   if (!RTMP_IsConnected (src->rtmp)) {
 603     if (!RTMP_Connect (src->rtmp, NULL)) {
 604       GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 605           ("Could not connect to RTMP stream \"%s\" for reading", src->uri));
 606       return FALSE;
 607     }
 608   }
 609
 610   return TRUE;
 611 }
 612
 613 /* open the file, do stuff necessary to go to PAUSED state */
 614 static gboolean
 615 gst_rtmp_src_start (GstBaseSrc * basesrc)
 616 {
 617   GstRTMPSrc *src;
 618
 619   src = GST_RTMP_SRC (basesrc);
 620
 621   if (!src->uri) {
 622     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL), ("No filename given"));
 623     return FALSE;
 624   }
 625
 626   src->cur_offset = 0;
 627   src->last_timestamp = 0;
 628   src->discont = TRUE;
 629
 630   src->rtmp = RTMP_Alloc ();
 631   if (!src->rtmp) {
 632     GST_ERROR_OBJECT (src, "Could not allocate librtmp's RTMP context");
 633     goto error;
 634   }
 635
 636   if (!gst_rtmp_src_connect (src))
 637     goto error;
 638
 639   return TRUE;
 640
 641 error:
 642   if (src->rtmp) {
 643     RTMP_Free (src->rtmp);
 644     src->rtmp = NULL;
 645   }
 646   return FALSE;
 647 }
 648
 649 static gboolean
 650 gst_rtmp_src_unlock (GstBaseSrc * basesrc)
 651 {
 652   GstRTMPSrc *rtmpsrc = GST_RTMP_SRC (basesrc);
 653
 654   GST_DEBUG_OBJECT (rtmpsrc, "unlock");
 655
 656   /* This closes the socket, which means that any pending socket calls
 657    * error out. */
 658   if (rtmpsrc->rtmp) {
 659     RTMP_Close (rtmpsrc->rtmp);
 660   }
 661
 662   return TRUE;
 663 }
 664
 665
 666 static gboolean
 667 gst_rtmp_src_stop (GstBaseSrc * basesrc)
 668 {
 669   GstRTMPSrc *src;
 670
 671   src = GST_RTMP_SRC (basesrc);
 672
 673   if (src->rtmp) {
 674     RTMP_Free (src->rtmp);
 675     src->rtmp = NULL;
 676   }
 677
 678   src->cur_offset = 0;
 679   src->last_timestamp = 0;
 680   src->discont = TRUE;
 681
 682   return TRUE;
 683 }