gstreamer/gst/rtmp2/rtmp/rtmpconnection.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;

  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;

  /* RTMP configuration */
  gsize in_chunk_size;
  gsize out_chunk_size;
  gsize in_window_ack_size;
  gsize out_window_ack_size;
  gsize total_input_bytes;
  gsize bytes_since_ack;
};


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_emit_error (GstRtmpConnection * self);
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_bytes_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_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_send_ack (GstRtmpConnection * connection);
static void
gst_rtmp_connection_send_ping_response (GstRtmpConnection * connection,
    guint32 event_data);

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_slice_new (Transaction);
  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_slice_free (Transaction, 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_slice_new (ExpectedCommand);
  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_slice_free (ExpectedCommand, 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_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, 0);

  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) g_bytes_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;
}

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);

  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_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);
}

GstRtmpConnection *
gst_rtmp_connection_new (GSocketConnection * connection)
{
  GstRtmpConnection *sc;

  sc = g_object_new (GST_TYPE_RTMP_CONNECTION, NULL);

  gst_rtmp_connection_set_socket_connection (sc, connection);

  return sc;
}

static void
cancel_all_commands (GstRtmpConnection * self)
{
  GList *l;

  for (l = self->transactions; l; l = g_list_next (l)) {
    Transaction *cc = l->data;
    GST_LOG ("calling transaction callback %s",
        GST_DEBUG_FUNCPTR_NAME (cc->func));
    cc->func ("<cancelled>", 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 ("calling expected command callback %s",
        GST_DEBUG_FUNCPTR_NAME (cc->func));
    cc->func ("<cancelled>", 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 ("Called from wrong thread");
  }

  g_cancellable_cancel (self->cancellable);
  cancel_all_commands (self);

  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;

  GST_TRACE ("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) {
    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 ("read IO error %d %s, continuing", code, error->message);
      g_error_free (error);
      return G_SOURCE_CONTINUE;
    }

    GST_ERROR ("read error: %s %d %s", g_quark_to_string (error->domain),
        code, error->message);
    g_error_free (error);
  } else if (ret == 0) {
    GST_INFO ("read EOF");
  }

  if (ret <= 0) {
    gst_rtmp_connection_emit_error (sc);
    return G_SOURCE_REMOVE;
  }

  GST_TRACE ("read %" G_GSIZE_FORMAT " bytes", ret);

  sc->total_input_bytes += ret;
  sc->bytes_since_ack += ret;
  if (sc->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;
  GBytes *bytes;

  if (self->writing) {
    return;
  }

  bytes = g_async_queue_try_pop (self->output_queue);
  if (!bytes) {
    return;
  }

  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_bytes_async (os, bytes,
      G_PRIORITY_DEFAULT, self->cancellable,
      gst_rtmp_connection_write_bytes_done, g_object_ref (self));
  g_bytes_unref (bytes);
}

static void
gst_rtmp_connection_emit_error (GstRtmpConnection * self)
{
  if (self->error) {
    return;
  }

  GST_INFO ("connection error");
  self->error = TRUE;

  cancel_all_commands (self);

  g_signal_emit (self, signals[SIGNAL_ERROR], 0);
}

static void
gst_rtmp_connection_write_bytes_done (GObject * obj,
    GAsyncResult * result, gpointer user_data)
{
  GOutputStream *os = G_OUTPUT_STREAM (obj);
  GstRtmpConnection *self = GST_RTMP_CONNECTION (user_data);
  GError *error = NULL;
  gboolean res;

  self->writing = FALSE;

  res = gst_rtmp_output_stream_write_all_bytes_finish (os, result, &error);
  if (!res) {
    if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
      GST_INFO ("write cancelled");
    } else {
      GST_ERROR ("write error: %s", error->message);
    }
    gst_rtmp_connection_emit_error (self);
    g_error_free (error);
    g_object_unref (self);
    return;
  }

  GST_LOG ("write completed");
  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 ("got %u < %u bytes, need more", len, need);
    return;
  }

  GST_TRACE ("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;

    default:
      if (sc->input_handler) {
        sc->input_handler (sc, buffer, sc->input_handler_user_data);
      }
      return;
  }
}

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 ("can't parse protocol control message");
    return;
  }

  GST_LOG ("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 ("new chunk size %" G_GUINT32_FORMAT, pc.param);
      connection->in_chunk_size = pc.param;
      break;

    case GST_RTMP_MESSAGE_TYPE_ABORT_MESSAGE:
      GST_ERROR ("unimplemented: chunk abort, stream_id = %" G_GUINT32_FORMAT,
          pc.param);
      break;

    case GST_RTMP_MESSAGE_TYPE_ACKNOWLEDGEMENT:
      /* We don't really send ack requests that we care about, so ignore */
      GST_DEBUG ("acknowledgement %" G_GUINT32_FORMAT, pc.param);
      break;

    case GST_RTMP_MESSAGE_TYPE_WINDOW_ACK_SIZE:
      GST_INFO ("window ack size: %" G_GUINT32_FORMAT, pc.param);
      connection->in_window_ack_size = pc.param;
      break;

    case GST_RTMP_MESSAGE_TYPE_SET_PEER_BANDWIDTH:
      GST_FIXME ("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 ("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 ("can't parse user control message");
    return;
  }

  GST_LOG ("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 ("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 ("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 ("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 ("ignoring ping response: %" G_GUINT32_FORMAT, uc.param);
      break;

    case GST_RTMP_USER_CONTROL_TYPE_BUFFER_EMPTY:
      GST_LOG ("ignoring buffer empty: %" G_GUINT32_FORMAT, uc.param);
      break;

    case GST_RTMP_USER_CONTROL_TYPE_BUFFER_READY:
      GST_LOG ("ignoring buffer ready: %" G_GUINT32_FORMAT, uc.param);
      break;

    default:
      GST_ERROR ("unimplemented user control type %d:%s", uc.type,
          gst_rtmp_user_control_type_get_nick (uc.type));
      break;
  }
}

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 ("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 ("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 ("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 ("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 ("Server sent response \"%s\" without transaction",
          GST_STR_NULL (command_name));
    }
  } else {
    GList *l;

    if (transaction_id != 0) {
      GST_FIXME ("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 ("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);
  g_object_unref (sc);
  return G_SOURCE_REMOVE;
}

static void
byte_array_take_buffer (GByteArray * byte_array, GstBuffer * buffer)
{
  GstMapInfo map;
  gboolean ret;
  ret = gst_buffer_map (buffer, &map, GST_MAP_READ);
  g_assert (ret);
  g_assert (byte_array->len + map.size <= (guint64) G_MAXUINT);
  g_byte_array_append (byte_array, map.data, map.size);
  gst_buffer_unmap (buffer, &map);
  gst_buffer_unref (buffer);
}

void
gst_rtmp_connection_queue_message (GstRtmpConnection * self, GstBuffer * buffer)
{
  GstRtmpMeta *meta;
  GstRtmpChunkStream *cstream;
  GstBuffer *out_buffer;
  GByteArray *out_ba;

  g_return_if_fail (GST_IS_RTMP_CONNECTION (self));
  g_return_if_fail (GST_IS_BUFFER (buffer));

  meta = gst_buffer_get_rtmp_meta (buffer);
  g_return_if_fail (meta);

  cstream = gst_rtmp_chunk_streams_get (self->output_streams, meta->cstream);
  g_return_if_fail (cstream);

  out_buffer = gst_rtmp_chunk_stream_serialize_start (cstream, buffer,
      self->out_chunk_size);
  g_return_if_fail (out_buffer);

  out_ba = g_byte_array_new ();

  while (out_buffer) {
    byte_array_take_buffer (out_ba, out_buffer);

    out_buffer = gst_rtmp_chunk_stream_serialize_next (cstream,
        self->out_chunk_size);
  }

  g_async_queue_push (self->output_queue, g_byte_array_free_to_bytes (out_ba));
  g_main_context_invoke (self->main_context, start_write, g_object_ref (self));
}

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;

  if (connection->thread != g_thread_self ()) {
    GST_ERROR ("Called from wrong thread");
  }

  GST_DEBUG ("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 ("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 ("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)
{
  GstRtmpProtocolControl pc = {
    .type = GST_RTMP_MESSAGE_TYPE_ACKNOWLEDGEMENT,
    .param = (guint32) connection->total_input_bytes,
  };

  gst_rtmp_connection_queue_message (connection,
      gst_rtmp_message_new_protocol_control (&pc));

  connection->bytes_since_ack = 0;
}

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_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));

  if (connection->out_window_ack_size == window_ack_size)
    return;

  connection->out_window_ack_size = window_ack_size;

  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);
}