rtsp-client: Avoid reuse of channel numbers for interleaved
[platform/upstream/gstreamer.git] / gst / rtsp-server / rtsp-client.c
index e46f587..2e124f8 100644 (file)
 #include "rtsp-sdp.h"
 #include "rtsp-params.h"
 
-#define GST_RTSP_CLIENT_GET_PRIVATE(obj)  \
-   (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTSP_CLIENT, GstRTSPClientPrivate))
+typedef enum
+{
+  TUNNEL_STATE_UNKNOWN,
+  TUNNEL_STATE_GET,
+  TUNNEL_STATE_POST
+} GstRTSPTunnelState;
 
 /* locking order:
  * send_lock, lock, tunnels_lock
@@ -66,13 +70,15 @@ struct _GstRTSPClientPrivate
   GstRTSPConnection *connection;
   GstRTSPWatch *watch;
   GMainContext *watch_context;
-  guint close_seq;
   gchar *server_ip;
   gboolean is_ipv6;
 
-  GstRTSPClientSendFunc send_func;      /* protected by send_lock */
-  gpointer send_data;           /* protected by send_lock */
-  GDestroyNotify send_notify;   /* protected by send_lock */
+  /* protected by send_lock */
+  GstRTSPClientSendFunc send_func;
+  gpointer send_data;
+  GDestroyNotify send_notify;
+  guint close_seq;
+  GArray *data_seqs;
 
   GstRTSPSessionPool *session_pool;
   gulong session_removed_id;
@@ -98,13 +104,18 @@ struct _GstRTSPClientPrivate
   GstRTSPVersion version;
 
   GHashTable *pipelined_requests;       /* pipelined_request_id -> session_id */
+  GstRTSPTunnelState tstate;
 };
 
+typedef struct
+{
+  guint8 channel;
+  guint seq;
+} DataSeq;
+
 static GMutex tunnels_lock;
 static GHashTable *tunnels;     /* protected by tunnels_lock */
 
-/* FIXME make this configurable. We don't want to do this yet because it will
- * be superceeded by a cache object later */
 #define WATCH_BACKLOG_SIZE              100
 
 #define DEFAULT_SESSION_POOL            NULL
@@ -184,15 +195,13 @@ static GstRTSPStatusCode default_pre_signal_handler (GstRTSPClient * client,
 static gboolean pre_signal_accumulator (GSignalInvocationHint * ihint,
     GValue * return_accu, const GValue * handler_return, gpointer data);
 
-G_DEFINE_TYPE (GstRTSPClient, gst_rtsp_client, G_TYPE_OBJECT);
+G_DEFINE_TYPE_WITH_PRIVATE (GstRTSPClient, gst_rtsp_client, G_TYPE_OBJECT);
 
 static void
 gst_rtsp_client_class_init (GstRTSPClientClass * klass)
 {
   GObjectClass *gobject_class;
 
-  g_type_class_add_private (klass, sizeof (GstRTSPClientPrivate));
-
   gobject_class = G_OBJECT_CLASS (klass);
 
   gobject_class->get_property = gst_rtsp_client_get_property;
@@ -248,7 +257,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-options-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -262,6 +271,11 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic, GST_TYPE_RTSP_STATUS_CODE, 1,
       GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::options-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_OPTIONS_REQUEST] =
       g_signal_new ("options-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass, options_request),
@@ -271,7 +285,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-describe-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -285,6 +299,11 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic, GST_TYPE_RTSP_STATUS_CODE, 1,
       GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::describe-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_DESCRIBE_REQUEST] =
       g_signal_new ("describe-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass, describe_request),
@@ -294,7 +313,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-setup-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -308,6 +327,11 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic, GST_TYPE_RTSP_STATUS_CODE, 1,
       GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::setup-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_SETUP_REQUEST] =
       g_signal_new ("setup-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass, setup_request),
@@ -317,7 +341,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-play-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -331,6 +355,11 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic, GST_TYPE_RTSP_STATUS_CODE, 1,
       GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::play-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_PLAY_REQUEST] =
       g_signal_new ("play-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass, play_request),
@@ -340,7 +369,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-pause-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -354,6 +383,11 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic, GST_TYPE_RTSP_STATUS_CODE, 1,
       GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::pause-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_PAUSE_REQUEST] =
       g_signal_new ("pause-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass, pause_request),
@@ -363,7 +397,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-teardown-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -377,6 +411,11 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic, GST_TYPE_RTSP_STATUS_CODE, 1,
       GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::teardown-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_TEARDOWN_REQUEST] =
       g_signal_new ("teardown-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass, teardown_request),
@@ -386,7 +425,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-set-parameter-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -400,6 +439,11 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic,
       GST_TYPE_RTSP_STATUS_CODE, 1, GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::set-parameter-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_SET_PARAMETER_REQUEST] =
       g_signal_new ("set-parameter-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass,
@@ -409,7 +453,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-get-parameter-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -423,12 +467,22 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic, GST_TYPE_RTSP_STATUS_CODE, 1,
       GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::get-parameter-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_GET_PARAMETER_REQUEST] =
       g_signal_new ("get-parameter-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass,
           get_parameter_request), NULL, NULL, g_cclosure_marshal_generic,
       G_TYPE_NONE, 1, GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::handle-response:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_HANDLE_RESPONSE] =
       g_signal_new ("handle-response", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass,
@@ -450,7 +504,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-announce-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -464,6 +518,11 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic, GST_TYPE_RTSP_STATUS_CODE, 1,
       GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::announce-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_ANNOUNCE_REQUEST] =
       g_signal_new ("announce-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass, announce_request),
@@ -473,7 +532,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::pre-record-request:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    *
    * Returns: a #GstRTSPStatusCode, GST_RTSP_STS_OK in case of success,
    *          otherwise an appropriate return code
@@ -487,6 +546,11 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
       g_cclosure_marshal_generic, GST_TYPE_RTSP_STATUS_CODE, 1,
       GST_TYPE_RTSP_CONTEXT);
 
+  /**
+   * GstRTSPClient::record-request:
+   * @client: a #GstRTSPClient
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
+   */
   gst_rtsp_client_signals[SIGNAL_RECORD_REQUEST] =
       g_signal_new ("record-request", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPClientClass, record_request),
@@ -496,7 +560,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
   /**
    * GstRTSPClient::check-requirements:
    * @client: a #GstRTSPClient
-   * @ctx: a #GstRTSPContext
+   * @ctx: (type GstRtspServer.RTSPContext): a #GstRTSPContext
    * @arr: a NULL-terminated array of strings
    *
    * Returns: a newly allocated string with comma-separated list of
@@ -521,7 +585,7 @@ gst_rtsp_client_class_init (GstRTSPClientClass * klass)
 static void
 gst_rtsp_client_init (GstRTSPClient * client)
 {
-  GstRTSPClientPrivate *priv = GST_RTSP_CLIENT_GET_PRIVATE (client);
+  GstRTSPClientPrivate *priv = gst_rtsp_client_get_instance_private (client);
 
   client->priv = priv;
 
@@ -529,12 +593,14 @@ gst_rtsp_client_init (GstRTSPClient * client)
   g_mutex_init (&priv->send_lock);
   g_mutex_init (&priv->watch_lock);
   priv->close_seq = 0;
+  priv->data_seqs = g_array_new (FALSE, FALSE, sizeof (DataSeq));
   priv->drop_backlog = DEFAULT_DROP_BACKLOG;
   priv->transports =
       g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL,
       g_object_unref);
   priv->pipelined_requests = g_hash_table_new_full (g_str_hash,
       g_str_equal, g_free, g_free);
+  priv->tstate = TUNNEL_STATE_UNKNOWN;
 }
 
 static GstRTSPFilterResult
@@ -695,6 +761,7 @@ gst_rtsp_client_finalize (GObject * obj)
   g_assert (priv->sessions == NULL);
   g_assert (priv->session_removed_id == 0);
 
+  g_array_unref (priv->data_seqs);
   g_hash_table_unref (priv->transports);
   g_hash_table_unref (priv->pipelined_requests);
 
@@ -996,12 +1063,88 @@ no_prepare:
   }
 }
 
+static inline DataSeq *
+get_data_seq_element (GstRTSPClient * client, guint8 channel)
+{
+  GstRTSPClientPrivate *priv = client->priv;
+  GArray *data_seqs = priv->data_seqs;
+  gint i = 0;
+
+  while (i < data_seqs->len) {
+    DataSeq *data_seq = &g_array_index (data_seqs, DataSeq, i);
+    if (data_seq->channel == channel)
+      return data_seq;
+    i++;
+  }
+
+  return NULL;
+}
+
+static void
+add_data_seq (GstRTSPClient * client, guint8 channel)
+{
+  GstRTSPClientPrivate *priv = client->priv;
+  DataSeq data_seq = {.channel = channel,.seq = 0 };
+
+  if (get_data_seq_element (client, channel) == NULL)
+    g_array_append_val (priv->data_seqs, data_seq);
+}
+
+static void
+set_data_seq (GstRTSPClient * client, guint8 channel, guint seq)
+{
+  DataSeq *data_seq;
+
+  data_seq = get_data_seq_element (client, channel);
+  g_assert_nonnull (data_seq);
+  data_seq->seq = seq;
+}
+
+static guint
+get_data_seq (GstRTSPClient * client, guint8 channel)
+{
+  DataSeq *data_seq;
+
+  data_seq = get_data_seq_element (client, channel);
+  g_assert_nonnull (data_seq);
+  return data_seq->seq;
+}
+
+static gboolean
+get_data_channel (GstRTSPClient * client, guint seq, guint8 * channel)
+{
+  GstRTSPClientPrivate *priv = client->priv;
+  GArray *data_seqs = priv->data_seqs;
+  gint i = 0;
+
+  while (i < data_seqs->len) {
+    DataSeq *data_seq = &g_array_index (data_seqs, DataSeq, i);
+    if (data_seq->seq == seq) {
+      *channel = data_seq->channel;
+      return TRUE;
+    }
+    i++;
+  }
+
+  return FALSE;
+}
+
+static gboolean
+do_close (gpointer user_data)
+{
+  GstRTSPClient *client = user_data;
+
+  gst_rtsp_client_close (client);
+
+  return G_SOURCE_REMOVE;
+}
+
 static gboolean
 do_send_data (GstBuffer * buffer, guint8 channel, GstRTSPClient * client)
 {
   GstRTSPClientPrivate *priv = client->priv;
   GstRTSPMessage message = { 0 };
-  GstRTSPResult res = GST_RTSP_OK;
+  gboolean ret = TRUE;
   GstMapInfo map_info;
   guint8 *data;
   guint usize;
@@ -1015,8 +1158,13 @@ do_send_data (GstBuffer * buffer, guint8 channel, GstRTSPClient * client)
   gst_rtsp_message_take_body (&message, map_info.data, map_info.size);
 
   g_mutex_lock (&priv->send_lock);
+  if (get_data_seq (client, channel) != 0) {
+    GST_WARNING ("already a queued data message for channel %d", channel);
+    g_mutex_unlock (&priv->send_lock);
+    return FALSE;
+  }
   if (priv->send_func)
-    res = priv->send_func (client, &message, FALSE, priv->send_data);
+    ret = priv->send_func (client, &message, FALSE, priv->send_data);
   g_mutex_unlock (&priv->send_lock);
 
   gst_rtsp_message_steal_body (&message, &data, &usize);
@@ -1024,7 +1172,17 @@ do_send_data (GstBuffer * buffer, guint8 channel, GstRTSPClient * client)
 
   gst_rtsp_message_unset (&message);
 
-  return res == GST_RTSP_OK;
+  if (!ret) {
+    GSource *idle_src;
+
+    /* close in watch context */
+    idle_src = g_idle_source_new ();
+    g_source_set_callback (idle_src, do_close, client, NULL);
+    g_source_attach (idle_src, priv->watch_context);
+    g_source_unref (idle_src);
+  }
+
+  return ret;
 }
 
 /**
@@ -1729,6 +1887,7 @@ parse_transport (const char *transport, GstRTSPStream * stream,
 
   /* loop through the transports, try to parse */
   for (i = 0; transports[i]; i++) {
+    g_strstrip (transports[i]);
     res = gst_rtsp_transport_parse (transports[i], tr);
     if (res != GST_RTSP_OK) {
       /* no valid transport, search some more */
@@ -1762,6 +1921,9 @@ default_configure_client_media (GstRTSPClient * client, GstRTSPMedia * media,
   GstRTSPMessage *request = ctx->request;
   gchar *blocksize_str;
 
+  if (!gst_rtsp_stream_is_sender (stream))
+    return TRUE;
+
   if (gst_rtsp_message_get_header (request, GST_RTSP_HDR_BLOCKSIZE,
           &blocksize_str, 0) == GST_RTSP_OK) {
     guint64 blocksize;
@@ -1802,33 +1964,64 @@ default_configure_client_transport (GstRTSPClient * client,
   /* we have a valid transport now, set the destination of the client. */
   if (ct->lower_transport == GST_RTSP_LOWER_TRANS_UDP_MCAST ||
       ct->lower_transport == GST_RTSP_LOWER_TRANS_UDP) {
-
     /* allocate UDP ports */
     GSocketFamily family;
     gboolean use_client_settings = FALSE;
 
     family = priv->is_ipv6 ? G_SOCKET_FAMILY_IPV6 : G_SOCKET_FAMILY_IPV4;
+
     if ((ct->lower_transport == GST_RTSP_LOWER_TRANS_UDP_MCAST) &&
         gst_rtsp_auth_check (GST_RTSP_AUTH_CHECK_TRANSPORT_CLIENT_SETTINGS) &&
-        (ct->destination != NULL))
+        (ct->destination != NULL)) {
+
+      if (!gst_rtsp_stream_verify_mcast_ttl (ctx->stream, ct->ttl))
+        goto error_ttl;
+
       use_client_settings = TRUE;
+    }
 
-    if (!gst_rtsp_stream_allocate_udp_sockets (ctx->stream, family, ct,
-            use_client_settings))
+    /* We need to allocate the sockets for both families before starting
+     * multiudpsink, otherwise multiudpsink won't accept new clients with
+     * a different family.
+     */
+    /* FIXME: could be more adequately solved by making it possible
+     * to set a socket on multiudpsink after it has already been started */
+    if (!gst_rtsp_stream_allocate_udp_sockets (ctx->stream,
+            G_SOCKET_FAMILY_IPV4, ct, use_client_settings)
+        && family == G_SOCKET_FAMILY_IPV4)
       goto error_allocating_ports;
 
-    if (ct->lower_transport == GST_RTSP_LOWER_TRANS_UDP_MCAST) {
-      GstRTSPAddress *addr = NULL;
+    if (!gst_rtsp_stream_allocate_udp_sockets (ctx->stream,
+            G_SOCKET_FAMILY_IPV6, ct, use_client_settings)
+        && family == G_SOCKET_FAMILY_IPV6)
+      goto error_allocating_ports;
 
+    if (ct->lower_transport == GST_RTSP_LOWER_TRANS_UDP_MCAST) {
       if (use_client_settings) {
-        /* the address has been successfully allocated, let's check if it's
-         * the one requested by the client */
-        addr = gst_rtsp_stream_reserve_address (ctx->stream, ct->destination,
-            ct->port.min, ct->port.max - ct->port.min + 1, ct->ttl);
+        /* FIXME: the address has been successfully allocated, however, in
+         * the use_client_settings case we need to verify that the allocated
+         * address is the one requested by the client and if this address is
+         * an allowed destination. Verifying this via the address pool in not
+         * the proper way as the address pool should only be used for choosing
+         * the server-selected address/port pairs. */
+        GSocket *rtp_socket;
+        guint ttl;
+
+        rtp_socket =
+            gst_rtsp_stream_get_rtp_multicast_socket (ctx->stream, family);
+        if (rtp_socket == NULL)
+          goto no_socket;
+        ttl = g_socket_get_multicast_ttl (rtp_socket);
+        g_object_unref (rtp_socket);
+        if (ct->ttl < ttl) {
+          /* use the maximum ttl that is requested by multicast clients */
+          GST_DEBUG ("requested ttl %u, but keeping ttl %u", ct->ttl, ttl);
+          ct->ttl = ttl;
+        }
 
-        if (addr == NULL)
-          goto no_address;
       } else {
+        GstRTSPAddress *addr = NULL;
+
         g_free (ct->destination);
         addr = gst_rtsp_stream_get_multicast_address (ctx->stream, family);
         if (addr == NULL)
@@ -1837,9 +2030,13 @@ default_configure_client_transport (GstRTSPClient * client,
         ct->port.min = addr->port;
         ct->port.max = addr->port + addr->n_ports - 1;
         ct->ttl = addr->ttl;
-
         gst_rtsp_address_free (addr);
       }
+
+      if (!gst_rtsp_stream_add_multicast_client_address (ctx->stream,
+              ct->destination, ct->port.min, ct->port.max, family))
+        goto error_mcast_transport;
+
     } else {
       GstRTSPUrl *url;
 
@@ -1887,11 +2084,27 @@ default_configure_client_transport (GstRTSPClient * client,
         gst_rtsp_session_media_alloc_channels (ctx->sessmedia,
             &ct->interleaved);
       }
+      /* alloc new channels if they are already taken */
+      while (g_hash_table_contains (priv->transports,
+              GINT_TO_POINTER (ct->interleaved.min))
+          || g_hash_table_contains (priv->transports,
+              GINT_TO_POINTER (ct->interleaved.max))) {
+        gst_rtsp_session_media_alloc_channels (ctx->sessmedia,
+            &ct->interleaved);
+        if (ct->interleaved.max > 255)
+          goto error_allocating_channels;
+      }
     }
   }
   return TRUE;
 
   /* ERRORS */
+error_ttl:
+  {
+    GST_ERROR_OBJECT (client,
+        "Failed to allocate UDP ports: invalid ttl value");
+    return FALSE;
+  }
 error_allocating_ports:
   {
     GST_ERROR_OBJECT (client, "Failed to allocate UDP ports");
@@ -1902,6 +2115,21 @@ no_address:
     GST_ERROR_OBJECT (client, "Failed to acquire address for stream");
     return FALSE;
   }
+no_socket:
+  {
+    GST_ERROR_OBJECT (client, "Failed to get UDP socket");
+    return FALSE;
+  }
+error_mcast_transport:
+  {
+    GST_ERROR_OBJECT (client, "Failed to add multicast client transport");
+    return FALSE;
+  }
+error_allocating_channels:
+  {
+    GST_ERROR_OBJECT (client, "Failed to allocate interleaved channels");
+    return FALSE;
+  }
 }
 
 static GstRTSPTransport *
@@ -1957,267 +2185,6 @@ make_server_transport (GstRTSPClient * client, GstRTSPMedia * media,
   return st;
 }
 
-#define AES_128_KEY_LEN 16
-#define AES_256_KEY_LEN 32
-
-#define HMAC_32_KEY_LEN 4
-#define HMAC_80_KEY_LEN 10
-
-static gboolean
-mikey_apply_policy (GstCaps * caps, GstMIKEYMessage * msg, guint8 policy)
-{
-  const gchar *srtp_cipher;
-  const gchar *srtp_auth;
-  const GstMIKEYPayload *sp;
-  guint i;
-
-  /* loop over Security policy until we find one containing policy */
-  for (i = 0;; i++) {
-    if ((sp = gst_mikey_message_find_payload (msg, GST_MIKEY_PT_SP, i)) == NULL)
-      break;
-
-    if (((GstMIKEYPayloadSP *) sp)->policy == policy)
-      break;
-  }
-
-  /* the default ciphers */
-  srtp_cipher = "aes-128-icm";
-  srtp_auth = "hmac-sha1-80";
-
-  /* now override the defaults with what is in the Security Policy */
-  if (sp != NULL) {
-    guint len;
-
-    /* collect all the params and go over them */
-    len = gst_mikey_payload_sp_get_n_params (sp);
-    for (i = 0; i < len; i++) {
-      const GstMIKEYPayloadSPParam *param =
-          gst_mikey_payload_sp_get_param (sp, i);
-
-      switch (param->type) {
-        case GST_MIKEY_SP_SRTP_ENC_ALG:
-          switch (param->val[0]) {
-            case 0:
-              srtp_cipher = "null";
-              break;
-            case 2:
-            case 1:
-              srtp_cipher = "aes-128-icm";
-              break;
-            default:
-              break;
-          }
-          break;
-        case GST_MIKEY_SP_SRTP_ENC_KEY_LEN:
-          switch (param->val[0]) {
-            case AES_128_KEY_LEN:
-              srtp_cipher = "aes-128-icm";
-              break;
-            case AES_256_KEY_LEN:
-              srtp_cipher = "aes-256-icm";
-              break;
-            default:
-              break;
-          }
-          break;
-        case GST_MIKEY_SP_SRTP_AUTH_ALG:
-          switch (param->val[0]) {
-            case 0:
-              srtp_auth = "null";
-              break;
-            case 2:
-            case 1:
-              srtp_auth = "hmac-sha1-80";
-              break;
-            default:
-              break;
-          }
-          break;
-        case GST_MIKEY_SP_SRTP_AUTH_KEY_LEN:
-          switch (param->val[0]) {
-            case HMAC_32_KEY_LEN:
-              srtp_auth = "hmac-sha1-32";
-              break;
-            case HMAC_80_KEY_LEN:
-              srtp_auth = "hmac-sha1-80";
-              break;
-            default:
-              break;
-          }
-          break;
-        case GST_MIKEY_SP_SRTP_SRTP_ENC:
-          break;
-        case GST_MIKEY_SP_SRTP_SRTCP_ENC:
-          break;
-        default:
-          break;
-      }
-    }
-  }
-  /* now configure the SRTP parameters */
-  gst_caps_set_simple (caps,
-      "srtp-cipher", G_TYPE_STRING, srtp_cipher,
-      "srtp-auth", G_TYPE_STRING, srtp_auth,
-      "srtcp-cipher", G_TYPE_STRING, srtp_cipher,
-      "srtcp-auth", G_TYPE_STRING, srtp_auth, NULL);
-
-  return TRUE;
-}
-
-static gboolean
-handle_mikey_data (GstRTSPClient * client, GstRTSPContext * ctx,
-    guint8 * data, gsize size)
-{
-  GstMIKEYMessage *msg;
-  guint i, n_cs;
-  GstCaps *caps = NULL;
-  GstMIKEYPayloadKEMAC *kemac;
-  const GstMIKEYPayloadKeyData *pkd;
-  GstBuffer *key;
-
-  /* the MIKEY message contains a CSB or crypto session bundle. It is a
-   * set of Crypto Sessions protected with the same master key.
-   * In the context of SRTP, an RTP and its RTCP stream is part of a
-   * crypto session */
-  if ((msg = gst_mikey_message_new_from_data (data, size, NULL, NULL)) == NULL)
-    goto parse_failed;
-
-  /* we can only handle SRTP crypto sessions for now */
-  if (msg->map_type != GST_MIKEY_MAP_TYPE_SRTP)
-    goto invalid_map_type;
-
-  /* get the number of crypto sessions. This maps SSRC to its
-   * security parameters */
-  n_cs = gst_mikey_message_get_n_cs (msg);
-  if (n_cs == 0)
-    goto no_crypto_sessions;
-
-  /* we also need keys */
-  if (!(kemac = (GstMIKEYPayloadKEMAC *) gst_mikey_message_find_payload
-          (msg, GST_MIKEY_PT_KEMAC, 0)))
-    goto no_keys;
-
-  /* we don't support encrypted keys */
-  if (kemac->enc_alg != GST_MIKEY_ENC_NULL
-      || kemac->mac_alg != GST_MIKEY_MAC_NULL)
-    goto unsupported_encryption;
-
-  /* get Key data sub-payload */
-  pkd = (const GstMIKEYPayloadKeyData *)
-      gst_mikey_payload_kemac_get_sub (&kemac->pt, 0);
-
-  key =
-      gst_buffer_new_wrapped (g_memdup (pkd->key_data, pkd->key_len),
-      pkd->key_len);
-
-  /* go over all crypto sessions and create the security policy for each
-   * SSRC */
-  for (i = 0; i < n_cs; i++) {
-    const GstMIKEYMapSRTP *map = gst_mikey_message_get_cs_srtp (msg, i);
-
-    caps = gst_caps_new_simple ("application/x-srtp",
-        "ssrc", G_TYPE_UINT, map->ssrc,
-        "roc", G_TYPE_UINT, map->roc, "srtp-key", GST_TYPE_BUFFER, key, NULL);
-    mikey_apply_policy (caps, msg, map->policy);
-
-    gst_rtsp_stream_update_crypto (ctx->stream, map->ssrc, caps);
-    gst_caps_unref (caps);
-  }
-  gst_mikey_message_unref (msg);
-  gst_buffer_unref (key);
-
-  return TRUE;
-
-  /* ERRORS */
-parse_failed:
-  {
-    GST_DEBUG_OBJECT (client, "failed to parse MIKEY message");
-    return FALSE;
-  }
-invalid_map_type:
-  {
-    GST_DEBUG_OBJECT (client, "invalid map type %d", msg->map_type);
-    goto cleanup_message;
-  }
-no_crypto_sessions:
-  {
-    GST_DEBUG_OBJECT (client, "no crypto sessions");
-    goto cleanup_message;
-  }
-no_keys:
-  {
-    GST_DEBUG_OBJECT (client, "no keys found");
-    goto cleanup_message;
-  }
-unsupported_encryption:
-  {
-    GST_DEBUG_OBJECT (client, "unsupported key encryption");
-    goto cleanup_message;
-  }
-cleanup_message:
-  {
-    gst_mikey_message_unref (msg);
-    return FALSE;
-  }
-}
-
-#define IS_STRIP_CHAR(c) (g_ascii_isspace ((guchar)(c)) || ((c) == '\"'))
-
-static void
-strip_chars (gchar * str)
-{
-  gchar *s;
-  gsize len;
-
-  len = strlen (str);
-  while (len--) {
-    if (!IS_STRIP_CHAR (str[len]))
-      break;
-    str[len] = '\0';
-  }
-  for (s = str; *s && IS_STRIP_CHAR (*s); s++);
-  memmove (str, s, len + 1);
-}
-
-/* KeyMgmt = "KeyMgmt" ":" key-mgmt-spec 0*("," key-mgmt-spec)
- * key-mgmt-spec = "prot" "=" KMPID ";" ["uri" "=" %x22 URI %x22 ";"]
- */
-static gboolean
-handle_keymgmt (GstRTSPClient * client, GstRTSPContext * ctx, gchar * keymgmt)
-{
-  gchar **specs;
-  gint i, j;
-
-  specs = g_strsplit (keymgmt, ",", 0);
-  for (i = 0; specs[i]; i++) {
-    gchar **split;
-
-    split = g_strsplit (specs[i], ";", 0);
-    for (j = 0; split[j]; j++) {
-      g_strstrip (split[j]);
-      if (g_str_has_prefix (split[j], "prot=")) {
-        g_strstrip (split[j] + 5);
-        if (!g_str_equal (split[j] + 5, "mikey"))
-          break;
-        GST_DEBUG ("found mikey");
-      } else if (g_str_has_prefix (split[j], "uri=")) {
-        strip_chars (split[j] + 4);
-        GST_DEBUG ("found uri '%s'", split[j] + 4);
-      } else if (g_str_has_prefix (split[j], "data=")) {
-        guchar *data;
-        gsize size;
-        strip_chars (split[j] + 5);
-        GST_DEBUG ("found data '%s'", split[j] + 5);
-        data = g_base64_decode_inplace (split[j] + 5, &size);
-        handle_mikey_data (client, ctx, data, size);
-      }
-    }
-    g_strfreev (split);
-  }
-  g_strfreev (specs);
-  return TRUE;
-}
-
 static gboolean
 rtsp_ctrl_timeout_cb (gpointer user_data)
 {
@@ -2259,6 +2226,76 @@ rtsp_ctrl_timeout_remove (GstRTSPClientPrivate * priv)
   g_mutex_unlock (&priv->lock);
 }
 
+static gchar *
+stream_make_keymgmt (GstRTSPClient * client, const gchar * location,
+    GstRTSPStream * stream)
+{
+  gchar *base64, *result = NULL;
+  GstMIKEYMessage *mikey_msg;
+  GstCaps *srtcpparams;
+  GstElement *rtcp_encoder;
+  gint srtcp_cipher, srtp_cipher;
+  gint srtcp_auth, srtp_auth;
+  GstBuffer *key;
+  GType ciphertype, authtype;
+  GEnumClass *cipher_enum, *auth_enum;
+  GEnumValue *srtcp_cipher_value, *srtp_cipher_value, *srtcp_auth_value,
+      *srtp_auth_value;
+
+  rtcp_encoder = gst_rtsp_stream_get_srtp_encoder (stream);
+
+  if (!rtcp_encoder)
+    goto done;
+
+  ciphertype = g_type_from_name ("GstSrtpCipherType");
+  authtype = g_type_from_name ("GstSrtpAuthType");
+
+  cipher_enum = g_type_class_ref (ciphertype);
+  auth_enum = g_type_class_ref (authtype);
+
+  /* We need to bring the encoder to READY so that it generates its key */
+  gst_element_set_state (rtcp_encoder, GST_STATE_READY);
+
+  g_object_get (rtcp_encoder, "rtcp-cipher", &srtcp_cipher, "rtcp-auth",
+      &srtcp_auth, "rtp-cipher", &srtp_cipher, "rtp-auth", &srtp_auth, "key",
+      &key, NULL);
+  g_object_unref (rtcp_encoder);
+
+  srtcp_cipher_value = g_enum_get_value (cipher_enum, srtcp_cipher);
+  srtp_cipher_value = g_enum_get_value (cipher_enum, srtp_cipher);
+  srtcp_auth_value = g_enum_get_value (auth_enum, srtcp_auth);
+  srtp_auth_value = g_enum_get_value (auth_enum, srtp_auth);
+
+  g_type_class_unref (cipher_enum);
+  g_type_class_unref (auth_enum);
+
+  srtcpparams = gst_caps_new_simple ("application/x-srtcp",
+      "srtcp-cipher", G_TYPE_STRING, srtcp_cipher_value->value_nick,
+      "srtcp-auth", G_TYPE_STRING, srtcp_auth_value->value_nick,
+      "srtp-cipher", G_TYPE_STRING, srtp_cipher_value->value_nick,
+      "srtp-auth", G_TYPE_STRING, srtp_auth_value->value_nick,
+      "srtp-key", GST_TYPE_BUFFER, key, NULL);
+
+  mikey_msg = gst_mikey_message_new_from_caps (srtcpparams);
+  if (mikey_msg) {
+    guint send_ssrc;
+
+    gst_rtsp_stream_get_ssrc (stream, &send_ssrc);
+    gst_mikey_message_add_cs_srtp (mikey_msg, 0, send_ssrc, 0);
+
+    base64 = gst_mikey_message_base64_encode (mikey_msg);
+    gst_mikey_message_unref (mikey_msg);
+
+    if (base64) {
+      result = gst_sdp_make_keymgmt (location, base64);
+      g_free (base64);
+    }
+  }
+
+done:
+  return result;
+}
+
 static gboolean
 handle_setup_request (GstRTSPClient * client, GstRTSPContext * ctx)
 {
@@ -2410,7 +2447,7 @@ handle_setup_request (GstRTSPClient * client, GstRTSPContext * ctx)
   /* parse the keymgmt */
   if (gst_rtsp_message_get_header (ctx->request, GST_RTSP_HDR_KEYMGMT,
           &keymgmt, 0) == GST_RTSP_OK) {
-    if (!handle_keymgmt (client, ctx, keymgmt))
+    if (!gst_rtsp_stream_handle_keymgmt (ctx->stream, keymgmt))
       goto keymgmt_error;
   }
 
@@ -2483,6 +2520,8 @@ handle_setup_request (GstRTSPClient * client, GstRTSPContext * ctx)
     g_hash_table_insert (priv->transports,
         GINT_TO_POINTER (ct->interleaved.max), trans);
     g_object_ref (trans);
+    add_data_seq (client, ct->interleaved.min);
+    add_data_seq (client, ct->interleaved.max);
   }
 
   /* create and serialize the server transport */
@@ -2915,6 +2954,7 @@ handle_announce_request (GstRTSPClient * client, GstRTSPContext * ctx)
   guint8 *data;
   guint size;
   GstRTSPStatusCode sig_result;
+  guint i, n_streams;
 
   klass = GST_RTSP_CLIENT_GET_CLASS (client);
 
@@ -2968,13 +3008,28 @@ handle_announce_request (GstRTSPClient * client, GstRTSPContext * ctx)
   if (!klass->handle_sdp (client, ctx, media, sdp))
     goto unhandled_sdp;
 
+  gst_rtsp_message_init_response (ctx->response, GST_RTSP_STS_OK,
+      gst_rtsp_status_as_text (GST_RTSP_STS_OK), ctx->request);
+
+  n_streams = gst_rtsp_media_n_streams (media);
+  for (i = 0; i < n_streams; i++) {
+    GstRTSPStream *stream = gst_rtsp_media_get_stream (media, i);
+    gchar *location =
+        g_strdup_printf ("rtsp://%s%s:8554/stream=%d", priv->server_ip, path,
+        i);
+    gchar *keymgmt = stream_make_keymgmt (client, location, stream);
+
+    if (keymgmt)
+      gst_rtsp_message_take_header (ctx->response, GST_RTSP_HDR_KEYMGMT,
+          keymgmt);
+
+    g_free (location);
+  }
+
   /* we suspend after the announce */
   gst_rtsp_media_suspend (media);
   g_object_unref (media);
 
-  gst_rtsp_message_init_response (ctx->response, GST_RTSP_STS_OK,
-      gst_rtsp_status_as_text (GST_RTSP_STS_OK), ctx->request);
-
   send_message (client, ctx, ctx->response, FALSE);
 
   g_signal_emit (client, gst_rtsp_client_signals[SIGNAL_ANNOUNCE_REQUEST],
@@ -3287,11 +3342,7 @@ client_session_removed (GstRTSPSessionPool * pool, GstRTSPSession * session,
   GST_INFO ("client %p: session %p removed", client, session);
 
   g_mutex_lock (&priv->lock);
-  if (priv->watch != NULL)
-    gst_rtsp_watch_set_send_backlog (priv->watch, 0, 0);
   client_unwatch_session (client, session, NULL);
-  if (priv->watch != NULL)
-    gst_rtsp_watch_set_send_backlog (priv->watch, 0, WATCH_BACKLOG_SIZE);
   g_mutex_unlock (&priv->lock);
 }
 
@@ -3481,37 +3532,6 @@ handle_request (GstRTSPClient * client, GstRTSPMessage * request)
   if (!check_request_requirements (ctx, &unsupported_reqs))
     goto unsupported_requirement;
 
-  /* the backlog must be unlimited while processing requests.
-   * the causes of this are two cases of deadlocks while streaming over TCP:
-   *
-   * 1. consider the scenario where the media pipeline's streaming thread
-   * is blocking in the appsink (taking the appsink's preroll lock) because
-   * the backlog is full. when a PAUSE request is received by the RTSP
-   * client thread then the the state of the session media ought to change
-   * to PAUSED. while most elements in the pipeline can change state this
-   * can never happen for the appsink since its preroll lock is taken by
-   * another thread.
-   *
-   * 2. consider the scenario where the media pipeline's streaming thread
-   * is blocking in the appsink new_sample callback (taking the send lock
-   * in RTSP client) because the backlog is full. when e.g. a GET request
-   * is received by the RTSP client thread then a response ought to be sent
-   * but this can never happen since it requires taking the send lock
-   * already taken by another thread.
-   *
-   * the reason that the backlog is never emptied is that the source used
-   * for dequeing messages from the backlog is never dispatched because it
-   * is attached to the same mainloop as the source receving RTSP requests and
-   * therefore run by the RTSP client thread which is alreayd blocking.
-   *
-   * without significant changes the easiest way to cope with this is to
-   * not block indefinitely when the backlog is full, but rather let the
-   * backlog grow in size. this in effect means that there can not be any
-   * upper boundary on its size.
-   */
-  if (priv->watch != NULL)
-    gst_rtsp_watch_set_send_backlog (priv->watch, 0, 0);
-
   /* now see what is asked and dispatch to a dedicated handler */
   switch (method) {
     case GST_RTSP_OPTIONS:
@@ -3550,19 +3570,12 @@ handle_request (GstRTSPClient * client, GstRTSPMessage * request)
       handle_record_request (client, ctx);
       break;
     case GST_RTSP_REDIRECT:
-      if (priv->watch != NULL)
-        gst_rtsp_watch_set_send_backlog (priv->watch, 0, WATCH_BACKLOG_SIZE);
       goto not_implemented;
     case GST_RTSP_INVALID:
     default:
-      if (priv->watch != NULL)
-        gst_rtsp_watch_set_send_backlog (priv->watch, 0, WATCH_BACKLOG_SIZE);
       goto bad_request;
   }
 
-  if (priv->watch != NULL)
-    gst_rtsp_watch_set_send_backlog (priv->watch, 0, WATCH_BACKLOG_SIZE);
-
 done:
   if (ctx == &sctx)
     gst_rtsp_context_pop_current (ctx);
@@ -3582,9 +3595,6 @@ not_supported:
   }
 invalid_command_for_version:
   {
-    if (priv->watch != NULL)
-      gst_rtsp_watch_set_send_backlog (priv->watch, 0, WATCH_BACKLOG_SIZE);
-
     GST_ERROR ("client %p: invalid command for version", client);
     send_generic_response (client, GST_RTSP_STS_BAD_REQUEST, ctx);
     goto done;
@@ -3729,6 +3739,37 @@ handle_data (GstRTSPClient * client, GstRTSPMessage * message)
   trans =
       g_hash_table_lookup (priv->transports, GINT_TO_POINTER ((gint) channel));
   if (trans) {
+    GSocketAddress *addr;
+
+    /* Only create the socket address once for the transport, we don't really
+     * want to do that for every single packet.
+     *
+     * The netaddress meta is later used by the RTP stack to know where
+     * packets came from and allows us to match it again to a stream transport
+     *
+     * In theory we could use the remote socket address of the RTSP connection
+     * here, but this would fail with a custom configure_client_transport()
+     * implementation.
+     */
+    if (!(addr =
+            g_object_get_data (G_OBJECT (trans), "rtsp-client.remote-addr"))) {
+      const GstRTSPTransport *tr;
+      GInetAddress *iaddr;
+
+      tr = gst_rtsp_stream_transport_get_transport (trans);
+      iaddr = g_inet_address_new_from_string (tr->destination);
+      if (iaddr) {
+        addr = g_inet_socket_address_new (iaddr, tr->client_port.min);
+        g_object_unref (iaddr);
+        g_object_set_data_full (G_OBJECT (trans), "rtsp-client.remote-addr",
+            addr, (GDestroyNotify) g_object_unref);
+      }
+    }
+
+    if (addr) {
+      gst_buffer_add_net_address_meta (buffer, addr);
+    }
+
     /* dispatch to the stream based on the channel number */
     GST_LOG_OBJECT (client, "%u bytes of data on channel %u", size, channel);
     gst_rtsp_stream_transport_recv_data (trans, channel, buffer);
@@ -3751,7 +3792,7 @@ invalid_length:
 /**
  * gst_rtsp_client_set_session_pool:
  * @client: a #GstRTSPClient
- * @pool: (transfer none): a #GstRTSPSessionPool
+ * @pool: (transfer none) (nullable): a #GstRTSPSessionPool
  *
  * Set @pool as the sessionpool for @client which it will use to find
  * or allocate sessions. the sessionpool is usually inherited from the server
@@ -3792,7 +3833,7 @@ gst_rtsp_client_set_session_pool (GstRTSPClient * client,
  *
  * Get the #GstRTSPSessionPool object that @client uses to manage its sessions.
  *
- * Returns: (transfer full): a #GstRTSPSessionPool, unref after usage.
+ * Returns: (transfer full) (nullable): a #GstRTSPSessionPool, unref after usage.
  */
 GstRTSPSessionPool *
 gst_rtsp_client_get_session_pool (GstRTSPClient * client)
@@ -3815,7 +3856,7 @@ gst_rtsp_client_get_session_pool (GstRTSPClient * client)
 /**
  * gst_rtsp_client_set_mount_points:
  * @client: a #GstRTSPClient
- * @mounts: (transfer none): a #GstRTSPMountPoints
+ * @mounts: (transfer none) (nullable): a #GstRTSPMountPoints
  *
  * Set @mounts as the mount points for @client which it will use to map urls
  * to media streams. These mount points are usually inherited from the server that
@@ -3850,7 +3891,7 @@ gst_rtsp_client_set_mount_points (GstRTSPClient * client,
  *
  * Get the #GstRTSPMountPoints object that @client uses to manage its sessions.
  *
- * Returns: (transfer full): a #GstRTSPMountPoints, unref after usage.
+ * Returns: (transfer full) (nullable): a #GstRTSPMountPoints, unref after usage.
  */
 GstRTSPMountPoints *
 gst_rtsp_client_get_mount_points (GstRTSPClient * client)
@@ -3873,7 +3914,7 @@ gst_rtsp_client_get_mount_points (GstRTSPClient * client)
 /**
  * gst_rtsp_client_set_auth:
  * @client: a #GstRTSPClient
- * @auth: (transfer none): a #GstRTSPAuth
+ * @auth: (transfer none) (nullable): a #GstRTSPAuth
  *
  * configure @auth to be used as the authentication manager of @client.
  */
@@ -3906,8 +3947,8 @@ gst_rtsp_client_set_auth (GstRTSPClient * client, GstRTSPAuth * auth)
  *
  * Get the #GstRTSPAuth used as the authentication manager of @client.
  *
- * Returns: (transfer full): the #GstRTSPAuth of @client. g_object_unref() after
- * usage.
+ * Returns: (transfer full) (nullable): the #GstRTSPAuth of @client.
+ * g_object_unref() after usage.
  */
 GstRTSPAuth *
 gst_rtsp_client_get_auth (GstRTSPClient * client)
@@ -3930,7 +3971,7 @@ gst_rtsp_client_get_auth (GstRTSPClient * client)
 /**
  * gst_rtsp_client_set_thread_pool:
  * @client: a #GstRTSPClient
- * @pool: (transfer none): a #GstRTSPThreadPool
+ * @pool: (transfer none) (nullable): a #GstRTSPThreadPool
  *
  * configure @pool to be used as the thread pool of @client.
  */
@@ -3963,7 +4004,7 @@ gst_rtsp_client_set_thread_pool (GstRTSPClient * client,
  *
  * Get the #GstRTSPThreadPool used as the thread pool of @client.
  *
- * Returns: (transfer full): the #GstRTSPThreadPool of @client. g_object_unref() after
+ * Returns: (transfer full) (nullable): the #GstRTSPThreadPool of @client. g_object_unref() after
  * usage.
  */
 GstRTSPThreadPool *
@@ -4055,7 +4096,7 @@ no_address:
  *
  * Get the #GstRTSPConnection of @client.
  *
- * Returns: (transfer none): the #GstRTSPConnection of @client.
+ * Returns: (transfer none) (nullable): the #GstRTSPConnection of @client.
  * The connection object returned remains valid until the client is freed.
  */
 GstRTSPConnection *
@@ -4184,33 +4225,48 @@ do_send_message (GstRTSPClient * client, GstRTSPMessage * message,
     gboolean close, gpointer user_data)
 {
   GstRTSPClientPrivate *priv = client->priv;
+  guint id = 0;
   GstRTSPResult ret;
-  GTimeVal time;
 
-  time.tv_sec = 1;
-  time.tv_usec = 0;
+  /* send the message */
+  ret = gst_rtsp_watch_send_message (priv->watch, message, &id);
+  if (ret != GST_RTSP_OK)
+    goto error;
 
-  do {
-    /* send the response and store the seq number so we can wait until it's
-     * written to the client to close the connection */
-    ret =
-        gst_rtsp_watch_send_message (priv->watch, message,
-        close ? &priv->close_seq : NULL);
-    if (ret == GST_RTSP_OK)
-      break;
+  /* if close flag is set, store the seq number so we can wait until it's
+   * written to the client to close the connection */
+  if (close)
+    priv->close_seq = id;
 
-    if (ret != GST_RTSP_ENOMEM)
-      goto error;
+  if (gst_rtsp_message_get_type (message) == GST_RTSP_MESSAGE_DATA) {
+    guint8 channel = 0;
+    GstRTSPResult r;
 
-    /* drop backlog */
-    if (priv->drop_backlog)
-      break;
+    r = gst_rtsp_message_parse_data (message, &channel);
+    if (r != GST_RTSP_OK) {
+      ret = r;
+      goto error;
+    }
 
-    /* queue was full, wait for more space */
-    GST_DEBUG_OBJECT (client, "waiting for backlog");
-    ret = gst_rtsp_watch_wait_backlog (priv->watch, &time);
-    GST_DEBUG_OBJECT (client, "Resend due to backlog full");
-  } while (ret != GST_RTSP_EINTR);
+    /* check if the message has been queued for transmission in watch */
+    if (id) {
+      /* store the seq number so we can wait until it has been sent */
+      GST_DEBUG_OBJECT (client, "wait for message %d, channel %d", id, channel);
+      set_data_seq (client, channel, id);
+    } else {
+      GstRTSPStreamTransport *trans;
+
+      trans =
+          g_hash_table_lookup (priv->transports,
+          GINT_TO_POINTER ((gint) channel));
+      if (trans) {
+        GST_DEBUG_OBJECT (client, "emit 'message-sent' signal");
+        g_mutex_unlock (&priv->send_lock);
+        gst_rtsp_stream_transport_message_sent (trans);
+        g_mutex_lock (&priv->send_lock);
+      }
+    }
+  }
 
   return ret == GST_RTSP_OK;
 
@@ -4218,7 +4274,7 @@ do_send_message (GstRTSPClient * client, GstRTSPMessage * message,
 error:
   {
     GST_DEBUG_OBJECT (client, "got error %d", ret);
-    return ret == GST_RTSP_OK;
+    return FALSE;
   }
 }
 
@@ -4234,13 +4290,33 @@ message_sent (GstRTSPWatch * watch, guint cseq, gpointer user_data)
 {
   GstRTSPClient *client = GST_RTSP_CLIENT (user_data);
   GstRTSPClientPrivate *priv = client->priv;
+  GstRTSPStreamTransport *trans = NULL;
+  guint8 channel = 0;
+  gboolean close = FALSE;
+
+  g_mutex_lock (&priv->send_lock);
+
+  if (get_data_channel (client, cseq, &channel)) {
+    trans = g_hash_table_lookup (priv->transports, GINT_TO_POINTER (channel));
+    set_data_seq (client, channel, 0);
+  }
 
   if (priv->close_seq && priv->close_seq == cseq) {
     GST_INFO ("client %p: send close message", client);
+    close = TRUE;
     priv->close_seq = 0;
-    gst_rtsp_client_close (client);
   }
 
+  g_mutex_unlock (&priv->send_lock);
+
+  if (trans) {
+    GST_DEBUG_OBJECT (client, "emit 'message-sent' signal");
+    gst_rtsp_stream_transport_message_sent (trans);
+  }
+
+  if (close)
+    gst_rtsp_client_close (client);
+
   return GST_RTSP_OK;
 }
 
@@ -4350,7 +4426,7 @@ tunnel_lost (GstRTSPWatch * watch, gpointer user_data)
   return GST_RTSP_OK;
 }
 
-static gboolean
+static GstRTSPStatusCode
 handle_tunnel (GstRTSPClient * client)
 {
   GstRTSPClientPrivate *priv = client->priv;
@@ -4386,6 +4462,8 @@ handle_tunnel (GstRTSPClient * client)
     g_mutex_lock (&opriv->watch_lock);
     if (opriv->watch == NULL)
       goto tunnel_closed;
+    if (opriv->tstate == priv->tstate)
+      goto tunnel_duplicate_id;
 
     GST_INFO ("client %p: found previous tunnel %p (old %p, new %p)", client,
         oclient, opriv->connection, priv->connection);
@@ -4402,20 +4480,27 @@ handle_tunnel (GstRTSPClient * client)
     gst_rtsp_client_set_send_func (client, NULL, NULL, NULL);
   }
 
-  return TRUE;
+  return GST_RTSP_STS_OK;
 
   /* ERRORS */
 no_tunnelid:
   {
     GST_ERROR ("client %p: no tunnelid provided", client);
-    return FALSE;
+    return GST_RTSP_STS_SERVICE_UNAVAILABLE;
   }
 tunnel_closed:
   {
     GST_ERROR ("client %p: tunnel session %s was closed", client, tunnelid);
     g_mutex_unlock (&opriv->watch_lock);
     g_object_unref (oclient);
-    return FALSE;
+    return GST_RTSP_STS_SERVICE_UNAVAILABLE;
+  }
+tunnel_duplicate_id:
+  {
+    GST_ERROR ("client %p: tunnel session %s was duplicate", client, tunnelid);
+    g_mutex_unlock (&opriv->watch_lock);
+    g_object_unref (oclient);
+    return GST_RTSP_STS_BAD_REQUEST;
   }
 }
 
@@ -4427,11 +4512,11 @@ tunnel_get (GstRTSPWatch * watch, gpointer user_data)
   GST_INFO ("client %p: tunnel get (connection %p)", client,
       client->priv->connection);
 
-  if (!handle_tunnel (client)) {
-    return GST_RTSP_STS_SERVICE_UNAVAILABLE;
-  }
+  g_mutex_lock (&client->priv->lock);
+  client->priv->tstate = TUNNEL_STATE_GET;
+  g_mutex_unlock (&client->priv->lock);
 
-  return GST_RTSP_STS_OK;
+  return handle_tunnel (client);
 }
 
 static GstRTSPResult
@@ -4442,9 +4527,12 @@ tunnel_post (GstRTSPWatch * watch, gpointer user_data)
   GST_INFO ("client %p: tunnel post (connection %p)", client,
       client->priv->connection);
 
-  if (!handle_tunnel (client)) {
+  g_mutex_lock (&client->priv->lock);
+  client->priv->tstate = TUNNEL_STATE_POST;
+  g_mutex_unlock (&client->priv->lock);
+
+  if (handle_tunnel (client) != GST_RTSP_STS_OK)
     return GST_RTSP_ERROR;
-  }
 
   return GST_RTSP_OK;
 }