Commit a13ac437 authored by Ole André Vadla Ravnås's avatar Ole André Vadla Ravnås Committed by Olivier Crête

agent: Add support for bytestream TCP

parent 7bd11cf0
......@@ -178,10 +178,10 @@ struct _NiceAgent
#endif
gchar *software_attribute; /* SOFTWARE attribute */
gboolean reliable; /* property: reliable */
gboolean bytestream_tcp; /* property: bytestream-tcp */
gboolean keepalive_conncheck; /* property: keepalive_conncheck */
GQueue pending_signals;
guint16 rfc4571_expecting_length;
gboolean use_ice_udp;
gboolean use_ice_tcp;
gboolean use_ice_trickle;
......
......@@ -82,9 +82,17 @@
#define MAX_TCP_MTU 1400 /* Use 1400 because of VPNs and we assume IEE 802.3 */
static void agent_consume_next_rfc4571_chunk (NiceAgent *agent,
NiceComponent *component, NiceInputMessage *messages, guint n_messages,
NiceInputMessageIter *iter);
static void
nice_debug_input_message_composition (const NiceInputMessage *messages,
guint n_messages);
static gsize append_buffer_to_input_messages (gboolean bytestream_tcp,
NiceInputMessage *messages, guint n_messages, NiceInputMessageIter *iter,
const guint8 *buffer, gsize buffer_length);
static gsize nice_input_message_iter_get_message_capacity (
NiceInputMessageIter *iter, NiceInputMessage *messages, guint n_messages);
static const gchar *_cand_type_to_sdp (NiceCandidateType type);
G_DEFINE_TYPE (NiceAgent, nice_agent, G_TYPE_OBJECT);
......@@ -151,6 +159,7 @@ static PseudoTcpWriteResult pseudo_tcp_socket_write_packet (PseudoTcpSocket *soc
const gchar *buffer, guint32 len, gpointer user_data);
static void adjust_tcp_clock (NiceAgent *agent, NiceStream *stream, NiceComponent *component);
static void nice_agent_constructed (GObject *object);
static void nice_agent_dispose (GObject *object);
static void nice_agent_get_property (GObject *object,
guint property_id, GValue *value, GParamSpec *pspec);
......@@ -352,6 +361,7 @@ nice_agent_class_init (NiceAgentClass *klass)
{
GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
gobject_class->constructed = nice_agent_constructed;
gobject_class->get_property = nice_agent_get_property;
gobject_class->set_property = nice_agent_set_property;
gobject_class->dispose = nice_agent_dispose;
......@@ -727,7 +737,7 @@ nice_agent_class_init (NiceAgentClass *klass)
/**
* NiceAgent:bytestream-tcp:
*
* This property defines whether receive/send over a TCP or pseudo-TCP, in
* This property defines whether receive/send operations over a TCP socket, in
* reliable mode, are considered as packetized or as bytestream.
* In unreliable mode, every send/recv is considered as packetized, and
* this property is ignored and cannot be set.
......@@ -737,15 +747,11 @@ nice_agent_class_init (NiceAgentClass *klass)
* </para>
* If the property is %TRUE, the stream is considered in bytestream mode
* and data can be read with any receive size. If the property is %FALSE, then
* the stream is considred packetized and each receive will return one packet
* the stream is considered packetized and each receive will return one packet
* of the same size as what was sent from the peer. If in packetized mode,
* then doing a receive with a size smaller than the packet, will cause the
* remaining bytes in the packet to be dropped, breaking the reliability
* of the stream.
* <para>
* This property is currently read-only, and will become read/write once
* bytestream mode will be supported.
* </para>
*
* Since: 0.1.8
*/
......@@ -753,9 +759,9 @@ nice_agent_class_init (NiceAgentClass *klass)
g_param_spec_boolean (
"bytestream-tcp",
"Bytestream TCP",
"Use bytestream mode for reliable TCP and Pseudo-TCP connections",
"Use bytestream mode for reliable TCP connections",
FALSE,
G_PARAM_READABLE));
G_PARAM_READWRITE));
/**
* NiceAgent:keepalive-conncheck:
......@@ -1283,6 +1289,7 @@ nice_agent_init (NiceAgent *agent)
agent->compatibility = NICE_COMPATIBILITY_RFC5245;
agent->reliable = FALSE;
agent->bytestream_tcp = FALSE;
agent->use_ice_udp = TRUE;
agent->use_ice_tcp = TRUE;
......@@ -1294,6 +1301,17 @@ nice_agent_init (NiceAgent *agent)
g_mutex_init (&agent->agent_mutex);
}
static void
nice_agent_constructed (GObject *object)
{
NiceAgent *agent = NICE_AGENT (object);
if (agent->reliable && agent->compatibility == NICE_COMPATIBILITY_GOOGLE)
agent->bytestream_tcp = TRUE;
G_OBJECT_CLASS (nice_agent_parent_class)->constructed (object);
}
NICEAPI_EXPORT NiceAgent *
nice_agent_new (GMainContext *ctx, NiceCompatibility compat)
......@@ -1318,6 +1336,7 @@ nice_agent_new_full (GMainContext *ctx,
"compatibility", compat,
"main-context", ctx,
"reliable", (flags & NICE_AGENT_OPTION_RELIABLE) ? TRUE : FALSE,
"bytestream-tcp", (flags & NICE_AGENT_OPTION_BYTESTREAM_TCP) ? TRUE : FALSE,
"nomination-mode", (flags & NICE_AGENT_OPTION_REGULAR_NOMINATION) ?
NICE_NOMINATION_MODE_REGULAR : NICE_NOMINATION_MODE_AGGRESSIVE,
"full-mode", (flags & NICE_AGENT_OPTION_LITE_MODE) ? FALSE : TRUE,
......@@ -1437,14 +1456,7 @@ nice_agent_get_property (
break;
case PROP_BYTESTREAM_TCP:
if (agent->reliable) {
if (agent->compatibility == NICE_COMPATIBILITY_GOOGLE)
g_value_set_boolean (value, TRUE);
else
g_value_set_boolean (value, FALSE);
} else {
g_value_set_boolean (value, FALSE);
}
g_value_set_boolean (value, agent->bytestream_tcp);
break;
case PROP_KEEPALIVE_CONNCHECK:
......@@ -1669,7 +1681,8 @@ nice_agent_set_property (
break;
case PROP_BYTESTREAM_TCP:
/* TODO: support bytestream mode and set property to writable */
if (agent->reliable && agent->compatibility != NICE_COMPATIBILITY_GOOGLE)
agent->bytestream_tcp = g_value_get_boolean (value);
break;
case PROP_KEEPALIVE_CONNCHECK:
......@@ -4194,6 +4207,9 @@ agent_recv_message_unlocked (
NiceSocket *nicesock,
NiceInputMessage *message)
{
NiceInputMessage *provided_message = message;
NiceInputMessage rfc4571_message;
GInputVector rfc4571_buf;
NiceAddress from;
RecvStatus retval;
gint sockret;
......@@ -4281,11 +4297,19 @@ agent_recv_message_unlocked (
/* In the case of a real ICE-TCP connection, we can use the socket as a
* bytestream and do the read here with caching of data being read
*/
gssize available = g_socket_get_available_bytes (nicesock->fileno);
guint headroom;
gboolean missing_cached_data, have_whole_frame;
/* TODO: Support bytestream reads */
message->length = 0;
sockret = 0;
message->length = 0;
headroom = nice_component_compute_rfc4571_headroom (component);
missing_cached_data = component->rfc4571_frame_size == 0 ||
headroom < component->rfc4571_frame_size;
if (missing_cached_data) {
gssize available = g_socket_get_available_bytes (nicesock->fileno);
if (available <= 0) {
sockret = available;
......@@ -4294,8 +4318,8 @@ agent_recv_message_unlocked (
* both conditions to make sure g_socket_is_connected returns the
* correct result, otherwise we end up closing valid connections
*/
if (g_socket_check_connect_result (nicesock->fileno, NULL) == FALSE ||
g_socket_is_connected (nicesock->fileno) == FALSE) {
if (!g_socket_check_connect_result (nicesock->fileno, NULL) ||
!g_socket_is_connected (nicesock->fileno)) {
/* If we receive a readable event on a TCP_BSD socket which is
* not connected, it means that it failed to connect, so we must
* return an error to make the socket fail/closed
......@@ -4309,84 +4333,67 @@ agent_recv_message_unlocked (
* data available or the peer closed the connection.
* The only way to know is to do a read, so we do here a peek and
* check the return value, if it's 0, it means the peer has closed
* the connection, so we must return an error instead of WOULD_BLOCK
* the connection, so we must return an error instead of
* WOULD_BLOCK
*/
if (g_socket_receive_message (nicesock->fileno, NULL,
NULL, 0, NULL, NULL, &flags, NULL, NULL) == 0)
sockret = -1;
}
} else if (agent->rfc4571_expecting_length == 0) {
if ((gsize) available >= sizeof(guint16)) {
guint16 rfc4571_frame;
GInputVector local_buf = { &rfc4571_frame, sizeof(guint16)};
NiceInputMessage local_message = { &local_buf, 1, message->from, 0};
} else {
GInputVector local_buf = {
component->rfc4571_buffer,
component->rfc4571_buffer_size
};
NiceInputMessage local_message = {
&local_buf, 1, &component->rfc4571_remote_addr, 0
};
sockret = nice_socket_recv_messages (nicesock, &local_message, 1);
if (sockret == 1 && local_message.length >= sizeof (guint16)) {
agent->rfc4571_expecting_length = ntohs (rfc4571_frame);
available = g_socket_get_available_bytes (nicesock->fileno);
if (headroom > 0) {
memmove (component->rfc4571_buffer,
component->rfc4571_buffer + component->rfc4571_frame_offset,
headroom);
local_buf.buffer = (guint8 *) local_buf.buffer + headroom;
local_buf.size -= headroom;
}
component->rfc4571_buffer_offset = headroom;
component->rfc4571_frame_offset = 0;
sockret = nice_socket_recv_messages (nicesock, &local_message, 1);
if (sockret == 1) {
component->rfc4571_buffer_offset += local_message.length;
headroom += local_message.length;
}
}
if (agent->rfc4571_expecting_length > 0) {
GInputVector *local_bufs;
NiceInputMessage local_message;
gsize off;
guint n_bufs = 0;
guint i;
/* Count the number of buffers. */
if (message->n_buffers == -1) {
for (i = 0; message->buffers[i].buffer != NULL; i++)
n_bufs++;
} else {
n_bufs = message->n_buffers;
if (component->rfc4571_frame_size == 0 &&
headroom >= sizeof (guint16)) {
component->rfc4571_frame_size = sizeof (guint16) + ntohs (
*((guint16 *) (component->rfc4571_buffer +
component->rfc4571_frame_offset)));
}
}
local_bufs = g_alloca (n_bufs * sizeof (GInputVector));
local_message.buffers = local_bufs;
local_message.from = message->from;
local_message.length = 0;
local_message.n_buffers = 0;
have_whole_frame = component->rfc4571_frame_size != 0 &&
headroom >= component->rfc4571_frame_size;
if (have_whole_frame) {
rfc4571_buf.buffer = component->rfc4571_buffer +
component->rfc4571_frame_offset + sizeof (guint16);
rfc4571_buf.size = component->rfc4571_frame_size - sizeof (guint16);
/* Only read up to the expected number of bytes in the frame */
off = 0;
for (i = 0; i < n_bufs; i++) {
if (message->buffers[i].size + off < message->length) {
/* Skip already full buffers */
off += message->buffers[i].size;
} else {
gssize diff = 0;
/* If we have a partially full buffer, offset the pointer */
if (off < message->length)
diff = message->length - off;
rfc4571_message.buffers = &rfc4571_buf;
rfc4571_message.n_buffers = 1;
rfc4571_message.from = provided_message->from;
rfc4571_message.length = rfc4571_buf.size;
/* Those buffers are filled */
local_bufs[local_message.n_buffers].buffer =
((char *) message->buffers[i].buffer) + diff;
local_bufs[local_message.n_buffers].size =
MIN (message->buffers[i].size - diff,
agent->rfc4571_expecting_length - off);
off += local_message.buffers[local_message.n_buffers].size;
local_message.n_buffers++;
message = &rfc4571_message;
*message->from = component->rfc4571_remote_addr;
/* If we have a big enough buffer, let's just stop */
if (off == message->length + agent->rfc4571_expecting_length)
break;
}
}
sockret = nice_socket_recv_messages (nicesock, &local_message, 1);
if (sockret == 1) {
message->length += local_message.length;
agent->rfc4571_expecting_length -= local_message.length;
if (agent->rfc4571_expecting_length != 0) {
retval = RECV_WOULD_BLOCK; /* EWOULDBLOCK */
nice_debug_verbose ("%s: Agent %p: TCP message incomplete",
G_STRFUNC, agent);
goto done;
}
}
sockret = 1;
} else {
if (sockret == 1)
sockret = 0;
}
}
}
......@@ -4539,6 +4546,16 @@ agent_recv_message_unlocked (
}
done:
if (message == &rfc4571_message) {
if (retval == RECV_SUCCESS) {
NiceInputMessageIter iter = { 0, 0, 0 };
agent_consume_next_rfc4571_chunk (agent, component, provided_message, 1,
&iter);
} else {
agent_consume_next_rfc4571_chunk (agent, component, NULL, 0, NULL);
}
}
/* Clear local modifications. */
if (message->from == &from) {
message->from = NULL;
......@@ -4547,6 +4564,75 @@ done:
return retval;
}
static void
agent_consume_next_rfc4571_chunk (NiceAgent *agent, NiceComponent *component,
NiceInputMessage *messages, guint n_messages, NiceInputMessageIter *iter)
{
gboolean fully_consumed;
if (messages != NULL) {
gsize bytes_unconsumed, bytes_copied;
bytes_unconsumed = component->rfc4571_frame_size - sizeof (guint16) -
component->rfc4571_consumed_size;
bytes_copied = append_buffer_to_input_messages (agent->bytestream_tcp,
messages, n_messages, iter, component->rfc4571_buffer +
component->rfc4571_frame_offset + component->rfc4571_frame_size -
bytes_unconsumed,
bytes_unconsumed);
component->rfc4571_consumed_size += bytes_copied;
fully_consumed = bytes_copied == bytes_unconsumed || !agent->bytestream_tcp;
} else {
fully_consumed = TRUE;
}
if (fully_consumed) {
guint headroom;
gboolean have_whole_next_frame;
component->rfc4571_frame_offset += component->rfc4571_frame_size;
component->rfc4571_frame_size = 0;
component->rfc4571_consumed_size = 0;
headroom = nice_component_compute_rfc4571_headroom (component);
if (headroom >= sizeof (guint16)) {
component->rfc4571_frame_size = sizeof (guint16) + ntohs (
*((guint16 *) (component->rfc4571_buffer +
component->rfc4571_frame_offset)));
have_whole_next_frame = headroom >= component->rfc4571_frame_size;
} else {
have_whole_next_frame = FALSE;
}
component->rfc4571_wakeup_needed = have_whole_next_frame;
} else {
component->rfc4571_wakeup_needed = TRUE;
}
}
static gboolean
agent_try_consume_next_rfc4571_chunk (NiceAgent *agent,
NiceComponent *component, NiceInputMessage *messages, guint n_messages,
NiceInputMessageIter *iter)
{
guint headroom;
if (component->rfc4571_frame_size == 0)
return FALSE;
headroom = nice_component_compute_rfc4571_headroom (component);
if (headroom < component->rfc4571_frame_size)
return FALSE;
agent_consume_next_rfc4571_chunk (agent, component, messages, n_messages,
iter);
return TRUE;
}
/* Print the composition of an array of messages. No-op if debugging is
* disabled. */
static void
......@@ -4657,6 +4743,48 @@ memcpy_buffer_to_input_message (NiceInputMessage *message,
return message->length;
}
static gsize
append_buffer_to_input_messages (gboolean bytestream_tcp,
NiceInputMessage *messages, guint n_messages, NiceInputMessageIter *iter,
const guint8 *buffer, gsize buffer_length)
{
NiceInputMessage *message = &messages[iter->message];
gsize buffer_offset;
if (iter->buffer == 0 && iter->offset == 0) {
message->length = 0;
}
for (buffer_offset = 0;
(message->n_buffers >= 0 && iter->buffer < (guint) message->n_buffers) ||
(message->n_buffers < 0 && message->buffers[iter->buffer].buffer != NULL);
iter->buffer++) {
GInputVector *v = &message->buffers[iter->buffer];
gsize len;
len = MIN (buffer_length - buffer_offset, v->size - iter->offset);
memcpy ((guint8 *) v->buffer + iter->offset, buffer + buffer_offset, len);
message->length += len;
iter->offset += len;
buffer_offset += len;
if (buffer_offset == buffer_length)
break;
iter->offset = 0;
}
if (!bytestream_tcp || nice_input_message_iter_get_message_capacity (iter,
messages, n_messages) == 0) {
iter->offset = 0;
iter->buffer = 0;
iter->message++;
}
return buffer_offset;
}
/* Concatenate all the buffers in the given @message into a single, newly
* allocated, monolithic buffer which is returned. The length of the new buffer
* is returned in @buffer_length, and should be equal to the length field of
......@@ -4764,6 +4892,28 @@ nice_input_message_iter_get_n_valid_messages (NiceInputMessageIter *iter)
return iter->message + 1;
}
static gsize
nice_input_message_iter_get_message_capacity (NiceInputMessageIter *iter,
NiceInputMessage *messages, guint n_messages)
{
NiceInputMessage *message = &messages[iter->message];
guint i;
gsize total;
if (iter->message == n_messages)
return 0;
total = 0;
for (i = iter->buffer;
(message->n_buffers >= 0 && i < (guint) message->n_buffers) ||
(message->n_buffers < 0 && message->buffers[i].buffer != NULL);
i++) {
total += message->buffers[i].size;
}
return total - iter->offset;
}
/**
* nice_input_message_iter_compare:
* @a: a #NiceInputMessageIter
......@@ -4784,9 +4934,8 @@ nice_input_message_iter_compare (const NiceInputMessageIter *a,
}
/* Will fill up @messages from the first free byte onwards (as determined using
* @iter). This may be used in reliable or non-reliable mode; in non-reliable
* mode it will always increment the message index after each buffer is
* consumed.
* @iter). This may be used in bytestream or packetized mode; in packetized mode
* it will always increment the message index after each buffer is consumed.
*
* Updates @iter in place. No errors can occur.
*
......@@ -4795,12 +4944,12 @@ nice_input_message_iter_compare (const NiceInputMessageIter *a,
*
* Must be called with the io_mutex held. */
static gint
pending_io_messages_recv_messages (NiceComponent *component, gboolean reliable,
NiceInputMessage *messages, guint n_messages, NiceInputMessageIter *iter)
pending_io_messages_recv_messages (NiceComponent *component,
gboolean bytestream_tcp, NiceInputMessage *messages, guint n_messages,
NiceInputMessageIter *iter)
{
gsize len;
IOCallbackData *data;
NiceInputMessage *message = &messages[iter->message];
gsize bytes_copied;
g_assert (component->io_callback_id == 0);
......@@ -4808,47 +4957,13 @@ pending_io_messages_recv_messages (NiceComponent *component, gboolean reliable,
if (data == NULL)
goto done;
if (iter->buffer == 0 && iter->offset == 0) {
message->length = 0;
}
for (;
(message->n_buffers >= 0 && iter->buffer < (guint) message->n_buffers) ||
(message->n_buffers < 0 && message->buffers[iter->buffer].buffer != NULL);
iter->buffer++) {
GInputVector *buffer = &message->buffers[iter->buffer];
do {
len = MIN (data->buf_len - data->offset, buffer->size - iter->offset);
memcpy ((guint8 *) buffer->buffer + iter->offset,
data->buf + data->offset, len);
nice_debug ("%s: Unbuffered %" G_GSIZE_FORMAT " bytes into "
"buffer %p (offset %" G_GSIZE_FORMAT ", length %" G_GSIZE_FORMAT
").", G_STRFUNC, len, buffer->buffer, iter->offset, buffer->size);
message->length += len;
iter->offset += len;
data->offset += len;
} while (iter->offset < buffer->size);
iter->offset = 0;
}
bytes_copied = append_buffer_to_input_messages (bytestream_tcp, messages,
n_messages, iter, data->buf + data->offset, data->buf_len - data->offset);
data->offset += bytes_copied;
/* Only if we managed to consume the whole buffer should it be popped off the
* queue; otherwise we’ll have another go at it later. */
if (data->offset == data->buf_len) {
if (!bytestream_tcp || data->offset == data->buf_len) {
g_queue_pop_head (&component->pending_io_messages);
io_callback_data_free (data);
/* If we’ve consumed an entire message from pending_io_messages, and
* are in non-reliable mode, move on to the next message in
* @messages. */
if (!reliable) {
iter->offset = 0;
iter->buffer = 0;
iter->message++;
}
}
done:
......@@ -4957,7 +5072,7 @@ nice_agent_recv_messages_blocking_or_nonblocking (NiceAgent *agent,
while (!received_enough &&
!g_queue_is_empty (&component->pending_io_messages)) {
pending_io_messages_recv_messages (component, agent->reliable,
pending_io_messages_recv_messages (component, agent->bytestream_tcp,
component->recv_messages, component->n_recv_messages,
&component->recv_messages_iter);
......@@ -4973,6 +5088,15 @@ nice_agent_recv_messages_blocking_or_nonblocking (NiceAgent *agent,
g_mutex_unlock (&component->io_mutex);
if (!received_enough && agent_try_consume_next_rfc4571_chunk (agent,
component, component->recv_messages, component->n_recv_messages,
&component->recv_messages_iter)) {
n_valid_messages = nice_input_message_iter_get_n_valid_messages (
&component->recv_messages_iter);
nice_component_set_io_callback (component, NULL, NULL, NULL, 0, NULL);
goto done;
}
/* For a reliable stream, grab any data from the pseudo-TCP input buffer
* before trying the sockets. */
if (agent->reliable &&
......@@ -5795,6 +5919,108 @@ component_io_cb (GSocket *gsocket, GIOCondition condition, gpointer user_data)
break;
}
has_io_callback = nice_component_has_io_callback (component);
}
} else if (agent->reliable &&
nice_socket_is_reliable (socket_source->socket)) {
NiceInputMessageIter *iter = &component->recv_messages_iter;
gsize total_bytes_received = 0;
while (has_io_callback ||
(component->recv_messages != NULL &&
!nice_input_message_iter_is_at_end (iter,
component->recv_messages, component->n_recv_messages))) {
GInputVector internal_buf = {
component->recv_buffer, component->recv_buffer_size
};
NiceInputMessage internal_message = {
&internal_buf, 1, NULL, 0
};
NiceInputMessage *msg;
guint n_bufs, i;
GInputVector *bufs;
RecvStatus retval = 0;
msg = has_io_callback
? &internal_message
: &component->recv_messages[iter->message];
if (msg->n_buffers == -1) {
n_bufs = 0;
for (i = 0; msg->buffers[i].buffer != NULL; i++)
n_bufs++;
} else {
n_bufs = msg->n_buffers;
}
bufs = g_newa (GInputVector, n_bufs);
memcpy (bufs, msg->buffers, n_bufs * sizeof (GInputVector));
msg->length = 0;
do {
NiceInputMessage m = { bufs, n_bufs, msg->from, 0 };
gsize off;
retval = agent_recv_message_unlocked (agent, stream, component,
socket_source->socket, &m);
if (retval == RECV_WOULD_BLOCK || retval == RECV_ERROR)
break;
if (retval == RECV_OOB)
continue;
msg->length += m.length;
total_bytes_received += m.length;
if (!agent->bytestream_tcp)
break;
off = 0;
for (i = 0; i < n_bufs; i++) {
GInputVector *buf = &bufs[i];
const gsize start = off;
const gsize end = start + buf->size;
if (m.length > start) {
const gsize consumed = MIN (m.length - start, buf->size);
buf->buffer = (guint8 *) buf->buffer + consumed;
buf->size -= consumed;
if (buf->size > 0)
break;
} else {
break;
}
off = end;
}
bufs += i;
n_bufs -= i;
} while (n_bufs > 0);
if (msg->length > 0) {
nice_debug_verbose ("%s: %p: received a valid message with %"
G_GSIZE_FORMAT " bytes", G_STRFUNC, agent, msg->length);
if (has_io_callback) {
nice_component_emit_io_callback (agent, component, msg->length);
} else {
iter->message++;
}
}
if (retval == RECV_WOULD_BLOCK) {
/* EWOULDBLOCK. */
break;
} else if (retval == RECV_ERROR) {
/* Other error. */
nice_debug ("%s: error receiving message", G_STRFUNC);
remove_source = TRUE;
break;
}
if (has_io_callback && g_source_is_destroyed (g_main_current_source ())) {
nice_debug ("Component IO source disappeared during the callback");
goto out;
}
has_io_callback = nice_component_has_io_callback (component);
}
} else if (has_io_callback) {
......
......@@ -409,6 +409,7 @@ typedef enum
* @NICE_AGENT_OPTION_SUPPORT_RENOMINATION: Enable renomination triggered by NOMINATION STUN attribute
* proposed here: https://tools.ietf.org/html/draft-thatcher-ice-renomination-00
* @NICE_AGENT_OPTION_CONSENT_FRESHNESS: Enable RFC 7675 consent freshness support. (Since: 0.1.19)
* @NICE_AGENT_OPTION_BYTESTREAM_TCP: Use bytestream mode for reliable TCP connections. (Since: 0.1.20)
*
* These are options that can be passed to nice_agent_new_full(). They set
* various properties on the agent. Not including them sets the property to
......@@ -424,6 +425,7 @@ typedef enum {
NICE_AGENT_OPTION_ICE_TRICKLE = 1 << 3,
NICE_AGENT_OPTION_SUPPORT_RENOMINATION = 1 << 4,
NICE_AGENT_OPTION_CONSENT_FRESHNESS = 1 << 5,
NICE_AGENT_OPTION_BYTESTREAM_TCP = 1 << 6,
} NiceAgentOption;
/**
......
......@@ -388,6 +388,7 @@ nice_component_close (NiceAgent *agent, NiceStream *stream, NiceComponent *cmp)
}
g_free (cmp->recv_buffer);
g_free (cmp->rfc4571_buffer);
}
/*
......@@ -1129,6 +1130,9 @@ nice_component_init (NiceComponent *component)
component->recv_buffer = g_malloc (MAX_BUFFER_SIZE);
component->recv_buffer_size = MAX_BUFFER_SIZE;
component->rfc4571_buffer_size = sizeof (guint16) + G_MAXUINT16;
component->rfc4571_buffer = g_malloc (component->rfc4571_buffer_size);
}
static void
......@@ -1304,6 +1308,9 @@ static gboolean
component_source_prepare (GSource *source, gint *timeout_)
{
ComponentSource *component_source = (ComponentSource *) source;
/* We can’t be sure if the ComponentSource itself needs to be dispatched until
* poll() is called on all the child sources. */
gboolean skip_poll = FALSE;
NiceAgent *agent;
NiceComponent *component;
GSList *parentl, *childl;
......@@ -1320,6 +1327,11 @@ component_source_prepare (GSource *source, gint *timeout_)
&component))
goto done;
if (component->rfc4571_wakeup_needed) {
component->rfc4571_wakeup_needed = FALSE;
skip_poll = TRUE;
goto done;
}
if (component->socket_sources_age ==
component_source->component_socket_sources_age)
......@@ -1393,9 +1405,7 @@ component_source_prepare (GSource *source, gint *timeout_)
agent_unlock_and_emit (agent);
g_object_unref (agent);
/* We can’t be sure if the ComponentSource itself needs to be dispatched until
* poll() is called on all the child sources. */
return FALSE;
return skip_poll;
}
static gboolean
......@@ -1659,3 +1669,9 @@ nice_component_get_sockets (NiceComponent *component)
return array;
}
guint
nice_component_compute_rfc4571_headroom (NiceComponent *component)
{
return component->rfc4571_buffer_offset - component->rfc4571_frame_offset;
}
......@@ -241,6 +241,16 @@ struct _NiceComponent {
*/
guint8 *recv_buffer;
guint recv_buffer_size;
/* ICE-TCP frame state */
guint8 *rfc4571_buffer;
guint rfc4571_buffer_offset;
guint rfc4571_buffer_size;
guint rfc4571_frame_offset;
guint rfc4571_frame_size;
guint rfc4571_consumed_size;
NiceAddress rfc4571_remote_addr;
gboolean rfc4571_wakeup_needed;
};
typedef struct {
......@@ -331,6 +341,9 @@ nice_component_verify_remote_candidate (NiceComponent *component,
GPtrArray *
nice_component_get_sockets (NiceComponent *component);
guint
nice_component_compute_rfc4571_headroom (NiceComponent *component);
G_END_DECLS
#endif /* _NICE_COMPONENT_H */
......
......@@ -23,6 +23,7 @@ nice_tests = [
'test-trickle',
'test-tcp',
'test-icetcp',
'test-bytestream-tcp',
'test-credentials',
'test-turn',
'test-drop-invalid',
......@@ -40,7 +41,7 @@ if cc.has_header('arpa/inet.h')
endif
foreach tname : nice_tests
if tname.startswith('test-io-stream') or tname.startswith('test-send-recv')
if tname.startswith('test-io-stream') or tname.startswith('test-send-recv') or tname == 'test-bytestream-tcp'
extra_src = ['test-io-stream-common.c']
else
extra_src = []
......
/*
* This file is part of the Nice GLib ICE library.
*
* (C) 2013 Collabora Ltd.
* Contact: Philip Withnall
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is the Nice GLib ICE library.
*
* The Initial Developers of the Original Code are Collabora Ltd and Nokia
* Corporation. All Rights Reserved.
*
* Contributors:
* Philip Withnall, Collabora Ltd.
*
* Alternatively, the contents of this file may be used under the terms of the
* the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
* case the provisions of LGPL are applicable instead of those above. If you
* wish to allow use of your version of this file only under the terms of the
* LGPL and not to allow others to use your version of this file under the
* MPL, indicate your decision by deleting the provisions above and replace
* them with the notice and other provisions required by the LGPL. If you do
* not delete the provisions above, a recipient may use your version of this
* file under either the MPL or the LGPL.
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include "agent.h"
#include "test-io-stream-common.h"
#include <stdlib.h>
#include <string.h>
#ifndef G_OS_WIN32
#include <unistd.h>
#endif
typedef struct {
gsize recv_count;
gsize *other_recv_count;
gsize send_count;
gsize *other_send_count;
} ThreadData;
static const gchar test_sequence[] = { '1', '2', '3', '4' };
static void
read_thread_cb (GInputStream *input_stream, TestIOStreamThreadData *data)
{
ThreadData *user_data = data->user_data;
guint8 buf[2];
GError *error = NULL;
g_input_stream_read_all (input_stream, buf, sizeof (buf), NULL, NULL, &error);
g_assert_no_error (error);
user_data->recv_count++;
g_assert_cmpmem (buf, sizeof (buf), test_sequence, sizeof (buf));
g_input_stream_read_all (input_stream, buf, sizeof (buf), NULL, NULL, &error);
g_assert_no_error (error);
user_data->recv_count++;
g_assert_cmpmem (buf, sizeof (buf), test_sequence + 2, sizeof (buf));
check_for_termination (data, &user_data->recv_count,
user_data->other_recv_count, &user_data->send_count, 2);
}
static void
write_thread_cb (GOutputStream *output_stream, TestIOStreamThreadData *data)
{
ThreadData *user_data = data->user_data;
GError *error = NULL;
g_output_stream_write_all (output_stream, test_sequence,
sizeof (test_sequence), NULL, NULL, &error);
g_assert_no_error (error);
user_data->send_count += 2;
}
int main (void)
{
ThreadData *l_data, *r_data;
const TestIOStreamCallbacks callbacks = {
read_thread_cb,
write_thread_cb,
NULL,
NULL,
};
#ifdef G_OS_WIN32
WSADATA w;
WSAStartup (0x0202, &w);
#endif
l_data = g_malloc0 (sizeof (ThreadData));
r_data = g_malloc0 (sizeof (ThreadData));
l_data->recv_count = 0;
l_data->other_recv_count = &r_data->recv_count;
l_data->send_count = 0;
l_data->other_send_count = &r_data->send_count;
r_data->recv_count = 0;
r_data->other_recv_count = &l_data->recv_count;
r_data->send_count = 0;
r_data->other_send_count = &l_data->send_count;
run_io_stream_test (30, TRUE, &callbacks, l_data, NULL, r_data, NULL,
TEST_IO_STREAM_OPTION_TCP_ONLY | TEST_IO_STREAM_OPTION_BYTESTREAM_TCP);
g_free (r_data);
g_free (l_data);
#ifdef G_OS_WIN32
WSACleanup ();
#endif
return 0;
}
......@@ -128,7 +128,7 @@ int main (void)
r_cancellation_thread = spawn_thread ("libnice R cancel",
cancellation_thread_cb, &r_data);
run_io_stream_test (30, TRUE, &callbacks, &l_data, NULL, &r_data, NULL);
run_io_stream_test (30, TRUE, &callbacks, &l_data, NULL, &r_data, NULL, 0);
g_thread_join (l_cancellation_thread);
g_thread_join (r_cancellation_thread);
......
......@@ -128,7 +128,8 @@ int main (void)
WSAStartup (0x0202, &w);
#endif
run_io_stream_test (30, TRUE, &callbacks, (gpointer) TRUE, NULL, NULL, NULL);
run_io_stream_test (30, TRUE, &callbacks, (gpointer) TRUE, NULL, NULL, NULL,
0);
#ifdef G_OS_WIN32
WSACleanup ();
......
......@@ -128,7 +128,8 @@ int main (void)
WSAStartup (0x0202, &w);
#endif
run_io_stream_test (30, TRUE, &callbacks, (gpointer) TRUE, NULL, NULL, NULL);
run_io_stream_test (30, TRUE, &callbacks, (gpointer) TRUE, NULL, NULL, NULL,
0);
#ifdef G_OS_WIN32
WSACleanup ();
......
......@@ -243,7 +243,8 @@ new_selected_pair_cb (NiceAgent *agent, guint stream_id, guint component_id,
static NiceAgent *
create_agent (gboolean controlling_mode, TestIOStreamThreadData *data,
GMainContext **main_context, GMainLoop **main_loop)
GMainContext **main_context, GMainLoop **main_loop,
TestIOStreamOption flags)
{
NiceAgent *agent;
NiceAddress base_addr;
......@@ -253,17 +254,29 @@ create_agent (gboolean controlling_mode, TestIOStreamThreadData *data,
*main_context = g_main_context_new ();
*main_loop = g_main_loop_new (*main_context, FALSE);
/* Use Google compatibility to ignore credentials. */
if (data->reliable)
agent = nice_agent_new_reliable (*main_context, NICE_COMPATIBILITY_GOOGLE);
agent = nice_agent_new_reliable (*main_context, NICE_COMPATIBILITY_RFC5245);
else
agent = nice_agent_new (*main_context, NICE_COMPATIBILITY_GOOGLE);
agent = nice_agent_new (*main_context, NICE_COMPATIBILITY_RFC5245);
g_object_set (G_OBJECT (agent),
"controlling-mode", controlling_mode,
"upnp", FALSE,
NULL);
if (flags & TEST_IO_STREAM_OPTION_TCP_ONLY) {
g_object_set (G_OBJECT (agent),
"ice-udp", FALSE,
"ice-tcp", TRUE,
NULL);
}
if (flags & TEST_IO_STREAM_OPTION_BYTESTREAM_TCP) {
g_object_set (G_OBJECT (agent),
"bytestream-tcp", TRUE,
NULL);
}
/* Specify which local interface to use. */
g_assert_true (nice_address_set_from_string (&base_addr, "127.0.0.1"));
nice_agent_add_local_address (agent, &base_addr);
......@@ -310,6 +323,28 @@ add_stream (NiceAgent *agent)
GUINT_TO_POINTER (stream_id));
}
static void
swap_credentials (NiceAgent *agent)
{
guint stream_id;
gchar *ufrag, *password;
NiceAgent *other_agent;
guint other_stream_id;
stream_id = GPOINTER_TO_UINT (
g_object_get_data (G_OBJECT (agent), "stream-id"));
nice_agent_get_local_credentials (agent, stream_id, &ufrag, &password);
other_agent = g_object_get_data (G_OBJECT (agent), "other-agent");
other_stream_id = GPOINTER_TO_UINT (
g_object_get_data (G_OBJECT (other_agent), "stream-id"));
nice_agent_set_remote_credentials (other_agent, other_stream_id, ufrag,
password);
g_free (ufrag);
g_free (password);
}
static void
run_agent (TestIOStreamThreadData *data, NiceAgent *agent)
{
......@@ -346,7 +381,8 @@ void
run_io_stream_test (guint deadlock_timeout, gboolean reliable,
const TestIOStreamCallbacks *callbacks,
gpointer l_user_data, GDestroyNotify l_user_data_free,
gpointer r_user_data, GDestroyNotify r_user_data_free)
gpointer r_user_data, GDestroyNotify r_user_data_free,
TestIOStreamOption flags)
{
GMainLoop *error_loop;
GThread *l_main_thread, *r_main_thread;
......@@ -396,9 +432,9 @@ run_io_stream_test (guint deadlock_timeout, gboolean reliable,
/* Create the L and R agents. */
l_data.agent = create_agent (TRUE, &l_data,
&l_data.main_context, &l_data.main_loop);
&l_data.main_context, &l_data.main_loop, flags);
r_data.agent = create_agent (FALSE, &r_data,
&r_data.main_context, &r_data.main_loop);
&r_data.main_context, &r_data.main_loop, flags);
g_object_set_data (G_OBJECT (l_data.agent), "other-agent", r_data.agent);
g_object_set_data (G_OBJECT (r_data.agent), "other-agent", l_data.agent);
......@@ -411,6 +447,8 @@ run_io_stream_test (guint deadlock_timeout, gboolean reliable,
add_stream (l_data.agent);
add_stream (r_data.agent);
swap_credentials (l_data.agent);
swap_credentials (r_data.agent);
run_agent (&l_data, l_data.agent);
run_agent (&r_data, r_data.agent);
......
......@@ -105,12 +105,19 @@ struct _TestIOStreamThreadData {
guint *start_count;
};
typedef enum
{
TEST_IO_STREAM_OPTION_TCP_ONLY = 1 << 0,
TEST_IO_STREAM_OPTION_BYTESTREAM_TCP = 1 << 1,
} TestIOStreamOption;
GThread *spawn_thread (const gchar *thread_name, GThreadFunc thread_func,
gpointer user_data);
void run_io_stream_test (guint deadlock_timeout, gboolean reliable,
const TestIOStreamCallbacks *callbacks,
gpointer l_user_data, GDestroyNotify l_user_data_free,
gpointer r_user_data, GDestroyNotify r_user_data_free);
gpointer r_user_data, GDestroyNotify r_user_data_free,
TestIOStreamOption flags);
void check_for_termination (TestIOStreamThreadData *data, gsize *recv_count,
gsize *other_recv_count, volatile gsize *send_count, gsize expected_recv_count);
......
......@@ -178,7 +178,7 @@ int main (void)
r_data->other_send_count = &l_data->send_count;
r_data->recv_offset = 0;
run_io_stream_test (30, TRUE, &callbacks, l_data, NULL, r_data, NULL);
run_io_stream_test (30, TRUE, &callbacks, l_data, NULL, r_data, NULL, 0);
g_free (r_data);
g_free (l_data);
......
......@@ -143,7 +143,7 @@ int main (void)
r_data->send_count = 0;
r_data->other_send_count = &l_data->send_count;
run_io_stream_test (30, TRUE, &callbacks, l_data, NULL, r_data, NULL);
run_io_stream_test (30, TRUE, &callbacks, l_data, NULL, r_data, NULL, 0);
/* Verify that correct number of local candidates were reported. */
g_assert_cmpuint (l_data->cand_count, ==, 1);
......
......@@ -1146,7 +1146,9 @@ test (gboolean reliable, StreamApi stream_api, gsize n_bytes, guint n_messages,
&l_data.received_bytes, &l_data.received_messages);
run_io_stream_test (deadlock_timeout, reliable, &callbacks[stream_api],
&l_data, NULL, &r_data, NULL);
&l_data, NULL, &r_data, NULL,
/* Ensure TCP has the same behavior as Pseudo-TCP in reliable mode: */
reliable ? TEST_IO_STREAM_OPTION_BYTESTREAM_TCP : 0);
test_data_clear (&r_data);
test_data_clear (&l_data);
......
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