Commit 535701b0 authored by Youness Alaoui's avatar Youness Alaoui Committed by Olivier Crête

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.
parent 06cb5a12
......@@ -229,6 +229,9 @@ compact_output_message (const NiceOutputMessage *message, gsize *buffer_length);
gsize
output_message_get_size (const NiceOutputMessage *message);
gssize agent_socket_send (NiceSocket *sock, const NiceAddress *addr, gsize len,
const gchar *buf);
/*
* nice_debug_init:
*
......
......@@ -4853,3 +4853,29 @@ nice_agent_forget_relays (NiceAgent *agent, guint stream_id, guint component_id)
return ret;
}
/* Helper function to allow us to send connchecks reliably.
* If the transport is reliable, then we request a reliable send, which will
* either send the data, or queue it in the case of unestablished http/socks5
* proxies or tcp-turn. If the transport is not reliable, then it could be an
* unreliable tcp-bsd, so we still try a reliable send to see if it can succeed
* meaning the message was queued, or if it failed, then it was either udp-bsd
* or turn and so we retry with a non reliable send and let the retransmissions
* take care of the rest.
* This is in order to avoid having to retransmit something if the underlying
* socket layer can queue the message and send it once a connection is
* established.
*/
gssize
agent_socket_send (NiceSocket *sock, const NiceAddress *addr, gsize len,
const gchar *buf)
{
if (nice_socket_is_reliable (sock)) {
return nice_socket_send_reliable (sock, addr, len, buf);
} else {
gssize ret = nice_socket_send_reliable (sock, addr, len, buf);
if (ret < 0)
ret = nice_socket_send (sock, addr, len, buf);
return ret;
}
}
......@@ -291,7 +291,7 @@ static gboolean priv_conn_check_tick_stream (Stream *stream, NiceAgent *agent, G
nice_debug ("Agent %p :STUN transaction retransmitted (timeout %dms).",
agent, timeout);
nice_socket_send (p->local->sockptr, &p->remote->addr,
agent_socket_send (p->local->sockptr, &p->remote->addr,
stun_message_length (&p->stun_message),
(gchar *)p->stun_buffer);
......@@ -518,7 +518,7 @@ static gboolean priv_conn_keepalive_retransmissions_tick (gpointer pointer)
}
case STUN_USAGE_TIMER_RETURN_RETRANSMIT:
/* Retransmit */
nice_socket_send (pair->local->sockptr, &pair->remote->addr,
agent_socket_send (pair->local->sockptr, &pair->remote->addr,
stun_message_length (&pair->keepalive.stun_message),
(gchar *)pair->keepalive.stun_buffer);
......@@ -615,7 +615,7 @@ static gboolean priv_conn_keepalive_tick_unlocked (NiceAgent *agent)
agent->media_after_tick = FALSE;
/* send the conncheck */
nice_socket_send (p->local->sockptr, &p->remote->addr,
agent_socket_send (p->local->sockptr, &p->remote->addr,
buf_len, (gchar *)p->keepalive.stun_buffer);
if (p->keepalive.tick_source != NULL) {
......@@ -642,7 +642,7 @@ static gboolean priv_conn_keepalive_tick_unlocked (NiceAgent *agent)
sizeof(p->keepalive.stun_buffer));
if (buf_len > 0) {
nice_socket_send (p->local->sockptr, &p->remote->addr, buf_len,
agent_socket_send (p->local->sockptr, &p->remote->addr, buf_len,
(gchar *)p->keepalive.stun_buffer);
nice_debug ("Agent %p : stun_bind_keepalive for pair %p res %d.",
......@@ -684,7 +684,7 @@ static gboolean priv_conn_keepalive_tick_unlocked (NiceAgent *agent)
/* send the conncheck */
nice_debug ("Agent %p : resending STUN on %s to keep the "
"candidate alive.", agent, candidate->foundation);
nice_socket_send (candidate->sockptr, &stun_server,
agent_socket_send (candidate->sockptr, &stun_server,
buffer_len, (gchar *)stun_buffer);
}
}
......@@ -766,7 +766,7 @@ static gboolean priv_turn_allocate_refresh_retransmissions_tick (gpointer pointe
}
case STUN_USAGE_TIMER_RETURN_RETRANSMIT:
/* Retransmit */
nice_socket_send (cand->nicesock, &cand->server,
agent_socket_send (cand->nicesock, &cand->server,
stun_message_length (&cand->stun_message), (gchar *)cand->stun_buffer);
cand->tick_source = agent_timeout_add_with_context (cand->agent,
......@@ -836,7 +836,7 @@ static void priv_turn_allocate_refresh_tick_unlocked (CandidateRefresh *cand)
STUN_TIMER_DEFAULT_MAX_RETRANSMISSIONS);
/* send the refresh */
nice_socket_send (cand->nicesock, &cand->server,
agent_socket_send (cand->nicesock, &cand->server,
buffer_len, (gchar *)cand->stun_buffer);
cand->tick_source = agent_timeout_add_with_context (cand->agent,
......@@ -1717,7 +1717,7 @@ int conn_check_send (NiceAgent *agent, CandidateCheckPair *pair)
STUN_TIMER_DEFAULT_MAX_RETRANSMISSIONS);
/* send the conncheck */
nice_socket_send (pair->local->sockptr, &pair->remote->addr,
agent_socket_send (pair->local->sockptr, &pair->remote->addr,
buffer_len, (gchar *)pair->stun_buffer);
timeout = stun_timer_remainder (&pair->timer);
......@@ -1926,7 +1926,7 @@ static void priv_reply_to_conn_check (NiceAgent *agent, Stream *stream, Componen
(int)use_candidate);
}
nice_socket_send (sockptr, toaddr, rbuf_len, (const gchar*)rbuf);
agent_socket_send (sockptr, toaddr, rbuf_len, (const gchar*)rbuf);
if (rcand) {
/* note: upon successful check, make the reserve check immediately */
......@@ -2829,7 +2829,7 @@ gboolean conn_check_handle_inbound_stun (NiceAgent *agent, Stream *stream,
rbuf_len = stun_agent_build_unknown_attributes_error (&agent->stun_agent,
&msg, rbuf, rbuf_len, &req);
if (rbuf_len != 0)
nice_socket_send (nicesock, from, rbuf_len, (const gchar*)rbuf);
agent_socket_send (nicesock, from, rbuf_len, (const gchar*)rbuf);
}
return TRUE;
}
......@@ -2842,7 +2842,7 @@ gboolean conn_check_handle_inbound_stun (NiceAgent *agent, Stream *stream,
rbuf_len = stun_agent_finish_message (&agent->stun_agent, &msg, NULL, 0);
if (rbuf_len > 0 && agent->compatibility != NICE_COMPATIBILITY_MSN &&
agent->compatibility != NICE_COMPATIBILITY_OC2007)
nice_socket_send (nicesock, from, rbuf_len, (const gchar*)rbuf);
agent_socket_send (nicesock, from, rbuf_len, (const gchar*)rbuf);
}
return TRUE;
}
......@@ -2853,7 +2853,7 @@ gboolean conn_check_handle_inbound_stun (NiceAgent *agent, Stream *stream,
rbuf_len = stun_agent_finish_message (&agent->stun_agent, &msg, NULL, 0);
if (rbuf_len > 0 && agent->compatibility != NICE_COMPATIBILITY_MSN &&
agent->compatibility != NICE_COMPATIBILITY_OC2007)
nice_socket_send (nicesock, from, rbuf_len, (const gchar*)rbuf);
agent_socket_send (nicesock, from, rbuf_len, (const gchar*)rbuf);
}
return TRUE;
}
......
......@@ -177,10 +177,10 @@ static void refresh_free_item (CandidateRefresh *cand)
stun_agent_forget_transaction (&cand->stun_agent, id);
/* send the refresh twice since we won't do retransmissions */
nice_socket_send (cand->nicesock, &cand->server,
agent_socket_send (cand->nicesock, &cand->server,
buffer_len, (gchar *)cand->stun_buffer);
if (!nice_socket_is_reliable (cand->nicesock)) {
nice_socket_send (cand->nicesock, &cand->server,
agent_socket_send (cand->nicesock, &cand->server,
buffer_len, (gchar *)cand->stun_buffer);
}
......@@ -918,7 +918,7 @@ static gboolean priv_discovery_tick_unlocked (gpointer pointer)
}
/* send the conncheck */
nice_socket_send (cand->nicesock, &cand->server,
agent_socket_send (cand->nicesock, &cand->server,
buffer_len, (gchar *)cand->stun_buffer);
/* case: success, start waiting for the result */
......@@ -974,7 +974,7 @@ static gboolean priv_discovery_tick_unlocked (gpointer pointer)
timeout);
/* retransmit */
nice_socket_send (cand->nicesock, &cand->server,
agent_socket_send (cand->nicesock, &cand->server,
stun_message_length (&cand->stun_message),
(gchar *)cand->stun_buffer);
......
......@@ -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)
......
......@@ -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)
{
......
......@@ -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)
{
......
......@@ -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);
......
......@@ -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;
......@@ -435,6 +439,28 @@ socket_send_messages (NiceSocket *sock, const NiceAddress *to,
n_messages);
} 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);
}
......
......@@ -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);
......
......@@ -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)
......
......@@ -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,
......
......@@ -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)
{
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment