mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-11-27 04:01:08 +00:00
Code Issues 1 Projects Releases Wiki Activity

rtmp2sink: handle EOS event and close stream

Browse source 
https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/issues/1285

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/merge_requests/1256>
This commit is contained in:
Nazar Mokrynskyi 2020-05-02 04:49:42 +03:00 committed by GStreamer Merge Bot
parent 66f9d37c37
commit 9a2828c216
3 changed files with 139 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

90
gst/rtmp2/gstrtmp2sink.c
View file

@ -88,6 +88,8 @@ typedef struct
GPtrArray *headers;
guint64 last_ts, base_ts; /* timestamp fixup */
GstRtmpStopCommands stop_commands;
} GstRtmp2Sink;
typedef struct
@ -106,6 +108,7 @@ static void gst_rtmp2_sink_uri_handler_init (GstURIHandlerInterface * iface);
/* GstBaseSink virtual functions */
static gboolean gst_rtmp2_sink_start (GstBaseSink * sink);
static gboolean gst_rtmp2_sink_stop (GstBaseSink * sink);
static gboolean gst_rtmp2_sink_event (GstBaseSink * sink, GstEvent * event);
static gboolean gst_rtmp2_sink_unlock (GstBaseSink * sink);
static gboolean gst_rtmp2_sink_unlock_stop (GstBaseSink * sink);
static GstFlowReturn gst_rtmp2_sink_render (GstBaseSink * sink,
@ -147,8 +150,35 @@ enum
PROP_PEAK_KBPS,
PROP_CHUNK_SIZE,
PROP_STATS,
PROP_STOP_COMMANDS,
};
#define DEFAULT_STOP_COMMANDS GST_RTMP_STOP_COMMAND_FCUNPUBLISH | \
GST_RTMP_STOP_COMMAND_DELETE_STREAM /* FCUnpublish + deleteStream */
#define GST_RTMP_STOP_COMMANDS_TYPE \
(gst_rtmp2_sink_stop_commands_get_type())
static GType
gst_rtmp2_sink_stop_commands_get_type (void)
{
static GType type = 0;
static const GFlagsValue types[] = {
{GST_RTMP_STOP_COMMAND_NONE, "No command", "none"},
{GST_RTMP_STOP_COMMAND_FCUNPUBLISH, "FCUnpublish", "fcunpublish"},
{GST_RTMP_STOP_COMMAND_CLOSE_STREAM, "closeStream", "closestream"},
{GST_RTMP_STOP_COMMAND_DELETE_STREAM, "deleteStream", "deletestream"},
{0, NULL, NULL},
};
if (g_once_init_enter (&type)) {
GType tmp = g_flags_register_static ("GstRtmp2SinkStopCommandsFlags",
types);
g_once_init_leave (&type, tmp);
}
return type;
}
/* pad templates */
static GstStaticPadTemplate gst_rtmp2_sink_sink_template =
@ -183,6 +213,7 @@ gst_rtmp2_sink_class_init (GstRtmp2SinkClass * klass)
gobject_class->finalize = gst_rtmp2_sink_finalize;
base_sink_class->start = GST_DEBUG_FUNCPTR (gst_rtmp2_sink_start);
base_sink_class->stop = GST_DEBUG_FUNCPTR (gst_rtmp2_sink_stop);
base_sink_class->event = GST_DEBUG_FUNCPTR (gst_rtmp2_sink_event);
base_sink_class->unlock = GST_DEBUG_FUNCPTR (gst_rtmp2_sink_unlock);
base_sink_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_rtmp2_sink_unlock_stop);
base_sink_class->render = GST_DEBUG_FUNCPTR (gst_rtmp2_sink_render);
@ -227,6 +258,12 @@ gst_rtmp2_sink_class_init (GstRtmp2SinkClass * klass)
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_STOP_COMMANDS,
g_param_spec_flags ("stop-commands", "Stop commands",
"RTMP commands to send on EOS event before closing connection",
GST_RTMP_STOP_COMMANDS_TYPE, DEFAULT_STOP_COMMANDS,
(GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
gst_type_mark_as_plugin_api (GST_TYPE_RTMP_LOCATION_HANDLER, 0);
GST_DEBUG_CATEGORY_INIT (gst_rtmp2_sink_debug_category, "rtmp2sink", 0,
"debug category for rtmp2sink element");
@ -249,6 +286,8 @@ gst_rtmp2_sink_init (GstRtmp2Sink * self)
self->headers = g_ptr_array_new_with_free_func
((GDestroyNotify) gst_mini_object_unref);
self->stop_commands = DEFAULT_STOP_COMMANDS;
}
static void
@ -360,6 +399,11 @@ gst_rtmp2_sink_set_property (GObject * object, guint property_id,
set_chunk_size (self);
g_mutex_unlock (&self->lock);
break;
case PROP_STOP_COMMANDS:
GST_OBJECT_LOCK (self);
self->stop_commands = g_value_get_flags (value);
GST_OBJECT_UNLOCK (self);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
break;
@ -457,6 +501,11 @@ gst_rtmp2_sink_get_property (GObject * object, guint property_id,
case PROP_STATS:
g_value_take_boxed (value, gst_rtmp2_sink_get_stats (self));
break;
case PROP_STOP_COMMANDS:
GST_OBJECT_LOCK (self);
g_value_set_flags (value, self->stop_commands);
GST_OBJECT_UNLOCK (self);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
break;
@ -556,6 +605,47 @@ gst_rtmp2_sink_stop (GstBaseSink * sink)
return TRUE;
}
static gboolean
stop_publish_invoker (gpointer user_data)
{
GstRtmp2Sink *self = user_data;
if (self->connection) {
GST_OBJECT_LOCK (self);
if (self->stop_commands != GST_RTMP_STOP_COMMAND_NONE) {
gst_rtmp_client_stop_publish (self->connection, self->location.stream,
self->stop_commands);
}
GST_OBJECT_UNLOCK (self);
}
return G_SOURCE_REMOVE;
}
static gboolean
gst_rtmp2_sink_event (GstBaseSink * sink, GstEvent * event)
{
GstEventType type;
GstRtmp2Sink *self = GST_RTMP2_SINK (sink);
type = GST_EVENT_TYPE (event);
switch (type) {
case GST_EVENT_EOS:
g_mutex_lock (&self->lock);
if (self->loop) {
GST_DEBUG_OBJECT (self, "Got EOS: stopping publish");
g_main_context_invoke (self->context, stop_publish_invoker, self);
}
g_mutex_unlock (&self->lock);
break;
default:
break;
}
return GST_BASE_SINK_CLASS (gst_rtmp2_sink_parent_class)->event (sink, event);;
}
static gboolean
gst_rtmp2_sink_unlock (GstBaseSink * sink)
{

38
gst/rtmp2/rtmp/rtmpclient.c
View file

@ -296,6 +296,8 @@ static void socket_connect_done (GObject * source, GAsyncResult * result,
static void handshake_done (GObject * source, GAsyncResult * result,
gpointer user_data);
static void send_connect (GTask * task);
static void send_stop (GstRtmpConnection * connection, const gchar * stream,
const GstRtmpStopCommands stop_commands);
static void send_secure_token_response (GTask * task,
GstRtmpConnection * connection, const gchar * challenge);
static void connection_error (GstRtmpConnection * connection,
@ -1346,3 +1348,39 @@ gst_rtmp_client_start_play_finish (GstRtmpConnection * connection,
{
return start_stream_finish (connection, result, stream_id, error);
}
void
gst_rtmp_client_stop_publish (GstRtmpConnection * connection,
const gchar * stream, const GstRtmpStopCommands stop_commands)
{
send_stop (connection, stream, stop_commands);
}
static void
send_stop (GstRtmpConnection * connection, const gchar * stream,
const GstRtmpStopCommands stop_commands)
{
GstAmfNode *command_object, *stream_name;
command_object = gst_amf_node_new_null ();
stream_name = gst_amf_node_new_string (stream, -1);
if (stop_commands & GST_RTMP_STOP_COMMAND_FCUNPUBLISH) {
GST_DEBUG ("Sending stop command 'FCUnpublish' for stream '%s'", stream);
gst_rtmp_connection_send_command (connection, NULL, NULL, 0,
"FCUnpublish", command_object, stream_name, NULL);
}
if (stop_commands & GST_RTMP_STOP_COMMAND_CLOSE_STREAM) {
GST_DEBUG ("Sending stop command 'closeStream' for stream '%s'", stream);
gst_rtmp_connection_send_command (connection, NULL, NULL, 0,
"closeStream", command_object, stream_name, NULL);
}
if (stop_commands & GST_RTMP_STOP_COMMAND_DELETE_STREAM) {
GST_DEBUG ("Sending stop command 'deleteStream' for stream '%s'", stream);
gst_rtmp_connection_send_command (connection, NULL, NULL, 0,
"deleteStream", command_object, stream_name, NULL);
}
gst_amf_node_free (stream_name);
gst_amf_node_free (command_object);
}

11
gst/rtmp2/rtmp/rtmpclient.h
View file

@ -73,6 +73,14 @@ typedef struct _GstRtmpLocation
gboolean publish;
} GstRtmpLocation;
typedef enum
{
GST_RTMP_STOP_COMMAND_NONE = 0,
GST_RTMP_STOP_COMMAND_FCUNPUBLISH = (1 << 0),
GST_RTMP_STOP_COMMAND_CLOSE_STREAM = (1 << 1),
GST_RTMP_STOP_COMMAND_DELETE_STREAM = (1 << 2)
} GstRtmpStopCommands;
void gst_rtmp_location_copy (GstRtmpLocation * dest,
const GstRtmpLocation * src);
void gst_rtmp_location_clear (GstRtmpLocation * uri);
@ -98,5 +106,8 @@ void gst_rtmp_client_start_play_async (GstRtmpConnection * connection,
gboolean gst_rtmp_client_start_play_finish (GstRtmpConnection * connection,
GAsyncResult * result, guint * stream_id, GError ** error);
void gst_rtmp_client_stop_publish (GstRtmpConnection * connection,
const gchar * stream, const GstRtmpStopCommands stop_commands);
G_END_DECLS
#endif