mirror of
https://gitlab.freedesktop.org/gstreamer/gstreamer.git
synced 2024-12-04 23:46:43 +00:00
501e53b033
There is currently no way for applications to know if the stream has been properly terminated by the server or if the network connection was disconnected as EOS is sent in both cases. Adding a property so connection errors can be reported as errors allowing applications to distinguish between both scenarios. Fix #2828 Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/5115>
1068 lines
30 KiB
C
1068 lines
30 KiB
C
/* GStreamer
|
|
* Copyright (C) 2014 David Schleef <ds@schleef.org>
|
|
* Copyright (C) 2017 Make.TV, Inc. <info@make.tv>
|
|
* Contact: Jan Alexander Steffens (heftig) <jsteffens@make.tv>
|
|
*
|
|
* This library is free software; you can redistribute it and/or
|
|
* modify it under the terms of the GNU Library General Public
|
|
* License as published by the Free Software Foundation; either
|
|
* version 2 of the License, or (at your option) any later version.
|
|
*
|
|
* This library is distributed in the hope that it will be useful,
|
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
|
* Library General Public License for more details.
|
|
*
|
|
* You should have received a copy of the GNU Library General Public
|
|
* License along with this library; if not, write to the
|
|
* Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
|
|
* Boston, MA 02110-1335, USA.
|
|
*/
|
|
/**
|
|
* SECTION:element-rtmp2src
|
|
*
|
|
* The rtmp2src element receives input streams from an RTMP server.
|
|
*
|
|
* <refsect2>
|
|
* <title>Example launch line</title>
|
|
* |[
|
|
* gst-launch -v rtmp2src ! decodebin ! fakesink
|
|
* ]|
|
|
* FIXME Describe what the pipeline does.
|
|
* </refsect2>
|
|
*/
|
|
|
|
#ifdef HAVE_CONFIG_H
|
|
#include "config.h"
|
|
#endif
|
|
|
|
#include "gstrtmp2elements.h"
|
|
#include "gstrtmp2src.h"
|
|
|
|
#include "gstrtmp2locationhandler.h"
|
|
#include "rtmp/rtmpclient.h"
|
|
#include "rtmp/rtmpmessage.h"
|
|
|
|
#include <gst/base/gstpushsrc.h>
|
|
#include <string.h>
|
|
|
|
GST_DEBUG_CATEGORY_STATIC (gst_rtmp2_src_debug_category);
|
|
#define GST_CAT_DEFAULT gst_rtmp2_src_debug_category
|
|
|
|
/* prototypes */
|
|
#define GST_RTMP2_SRC(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_RTMP2_SRC,GstRtmp2Src))
|
|
#define GST_IS_RTMP2_SRC(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_RTMP2_SRC))
|
|
|
|
typedef struct
|
|
{
|
|
GstPushSrc parent_instance;
|
|
|
|
/* properties */
|
|
GstRtmpLocation location;
|
|
gboolean async_connect;
|
|
GstStructure *stats;
|
|
guint idle_timeout;
|
|
gboolean no_eof_is_error;
|
|
|
|
/* If both self->lock and OBJECT_LOCK are needed,
|
|
* self->lock must be taken first */
|
|
GMutex lock;
|
|
GCond cond;
|
|
|
|
gboolean running, flushing;
|
|
gboolean timeout;
|
|
gboolean started;
|
|
/* TRUE if there was an error with the connection to the RTMP server */
|
|
gboolean connection_error;
|
|
|
|
GstTask *task;
|
|
GRecMutex task_lock;
|
|
|
|
GMainLoop *loop;
|
|
GMainContext *context;
|
|
|
|
GCancellable *cancellable;
|
|
GstRtmpConnection *connection;
|
|
guint32 stream_id;
|
|
|
|
GstBuffer *message;
|
|
gboolean sent_header;
|
|
GstClockTime last_ts;
|
|
} GstRtmp2Src;
|
|
|
|
typedef struct
|
|
{
|
|
GstPushSrcClass parent_class;
|
|
} GstRtmp2SrcClass;
|
|
|
|
/* GObject virtual functions */
|
|
static void gst_rtmp2_src_set_property (GObject * object,
|
|
guint property_id, const GValue * value, GParamSpec * pspec);
|
|
static void gst_rtmp2_src_get_property (GObject * object,
|
|
guint property_id, GValue * value, GParamSpec * pspec);
|
|
static void gst_rtmp2_src_finalize (GObject * object);
|
|
static void gst_rtmp2_src_uri_handler_init (GstURIHandlerInterface * iface);
|
|
|
|
/* GstBaseSrc virtual functions */
|
|
static gboolean gst_rtmp2_src_start (GstBaseSrc * src);
|
|
static gboolean gst_rtmp2_src_stop (GstBaseSrc * src);
|
|
static gboolean gst_rtmp2_src_unlock (GstBaseSrc * src);
|
|
static gboolean gst_rtmp2_src_unlock_stop (GstBaseSrc * src);
|
|
static GstFlowReturn gst_rtmp2_src_create (GstBaseSrc * src, guint64 offset,
|
|
guint size, GstBuffer ** outbuf);
|
|
static gboolean gst_rtmp2_src_query (GstBaseSrc * src, GstQuery * query);
|
|
|
|
/* Internal API */
|
|
static void gst_rtmp2_src_task_func (gpointer user_data);
|
|
static void client_connect_done (GObject * source, GAsyncResult * result,
|
|
gpointer user_data);
|
|
static void start_play_done (GObject * object, GAsyncResult * result,
|
|
gpointer user_data);
|
|
static void connect_task_done (GObject * object, GAsyncResult * result,
|
|
gpointer user_data);
|
|
|
|
static GstStructure *gst_rtmp2_src_get_stats (GstRtmp2Src * self);
|
|
|
|
enum
|
|
{
|
|
PROP_0,
|
|
PROP_LOCATION,
|
|
PROP_SCHEME,
|
|
PROP_HOST,
|
|
PROP_PORT,
|
|
PROP_APPLICATION,
|
|
PROP_STREAM,
|
|
PROP_SECURE_TOKEN,
|
|
PROP_USERNAME,
|
|
PROP_PASSWORD,
|
|
PROP_AUTHMOD,
|
|
PROP_TIMEOUT,
|
|
PROP_TLS_VALIDATION_FLAGS,
|
|
PROP_FLASH_VERSION,
|
|
PROP_ASYNC_CONNECT,
|
|
PROP_STATS,
|
|
PROP_IDLE_TIMEOUT,
|
|
PROP_NO_EOF_IS_ERROR,
|
|
};
|
|
|
|
#define DEFAULT_IDLE_TIMEOUT 0
|
|
|
|
/* pad templates */
|
|
|
|
static GstStaticPadTemplate gst_rtmp2_src_src_template =
|
|
GST_STATIC_PAD_TEMPLATE ("src",
|
|
GST_PAD_SRC,
|
|
GST_PAD_ALWAYS,
|
|
GST_STATIC_CAPS ("video/x-flv")
|
|
);
|
|
|
|
/* class initialization */
|
|
|
|
G_DEFINE_TYPE_WITH_CODE (GstRtmp2Src, gst_rtmp2_src, GST_TYPE_PUSH_SRC,
|
|
G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER,
|
|
gst_rtmp2_src_uri_handler_init);
|
|
G_IMPLEMENT_INTERFACE (GST_TYPE_RTMP_LOCATION_HANDLER, NULL));
|
|
GST_ELEMENT_REGISTER_DEFINE_WITH_CODE (rtmp2src, "rtmp2src",
|
|
GST_RANK_PRIMARY + 1, GST_TYPE_RTMP2_SRC, rtmp2_element_init (plugin));
|
|
|
|
static void
|
|
gst_rtmp2_src_class_init (GstRtmp2SrcClass * klass)
|
|
{
|
|
GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
|
|
GstBaseSrcClass *base_src_class = GST_BASE_SRC_CLASS (klass);
|
|
|
|
gst_element_class_add_static_pad_template (GST_ELEMENT_CLASS (klass),
|
|
&gst_rtmp2_src_src_template);
|
|
|
|
gst_element_class_set_static_metadata (GST_ELEMENT_CLASS (klass),
|
|
"RTMP source element", "Source", "Source element for RTMP streams",
|
|
"Make.TV, Inc. <info@make.tv>");
|
|
|
|
gobject_class->set_property = gst_rtmp2_src_set_property;
|
|
gobject_class->get_property = gst_rtmp2_src_get_property;
|
|
gobject_class->finalize = gst_rtmp2_src_finalize;
|
|
base_src_class->start = GST_DEBUG_FUNCPTR (gst_rtmp2_src_start);
|
|
base_src_class->stop = GST_DEBUG_FUNCPTR (gst_rtmp2_src_stop);
|
|
base_src_class->unlock = GST_DEBUG_FUNCPTR (gst_rtmp2_src_unlock);
|
|
base_src_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_rtmp2_src_unlock_stop);
|
|
base_src_class->create = GST_DEBUG_FUNCPTR (gst_rtmp2_src_create);
|
|
base_src_class->query = GST_DEBUG_FUNCPTR (gst_rtmp2_src_query);
|
|
|
|
g_object_class_override_property (gobject_class, PROP_LOCATION, "location");
|
|
g_object_class_override_property (gobject_class, PROP_SCHEME, "scheme");
|
|
g_object_class_override_property (gobject_class, PROP_HOST, "host");
|
|
g_object_class_override_property (gobject_class, PROP_PORT, "port");
|
|
g_object_class_override_property (gobject_class, PROP_APPLICATION,
|
|
"application");
|
|
g_object_class_override_property (gobject_class, PROP_STREAM, "stream");
|
|
g_object_class_override_property (gobject_class, PROP_SECURE_TOKEN,
|
|
"secure-token");
|
|
g_object_class_override_property (gobject_class, PROP_USERNAME, "username");
|
|
g_object_class_override_property (gobject_class, PROP_PASSWORD, "password");
|
|
g_object_class_override_property (gobject_class, PROP_AUTHMOD, "authmod");
|
|
g_object_class_override_property (gobject_class, PROP_TIMEOUT, "timeout");
|
|
g_object_class_override_property (gobject_class, PROP_TLS_VALIDATION_FLAGS,
|
|
"tls-validation-flags");
|
|
g_object_class_override_property (gobject_class, PROP_FLASH_VERSION,
|
|
"flash-version");
|
|
|
|
g_object_class_install_property (gobject_class, PROP_ASYNC_CONNECT,
|
|
g_param_spec_boolean ("async-connect", "Async connect",
|
|
"Connect on READY, otherwise on first push", TRUE,
|
|
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
|
|
|
|
g_object_class_install_property (gobject_class, PROP_STATS,
|
|
g_param_spec_boxed ("stats", "Stats", "Retrieve a statistics structure",
|
|
GST_TYPE_STRUCTURE, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
|
|
|
|
g_object_class_install_property (gobject_class, PROP_IDLE_TIMEOUT,
|
|
g_param_spec_uint ("idle-timeout", "Idle timeout",
|
|
"The maximum allowed time in seconds for valid packets not to arrive "
|
|
"from the peer (0 = no timeout)",
|
|
0, G_MAXUINT, DEFAULT_IDLE_TIMEOUT,
|
|
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
|
|
|
|
/**
|
|
* GstRtmp2Src:no-eof-is-error:
|
|
*
|
|
* If set, an error is raised if the connection is closed without receiving an EOF RTMP message first.
|
|
" If not set, those are reported using EOS.
|
|
*
|
|
* Since: 1.24
|
|
*/
|
|
g_object_class_install_property (gobject_class, PROP_NO_EOF_IS_ERROR,
|
|
g_param_spec_boolean ("no-eof-is-error",
|
|
"No EOF is error",
|
|
"If set, an error is raised if the connection is closed without receiving an EOF RTMP message first. "
|
|
"If not set, those are reported using EOS", FALSE,
|
|
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
|
|
|
|
GST_DEBUG_CATEGORY_INIT (gst_rtmp2_src_debug_category, "rtmp2src", 0,
|
|
"debug category for rtmp2src element");
|
|
}
|
|
|
|
static void
|
|
gst_rtmp2_src_init (GstRtmp2Src * self)
|
|
{
|
|
self->async_connect = TRUE;
|
|
self->idle_timeout = DEFAULT_IDLE_TIMEOUT;
|
|
|
|
g_mutex_init (&self->lock);
|
|
g_cond_init (&self->cond);
|
|
|
|
self->task = gst_task_new (gst_rtmp2_src_task_func, self, NULL);
|
|
g_rec_mutex_init (&self->task_lock);
|
|
gst_task_set_lock (self->task, &self->task_lock);
|
|
}
|
|
|
|
static void
|
|
gst_rtmp2_src_uri_handler_init (GstURIHandlerInterface * iface)
|
|
{
|
|
gst_rtmp_location_handler_implement_uri_handler (iface, GST_URI_SRC);
|
|
}
|
|
|
|
static void
|
|
gst_rtmp2_src_set_property (GObject * object, guint property_id,
|
|
const GValue * value, GParamSpec * pspec)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (object);
|
|
|
|
switch (property_id) {
|
|
case PROP_LOCATION:
|
|
gst_rtmp_location_handler_set_uri (GST_RTMP_LOCATION_HANDLER (self),
|
|
g_value_get_string (value));
|
|
break;
|
|
case PROP_SCHEME:
|
|
GST_OBJECT_LOCK (self);
|
|
self->location.scheme = g_value_get_enum (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_HOST:
|
|
GST_OBJECT_LOCK (self);
|
|
g_free (self->location.host);
|
|
self->location.host = g_value_dup_string (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_PORT:
|
|
GST_OBJECT_LOCK (self);
|
|
self->location.port = g_value_get_int (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_APPLICATION:
|
|
GST_OBJECT_LOCK (self);
|
|
g_free (self->location.application);
|
|
self->location.application = g_value_dup_string (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_STREAM:
|
|
GST_OBJECT_LOCK (self);
|
|
g_free (self->location.stream);
|
|
self->location.stream = g_value_dup_string (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_SECURE_TOKEN:
|
|
GST_OBJECT_LOCK (self);
|
|
g_free (self->location.secure_token);
|
|
self->location.secure_token = g_value_dup_string (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_USERNAME:
|
|
GST_OBJECT_LOCK (self);
|
|
g_free (self->location.username);
|
|
self->location.username = g_value_dup_string (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_PASSWORD:
|
|
GST_OBJECT_LOCK (self);
|
|
g_free (self->location.password);
|
|
self->location.password = g_value_dup_string (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_AUTHMOD:
|
|
GST_OBJECT_LOCK (self);
|
|
self->location.authmod = g_value_get_enum (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_TIMEOUT:
|
|
GST_OBJECT_LOCK (self);
|
|
self->location.timeout = g_value_get_uint (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_TLS_VALIDATION_FLAGS:
|
|
GST_OBJECT_LOCK (self);
|
|
self->location.tls_flags = g_value_get_flags (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_FLASH_VERSION:
|
|
GST_OBJECT_LOCK (self);
|
|
g_free (self->location.flash_ver);
|
|
self->location.flash_ver = g_value_dup_string (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_ASYNC_CONNECT:
|
|
GST_OBJECT_LOCK (self);
|
|
self->async_connect = g_value_get_boolean (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_IDLE_TIMEOUT:
|
|
GST_OBJECT_LOCK (self);
|
|
self->idle_timeout = g_value_get_uint (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_NO_EOF_IS_ERROR:
|
|
GST_OBJECT_LOCK (self);
|
|
self->no_eof_is_error = g_value_get_boolean (value);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
default:
|
|
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
|
|
break;
|
|
}
|
|
}
|
|
|
|
static void
|
|
gst_rtmp2_src_get_property (GObject * object, guint property_id,
|
|
GValue * value, GParamSpec * pspec)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (object);
|
|
|
|
switch (property_id) {
|
|
case PROP_LOCATION:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_take_string (value, gst_rtmp_location_get_string (&self->location,
|
|
TRUE));
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_SCHEME:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_enum (value, self->location.scheme);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_HOST:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_string (value, self->location.host);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_PORT:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_int (value, self->location.port);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_APPLICATION:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_string (value, self->location.application);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_STREAM:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_string (value, self->location.stream);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_SECURE_TOKEN:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_string (value, self->location.secure_token);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_USERNAME:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_string (value, self->location.username);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_PASSWORD:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_string (value, self->location.password);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_AUTHMOD:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_enum (value, self->location.authmod);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_TIMEOUT:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_uint (value, self->location.timeout);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_TLS_VALIDATION_FLAGS:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_flags (value, self->location.tls_flags);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_FLASH_VERSION:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_string (value, self->location.flash_ver);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_ASYNC_CONNECT:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_boolean (value, self->async_connect);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_STATS:
|
|
g_value_take_boxed (value, gst_rtmp2_src_get_stats (self));
|
|
break;
|
|
case PROP_IDLE_TIMEOUT:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_uint (value, self->idle_timeout);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
case PROP_NO_EOF_IS_ERROR:
|
|
GST_OBJECT_LOCK (self);
|
|
g_value_set_boolean (value, self->no_eof_is_error);
|
|
GST_OBJECT_UNLOCK (self);
|
|
break;
|
|
default:
|
|
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
|
|
break;
|
|
}
|
|
}
|
|
|
|
static void
|
|
gst_rtmp2_src_finalize (GObject * object)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (object);
|
|
|
|
gst_buffer_replace (&self->message, NULL);
|
|
|
|
g_clear_object (&self->cancellable);
|
|
g_clear_object (&self->connection);
|
|
|
|
g_clear_object (&self->task);
|
|
g_rec_mutex_clear (&self->task_lock);
|
|
|
|
g_mutex_clear (&self->lock);
|
|
g_cond_clear (&self->cond);
|
|
|
|
g_clear_pointer (&self->stats, gst_structure_free);
|
|
gst_rtmp_location_clear (&self->location);
|
|
|
|
G_OBJECT_CLASS (gst_rtmp2_src_parent_class)->finalize (object);
|
|
}
|
|
|
|
static gboolean
|
|
gst_rtmp2_src_start (GstBaseSrc * src)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (src);
|
|
gboolean async;
|
|
|
|
GST_OBJECT_LOCK (self);
|
|
async = self->async_connect;
|
|
GST_OBJECT_UNLOCK (self);
|
|
|
|
GST_INFO_OBJECT (self, "Starting (%s)", async ? "async" : "delayed");
|
|
|
|
g_clear_object (&self->cancellable);
|
|
|
|
self->running = TRUE;
|
|
self->cancellable = g_cancellable_new ();
|
|
self->stream_id = 0;
|
|
self->sent_header = FALSE;
|
|
self->last_ts = GST_CLOCK_TIME_NONE;
|
|
self->timeout = FALSE;
|
|
self->started = FALSE;
|
|
self->connection_error = FALSE;
|
|
|
|
if (async) {
|
|
gst_task_start (self->task);
|
|
}
|
|
|
|
return TRUE;
|
|
}
|
|
|
|
static gboolean
|
|
quit_invoker (gpointer user_data)
|
|
{
|
|
g_main_loop_quit (user_data);
|
|
return G_SOURCE_REMOVE;
|
|
}
|
|
|
|
static void
|
|
stop_task (GstRtmp2Src * self)
|
|
{
|
|
gst_task_stop (self->task);
|
|
self->running = FALSE;
|
|
|
|
if (self->cancellable) {
|
|
GST_DEBUG_OBJECT (self, "Cancelling");
|
|
g_cancellable_cancel (self->cancellable);
|
|
}
|
|
|
|
if (self->loop) {
|
|
GST_DEBUG_OBJECT (self, "Stopping loop");
|
|
g_main_context_invoke_full (self->context, G_PRIORITY_DEFAULT_IDLE,
|
|
quit_invoker, g_main_loop_ref (self->loop),
|
|
(GDestroyNotify) g_main_loop_unref);
|
|
}
|
|
|
|
g_cond_broadcast (&self->cond);
|
|
}
|
|
|
|
static gboolean
|
|
gst_rtmp2_src_stop (GstBaseSrc * src)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (src);
|
|
|
|
GST_DEBUG_OBJECT (self, "stop");
|
|
|
|
g_mutex_lock (&self->lock);
|
|
stop_task (self);
|
|
g_mutex_unlock (&self->lock);
|
|
|
|
gst_task_join (self->task);
|
|
|
|
return TRUE;
|
|
}
|
|
|
|
static gboolean
|
|
gst_rtmp2_src_unlock (GstBaseSrc * src)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (src);
|
|
|
|
GST_DEBUG_OBJECT (self, "unlock");
|
|
|
|
g_mutex_lock (&self->lock);
|
|
self->flushing = TRUE;
|
|
g_cond_broadcast (&self->cond);
|
|
g_mutex_unlock (&self->lock);
|
|
|
|
return TRUE;
|
|
}
|
|
|
|
static gboolean
|
|
gst_rtmp2_src_unlock_stop (GstBaseSrc * src)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (src);
|
|
|
|
GST_DEBUG_OBJECT (self, "unlock_stop");
|
|
|
|
g_mutex_lock (&self->lock);
|
|
self->flushing = FALSE;
|
|
g_mutex_unlock (&self->lock);
|
|
|
|
return TRUE;
|
|
}
|
|
|
|
static gboolean
|
|
on_timeout (GstRtmp2Src * self)
|
|
{
|
|
g_mutex_lock (&self->lock);
|
|
self->timeout = TRUE;
|
|
g_cond_broadcast (&self->cond);
|
|
g_mutex_unlock (&self->lock);
|
|
|
|
return G_SOURCE_REMOVE;
|
|
}
|
|
|
|
static GstFlowReturn
|
|
gst_rtmp2_src_create (GstBaseSrc * src, guint64 offset, guint size,
|
|
GstBuffer ** outbuf)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (src);
|
|
GstBuffer *message, *buffer;
|
|
GstRtmpMeta *meta;
|
|
guint32 timestamp = 0;
|
|
GSource *timeout = NULL;
|
|
GstFlowReturn ret = GST_FLOW_OK;
|
|
|
|
static const guint8 flv_header_data[] = {
|
|
0x46, 0x4c, 0x56, 0x01, 0x01, 0x00, 0x00, 0x00,
|
|
0x09, 0x00, 0x00, 0x00, 0x00,
|
|
};
|
|
|
|
GST_LOG_OBJECT (self, "create");
|
|
|
|
g_mutex_lock (&self->lock);
|
|
|
|
if (self->running) {
|
|
gst_task_start (self->task);
|
|
}
|
|
|
|
/* wait until GMainLoop begins running so that we can attach
|
|
* timeout source safely.
|
|
* If the task stopped meanwhile, "running" will be FALSE
|
|
* than stop_task() will wake up us as well
|
|
*/
|
|
while ((!self->started && self->running) && (!self->loop
|
|
|| !g_main_loop_is_running (self->loop)))
|
|
g_cond_wait (&self->cond, &self->lock);
|
|
|
|
GST_OBJECT_LOCK (self);
|
|
if (self->idle_timeout && self->context) {
|
|
timeout = g_timeout_source_new_seconds (self->idle_timeout);
|
|
|
|
g_source_set_callback (timeout, (GSourceFunc) on_timeout, self, NULL);
|
|
g_source_attach (timeout, self->context);
|
|
}
|
|
GST_OBJECT_UNLOCK (self);
|
|
|
|
while (!self->message) {
|
|
if (!self->running) {
|
|
if (self->no_eof_is_error && self->connection_error) {
|
|
GST_DEBUG_OBJECT (self,
|
|
"stopped because of connection error, return ERROR");
|
|
ret = GST_FLOW_ERROR;
|
|
} else {
|
|
GST_DEBUG_OBJECT (self, "stopped, return EOS");
|
|
ret = GST_FLOW_EOS;
|
|
}
|
|
|
|
goto out;
|
|
}
|
|
if (self->flushing) {
|
|
ret = GST_FLOW_FLUSHING;
|
|
goto out;
|
|
}
|
|
if (self->timeout) {
|
|
GST_DEBUG_OBJECT (self, "Idle timeout, return EOS");
|
|
ret = GST_FLOW_EOS;
|
|
goto out;
|
|
}
|
|
g_cond_wait (&self->cond, &self->lock);
|
|
}
|
|
|
|
if (timeout) {
|
|
g_source_destroy (timeout);
|
|
g_source_unref (timeout);
|
|
}
|
|
|
|
message = self->message;
|
|
self->message = NULL;
|
|
g_cond_signal (&self->cond);
|
|
g_mutex_unlock (&self->lock);
|
|
|
|
meta = gst_buffer_get_rtmp_meta (message);
|
|
if (!meta) {
|
|
GST_ELEMENT_ERROR (self, CORE, FAILED,
|
|
("Internal error: No RTMP meta on buffer"),
|
|
("No RTMP meta on %" GST_PTR_FORMAT, message));
|
|
gst_buffer_unref (message);
|
|
return GST_FLOW_ERROR;
|
|
}
|
|
|
|
if (GST_BUFFER_DTS_IS_VALID (message)) {
|
|
GstClockTime last_ts = self->last_ts, ts = GST_BUFFER_DTS (message);
|
|
|
|
if (GST_CLOCK_TIME_IS_VALID (last_ts) && last_ts > ts) {
|
|
GST_LOG_OBJECT (self, "Timestamp regression: %" GST_TIME_FORMAT
|
|
" > %" GST_TIME_FORMAT, GST_TIME_ARGS (last_ts), GST_TIME_ARGS (ts));
|
|
}
|
|
|
|
self->last_ts = ts;
|
|
timestamp = ts / GST_MSECOND;
|
|
}
|
|
|
|
buffer = gst_buffer_copy_region (message, GST_BUFFER_COPY_MEMORY, 0, -1);
|
|
|
|
{
|
|
guint8 *tag_header = g_malloc (11);
|
|
GstMemory *memory =
|
|
gst_memory_new_wrapped (0, tag_header, 11, 0, 11, tag_header, g_free);
|
|
GST_WRITE_UINT8 (tag_header, meta->type);
|
|
GST_WRITE_UINT24_BE (tag_header + 1, meta->size);
|
|
GST_WRITE_UINT24_BE (tag_header + 4, timestamp);
|
|
GST_WRITE_UINT8 (tag_header + 7, timestamp >> 24);
|
|
GST_WRITE_UINT24_BE (tag_header + 8, 0);
|
|
gst_buffer_prepend_memory (buffer, memory);
|
|
}
|
|
|
|
{
|
|
guint8 *tag_footer = g_malloc (4);
|
|
GstMemory *memory =
|
|
gst_memory_new_wrapped (0, tag_footer, 4, 0, 4, tag_footer, g_free);
|
|
GST_WRITE_UINT32_BE (tag_footer, meta->size + 11);
|
|
gst_buffer_append_memory (buffer, memory);
|
|
}
|
|
|
|
if (!self->sent_header) {
|
|
GstMemory *memory = gst_memory_new_wrapped (GST_MEMORY_FLAG_READONLY,
|
|
(guint8 *) flv_header_data, sizeof flv_header_data, 0,
|
|
sizeof flv_header_data, NULL, NULL);
|
|
gst_buffer_prepend_memory (buffer, memory);
|
|
self->sent_header = TRUE;
|
|
}
|
|
|
|
GST_BUFFER_DTS (buffer) = self->last_ts;
|
|
|
|
*outbuf = buffer;
|
|
|
|
gst_buffer_unref (message);
|
|
return GST_FLOW_OK;
|
|
|
|
out:
|
|
if (timeout) {
|
|
g_source_destroy (timeout);
|
|
g_source_unref (timeout);
|
|
}
|
|
/* Keep the unlock after the destruction of the timeout source to workaround
|
|
* https://gitlab.gnome.org/GNOME/glib/-/issues/803
|
|
*/
|
|
g_mutex_unlock (&self->lock);
|
|
|
|
return ret;
|
|
}
|
|
|
|
static gboolean
|
|
gst_rtmp2_src_query (GstBaseSrc * basesrc, GstQuery * query)
|
|
{
|
|
gboolean ret = FALSE;
|
|
|
|
switch (GST_QUERY_TYPE (query)) {
|
|
case GST_QUERY_SCHEDULING:{
|
|
gst_query_set_scheduling (query,
|
|
GST_SCHEDULING_FLAG_SEQUENTIAL |
|
|
GST_SCHEDULING_FLAG_BANDWIDTH_LIMITED, 1, -1, 0);
|
|
gst_query_add_scheduling_mode (query, GST_PAD_MODE_PUSH);
|
|
|
|
ret = TRUE;
|
|
break;
|
|
}
|
|
default:
|
|
ret = FALSE;
|
|
break;
|
|
}
|
|
|
|
if (!ret)
|
|
ret =
|
|
GST_BASE_SRC_CLASS (gst_rtmp2_src_parent_class)->query (basesrc, query);
|
|
|
|
return ret;
|
|
}
|
|
|
|
static gboolean
|
|
main_loop_running_cb (GstRtmp2Src * self)
|
|
{
|
|
GST_TRACE_OBJECT (self, "Main loop running now");
|
|
|
|
g_mutex_lock (&self->lock);
|
|
self->started = TRUE;
|
|
g_cond_broadcast (&self->cond);
|
|
g_mutex_unlock (&self->lock);
|
|
|
|
return G_SOURCE_REMOVE;
|
|
}
|
|
|
|
/* Mainloop task */
|
|
static void
|
|
gst_rtmp2_src_task_func (gpointer user_data)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (user_data);
|
|
GMainContext *context;
|
|
GMainLoop *loop;
|
|
GTask *connector;
|
|
GSource *source;
|
|
|
|
GST_DEBUG_OBJECT (self, "gst_rtmp2_src_task starting");
|
|
g_mutex_lock (&self->lock);
|
|
|
|
context = self->context = g_main_context_new ();
|
|
g_main_context_push_thread_default (context);
|
|
loop = self->loop = g_main_loop_new (context, TRUE);
|
|
|
|
source = g_idle_source_new ();
|
|
g_source_set_callback (source, (GSourceFunc) main_loop_running_cb, self,
|
|
NULL);
|
|
g_source_attach (source, self->context);
|
|
g_source_unref (source);
|
|
|
|
connector = g_task_new (self, self->cancellable, connect_task_done, NULL);
|
|
|
|
g_clear_pointer (&self->stats, gst_structure_free);
|
|
|
|
GST_OBJECT_LOCK (self);
|
|
gst_rtmp_client_connect_async (&self->location, self->cancellable,
|
|
client_connect_done, connector);
|
|
GST_OBJECT_UNLOCK (self);
|
|
|
|
/* Run loop */
|
|
g_mutex_unlock (&self->lock);
|
|
g_main_loop_run (loop);
|
|
g_mutex_lock (&self->lock);
|
|
|
|
if (self->connection) {
|
|
self->stats = gst_rtmp_connection_get_stats (self->connection);
|
|
}
|
|
|
|
g_clear_pointer (&self->loop, g_main_loop_unref);
|
|
g_clear_pointer (&self->connection, gst_rtmp_connection_close_and_unref);
|
|
g_cond_broadcast (&self->cond);
|
|
|
|
/* Run loop cleanup */
|
|
g_mutex_unlock (&self->lock);
|
|
while (g_main_context_pending (context)) {
|
|
GST_DEBUG_OBJECT (self, "iterating main context to clean up");
|
|
g_main_context_iteration (context, FALSE);
|
|
}
|
|
g_main_context_pop_thread_default (context);
|
|
g_mutex_lock (&self->lock);
|
|
|
|
g_clear_pointer (&self->context, g_main_context_unref);
|
|
gst_buffer_replace (&self->message, NULL);
|
|
|
|
g_mutex_unlock (&self->lock);
|
|
GST_DEBUG_OBJECT (self, "gst_rtmp2_src_task exiting");
|
|
}
|
|
|
|
static void
|
|
client_connect_done (GObject * source, GAsyncResult * result,
|
|
gpointer user_data)
|
|
{
|
|
GTask *task = user_data;
|
|
GstRtmp2Src *self = g_task_get_source_object (task);
|
|
GError *error = NULL;
|
|
GstRtmpConnection *connection;
|
|
|
|
connection = gst_rtmp_client_connect_finish (result, &error);
|
|
if (!connection) {
|
|
g_task_return_error (task, error);
|
|
g_object_unref (task);
|
|
return;
|
|
}
|
|
|
|
g_task_set_task_data (task, connection, g_object_unref);
|
|
|
|
if (g_task_return_error_if_cancelled (task)) {
|
|
g_object_unref (task);
|
|
return;
|
|
}
|
|
|
|
GST_OBJECT_LOCK (self);
|
|
gst_rtmp_client_start_play_async (connection, self->location.stream,
|
|
g_task_get_cancellable (task), start_play_done, task);
|
|
GST_OBJECT_UNLOCK (self);
|
|
}
|
|
|
|
static void
|
|
start_play_done (GObject * source, GAsyncResult * result, gpointer user_data)
|
|
{
|
|
GTask *task = G_TASK (user_data);
|
|
GstRtmp2Src *self = g_task_get_source_object (task);
|
|
GstRtmpConnection *connection = g_task_get_task_data (task);
|
|
GError *error = NULL;
|
|
|
|
if (g_task_return_error_if_cancelled (task)) {
|
|
g_object_unref (task);
|
|
return;
|
|
}
|
|
|
|
if (gst_rtmp_client_start_play_finish (connection, result,
|
|
&self->stream_id, &error)) {
|
|
g_task_return_pointer (task, g_object_ref (connection),
|
|
gst_rtmp_connection_close_and_unref);
|
|
} else {
|
|
g_task_return_error (task, error);
|
|
}
|
|
|
|
g_task_set_task_data (task, NULL, NULL);
|
|
g_object_unref (task);
|
|
}
|
|
|
|
static void
|
|
got_message (GstRtmpConnection * connection, GstBuffer * buffer,
|
|
gpointer user_data)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (user_data);
|
|
GstRtmpMeta *meta = gst_buffer_get_rtmp_meta (buffer);
|
|
guint32 min_size = 1;
|
|
|
|
g_return_if_fail (meta);
|
|
|
|
if (meta->mstream != self->stream_id) {
|
|
GST_DEBUG_OBJECT (self, "Ignoring %s message with stream %" G_GUINT32_FORMAT
|
|
" != %" G_GUINT32_FORMAT, gst_rtmp_message_type_get_nick (meta->type),
|
|
meta->mstream, self->stream_id);
|
|
return;
|
|
}
|
|
|
|
switch (meta->type) {
|
|
case GST_RTMP_MESSAGE_TYPE_VIDEO:
|
|
min_size = 6;
|
|
break;
|
|
|
|
case GST_RTMP_MESSAGE_TYPE_AUDIO:
|
|
min_size = 2;
|
|
break;
|
|
|
|
case GST_RTMP_MESSAGE_TYPE_DATA_AMF0:
|
|
break;
|
|
|
|
default:
|
|
GST_DEBUG_OBJECT (self, "Ignoring %s message, wrong type",
|
|
gst_rtmp_message_type_get_nick (meta->type));
|
|
return;
|
|
}
|
|
|
|
if (meta->size < min_size) {
|
|
GST_DEBUG_OBJECT (self, "Ignoring too small %s message (%" G_GUINT32_FORMAT
|
|
" < %" G_GUINT32_FORMAT ")",
|
|
gst_rtmp_message_type_get_nick (meta->type), meta->size, min_size);
|
|
return;
|
|
}
|
|
|
|
g_mutex_lock (&self->lock);
|
|
while (self->message) {
|
|
if (!self->running) {
|
|
goto out;
|
|
}
|
|
g_cond_wait (&self->cond, &self->lock);
|
|
}
|
|
|
|
self->message = gst_buffer_ref (buffer);
|
|
g_cond_signal (&self->cond);
|
|
|
|
out:
|
|
g_mutex_unlock (&self->lock);
|
|
return;
|
|
}
|
|
|
|
static void
|
|
error_callback (GstRtmpConnection * connection, const GError * error,
|
|
GstRtmp2Src * self)
|
|
{
|
|
g_mutex_lock (&self->lock);
|
|
if (self->cancellable) {
|
|
g_cancellable_cancel (self->cancellable);
|
|
} else if (self->loop) {
|
|
GST_INFO_OBJECT (self, "Connection error: %s %d %s",
|
|
g_quark_to_string (error->domain), error->code, error->message);
|
|
self->connection_error = TRUE;
|
|
stop_task (self);
|
|
}
|
|
g_mutex_unlock (&self->lock);
|
|
}
|
|
|
|
static void
|
|
control_callback (GstRtmpConnection * connection, gint uc_type,
|
|
guint stream_id, GstRtmp2Src * self)
|
|
{
|
|
GST_INFO_OBJECT (self, "stream %u got %s", stream_id,
|
|
gst_rtmp_user_control_type_get_nick (uc_type));
|
|
|
|
if (uc_type == GST_RTMP_USER_CONTROL_TYPE_STREAM_EOF && stream_id == 1) {
|
|
GST_INFO_OBJECT (self, "went EOS");
|
|
stop_task (self);
|
|
}
|
|
}
|
|
|
|
static void
|
|
send_connect_error (GstRtmp2Src * self, GError * error)
|
|
{
|
|
if (!error) {
|
|
GST_ERROR_OBJECT (self, "Connect failed with NULL error");
|
|
GST_ELEMENT_ERROR (self, RESOURCE, FAILED, ("Failed to connect"), (NULL));
|
|
return;
|
|
}
|
|
|
|
if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
|
|
GST_DEBUG_OBJECT (self, "Connection was cancelled: %s", error->message);
|
|
return;
|
|
}
|
|
|
|
GST_ERROR_OBJECT (self, "Failed to connect: %s %d %s",
|
|
g_quark_to_string (error->domain), error->code, error->message);
|
|
|
|
if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_PERMISSION_DENIED)) {
|
|
GST_ELEMENT_ERROR (self, RESOURCE, NOT_AUTHORIZED,
|
|
("Not authorized to connect: %s", error->message), (NULL));
|
|
} else if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CONNECTION_REFUSED)) {
|
|
GST_ELEMENT_ERROR (self, RESOURCE, OPEN_READ,
|
|
("Connection refused: %s", error->message), (NULL));
|
|
} else {
|
|
GST_ELEMENT_ERROR (self, RESOURCE, FAILED,
|
|
("Failed to connect: %s", error->message),
|
|
("domain %s, code %d", g_quark_to_string (error->domain), error->code));
|
|
}
|
|
}
|
|
|
|
static void
|
|
connect_task_done (GObject * object, GAsyncResult * result, gpointer user_data)
|
|
{
|
|
GstRtmp2Src *self = GST_RTMP2_SRC (object);
|
|
GTask *task = G_TASK (result);
|
|
GError *error = NULL;
|
|
|
|
g_mutex_lock (&self->lock);
|
|
|
|
g_warn_if_fail (g_task_is_valid (task, object));
|
|
|
|
if (self->cancellable == g_task_get_cancellable (task)) {
|
|
g_clear_object (&self->cancellable);
|
|
}
|
|
|
|
self->connection = g_task_propagate_pointer (task, &error);
|
|
if (self->connection) {
|
|
gst_rtmp_connection_set_input_handler (self->connection,
|
|
got_message, g_object_ref (self), g_object_unref);
|
|
g_signal_connect_object (self->connection, "error",
|
|
G_CALLBACK (error_callback), self, 0);
|
|
g_signal_connect_object (self->connection, "stream-control",
|
|
G_CALLBACK (control_callback), self, 0);
|
|
} else {
|
|
send_connect_error (self, error);
|
|
self->connection_error = TRUE;
|
|
stop_task (self);
|
|
g_error_free (error);
|
|
}
|
|
|
|
g_cond_broadcast (&self->cond);
|
|
g_mutex_unlock (&self->lock);
|
|
}
|
|
|
|
static GstStructure *
|
|
gst_rtmp2_src_get_stats (GstRtmp2Src * self)
|
|
{
|
|
GstStructure *s;
|
|
|
|
g_mutex_lock (&self->lock);
|
|
|
|
if (self->connection) {
|
|
s = gst_rtmp_connection_get_stats (self->connection);
|
|
} else if (self->stats) {
|
|
s = gst_structure_copy (self->stats);
|
|
} else {
|
|
s = gst_rtmp_connection_get_null_stats ();
|
|
}
|
|
|
|
g_mutex_unlock (&self->lock);
|
|
|
|
return s;
|
|
}
|