agent: Close pseudo-TCP streams when closing G[IO|Input|Output]Streams

This means that users of the Nice[Input|Output|IO]Stream API can easily
close TCP connections without having to hack around with libnice
internals.

agent/inputstream.c

@@ -50,8 +50,8 @@
  * component triple, and will be closed as soon as that stream is removed from
  * the agent (e.g. if nice_agent_remove_stream() is called from another thread).
  * If g_input_stream_close() is called on a #NiceInputStream, the input stream
- * will be marked as closed, but the underlying #NiceAgent stream will not be
- * removed. Use nice_agent_remove_stream() to do that.
+ * and underlying #NiceAgent stream will be closed, but the underlying stream
+ * will not be removed. Use nice_agent_remove_stream() to do that.
  *
  * Since: 0.1.5
  */
@@ -96,6 +96,8 @@ static void nice_input_stream_set_property (GObject *object, guint prop_id,
     const GValue *value, GParamSpec *pspec);
 static gssize nice_input_stream_read (GInputStream *stream, void *buffer,
     gsize count, GCancellable *cancellable, GError **error);
+static gboolean nice_input_stream_close (GInputStream *stream,
+    GCancellable *cancellable, GError **error);
 static gboolean nice_input_stream_is_readable (GPollableInputStream *stream);
 static gssize nice_input_stream_read_nonblocking (GPollableInputStream *stream,
     void *buffer, gsize count, GError **error);
@@ -115,6 +117,7 @@ nice_input_stream_class_init (NiceInputStreamClass *klass)
   gobject_class->dispose = nice_input_stream_dispose;
 
   stream_class->read_fn = nice_input_stream_read;
+  stream_class->close_fn = nice_input_stream_close;
 
   /***
    * NiceInputStream:agent:
@@ -174,6 +177,11 @@ nice_input_stream_dispose (GObject *object)
   NiceInputStream *self = NICE_INPUT_STREAM (object);
   NiceAgent *agent;
 
+  /* Ensure the stream is closed first, otherwise the agent can’t be found in
+   * the close handler called by the parent implementation. */
+  if (!g_input_stream_is_closed (G_INPUT_STREAM (object)))
+    g_input_stream_close (G_INPUT_STREAM (object), NULL, NULL);
+
   agent = g_weak_ref_get (&self->priv->agent_ref);
   if (agent != NULL) {
     g_signal_handlers_disconnect_by_func (agent, streams_removed_cb, self);
@@ -321,6 +329,36 @@ nice_input_stream_read (GInputStream *stream, void *buffer, gsize count,
   return len;
 }
 
+static gboolean
+nice_input_stream_close (GInputStream *stream, GCancellable *cancellable,
+    GError **error)
+{
+  NiceInputStreamPrivate *priv = NICE_INPUT_STREAM (stream)->priv;
+  Component *component = NULL;
+  Stream *_stream = NULL;
+  NiceAgent *agent;  /* owned */
+
+  /* Has the agent disappeared? */
+  agent = g_weak_ref_get (&priv->agent_ref);
+  if (agent == NULL)
+    return TRUE;
+
+  agent_lock ();
+
+  /* Shut down the read side of the pseudo-TCP stream, if it still exists. */
+  if (agent_find_component (agent, priv->stream_id, priv->component_id,
+          &_stream, &component) &&
+      component->tcp != NULL) {
+    pseudo_tcp_socket_shutdown (component->tcp, PSEUDO_TCP_SHUTDOWN_RD);
+  }
+
+  agent_unlock ();
+
+  g_object_unref (agent);
+
+  return TRUE;
+}
+
 static gboolean
 nice_input_stream_is_readable (GPollableInputStream *stream)
 {

agent/iostream.c

@@ -50,9 +50,10 @@
  * A single #NiceIOStream can only be used with a single agent, stream and
  * component triple, and will be closed as soon as that stream is removed from
  * the agent (e.g. if nice_agent_remove_stream() is called from another thread).
- * If g_io_stream_close() is called on a #NiceIOStream, the I/O stream will be
- * marked as closed in both directions, but the underlying #NiceAgent stream
- * will not be removed. Use nice_agent_remove_stream() to do that.
+ * If g_io_stream_close() is called on a #NiceIOStream, the I/O stream and
+ * underlying #NiceAgent stream will be closed in both directions, but the
+ * underlying stream will not be removed. Use nice_agent_remove_stream() to do
+ * that.
  *
  * Since: 0.1.5
  */

agent/outputstream.c

@@ -50,8 +50,8 @@
  * component triple, and will be closed as soon as that stream is removed from
  * the agent (e.g. if nice_agent_remove_stream() is called from another thread).
  * If g_output_stream_close() is called on a #NiceOutputStream, the output
- * stream will be marked as closed, but the underlying #NiceAgent stream will
- * not be removed. Use nice_agent_remove_stream() to do that.
+ * stream and underlying #NiceAgent stream will be closed, but the underlying
+ * stream will not be removed. Use nice_agent_remove_stream() to do that.
  *
  * The output stream can only be used once the
  * #NiceAgent::reliable-transport-writable signal has been received for the
@@ -104,6 +104,8 @@ static void nice_output_stream_set_property (GObject *object, guint prop_id,
 
 static gssize nice_output_stream_write (GOutputStream *stream,
     const void *buffer, gsize count, GCancellable *cancellable, GError **error);
+static gboolean nice_output_stream_close (GOutputStream *stream,
+    GCancellable *cancellable, GError **error);
 
 static gboolean nice_output_stream_is_writable (GPollableOutputStream *stream);
 static gssize nice_output_stream_write_nonblocking (
@@ -122,6 +124,7 @@ nice_output_stream_class_init (NiceOutputStreamClass *klass)
   g_type_class_add_private (klass, sizeof (NiceOutputStreamPrivate));
 
   stream_class->write_fn = nice_output_stream_write;
+  stream_class->close_fn = nice_output_stream_close;
 
   gobject_class->set_property = nice_output_stream_set_property;
   gobject_class->get_property = nice_output_stream_get_property;
@@ -185,6 +188,11 @@ nice_output_stream_dispose (GObject *object)
   NiceOutputStream *self = NICE_OUTPUT_STREAM (object);
   NiceAgent *agent;
 
+  /* Ensure the stream is closed first, otherwise the agent can’t be found in
+   * the close handler called by the parent implementation. */
+  if (!g_output_stream_is_closed (G_OUTPUT_STREAM (object)))
+    g_output_stream_close (G_OUTPUT_STREAM (object), NULL, NULL);
+
   agent = g_weak_ref_get (&self->priv->agent_ref);
   if (agent != NULL) {
     g_signal_handlers_disconnect_by_func (agent, streams_removed_cb, self);
@@ -463,6 +471,36 @@ nice_output_stream_write (GOutputStream *stream, const void *buffer, gsize count
   return len;
 }
 
+static gboolean
+nice_output_stream_close (GOutputStream *stream, GCancellable *cancellable,
+    GError **error)
+{
+  NiceOutputStreamPrivate *priv = NICE_OUTPUT_STREAM (stream)->priv;
+  Component *component = NULL;
+  Stream *_stream = NULL;
+  NiceAgent *agent;  /* owned */
+
+  /* Has the agent disappeared? */
+  agent = g_weak_ref_get (&priv->agent_ref);
+  if (agent == NULL)
+    return TRUE;
+
+  agent_lock ();
+
+  /* Shut down the write side of the pseudo-TCP stream. */
+  if (agent_find_component (agent, priv->stream_id, priv->component_id,
+          &_stream, &component) &&
+      component->tcp != NULL) {
+    pseudo_tcp_socket_shutdown (component->tcp, PSEUDO_TCP_SHUTDOWN_WR);
+  }
+
+  agent_unlock ();
+
+  g_object_unref (agent);
+
+  return TRUE;
+}
+
 static gboolean
 nice_output_stream_is_writable (GPollableOutputStream *stream)
 {