Add a nice_socket_send_reliable API for internal use.

One issue with tcp-bsd is that it will queue messages when the tcp socket
is not writable, but it will also drop messages when the queue is full.
If we want to do proper reliable ice-tcp, we need to make sure that
messages don't get dropped, also, this would affect http/socks5/pseudossl
if their messages get dropped. For ice-tcp, when the socket is not writable
we want to return 0, not queue the message.
The change here is to allow connchecks and other important protocol messages
to be sent as 'reliable' on tcp transports by queuing them and in the various
socket layers and to never drop them, but all user messages will be dropped.

if the tcp socket is not writable.

9 files changed, 304 insertions, 57 deletions

diff --git a/socket/http.c b/socket/http.c
index 0041ca6..0268fbb 100644
--- a/socket/http.c
+++ b/socket/http.c
@@ -95,6 +95,8 @@ static gint socket_recv_messages (NiceSocket *sock,
     NiceInputMessage *recv_messages, guint n_recv_messages);
 static gint socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *messages, guint n_messages);
+static gint socket_send_messages_reliable (NiceSocket *sock,
+    const NiceAddress *to, const NiceOutputMessage *messages, guint n_messages);
 static gboolean socket_is_reliable (NiceSocket *sock);
 
 static void add_to_be_sent (NiceSocket *sock, const NiceAddress *to,
@@ -127,6 +129,7 @@ nice_http_socket_new (NiceSocket *base_socket,
     sock->fileno = priv->base_socket->fileno;
     sock->addr = priv->base_socket->addr;
     sock->send_messages = socket_send_messages;
+    sock->send_messages_reliable = socket_send_messages_reliable;
     sock->recv_messages = socket_recv_messages;
     sock->is_reliable = socket_is_reliable;
     sock->close = socket_close;
@@ -167,7 +170,8 @@ nice_http_socket_new (NiceSocket *base_socket,
       local_messages.buffers = &local_bufs;
       local_messages.n_buffers = 1;
 
-      nice_socket_send_messages (priv->base_socket, NULL, &local_messages, 1);
+      nice_socket_send_messages_reliable (priv->base_socket, NULL,
+          &local_messages, 1);
       priv->state = HTTP_STATE_INIT;
       g_free (msg);
     }
@@ -545,7 +549,9 @@ retry:
 
         /* Send the pending data */
         while ((tbs = g_queue_pop_head (&priv->send_queue))) {
-          nice_socket_send (priv->base_socket, &tbs->to, tbs->length, tbs->buf);
+          /* We only queue reliable data */
+          nice_socket_send_reliable (priv->base_socket, &tbs->to,
+              tbs->length, tbs->buf);
           g_free (tbs->buf);
           g_slice_free (struct to_be_sent, tbs);
         }
@@ -588,12 +594,33 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
   } else if (priv->state == HTTP_STATE_ERROR) {
     return -1;
   } else {
-    add_to_be_sent (sock, to, messages, n_messages);
+    return 0;
   }
 
   return n_messages;
 }
 
+static gint
+socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
+    const NiceOutputMessage *messages, guint n_messages)
+{
+  HttpPriv *priv = sock->priv;
+
+  if (priv->state == HTTP_STATE_CONNECTED) {
+    /* Fast path. */
+    if (!priv->base_socket)
+      return -1;
+
+    return nice_socket_send_messages_reliable (priv->base_socket, to, messages,
+        n_messages);
+  } else if (priv->state == HTTP_STATE_ERROR) {
+    return -1;
+  } else {
+    add_to_be_sent (sock, to, messages, n_messages);
+  }
+
+  return n_messages;
+}
 
 static gboolean
 socket_is_reliable (NiceSocket *sock)
diff --git a/socket/pseudossl.c b/socket/pseudossl.c
index 5d0965c..6cf83d8 100644
--- a/socket/pseudossl.c
+++ b/socket/pseudossl.c
@@ -94,6 +94,8 @@ static gint socket_recv_messages (NiceSocket *sock,
     NiceInputMessage *recv_messages, guint n_recv_messages);
 static gint socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *messages, guint n_messages);
+static gint socket_send_messages_reliable (NiceSocket *sock,
+    const NiceAddress *to, const NiceOutputMessage *messages, guint n_messages);
 static gboolean socket_is_reliable (NiceSocket *sock);
 
 static void add_to_be_sent (NiceSocket *sock, const NiceAddress *to,
@@ -115,13 +117,14 @@ nice_pseudossl_socket_new (NiceSocket *base_socket)
   sock->fileno = priv->base_socket->fileno;
   sock->addr = priv->base_socket->addr;
   sock->send_messages = socket_send_messages;
+  sock->send_messages_reliable = socket_send_messages_reliable;
   sock->recv_messages = socket_recv_messages;
   sock->is_reliable = socket_is_reliable;
   sock->close = socket_close;
 
   /* We send 'to' NULL because it will always be to an already connected
    * TCP base socket, which ignores the destination */
-  nice_socket_send (priv->base_socket, NULL,
+  nice_socket_send_reliable (priv->base_socket, NULL,
       sizeof(SSL_CLIENT_HANDSHAKE), SSL_CLIENT_HANDSHAKE);
 
   return sock;
@@ -175,7 +178,8 @@ socket_recv_messages (NiceSocket *sock,
       struct to_be_sent *tbs = NULL;
       priv->handshaken = TRUE;
       while ((tbs = g_queue_pop_head (&priv->send_queue))) {
-        nice_socket_send (priv->base_socket, &tbs->to, tbs->length,
+        /* We only queue reliable data */
+        nice_socket_send_reliable (priv->base_socket, &tbs->to, tbs->length,
             (const gchar *) tbs->buf);
         g_free (tbs->buf);
         g_slice_free (struct to_be_sent, tbs);
@@ -206,12 +210,32 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     return nice_socket_send_messages (priv->base_socket, to, messages,
         n_messages);
   } else {
-    add_to_be_sent (sock, to, messages, n_messages);
+    return 0;
   }
   return n_messages;
 }
 
 
+static gint
+socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
+    const NiceOutputMessage *messages, guint n_messages)
+{
+  PseudoSSLPriv *priv = sock->priv;
+
+  if (priv->handshaken) {
+    /* Fast path: pass directly through to the base socket once the handshake is
+     * complete. */
+    if (priv->base_socket == NULL)
+      return -1;
+
+    return nice_socket_send_messages_reliable (priv->base_socket, to, messages,
+        n_messages);
+  } else {
+    add_to_be_sent (sock, to, messages, n_messages);
+  }
+  return n_messages;
+}
+
 static gboolean
 socket_is_reliable (NiceSocket *sock)
 {
diff --git a/socket/socket.c b/socket/socket.c
index 92ee15a..d7ce0a6 100644
--- a/socket/socket.c
+++ b/socket/socket.c
@@ -138,6 +138,48 @@ nice_socket_send_messages (NiceSocket *sock, const NiceAddress *to,
   return sock->send_messages (sock, to, messages, n_messages);
 }
 
+/**
+ * nice_socket_send_messages_reliable:
+ * @sock: a #NiceSocket
+ * @messages: (array length=n_messages) (in caller-allocates):
+ * array of #NiceOutputMessages containing the messages to send
+ * @n_messages: number of elements in the @messages array
+ *
+ * Send @n_messages on the socket, in a reliable, non-blocking fashion.
+ * The total size of the buffers in each #NiceOutputMessage
+ * must be at most the maximum UDP payload size (65535 bytes), or excess
+ * bytes will be silently dropped.
+ *
+ * On success, the number of messages transmitted from @messages is returned,
+ * which will be equal to @n_messages. If the call would have blocked part-way
+ * though, the remaining bytes will be queued for sending later.
+ * On failure, a negative value is returned, but no further error information
+ * is available. Calling this function on a socket which has closed is an error,
+ * and a negative value is returned. Calling this function on a socket which
+ * is not TCP or does not have a TCP base socket, will result in an error.
+ *
+ * If #NiceOutputMessage::to is specified for a message, that will be used as
+ * the destination address for the message. Otherwise, if %NULL, the default
+ * destination for @sock will be used.
+ *
+ * Every field of every #NiceOutputMessage is guaranteed to be unmodified when
+ * this function returns.
+ *
+ * Returns: number of messages successfully sent from @messages, or a negative
+ * value on error
+ *
+ * Since: 0.1.5
+ */
+gint
+nice_socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
+    const NiceOutputMessage *messages, guint n_messages)
+{
+  g_return_val_if_fail (sock != NULL, -1);
+  g_return_val_if_fail (n_messages == 0 || messages != NULL, -1);
+
+  return sock->send_messages_reliable (sock, to, messages, n_messages);
+}
+
 /* Convenience wrapper around nice_socket_send_messages(). Returns the number of
  * bytes sent on success (which will be @len), zero if sending would block, or
  * -1 on error. */
@@ -155,6 +197,20 @@ nice_socket_send (NiceSocket *sock, const NiceAddress *to, gsize len,
   return ret;
 }
 
+gssize
+nice_socket_send_reliable (NiceSocket *sock, const NiceAddress *to, gsize len,
+    const gchar *buf)
+{
+  GOutputVector local_buf = { buf, len };
+  NiceOutputMessage local_message = { &local_buf, 1};
+  gint ret;
+
+  ret = sock->send_messages_reliable (sock, to, &local_message, 1);
+  if (ret == 1)
+    return len;
+  return ret;
+}
+
 gboolean
 nice_socket_is_reliable (NiceSocket *sock)
 {
diff --git a/socket/socket.h b/socket/socket.h
index 18e5fb5..e6eb27e 100644
--- a/socket/socket.h
+++ b/socket/socket.h
@@ -80,6 +80,8 @@ struct _NiceSocket
   /* As above, @n_messages may be zero. Iff so, @messages may be %NULL. */
   gint (*send_messages) (NiceSocket *sock, const NiceAddress *to,
       const NiceOutputMessage *messages, guint n_messages);
+  gint (*send_messages_reliable) (NiceSocket *sock, const NiceAddress *to,
+      const NiceOutputMessage *messages, guint n_messages);
   gboolean (*is_reliable) (NiceSocket *sock);
   void (*close) (NiceSocket *sock);
   void *priv;
@@ -93,9 +95,15 @@ nice_socket_recv_messages (NiceSocket *sock,
 gint
 nice_socket_send_messages (NiceSocket *sock, const NiceAddress *addr,
     const NiceOutputMessage *messages, guint n_messages);
+gint
+nice_socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *addr,
+    const NiceOutputMessage *messages, guint n_messages);
 gssize
 nice_socket_send (NiceSocket *sock, const NiceAddress *addr, gsize len,
     const gchar *buf);
+gssize
+nice_socket_send_reliable (NiceSocket *sock, const NiceAddress *addr, gsize len,
+    const gchar *buf);
 
 gboolean
 nice_socket_is_reliable (NiceSocket *sock);
diff --git a/socket/socks5.c b/socket/socks5.c
index 1eb2726..af2d3d0 100644
--- a/socket/socks5.c
+++ b/socket/socks5.c
@@ -81,6 +81,8 @@ static gint socket_recv_messages (NiceSocket *sock,
     NiceInputMessage *recv_messages, guint n_recv_messages);
 static gint socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *messages, guint n_messages);
+static gint socket_send_messages_reliable (NiceSocket *sock,
+    const NiceAddress *to, const NiceOutputMessage *messages, guint n_messages);
 static gboolean socket_is_reliable (NiceSocket *sock);
 
 static void add_to_be_sent (NiceSocket *sock, const NiceAddress *to,
@@ -108,6 +110,7 @@ nice_socks5_socket_new (NiceSocket *base_socket,
     sock->fileno = priv->base_socket->fileno;
     sock->addr = priv->base_socket->addr;
     sock->send_messages = socket_send_messages;
+    sock->send_messages_reliable = socket_send_messages_reliable;
     sock->recv_messages = socket_recv_messages;
     sock->is_reliable = socket_is_reliable;
     sock->close = socket_close;
@@ -131,7 +134,7 @@ nice_socks5_socket_new (NiceSocket *base_socket,
 
       /* We send 'to' NULL because it will always be to an already connected
        * TCP base socket, which ignores the destination */
-      nice_socket_send (priv->base_socket, NULL, len, msg);
+      nice_socket_send_reliable (priv->base_socket, NULL, len, msg);
       priv->state = SOCKS_STATE_INIT;
     }
   }
@@ -239,7 +242,7 @@ socket_recv_messages (NiceSocket *sock,
                   memcpy (msg + len, priv->password, plen); /* Password */
                 len += plen;
 
-                nice_socket_send (priv->base_socket, NULL, len, msg);
+                nice_socket_send_reliable (priv->base_socket, NULL, len, msg);
                 priv->state = SOCKS_STATE_AUTH;
               } else {
                 /* Authentication required but no auth info available */
@@ -331,7 +334,8 @@ socket_recv_messages (NiceSocket *sock,
                       goto error;
                   }
                   while ((tbs = g_queue_pop_head (&priv->send_queue))) {
-                    nice_socket_send (priv->base_socket, &tbs->to,
+                    /* We only queue reliable data */
+                    nice_socket_send_reliable (priv->base_socket, &tbs->to,
                         tbs->length, (const gchar *) tbs->buf);
                     g_free (tbs->buf);
                     g_slice_free (struct to_be_sent, tbs);
@@ -405,7 +409,7 @@ socket_recv_messages (NiceSocket *sock,
       len += 2;
     }
 
-    nice_socket_send (priv->base_socket, NULL, len, msg);
+    nice_socket_send_reliable (priv->base_socket, NULL, len, msg);
     priv->state = SOCKS_STATE_CONNECT;
 
     return 0;
@@ -436,6 +440,28 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
   } else if (priv->state == SOCKS_STATE_ERROR) {
     return -1;
   } else {
+    return 0;
+  }
+  return n_messages;
+}
+
+
+static gint
+socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
+    const NiceOutputMessage *messages, guint n_messages)
+{
+  Socks5Priv *priv = sock->priv;
+
+  if (priv->state == SOCKS_STATE_CONNECTED) {
+    /* Fast path: pass through to the base socket once connected. */
+    if (priv->base_socket == NULL)
+      return -1;
+
+    return nice_socket_send_messages_reliable (priv->base_socket, to, messages,
+        n_messages);
+  } else if (priv->state == SOCKS_STATE_ERROR) {
+    return -1;
+  } else {
     add_to_be_sent (sock, to, messages, n_messages);
   }
   return n_messages;
diff --git a/socket/tcp-bsd.c b/socket/tcp-bsd.c
index 620c796..da6fc08 100644
--- a/socket/tcp-bsd.c
+++ b/socket/tcp-bsd.c
@@ -65,7 +65,6 @@ typedef struct {
 struct to_be_sent {
   guint8 *buf;
   gsize length;
-  gboolean can_drop;
 };
 
 #define MAX_QUEUE_LENGTH 20
@@ -75,6 +74,8 @@ static gint socket_recv_messages (NiceSocket *sock,
     NiceInputMessage *recv_messages, guint n_recv_messages);
 static gint socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *messages, guint n_messages);
+static gint socket_send_messages_reliable (NiceSocket *sock,
+    const NiceAddress *to, const NiceOutputMessage *messages, guint n_messages);
 static gboolean socket_is_reliable (NiceSocket *sock);
 
 
@@ -177,6 +178,7 @@ nice_tcp_bsd_socket_new (GMainContext *ctx, NiceAddress *addr, gboolean reliable
   sock->type = NICE_SOCKET_TYPE_TCP_BSD;
   sock->fileno = gsock;
   sock->send_messages = socket_send_messages;
+  sock->send_messages_reliable = socket_send_messages_reliable;
   sock->recv_messages = socket_recv_messages;
   sock->is_reliable = socket_is_reliable;
   sock->close = socket_close;
@@ -260,7 +262,8 @@ socket_recv_messages (NiceSocket *sock,
 }
 
 static gssize
-socket_send_message (NiceSocket *sock, const NiceOutputMessage *message)
+socket_send_message (NiceSocket *sock,
+    const NiceOutputMessage *message, gboolean reliable)
 {
   TcpPriv *priv = sock->priv;
   gssize ret;
@@ -294,32 +297,16 @@ socket_send_message (NiceSocket *sock, const NiceOutputMessage *message)
       add_to_be_sent (sock, message, ret, message_len, TRUE);
       ret = message_len;
     }
-  } else if (priv->reliable) {
-    /* Reliable TCP, so we shouldn't drop any messages or queue them */
-    ret = 0;
   } else {
-    /* FIXME: This dropping will break http/socks5/etc
-     * We probably need a way to the upper layer to control reliability
-     */
-    /* If the queue is too long, drop whatever packets we can. */
-    if (g_queue_get_length (&priv->send_queue) >= MAX_QUEUE_LENGTH) {
-      guint peek_idx = 0;
-      struct to_be_sent *tbs = NULL;
-
-      while ((tbs = g_queue_peek_nth (&priv->send_queue, peek_idx)) != NULL) {
-        if (tbs->can_drop) {
-          tbs = g_queue_pop_nth (&priv->send_queue, peek_idx);
-          free_to_be_sent (tbs);
-          break;
-        } else {
-          peek_idx++;
-        }
-      }
+    /* Only queue if we're sending reliably  */
+    if (reliable) {
+      /* Queue the message and send it later. */
+      add_to_be_sent (sock, message, 0, message_len, FALSE);
+      ret = message_len;
+    } else {
+      /* non reliable send, so we shouldn't queue the message */
+      ret = 0;
     }
-
-    /* Queue the message and send it later. */
-    add_to_be_sent (sock, message, 0, message_len, FALSE);
-    ret = message_len;
   }
 
   return ret;
@@ -338,7 +325,7 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *message = &messages[i];
     gssize len;
 
-    len = socket_send_message (sock, message);
+    len = socket_send_message (sock, message, FALSE);
 
     if (len < 0) {
       /* Error. */
@@ -354,6 +341,22 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
   return i;
 }
 
+static gint
+socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
+    const NiceOutputMessage *messages, guint n_messages)
+{
+  guint i;
+
+  for (i = 0; i < n_messages; i++) {
+    if (socket_send_message (sock, &messages[i], TRUE) < 0) {
+      /* Error. */
+      return -1;
+    }
+  }
+
+  return i;
+}
+
 static gboolean
 socket_is_reliable (NiceSocket *sock)
 {
@@ -458,7 +461,6 @@ add_to_be_sent (NiceSocket *sock, const NiceOutputMessage *message,
   tbs = g_slice_new0 (struct to_be_sent);
   tbs->length = message_len - message_offset;
   tbs->buf = g_malloc (tbs->length);
-  tbs->can_drop = !head;
 
   if (head)
     g_queue_push_head (&priv->send_queue, tbs);
diff --git a/socket/tcp-turn.c b/socket/tcp-turn.c
index 9c7cb99..82883d3 100644
--- a/socket/tcp-turn.c
+++ b/socket/tcp-turn.c
@@ -71,6 +71,8 @@ static gint socket_recv_messages (NiceSocket *sock,
     NiceInputMessage *recv_messages, guint n_recv_messages);
 static gint socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *messages, guint n_messages);
+static gint socket_send_messages_reliable (NiceSocket *sock,
+    const NiceAddress *to, const NiceOutputMessage *messages, guint n_messages);
 static gboolean socket_is_reliable (NiceSocket *sock);
 
 NiceSocket *
@@ -88,6 +90,7 @@ nice_tcp_turn_socket_new (NiceSocket *base_socket,
   sock->fileno = priv->base_socket->fileno;
   sock->addr = priv->base_socket->addr;
   sock->send_messages = socket_send_messages;
+  sock->send_messages_reliable = socket_send_messages_reliable;
   sock->recv_messages = socket_recv_messages;
   sock->is_reliable = socket_is_reliable;
   sock->close = socket_close;
@@ -225,7 +228,7 @@ socket_recv_messages (NiceSocket *nicesock,
 
 static gssize
 socket_send_message (NiceSocket *sock, const NiceAddress *to,
-    const NiceOutputMessage *message)
+    const NiceOutputMessage *message, gboolean reliable)
 {
   TurnTcpPriv *priv = sock->priv;
   guint8 padbuf[3] = {0, 0, 0};
@@ -276,7 +279,11 @@ socket_send_message (NiceSocket *sock, const NiceAddress *to,
   }
 
 
-  ret = nice_socket_send_messages (priv->base_socket, to, &local_message, 1);
+  if (reliable)
+    ret = nice_socket_send_messages_reliable (priv->base_socket, to,
+        &local_message, 1);
+  else
+    ret = nice_socket_send_messages (priv->base_socket, to, &local_message, 1);
 
   if (ret == 1)
     ret = output_message_get_size (&local_message);
@@ -296,7 +303,7 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *message = &messages[i];
     gssize len;
 
-    len = socket_send_message (sock, to, message);
+    len = socket_send_message (sock, to, message, FALSE);
 
     if (len < 0) {
       /* Error. */
@@ -312,6 +319,27 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
   return i;
 }
 
+static gint
+socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
+    const NiceOutputMessage *messages, guint n_messages)
+{
+  guint i;
+
+  for (i = 0; i < n_messages; i++) {
+    const NiceOutputMessage *message = &messages[i];
+    gssize len;
+
+    len = socket_send_message (sock, to, message, TRUE);
+
+    if (len < 0) {
+      /* Error. */
+      return len;
+    }
+  }
+
+  return i;
+}
+
 
 static gboolean
 socket_is_reliable (NiceSocket *sock)
diff --git a/socket/turn.c b/socket/turn.c
index e31cb02..8bfb4dc 100644
--- a/socket/turn.c
+++ b/socket/turn.c
@@ -111,6 +111,7 @@ typedef struct {
 typedef struct {
   gchar *data;
   guint data_len;
+  gboolean reliable;
 } SendData;
 
 static void socket_close (NiceSocket *sock);
@@ -118,6 +119,8 @@ static gint socket_recv_messages (NiceSocket *sock,
     NiceInputMessage *recv_messages, guint n_recv_messages);
 static gint socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *messages, guint n_messages);
+static gint socket_send_messages_reliable (NiceSocket *sock,
+    const NiceAddress *to, const NiceOutputMessage *messages, guint n_messages);
 static gboolean socket_is_reliable (NiceSocket *sock);
 
 static void priv_process_pending_bindings (TurnPriv *priv);
@@ -232,6 +235,7 @@ nice_turn_socket_new (GMainContext *ctx, NiceAddress *addr,
   sock->fileno = base_socket->fileno;
   sock->addr = *addr;
   sock->send_messages = socket_send_messages;
+  sock->send_messages_reliable = socket_send_messages_reliable;
   sock->recv_messages = socket_recv_messages;
   sock->is_reliable = socket_is_reliable;
   sock->close = socket_close;
@@ -528,7 +532,7 @@ priv_clear_permissions (TurnPriv *priv)
 
 static void
 socket_enqueue_data(TurnPriv *priv, const NiceAddress *to,
-    guint len, const gchar *buf)
+    guint len, const gchar *buf, gboolean reliable)
 {
   SendData *data = g_slice_new0 (SendData);
   GQueue *queue = g_hash_table_lookup (priv->send_data_queues, to);
@@ -541,6 +545,7 @@ socket_enqueue_data(TurnPriv *priv, const NiceAddress *to,
 
   data->data = g_memdup(buf, len);
   data->data_len = len;
+  data->reliable = reliable;
   g_queue_push_tail (queue, data);
 }
 
@@ -555,8 +560,12 @@ socket_dequeue_all_data (TurnPriv *priv, const NiceAddress *to)
           (SendData *) g_queue_pop_head(send_queue);
 
       nice_debug ("dequeuing data");
-      nice_socket_send (priv->base_socket, &priv->server_addr, data->data_len,
-          data->data);
+      if (data->reliable)
+        nice_socket_send_reliable (priv->base_socket, &priv->server_addr,
+            data->data_len, data->data);
+      else
+        nice_socket_send (priv->base_socket, &priv->server_addr, data->data_len,
+            data->data);
 
       g_free (data->data);
       g_slice_free (SendData, data);
@@ -570,7 +579,7 @@ socket_dequeue_all_data (TurnPriv *priv, const NiceAddress *to)
 
 static gssize
 socket_send_message (NiceSocket *sock, const NiceAddress *to,
-    const NiceOutputMessage *message)
+    const NiceOutputMessage *message, gboolean reliable)
 {
   TurnPriv *priv = (TurnPriv *) sock->priv;
   StunMessage msg;
@@ -631,8 +640,13 @@ socket_send_message (NiceSocket *sock, const NiceAddress *to,
         goto error;
       }
     } else {
-      ret = nice_socket_send_messages (priv->base_socket, &priv->server_addr,
-          message, 1);
+      if (reliable)
+        ret = nice_socket_send_messages_reliable (priv->base_socket,
+            &priv->server_addr, message, 1);
+      else
+        ret = nice_socket_send_messages (priv->base_socket, &priv->server_addr,
+            message, 1);
+
       if (ret == 1)
         return output_message_get_size (message);
       return ret;
@@ -722,15 +736,20 @@ socket_send_message (NiceSocket *sock, const NiceAddress *to,
 
       /* enque data */
       nice_debug ("enqueuing data");
-      socket_enqueue_data(priv, to, msg_len, (gchar *)buffer);
+      socket_enqueue_data(priv, to, msg_len, (gchar *)buffer, reliable);
 
       return msg_len;
     } else {
       GOutputVector local_buf = { buffer, msg_len };
       NiceOutputMessage local_message = {&local_buf, 1};
 
-      ret = nice_socket_send_messages (priv->base_socket, &priv->server_addr,
-          &local_message, 1);
+      if (reliable)
+        ret = nice_socket_send_messages_reliable (priv->base_socket,
+            &priv->server_addr, &local_message, 1);
+      else
+        ret = nice_socket_send_messages (priv->base_socket, &priv->server_addr,
+            &local_message, 1);
+
       if (ret == 1)
         return msg_len;
       return ret;
@@ -738,7 +757,10 @@ socket_send_message (NiceSocket *sock, const NiceAddress *to,
   }
 
   /* Error condition pass through to the base socket. */
-  ret = nice_socket_send_messages (priv->base_socket, to, message, 1);
+  if (reliable)
+    ret = nice_socket_send_messages_reliable (priv->base_socket, to, message, 1);
+  else
+    ret = nice_socket_send_messages (priv->base_socket, to, message, 1);
   if (ret == 1)
     return output_message_get_size (message);
   return ret;
@@ -756,7 +778,7 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *message = &messages[i];
     gssize len;
 
-    len = socket_send_message (sock, to, message);
+    len = socket_send_message (sock, to, message, FALSE);
 
     if (len < 0) {
       /* Error. */
@@ -772,6 +794,39 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
   return i;
 }
 
+static gint
+socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
+    const NiceOutputMessage *messages, guint n_messages)
+{
+  TurnPriv *priv = (TurnPriv *) sock->priv;
+  guint i;
+
+  /* TURN can depend either on tcp-turn or udp-bsd as a base socket
+   * if we allow reliable send and need to create permissions and we queue the
+   * data, then we must be sure that the reliable send will succeed later, so
+   * we check for udp-bsd here as the base socket and don't allow it.
+   */
+  if (priv->base_socket->type == NICE_SOCKET_TYPE_UDP_BSD)
+    return -1;
+
+  for (i = 0; i < n_messages; i++) {
+    const NiceOutputMessage *message = &messages[i];
+    gssize len;
+
+    len = socket_send_message (sock, to, message, TRUE);
+
+    if (len < 0) {
+      /* Error. */
+      return len;
+    } else if (len == 0) {
+      /* EWOULDBLOCK. */
+      break;
+    }
+  }
+
+  return i;
+}
+
 static gboolean
 socket_is_reliable (NiceSocket *sock)
 {
@@ -1601,13 +1656,16 @@ priv_send_turn_message (TurnPriv *priv, TURNMessage *msg)
     priv->current_binding_msg = NULL;
   }
 
-  nice_socket_send (priv->base_socket, &priv->server_addr,
-      stun_len, (gchar *)msg->buffer);
-
   if (nice_socket_is_reliable (priv->base_socket)) {
+    nice_socket_send_reliable (priv->base_socket, &priv->server_addr,
+        stun_len, (gchar *)msg->buffer);
     stun_timer_start_reliable (&msg->timer,
         STUN_TIMER_DEFAULT_RELIABLE_TIMEOUT);
   } else {
+    if (nice_socket_send_reliable (priv->base_socket, &priv->server_addr,
+            stun_len, (gchar *)msg->buffer) < 0)
+      nice_socket_send (priv->base_socket, &priv->server_addr,
+          stun_len, (gchar *)msg->buffer);
     stun_timer_start (&msg->timer, STUN_TIMER_DEFAULT_TIMEOUT,
         STUN_TIMER_DEFAULT_MAX_RETRANSMISSIONS);
   }
@@ -1660,8 +1718,16 @@ priv_send_create_permission(TurnPriv *priv, StunMessage *resp,
       STUN_USAGE_TURN_COMPATIBILITY_RFC5766);
 
   if (msg_buf_len > 0) {
-    res = nice_socket_send (priv->base_socket, &priv->server_addr,
-        msg_buf_len, (gchar *) msg->buffer);
+    if (nice_socket_is_reliable (priv->base_socket)) {
+      res = nice_socket_send_reliable (priv->base_socket, &priv->server_addr,
+          msg_buf_len, (gchar *) msg->buffer);
+    } else {
+      res = nice_socket_send_reliable (priv->base_socket, &priv->server_addr,
+          msg_buf_len, (gchar *) msg->buffer);
+      if (res < 0)
+        res = nice_socket_send (priv->base_socket, &priv->server_addr,
+            msg_buf_len, (gchar *) msg->buffer);
+    }
 
     if (nice_socket_is_reliable (priv->base_socket)) {
       stun_timer_start_reliable (&msg->timer,
diff --git a/socket/udp-bsd.c b/socket/udp-bsd.c
index 0463405..7075d3f 100644
--- a/socket/udp-bsd.c
+++ b/socket/udp-bsd.c
@@ -61,6 +61,8 @@ static gint socket_recv_messages (NiceSocket *sock,
     NiceInputMessage *recv_messages, guint n_recv_messages);
 static gint socket_send_messages (NiceSocket *sock, const NiceAddress *to,
     const NiceOutputMessage *messages, guint n_messages);
+static gint socket_send_messages_reliable (NiceSocket *sock,
+    const NiceAddress *to, const NiceOutputMessage *messages, guint n_messages);
 static gboolean socket_is_reliable (NiceSocket *sock);
 
 struct UdpBsdSocketPrivate
@@ -144,6 +146,7 @@ nice_udp_bsd_socket_new (NiceAddress *addr)
   sock->type = NICE_SOCKET_TYPE_UDP_BSD;
   sock->fileno = gsock;
   sock->send_messages = socket_send_messages;
+  sock->send_messages_reliable = socket_send_messages_reliable;
   sock->recv_messages = socket_recv_messages;
   sock->is_reliable = socket_is_reliable;
   sock->close = socket_close;
@@ -293,6 +296,13 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
   return i;
 }
 
+static gint
+socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
+    const NiceOutputMessage *messages, guint n_messages)
+{
+  return -1;
+}
+
 static gboolean
 socket_is_reliable (NiceSocket *sock)
 {