Commit ee189732 authored by Olivier Crête's avatar Olivier Crête

outputstream: Add a cancellable to get out of blocking write if the stream is removed

parent 0cca1560
...@@ -92,6 +92,8 @@ struct _NiceOutputStreamPrivate ...@@ -92,6 +92,8 @@ struct _NiceOutputStreamPrivate
GWeakRef/*<NiceAgent>*/ agent_ref; GWeakRef/*<NiceAgent>*/ agent_ref;
guint stream_id; guint stream_id;
guint component_id; guint component_id;
GCancellable *closed_cancellable;
}; };
static void nice_output_stream_dispose (GObject *object); static void nice_output_stream_dispose (GObject *object);
...@@ -191,6 +193,8 @@ nice_output_stream_dispose (GObject *object) ...@@ -191,6 +193,8 @@ nice_output_stream_dispose (GObject *object)
g_weak_ref_clear (&self->priv->agent_ref); g_weak_ref_clear (&self->priv->agent_ref);
g_clear_object (&self->priv->closed_cancellable);
G_OBJECT_CLASS (nice_output_stream_parent_class)->dispose (object); G_OBJECT_CLASS (nice_output_stream_parent_class)->dispose (object);
} }
...@@ -258,6 +262,7 @@ nice_output_stream_init (NiceOutputStream *stream) ...@@ -258,6 +262,7 @@ nice_output_stream_init (NiceOutputStream *stream)
NiceOutputStreamPrivate); NiceOutputStreamPrivate);
g_weak_ref_init (&stream->priv->agent_ref, NULL); g_weak_ref_init (&stream->priv->agent_ref, NULL);
stream->priv->closed_cancellable = g_cancellable_new ();
} }
static void static void
...@@ -351,7 +356,7 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count ...@@ -351,7 +356,7 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count
gssize len = 0; gssize len = 0;
gint n_sent; gint n_sent;
NiceAgent *agent = NULL; /* owned */ NiceAgent *agent = NULL; /* owned */
gulong cancel_id = 0, writeable_id; gulong cancel_id = 0, closed_cancel_id, writeable_id;
WriteData *write_data; WriteData *write_data;
/* Closed streams are not writeable. */ /* Closed streams are not writeable. */
...@@ -379,7 +384,7 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count ...@@ -379,7 +384,7 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count
* GCond solution; would be much better for nice_agent_send() to block * GCond solution; would be much better for nice_agent_send() to block
* properly in the main loop. */ * properly in the main loop. */
write_data = g_slice_new0 (WriteData); write_data = g_slice_new0 (WriteData);
g_atomic_int_set (&write_data->ref_count, 3); g_atomic_int_set (&write_data->ref_count, 4);
g_mutex_init (&write_data->mutex); g_mutex_init (&write_data->mutex);
g_cond_init (&write_data->cond); g_cond_init (&write_data->cond);
...@@ -390,6 +395,10 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count ...@@ -390,6 +395,10 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count
(GDestroyNotify) write_data_unref); (GDestroyNotify) write_data_unref);
} }
closed_cancel_id = g_cancellable_connect (self->priv->closed_cancellable,
(GCallback) write_cancelled_cb, write_data,
(GDestroyNotify) write_data_unref);
g_mutex_lock (&write_data->mutex); g_mutex_lock (&write_data->mutex);
writeable_id = g_signal_connect_data (G_OBJECT (agent), writeable_id = g_signal_connect_data (G_OBJECT (agent),
...@@ -403,7 +412,8 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count ...@@ -403,7 +412,8 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count
* it will take the agent lock which will cause a deadlock if one of * it will take the agent lock which will cause a deadlock if one of
* the callbacks is called. * the callbacks is called.
*/ */
if (g_cancellable_is_cancelled (cancellable)) if (g_cancellable_is_cancelled (cancellable) ||
g_cancellable_is_cancelled (self->priv->closed_cancellable))
break; break;
write_data->writable = FALSE; write_data->writable = FALSE;
...@@ -428,10 +438,15 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count ...@@ -428,10 +438,15 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count
if (cancel_id) if (cancel_id)
g_cancellable_disconnect (cancellable, cancel_id); g_cancellable_disconnect (cancellable, cancel_id);
g_cancellable_disconnect (self->priv->closed_cancellable, closed_cancel_id);
if (len == 0) { if (len == 0) {
g_cancellable_set_error_if_cancelled (cancellable, error);
len = -1; len = -1;
if (!g_cancellable_set_error_if_cancelled (cancellable, error)) {
if (g_cancellable_is_cancelled (self->priv->closed_cancellable))
g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
"Stream has been removed from agent");
}
} }
write_data_unref (write_data); write_data_unref (write_data);
...@@ -608,6 +623,8 @@ streams_removed_cb (NiceAgent *agent, guint *stream_ids, gpointer user_data) ...@@ -608,6 +623,8 @@ streams_removed_cb (NiceAgent *agent, guint *stream_ids, gpointer user_data)
for (i = 0; stream_ids[i] != 0; i++) { for (i = 0; stream_ids[i] != 0; i++) {
if (stream_ids[i] == self->priv->stream_id) { if (stream_ids[i] == self->priv->stream_id) {
/* The socket has been closed. */ /* The socket has been closed. */
g_cancellable_cancel (self->priv->closed_cancellable);
g_output_stream_close (G_OUTPUT_STREAM (self), NULL, NULL); g_output_stream_close (G_OUTPUT_STREAM (self), NULL, NULL);
break; break;
} }
......
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