gstreamer/subprojects/gst-plugins-bad/gst/rtmp2/rtmp/rtmpconnection.c

1401 lines
40 KiB
C

/* GStreamer RTMP Library
* Copyright (C) 2013 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.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <gst/gst.h>
#include <string.h>
#include <math.h>
#include "rtmpconnection.h"
#include "rtmpchunkstream.h"
#include "rtmpmessage.h"
#include "rtmputils.h"
#include "amf.h"
GST_DEBUG_CATEGORY_STATIC (gst_rtmp_connection_debug_category);
#define GST_CAT_DEFAULT gst_rtmp_connection_debug_category
#define READ_SIZE 8192
typedef void (*GstRtmpConnectionCallback) (GstRtmpConnection * connection);
struct _GstRtmpConnection
{
GObject parent_instance;
/* should be properties */
gboolean input_paused;
gboolean error;
/* private */
GThread *thread;
GSocketConnection *connection;
GCancellable *cancellable;
GSocketClient *socket_client;
GAsyncQueue *output_queue;
GMainContext *main_context;
GCancellable *outer_cancellable;
gulong cancel_handler_id;
GSource *input_source;
GByteArray *input_bytes;
guint input_needed_bytes;
GstRtmpChunkStreams *input_streams, *output_streams;
GList *transactions;
GList *expected_commands;
guint transaction_count;
GstRtmpConnectionMessageFunc input_handler;
gpointer input_handler_user_data;
GDestroyNotify input_handler_user_data_destroy;
GstRtmpConnectionFunc output_handler;
gpointer output_handler_user_data;
GDestroyNotify output_handler_user_data_destroy;
gboolean writing;
/* Protects the values below during concurrent access.
* - Taken by the loop thread when writing, but not reading.
* - Taken by other threads when reading (calling get_stats).
*/
GMutex stats_lock;
/* RTMP configuration */
guint32 in_chunk_size;
guint32 out_chunk_size, out_chunk_size_pending;
guint32 in_window_ack_size;
guint32 out_window_ack_size, out_window_ack_size_pending;
guint64 in_bytes_total;
guint64 out_bytes_total;
guint64 in_bytes_acked;
guint64 out_bytes_acked;
};
typedef struct
{
GObjectClass parent_class;
} GstRtmpConnectionClass;
/* prototypes */
static void gst_rtmp_connection_dispose (GObject * object);
static void gst_rtmp_connection_finalize (GObject * object);
static void gst_rtmp_connection_set_cancellable (GstRtmpConnection * self,
GCancellable * cancellable);
static void gst_rtmp_connection_emit_error (GstRtmpConnection * self,
GError * error);
static gboolean gst_rtmp_connection_input_ready (GInputStream * is,
gpointer user_data);
static void gst_rtmp_connection_start_write (GstRtmpConnection * self);
static void gst_rtmp_connection_write_buffer_done (GObject * obj,
GAsyncResult * result, gpointer user_data);
static void gst_rtmp_connection_start_read (GstRtmpConnection * sc,
guint needed_bytes);
static void gst_rtmp_connection_try_read (GstRtmpConnection * sc);
static void gst_rtmp_connection_do_read (GstRtmpConnection * sc);
static void gst_rtmp_connection_handle_aggregate (GstRtmpConnection *
connection, GstBuffer * buffer);
static void gst_rtmp_connection_handle_protocol_control (GstRtmpConnection *
connection, GstBuffer * buffer);
static void gst_rtmp_connection_handle_cm (GstRtmpConnection * connection,
GstBuffer * buffer);
static void gst_rtmp_connection_handle_user_control (GstRtmpConnection * sc,
GstBuffer * buffer);
static void gst_rtmp_connection_handle_message (GstRtmpConnection * sc,
GstBuffer * buffer);
static void gst_rtmp_connection_handle_set_chunk_size (GstRtmpConnection * self,
guint32 in_chunk_size);
static void gst_rtmp_connection_handle_ack (GstRtmpConnection * self,
guint32 bytes);
static void gst_rtmp_connection_handle_window_ack_size (GstRtmpConnection *
self, guint32 in_chunk_size);
static void gst_rtmp_connection_send_ack (GstRtmpConnection * connection);
static void
gst_rtmp_connection_send_ping_response (GstRtmpConnection * connection,
guint32 event_data);
static gboolean
gst_rtmp_connection_prepare_protocol_control (GstRtmpConnection * self,
GstBuffer * buffer);
static void
gst_rtmp_connection_apply_protocol_control (GstRtmpConnection * self);
typedef struct
{
gdouble transaction_id;
GstRtmpCommandCallback func;
gpointer user_data;
} Transaction;
static Transaction *
transaction_new (gdouble transaction_id, GstRtmpCommandCallback func,
gpointer user_data)
{
Transaction *data = g_new (Transaction, 1);
data->transaction_id = transaction_id;
data->func = func;
data->user_data = user_data;
return data;
}
static void
transaction_free (gpointer ptr)
{
Transaction *data = ptr;
g_free (data);
}
typedef struct
{
guint32 stream_id;
gchar *command_name;
GstRtmpCommandCallback func;
gpointer user_data;
} ExpectedCommand;
static ExpectedCommand *
expected_command_new (guint32 stream_id, const gchar * command_name,
GstRtmpCommandCallback func, gpointer user_data)
{
ExpectedCommand *data = g_new (ExpectedCommand, 1);
data->stream_id = stream_id;
data->command_name = g_strdup (command_name);
data->func = func;
data->user_data = user_data;
return data;
}
static void
expected_command_free (gpointer ptr)
{
ExpectedCommand *data = ptr;
g_free (data->command_name);
g_free (data);
}
enum
{
SIGNAL_ERROR,
SIGNAL_STREAM_CONTROL,
N_SIGNALS
};
static guint signals[N_SIGNALS] = { 0, };
/* singletons */
static GstMemory *set_data_frame_value;
static void
init_set_data_frame_value (void)
{
GstAmfNode *node = gst_amf_node_new_string ("@setDataFrame", -1);
GBytes *bytes = gst_amf_node_serialize (node);
gsize size;
const gchar *data = g_bytes_get_data (bytes, &size);
set_data_frame_value = gst_memory_new_wrapped (GST_MEMORY_FLAG_READONLY,
(gpointer) data, size, 0, size, bytes, (GDestroyNotify) g_bytes_unref);
GST_MINI_OBJECT_FLAG_SET (set_data_frame_value,
GST_MINI_OBJECT_FLAG_MAY_BE_LEAKED);
gst_amf_node_free (node);
}
/* class initialization */
G_DEFINE_TYPE_WITH_CODE (GstRtmpConnection, gst_rtmp_connection,
G_TYPE_OBJECT,
GST_DEBUG_CATEGORY_INIT (gst_rtmp_connection_debug_category,
"rtmpconnection", 0, "debug category for GstRtmpConnection class");
init_set_data_frame_value ());
static void
gst_rtmp_connection_class_init (GstRtmpConnectionClass * klass)
{
GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
gobject_class->dispose = gst_rtmp_connection_dispose;
gobject_class->finalize = gst_rtmp_connection_finalize;
signals[SIGNAL_ERROR] = g_signal_new ("error", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST, 0, NULL, NULL, NULL, G_TYPE_NONE, 1, G_TYPE_ERROR);
signals[SIGNAL_STREAM_CONTROL] = g_signal_new ("stream-control",
G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST, 0, NULL, NULL, NULL,
G_TYPE_NONE, 2, G_TYPE_INT, G_TYPE_UINT);
GST_DEBUG_REGISTER_FUNCPTR (gst_rtmp_connection_do_read);
}
static void
gst_rtmp_connection_init (GstRtmpConnection * rtmpconnection)
{
rtmpconnection->cancellable = g_cancellable_new ();
rtmpconnection->output_queue =
g_async_queue_new_full ((GDestroyNotify) gst_buffer_unref);
rtmpconnection->input_streams = gst_rtmp_chunk_streams_new ();
rtmpconnection->output_streams = gst_rtmp_chunk_streams_new ();
rtmpconnection->in_chunk_size = GST_RTMP_DEFAULT_CHUNK_SIZE;
rtmpconnection->out_chunk_size = GST_RTMP_DEFAULT_CHUNK_SIZE;
rtmpconnection->input_bytes = g_byte_array_sized_new (2 * READ_SIZE);
rtmpconnection->input_needed_bytes = 1;
g_mutex_init (&rtmpconnection->stats_lock);
}
void
gst_rtmp_connection_dispose (GObject * object)
{
GstRtmpConnection *rtmpconnection = GST_RTMP_CONNECTION (object);
GST_DEBUG_OBJECT (rtmpconnection, "dispose");
/* clean up as possible. may be called multiple times */
gst_rtmp_connection_close (rtmpconnection);
g_cancellable_cancel (rtmpconnection->cancellable);
gst_rtmp_connection_set_input_handler (rtmpconnection, NULL, NULL, NULL);
gst_rtmp_connection_set_output_handler (rtmpconnection, NULL, NULL, NULL);
gst_rtmp_connection_set_cancellable (rtmpconnection, NULL);
G_OBJECT_CLASS (gst_rtmp_connection_parent_class)->dispose (object);
}
void
gst_rtmp_connection_finalize (GObject * object)
{
GstRtmpConnection *rtmpconnection = GST_RTMP_CONNECTION (object);
GST_DEBUG_OBJECT (rtmpconnection, "finalize");
/* clean up object here */
g_mutex_clear (&rtmpconnection->stats_lock);
g_clear_object (&rtmpconnection->cancellable);
g_clear_object (&rtmpconnection->connection);
g_clear_pointer (&rtmpconnection->output_queue, g_async_queue_unref);
g_clear_pointer (&rtmpconnection->input_streams, gst_rtmp_chunk_streams_free);
g_clear_pointer (&rtmpconnection->output_streams,
gst_rtmp_chunk_streams_free);
g_clear_pointer (&rtmpconnection->input_bytes, g_byte_array_unref);
g_clear_pointer (&rtmpconnection->main_context, g_main_context_unref);
g_clear_pointer (&rtmpconnection->thread, g_thread_unref);
G_OBJECT_CLASS (gst_rtmp_connection_parent_class)->finalize (object);
}
GSocket *
gst_rtmp_connection_get_socket (GstRtmpConnection * sc)
{
return g_socket_connection_get_socket (sc->connection);
}
static void
gst_rtmp_connection_set_socket_connection (GstRtmpConnection * sc,
GSocketConnection * connection)
{
GInputStream *is;
sc->thread = g_thread_ref (g_thread_self ());
sc->main_context = g_main_context_ref_thread_default ();
sc->connection = g_object_ref (connection);
/* refs the socket because it's creating an input stream, which holds a ref */
is = g_io_stream_get_input_stream (G_IO_STREAM (sc->connection));
/* refs the socket because it's creating a socket source */
g_warn_if_fail (!sc->input_source);
sc->input_source =
g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM (is),
sc->cancellable);
g_source_set_callback (sc->input_source,
(GSourceFunc) gst_rtmp_connection_input_ready, g_object_ref (sc),
g_object_unref);
g_source_attach (sc->input_source, sc->main_context);
}
static void
gst_rtmp_connection_set_cancellable (GstRtmpConnection * self,
GCancellable * cancellable)
{
g_cancellable_disconnect (self->outer_cancellable, self->cancel_handler_id);
g_clear_object (&self->outer_cancellable);
self->cancel_handler_id = 0;
if (cancellable == NULL)
return;
self->outer_cancellable = g_object_ref (cancellable);
self->cancel_handler_id =
g_cancellable_connect (cancellable, G_CALLBACK (g_cancellable_cancel),
g_object_ref (self->cancellable), g_object_unref);
}
GstRtmpConnection *
gst_rtmp_connection_new (GSocketConnection * connection,
GCancellable * cancellable)
{
GstRtmpConnection *sc;
sc = g_object_new (GST_TYPE_RTMP_CONNECTION, NULL);
gst_rtmp_connection_set_socket_connection (sc, connection);
gst_rtmp_connection_set_cancellable (sc, cancellable);
return sc;
}
static void
cancel_all_commands (GstRtmpConnection * self, const gchar * reason)
{
GList *l;
for (l = self->transactions; l; l = g_list_next (l)) {
Transaction *cc = l->data;
GST_LOG_OBJECT (self, "calling transaction callback %s",
GST_DEBUG_FUNCPTR_NAME (cc->func));
cc->func (reason, NULL, cc->user_data);
}
g_list_free_full (self->transactions, transaction_free);
self->transactions = NULL;
for (l = self->expected_commands; l; l = g_list_next (l)) {
ExpectedCommand *cc = l->data;
GST_LOG_OBJECT (self, "calling expected command callback %s",
GST_DEBUG_FUNCPTR_NAME (cc->func));
cc->func (reason, NULL, cc->user_data);
}
g_list_free_full (self->expected_commands, expected_command_free);
self->expected_commands = NULL;
}
void
gst_rtmp_connection_close (GstRtmpConnection * self)
{
if (self->thread != g_thread_self ()) {
GST_ERROR_OBJECT (self, "Called from wrong thread");
}
g_cancellable_cancel (self->cancellable);
cancel_all_commands (self, "connection closed locally");
if (self->input_source) {
g_source_destroy (self->input_source);
g_clear_pointer (&self->input_source, g_source_unref);
}
if (self->connection) {
g_io_stream_close_async (G_IO_STREAM (self->connection),
G_PRIORITY_DEFAULT, NULL, NULL, NULL);
}
}
void
gst_rtmp_connection_close_and_unref (gpointer ptr)
{
GstRtmpConnection *connection;
g_return_if_fail (ptr);
connection = GST_RTMP_CONNECTION (ptr);
gst_rtmp_connection_close (connection);
g_object_unref (connection);
}
void
gst_rtmp_connection_set_input_handler (GstRtmpConnection * sc,
GstRtmpConnectionMessageFunc callback, gpointer user_data,
GDestroyNotify user_data_destroy)
{
if (sc->input_handler_user_data_destroy) {
sc->input_handler_user_data_destroy (sc->input_handler_user_data);
}
sc->input_handler = callback;
sc->input_handler_user_data = user_data;
sc->input_handler_user_data_destroy = user_data_destroy;
}
void
gst_rtmp_connection_set_output_handler (GstRtmpConnection * sc,
GstRtmpConnectionFunc callback, gpointer user_data,
GDestroyNotify user_data_destroy)
{
if (sc->output_handler_user_data_destroy) {
sc->output_handler_user_data_destroy (sc->output_handler_user_data);
}
sc->output_handler = callback;
sc->output_handler_user_data = user_data;
sc->output_handler_user_data_destroy = user_data_destroy;
}
static gboolean
gst_rtmp_connection_input_ready (GInputStream * is, gpointer user_data)
{
GstRtmpConnection *sc = user_data;
gssize ret;
guint oldsize;
GError *error = NULL;
guint64 bytes_since_ack;
GST_TRACE_OBJECT (sc, "input ready");
oldsize = sc->input_bytes->len;
g_byte_array_set_size (sc->input_bytes, oldsize + READ_SIZE);
ret =
g_pollable_input_stream_read_nonblocking (G_POLLABLE_INPUT_STREAM (is),
sc->input_bytes->data + oldsize, READ_SIZE, sc->cancellable, &error);
g_byte_array_set_size (sc->input_bytes, oldsize + (ret > 0 ? ret : 0));
if (ret == 0) {
error = g_error_new (G_IO_ERROR, G_IO_ERROR_CONNECTION_CLOSED,
"connection closed remotely");
ret = -1;
}
if (ret < 0) {
gint code = error->code;
if (error->domain == G_IO_ERROR && (code == G_IO_ERROR_WOULD_BLOCK ||
code == G_IO_ERROR_TIMED_OUT || code == G_IO_ERROR_AGAIN)) {
/* should retry */
GST_DEBUG_OBJECT (sc, "read IO error %d %s, continuing",
code, error->message);
g_error_free (error);
return G_SOURCE_CONTINUE;
}
GST_ERROR_OBJECT (sc, "read error: %s %d %s",
g_quark_to_string (error->domain), code, error->message);
gst_rtmp_connection_emit_error (sc, error);
return G_SOURCE_REMOVE;
}
GST_TRACE_OBJECT (sc, "read %" G_GSIZE_FORMAT " bytes", ret);
g_mutex_lock (&sc->stats_lock);
sc->in_bytes_total += ret;
g_mutex_unlock (&sc->stats_lock);
bytes_since_ack = sc->in_bytes_total - sc->in_bytes_acked;
if (sc->in_window_ack_size && bytes_since_ack >= sc->in_window_ack_size) {
gst_rtmp_connection_send_ack (sc);
}
gst_rtmp_connection_try_read (sc);
return G_SOURCE_CONTINUE;
}
static void
gst_rtmp_connection_start_write (GstRtmpConnection * self)
{
GOutputStream *os;
GstBuffer *message, *chunks;
GstRtmpMeta *meta;
GstRtmpChunkStream *cstream;
if (self->writing) {
return;
}
message = g_async_queue_try_pop (self->output_queue);
if (!message) {
return;
}
meta = gst_buffer_get_rtmp_meta (message);
if (!meta) {
GST_ERROR_OBJECT (self, "No RTMP meta on %" GST_PTR_FORMAT, message);
goto out;
}
if (gst_rtmp_message_is_protocol_control (message)) {
if (!gst_rtmp_connection_prepare_protocol_control (self, message)) {
GST_ERROR_OBJECT (self,
"Failed to prepare protocol control %" GST_PTR_FORMAT, message);
goto out;
}
}
cstream = gst_rtmp_chunk_streams_get (self->output_streams, meta->cstream);
if (!cstream) {
GST_ERROR_OBJECT (self, "Failed to get chunk stream for %" GST_PTR_FORMAT,
message);
goto out;
}
chunks = gst_rtmp_chunk_stream_serialize_all (cstream, message,
self->out_chunk_size);
if (!chunks) {
GST_ERROR_OBJECT (self, "Failed to serialize %" GST_PTR_FORMAT, message);
goto out;
}
self->writing = TRUE;
if (self->output_handler) {
self->output_handler (self, self->output_handler_user_data);
}
os = g_io_stream_get_output_stream (G_IO_STREAM (self->connection));
gst_rtmp_output_stream_write_all_buffer_async (os, chunks, G_PRIORITY_DEFAULT,
self->cancellable, gst_rtmp_connection_write_buffer_done,
g_object_ref (self));
gst_buffer_unref (chunks);
out:
gst_buffer_unref (message);
}
static void
gst_rtmp_connection_emit_error (GstRtmpConnection * self, GError * error)
{
if (!self->error) {
self->error = TRUE;
cancel_all_commands (self, error->message);
g_signal_emit (self, signals[SIGNAL_ERROR], 0, error);
}
g_error_free (error);
}
static void
gst_rtmp_connection_write_buffer_done (GObject * obj,
GAsyncResult * result, gpointer user_data)
{
GOutputStream *os = G_OUTPUT_STREAM (obj);
GstRtmpConnection *self = GST_RTMP_CONNECTION (user_data);
gsize bytes_written = 0;
GError *error = NULL;
gboolean res;
self->writing = FALSE;
res = gst_rtmp_output_stream_write_all_buffer_finish (os, result,
&bytes_written, &error);
g_mutex_lock (&self->stats_lock);
self->out_bytes_total += bytes_written;
g_mutex_unlock (&self->stats_lock);
if (!res) {
if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
GST_INFO_OBJECT (self,
"write cancelled (wrote %" G_GSIZE_FORMAT " bytes)", bytes_written);
} else {
GST_ERROR_OBJECT (self,
"write error: %s (wrote %" G_GSIZE_FORMAT " bytes)",
error->message, bytes_written);
}
gst_rtmp_connection_emit_error (self, error);
g_object_unref (self);
return;
}
GST_LOG_OBJECT (self, "write completed; wrote %" G_GSIZE_FORMAT " bytes",
bytes_written);
gst_rtmp_connection_apply_protocol_control (self);
gst_rtmp_connection_start_write (self);
g_object_unref (self);
}
static void
gst_rtmp_connection_start_read (GstRtmpConnection * connection,
guint needed_bytes)
{
g_return_if_fail (needed_bytes > 0);
connection->input_needed_bytes = needed_bytes;
gst_rtmp_connection_try_read (connection);
}
static void
gst_rtmp_connection_try_read (GstRtmpConnection * connection)
{
guint need = connection->input_needed_bytes,
len = connection->input_bytes->len;
if (len < need) {
GST_TRACE_OBJECT (connection, "got %u < %u bytes, need more", len, need);
return;
}
GST_TRACE_OBJECT (connection, "got %u >= %u bytes, proceeding", len, need);
gst_rtmp_connection_do_read (connection);
}
static void
gst_rtmp_connection_take_input_bytes (GstRtmpConnection * sc, gsize size,
GBytes ** outbytes)
{
g_return_if_fail (size <= sc->input_bytes->len);
if (outbytes) {
*outbytes = g_bytes_new (sc->input_bytes->data, size);
}
g_byte_array_remove_range (sc->input_bytes, 0, size);
}
static void
gst_rtmp_connection_do_read (GstRtmpConnection * sc)
{
GByteArray *input_bytes = sc->input_bytes;
gsize needed_bytes = 1;
while (1) {
GstRtmpChunkStream *cstream;
guint32 chunk_stream_id, header_size, next_size;
guint8 *data;
chunk_stream_id = gst_rtmp_chunk_stream_parse_id (input_bytes->data,
input_bytes->len);
if (!chunk_stream_id) {
needed_bytes = input_bytes->len + 1;
break;
}
cstream = gst_rtmp_chunk_streams_get (sc->input_streams, chunk_stream_id);
header_size = gst_rtmp_chunk_stream_parse_header (cstream,
input_bytes->data, input_bytes->len);
if (input_bytes->len < header_size) {
needed_bytes = header_size;
break;
}
next_size = gst_rtmp_chunk_stream_parse_payload (cstream,
sc->in_chunk_size, &data);
if (input_bytes->len < header_size + next_size) {
needed_bytes = header_size + next_size;
break;
}
memcpy (data, input_bytes->data + header_size, next_size);
gst_rtmp_connection_take_input_bytes (sc, header_size + next_size, NULL);
next_size = gst_rtmp_chunk_stream_wrote_payload (cstream,
sc->in_chunk_size);
if (next_size == 0) {
GstBuffer *buffer = gst_rtmp_chunk_stream_parse_finish (cstream);
gst_rtmp_connection_handle_message (sc, buffer);
gst_buffer_unref (buffer);
}
}
gst_rtmp_connection_start_read (sc, needed_bytes);
}
static void
gst_rtmp_connection_handle_message (GstRtmpConnection * sc, GstBuffer * buffer)
{
if (gst_rtmp_message_is_protocol_control (buffer)) {
gst_rtmp_connection_handle_protocol_control (sc, buffer);
return;
}
if (gst_rtmp_message_is_user_control (buffer)) {
gst_rtmp_connection_handle_user_control (sc, buffer);
return;
}
switch (gst_rtmp_message_get_type (buffer)) {
case GST_RTMP_MESSAGE_TYPE_COMMAND_AMF0:
gst_rtmp_connection_handle_cm (sc, buffer);
return;
case GST_RTMP_MESSAGE_TYPE_AGGREGATE:
gst_rtmp_connection_handle_aggregate (sc, buffer);
break;
default:
if (sc->input_handler) {
sc->input_handler (sc, buffer, sc->input_handler_user_data);
}
return;
}
}
static void
gst_rtmp_connection_handle_aggregate (GstRtmpConnection * connection,
GstBuffer * buffer)
{
GstRtmpMeta *meta;
GstMapInfo map;
gsize pos = 0;
guint32 first_ts = 0;
meta = gst_buffer_get_rtmp_meta (buffer);
g_return_if_fail (meta);
gst_buffer_map (buffer, &map, GST_MAP_READ);
GST_TRACE_OBJECT (connection, "got aggregate message");
/* Parse Aggregate Messages as described in rtmp_specification_1.0.pdf page 26
* The payload is part of a FLV file.
*
* WARNING: This spec defines the payload to use an "RTMP message format"
* which misidentifies the format of the timestamps and omits the size of the
* backpointers. */
while (pos < map.size) {
gsize remaining = map.size - pos;
GstBuffer *submessage;
GstRtmpMeta *submeta;
GstRtmpFlvTagHeader header;
if (!gst_rtmp_flv_tag_parse_header (&header, map.data + pos, remaining)) {
GST_ERROR_OBJECT (connection,
"aggregate contains incomplete header; want %d, got %" G_GSIZE_FORMAT,
GST_RTMP_FLV_TAG_HEADER_SIZE, remaining);
break;
}
if (remaining < header.total_size) {
GST_ERROR_OBJECT (connection,
"aggregate contains incomplete message; want %" G_GSIZE_FORMAT
", got %" G_GSIZE_FORMAT, header.total_size, remaining);
break;
}
submessage = gst_buffer_copy_region (buffer, GST_BUFFER_COPY_FLAGS |
GST_BUFFER_COPY_META | GST_BUFFER_COPY_MEMORY,
pos + GST_RTMP_FLV_TAG_HEADER_SIZE, header.payload_size);
GST_BUFFER_DTS (submessage) = GST_BUFFER_DTS (buffer);
GST_BUFFER_OFFSET (submessage) = GST_BUFFER_OFFSET (buffer) + pos;
GST_BUFFER_OFFSET_END (submessage) =
GST_BUFFER_OFFSET (submessage) + header.total_size;
submeta = gst_buffer_get_rtmp_meta (submessage);
g_assert (submeta);
submeta->type = header.type;
submeta->size = header.payload_size;
if (pos == 0) {
first_ts = header.timestamp;
} else {
guint32 ts_offset = header.timestamp - first_ts;
submeta->ts_delta += ts_offset;
GST_BUFFER_DTS (submessage) += ts_offset * GST_MSECOND;
GST_BUFFER_FLAG_UNSET (submessage, GST_BUFFER_FLAG_DISCONT);
}
gst_rtmp_buffer_dump (submessage, "<<< submessage");
gst_rtmp_connection_handle_message (connection, submessage);
gst_buffer_unref (submessage);
pos += header.total_size;
}
gst_buffer_unmap (buffer, &map);
}
static void
gst_rtmp_connection_handle_protocol_control (GstRtmpConnection * connection,
GstBuffer * buffer)
{
GstRtmpProtocolControl pc;
if (!gst_rtmp_message_parse_protocol_control (buffer, &pc)) {
GST_ERROR_OBJECT (connection, "can't parse protocol control message");
return;
}
GST_LOG_OBJECT (connection, "got protocol control message %d:%s", pc.type,
gst_rtmp_message_type_get_nick (pc.type));
switch (pc.type) {
case GST_RTMP_MESSAGE_TYPE_SET_CHUNK_SIZE:
GST_INFO_OBJECT (connection, "incoming chunk size %" G_GUINT32_FORMAT,
pc.param);
gst_rtmp_connection_handle_set_chunk_size (connection, pc.param);
break;
case GST_RTMP_MESSAGE_TYPE_ABORT_MESSAGE:
GST_ERROR_OBJECT (connection, "unimplemented: chunk abort, stream_id = %"
G_GUINT32_FORMAT, pc.param);
break;
case GST_RTMP_MESSAGE_TYPE_ACKNOWLEDGEMENT:
GST_DEBUG_OBJECT (connection, "acknowledgement %" G_GUINT32_FORMAT,
pc.param);
gst_rtmp_connection_handle_ack (connection, pc.param);
break;
case GST_RTMP_MESSAGE_TYPE_WINDOW_ACK_SIZE:
GST_INFO_OBJECT (connection,
"incoming window ack size: %" G_GUINT32_FORMAT, pc.param);
gst_rtmp_connection_handle_window_ack_size (connection, pc.param);
break;
case GST_RTMP_MESSAGE_TYPE_SET_PEER_BANDWIDTH:
GST_FIXME_OBJECT (connection, "set peer bandwidth: %" G_GUINT32_FORMAT
", %" G_GUINT32_FORMAT, pc.param, pc.param2);
/* FIXME this is not correct, but close enough */
gst_rtmp_connection_request_window_size (connection, pc.param);
break;
default:
GST_ERROR_OBJECT (connection, "unimplemented protocol control type %d:%s",
pc.type, gst_rtmp_message_type_get_nick (pc.type));
break;
}
}
static void
gst_rtmp_connection_handle_user_control (GstRtmpConnection * connection,
GstBuffer * buffer)
{
GstRtmpUserControl uc;
if (!gst_rtmp_message_parse_user_control (buffer, &uc)) {
GST_ERROR_OBJECT (connection, "can't parse user control message");
return;
}
GST_LOG_OBJECT (connection, "got user control message %d:%s", uc.type,
gst_rtmp_user_control_type_get_nick (uc.type));
switch (uc.type) {
case GST_RTMP_USER_CONTROL_TYPE_STREAM_BEGIN:
case GST_RTMP_USER_CONTROL_TYPE_STREAM_EOF:
case GST_RTMP_USER_CONTROL_TYPE_STREAM_DRY:
case GST_RTMP_USER_CONTROL_TYPE_STREAM_IS_RECORDED:
GST_INFO_OBJECT (connection, "stream %u got %s", uc.param,
gst_rtmp_user_control_type_get_nick (uc.type));
g_signal_emit (connection, signals[SIGNAL_STREAM_CONTROL], 0,
uc.type, uc.param);
break;
case GST_RTMP_USER_CONTROL_TYPE_SET_BUFFER_LENGTH:
GST_FIXME_OBJECT (connection, "ignoring set buffer length: %"
G_GUINT32_FORMAT ", %" G_GUINT32_FORMAT " ms", uc.param, uc.param2);
break;
case GST_RTMP_USER_CONTROL_TYPE_PING_REQUEST:
GST_DEBUG_OBJECT (connection, "ping request: %" G_GUINT32_FORMAT,
uc.param);
gst_rtmp_connection_send_ping_response (connection, uc.param);
break;
case GST_RTMP_USER_CONTROL_TYPE_PING_RESPONSE:
GST_DEBUG_OBJECT (connection,
"ignoring ping response: %" G_GUINT32_FORMAT, uc.param);
break;
case GST_RTMP_USER_CONTROL_TYPE_BUFFER_EMPTY:
GST_LOG_OBJECT (connection, "ignoring buffer empty: %" G_GUINT32_FORMAT,
uc.param);
break;
case GST_RTMP_USER_CONTROL_TYPE_BUFFER_READY:
GST_LOG_OBJECT (connection, "ignoring buffer ready: %" G_GUINT32_FORMAT,
uc.param);
break;
default:
GST_ERROR_OBJECT (connection, "unimplemented user control type %d:%s",
uc.type, gst_rtmp_user_control_type_get_nick (uc.type));
break;
}
}
static void
gst_rtmp_connection_handle_set_chunk_size (GstRtmpConnection * self,
guint32 chunk_size)
{
if (chunk_size < GST_RTMP_MINIMUM_CHUNK_SIZE) {
GST_ERROR_OBJECT (self,
"peer requested chunk size %" G_GUINT32_FORMAT "; too small",
chunk_size);
return;
}
if (chunk_size > GST_RTMP_MAXIMUM_CHUNK_SIZE) {
GST_ERROR_OBJECT (self,
"peer requested chunk size %" G_GUINT32_FORMAT "; too large",
chunk_size);
return;
}
if (chunk_size < GST_RTMP_DEFAULT_CHUNK_SIZE) {
GST_WARNING_OBJECT (self,
"peer requested small chunk size %" G_GUINT32_FORMAT, chunk_size);
}
g_mutex_lock (&self->stats_lock);
self->in_chunk_size = chunk_size;
g_mutex_unlock (&self->stats_lock);
}
static void
gst_rtmp_connection_handle_ack (GstRtmpConnection * self, guint32 bytes)
{
guint64 last_ack, new_ack;
guint32 last_ack_low, last_ack_high;
last_ack = self->out_bytes_acked;
last_ack_low = last_ack & G_MAXUINT32;
last_ack_high = (last_ack >> 32) & G_MAXUINT32;
if (bytes < last_ack_low) {
GST_WARNING_OBJECT (self,
"Acknowledgement bytes regression, assuming rollover: %"
G_GUINT32_FORMAT " < %" G_GUINT32_FORMAT, bytes, last_ack_low);
last_ack_high += 1;
}
new_ack = (((guint64) last_ack_high) << 32) | bytes;
GST_LOG_OBJECT (self, "Peer acknowledged %" G_GUINT64_FORMAT " bytes",
new_ack - last_ack);
g_mutex_lock (&self->stats_lock);
self->out_bytes_acked = new_ack;
g_mutex_unlock (&self->stats_lock);
}
static void
gst_rtmp_connection_handle_window_ack_size (GstRtmpConnection * self,
guint32 window_ack_size)
{
if (window_ack_size < GST_RTMP_DEFAULT_WINDOW_ACK_SIZE) {
GST_WARNING_OBJECT (self,
"peer requested small window ack size %" G_GUINT32_FORMAT,
window_ack_size);
}
g_mutex_lock (&self->stats_lock);
self->in_window_ack_size = window_ack_size;
g_mutex_unlock (&self->stats_lock);
}
static gboolean
is_command_response (const gchar * command_name)
{
return g_strcmp0 (command_name, "_result") == 0 ||
g_strcmp0 (command_name, "_error") == 0;
}
static void
gst_rtmp_connection_handle_cm (GstRtmpConnection * sc, GstBuffer * buffer)
{
GstRtmpMeta *meta;
gchar *command_name;
gdouble transaction_id;
GPtrArray *args;
meta = gst_buffer_get_rtmp_meta (buffer);
g_return_if_fail (meta);
{
GstMapInfo map;
gst_buffer_map (buffer, &map, GST_MAP_READ);
args = gst_amf_parse_command (map.data, map.size, &transaction_id,
&command_name);
gst_buffer_unmap (buffer, &map);
}
if (!args) {
return;
}
if (!isfinite (transaction_id) || transaction_id < 0 ||
transaction_id > G_MAXUINT) {
GST_WARNING_OBJECT (sc,
"Server sent command \"%s\" with extreme transaction ID %.0f",
GST_STR_NULL (command_name), transaction_id);
} else if (transaction_id > sc->transaction_count) {
GST_WARNING_OBJECT (sc,
"Server sent command \"%s\" with unused transaction ID (%.0f > %u)",
GST_STR_NULL (command_name), transaction_id, sc->transaction_count);
sc->transaction_count = transaction_id;
}
GST_DEBUG_OBJECT (sc,
"got control message \"%s\" transaction %.0f size %"
G_GUINT32_FORMAT, GST_STR_NULL (command_name), transaction_id,
meta->size);
if (is_command_response (command_name)) {
if (transaction_id != 0) {
GList *l;
for (l = sc->transactions; l; l = g_list_next (l)) {
Transaction *t = l->data;
if (t->transaction_id != transaction_id) {
continue;
}
GST_LOG_OBJECT (sc, "calling transaction callback %s",
GST_DEBUG_FUNCPTR_NAME (t->func));
sc->transactions = g_list_remove_link (sc->transactions, l);
t->func (command_name, args, t->user_data);
g_list_free_full (l, transaction_free);
break;
}
} else {
GST_WARNING_OBJECT (sc, "Server sent response \"%s\" without transaction",
GST_STR_NULL (command_name));
}
} else {
GList *l;
if (transaction_id != 0) {
GST_FIXME_OBJECT (sc, "Server sent command \"%s\" expecting reply",
GST_STR_NULL (command_name));
}
for (l = sc->expected_commands; l; l = g_list_next (l)) {
ExpectedCommand *ec = l->data;
if (ec->stream_id != meta->mstream) {
continue;
}
if (g_strcmp0 (ec->command_name, command_name)) {
continue;
}
GST_LOG_OBJECT (sc, "calling expected command callback %s",
GST_DEBUG_FUNCPTR_NAME (ec->func));
sc->expected_commands = g_list_remove_link (sc->expected_commands, l);
ec->func (command_name, args, ec->user_data);
g_list_free_full (l, expected_command_free);
break;
}
}
g_free (command_name);
g_ptr_array_unref (args);
}
static gboolean
start_write (gpointer user_data)
{
GstRtmpConnection *sc = user_data;
gst_rtmp_connection_start_write (sc);
return G_SOURCE_REMOVE;
}
void
gst_rtmp_connection_queue_message (GstRtmpConnection * self, GstBuffer * buffer)
{
g_return_if_fail (GST_IS_RTMP_CONNECTION (self));
g_return_if_fail (GST_IS_BUFFER (buffer));
g_async_queue_push (self->output_queue, buffer);
g_main_context_invoke_full (self->main_context, G_PRIORITY_DEFAULT,
start_write, g_object_ref (self), g_object_unref);
}
guint
gst_rtmp_connection_get_num_queued (GstRtmpConnection * connection)
{
return g_async_queue_length (connection->output_queue);
}
guint
gst_rtmp_connection_send_command (GstRtmpConnection * connection,
GstRtmpCommandCallback response_command, gpointer user_data,
guint32 stream_id, const gchar * command_name, const GstAmfNode * argument,
...)
{
GstBuffer *buffer;
gdouble transaction_id = 0;
va_list ap;
GBytes *payload;
guint8 *data;
gsize size;
g_return_val_if_fail (GST_IS_RTMP_CONNECTION (connection), 0);
if (connection->thread != g_thread_self ()) {
GST_ERROR_OBJECT (connection, "Called from wrong thread");
}
GST_DEBUG_OBJECT (connection,
"Sending command '%s' on stream id %" G_GUINT32_FORMAT,
command_name, stream_id);
if (response_command) {
Transaction *t;
transaction_id = ++connection->transaction_count;
GST_LOG_OBJECT (connection, "Registering %s for transid %.0f",
GST_DEBUG_FUNCPTR_NAME (response_command), transaction_id);
t = transaction_new (transaction_id, response_command, user_data);
connection->transactions = g_list_append (connection->transactions, t);
}
va_start (ap, argument);
payload = gst_amf_serialize_command_valist (transaction_id,
command_name, argument, ap);
va_end (ap);
data = g_bytes_unref_to_data (payload, &size);
buffer = gst_rtmp_message_new_wrapped (GST_RTMP_MESSAGE_TYPE_COMMAND_AMF0,
3, stream_id, data, size);
gst_rtmp_connection_queue_message (connection, buffer);
return transaction_id;
}
void
gst_rtmp_connection_expect_command (GstRtmpConnection * connection,
GstRtmpCommandCallback response_command, gpointer user_data,
guint32 stream_id, const gchar * command_name)
{
ExpectedCommand *ec;
g_return_if_fail (response_command);
g_return_if_fail (command_name);
g_return_if_fail (!is_command_response (command_name));
GST_LOG_OBJECT (connection,
"Registering %s for stream id %" G_GUINT32_FORMAT " name \"%s\"",
GST_DEBUG_FUNCPTR_NAME (response_command), stream_id, command_name);
ec = expected_command_new (stream_id, command_name, response_command,
user_data);
connection->expected_commands =
g_list_append (connection->expected_commands, ec);
}
static void
gst_rtmp_connection_send_ack (GstRtmpConnection * connection)
{
guint64 in_bytes_total = connection->in_bytes_total;
GstRtmpProtocolControl pc = {
.type = GST_RTMP_MESSAGE_TYPE_ACKNOWLEDGEMENT,
.param = (guint32) in_bytes_total,
};
gst_rtmp_connection_queue_message (connection,
gst_rtmp_message_new_protocol_control (&pc));
g_mutex_lock (&connection->stats_lock);
connection->in_bytes_acked = in_bytes_total;
g_mutex_unlock (&connection->stats_lock);
}
static void
gst_rtmp_connection_send_ping_response (GstRtmpConnection * connection,
guint32 event_data)
{
GstRtmpUserControl uc = {
.type = GST_RTMP_USER_CONTROL_TYPE_PING_RESPONSE,
.param = event_data,
};
gst_rtmp_connection_queue_message (connection,
gst_rtmp_message_new_user_control (&uc));
}
void
gst_rtmp_connection_set_chunk_size (GstRtmpConnection * connection,
guint32 chunk_size)
{
GstRtmpProtocolControl pc = {
.type = GST_RTMP_MESSAGE_TYPE_SET_CHUNK_SIZE,
.param = chunk_size,
};
g_return_if_fail (GST_IS_RTMP_CONNECTION (connection));
gst_rtmp_connection_queue_message (connection,
gst_rtmp_message_new_protocol_control (&pc));
}
void
gst_rtmp_connection_request_window_size (GstRtmpConnection * connection,
guint32 window_ack_size)
{
GstRtmpProtocolControl pc = {
.type = GST_RTMP_MESSAGE_TYPE_WINDOW_ACK_SIZE,
.param = window_ack_size,
};
g_return_if_fail (GST_IS_RTMP_CONNECTION (connection));
gst_rtmp_connection_queue_message (connection,
gst_rtmp_message_new_protocol_control (&pc));
}
void
gst_rtmp_connection_set_data_frame (GstRtmpConnection * connection,
GstBuffer * buffer)
{
g_return_if_fail (GST_IS_RTMP_CONNECTION (connection));
g_return_if_fail (GST_IS_BUFFER (buffer));
gst_buffer_prepend_memory (buffer, gst_memory_ref (set_data_frame_value));
gst_rtmp_connection_queue_message (connection, buffer);
}
static gboolean
gst_rtmp_connection_prepare_protocol_control (GstRtmpConnection * self,
GstBuffer * buffer)
{
GstRtmpProtocolControl pc;
if (!gst_rtmp_message_parse_protocol_control (buffer, &pc)) {
GST_ERROR_OBJECT (self, "can't parse protocol control message");
return FALSE;
}
switch (pc.type) {
case GST_RTMP_MESSAGE_TYPE_SET_CHUNK_SIZE:{
guint32 chunk_size = pc.param;
GST_INFO_OBJECT (self, "pending chunk size %" G_GUINT32_FORMAT,
chunk_size);
if (chunk_size < GST_RTMP_MINIMUM_CHUNK_SIZE) {
GST_ERROR_OBJECT (self,
"requested chunk size %" G_GUINT32_FORMAT " is too small",
chunk_size);
return FALSE;
}
if (chunk_size > GST_RTMP_MAXIMUM_CHUNK_SIZE) {
GST_ERROR_OBJECT (self,
"requested chunk size %" G_GUINT32_FORMAT " is too large",
chunk_size);
return FALSE;
}
if (chunk_size < GST_RTMP_DEFAULT_CHUNK_SIZE) {
GST_WARNING_OBJECT (self,
"requesting small chunk size %" G_GUINT32_FORMAT, chunk_size);
}
self->out_chunk_size_pending = pc.param;
break;
}
case GST_RTMP_MESSAGE_TYPE_WINDOW_ACK_SIZE:{
guint32 window_ack_size = pc.param;
GST_INFO_OBJECT (self, "pending window ack size: %" G_GUINT32_FORMAT,
window_ack_size);
if (window_ack_size < GST_RTMP_DEFAULT_WINDOW_ACK_SIZE) {
GST_WARNING_OBJECT (self,
"requesting small window ack size %" G_GUINT32_FORMAT,
window_ack_size);
}
self->out_window_ack_size_pending = window_ack_size;
break;
}
default:
break;
}
return TRUE;
}
static void
gst_rtmp_connection_apply_protocol_control (GstRtmpConnection * self)
{
guint32 chunk_size, window_ack_size;
chunk_size = self->out_chunk_size_pending;
if (chunk_size) {
self->out_chunk_size_pending = 0;
g_mutex_lock (&self->stats_lock);
self->out_chunk_size = chunk_size;
g_mutex_unlock (&self->stats_lock);
GST_INFO_OBJECT (self, "applied chunk size %" G_GUINT32_FORMAT, chunk_size);
}
window_ack_size = self->out_window_ack_size_pending;
if (window_ack_size) {
self->out_window_ack_size_pending = 0;
g_mutex_lock (&self->stats_lock);
self->out_window_ack_size = window_ack_size;
g_mutex_unlock (&self->stats_lock);
GST_INFO_OBJECT (self, "applied window ack size %" G_GUINT32_FORMAT,
window_ack_size);
}
}
static GstStructure *
get_stats (GstRtmpConnection * self)
{
return gst_structure_new ("GstRtmpConnectionStats",
"in-chunk-size", G_TYPE_UINT, self ? self->in_chunk_size : 0,
"out-chunk-size", G_TYPE_UINT, self ? self->out_chunk_size : 0,
"in-window-ack-size", G_TYPE_UINT, self ? self->in_window_ack_size : 0,
"out-window-ack-size", G_TYPE_UINT, self ? self->out_window_ack_size : 0,
"in-bytes-total", G_TYPE_UINT64, self ? self->in_bytes_total : 0,
"out-bytes-total", G_TYPE_UINT64, self ? self->out_bytes_total : 0,
"in-bytes-acked", G_TYPE_UINT64, self ? self->in_bytes_acked : 0,
"out-bytes-acked", G_TYPE_UINT64, self ? self->out_bytes_acked : 0, NULL);
}
GstStructure *
gst_rtmp_connection_get_null_stats (void)
{
return get_stats (NULL);
}
GstStructure *
gst_rtmp_connection_get_stats (GstRtmpConnection * self)
{
GstStructure *s;
g_return_val_if_fail (GST_IS_RTMP_CONNECTION (self), NULL);
g_mutex_lock (&self->stats_lock);
s = get_stats (self);
g_mutex_unlock (&self->stats_lock);
return s;
}