more work on cleanup

src/client/pv-context.c

@@ -405,6 +405,12 @@ subscription_cb (PvSubscribe         *subscribe,
       break;
 
     case PV_SUBSCRIPTION_FLAGS_CLIENT:
+      if (event == PV_SUBSCRIPTION_EVENT_REMOVE) {
+        if (object == priv->client) {
+          priv->error = g_error_new_literal (G_IO_ERROR, G_IO_ERROR_CLOSED, "Client disappeared");
+          context_set_state (context, PV_CONTEXT_STATE_ERROR);
+        }
+      }
       break;
 
     case PV_SUBSCRIPTION_FLAGS_SOURCE:

src/client/pv-stream.c

@@ -34,6 +34,8 @@ struct _PvStreamPrivate
   gchar *name;
   GVariant *properties;
 
+  guint id;
+
   PvStreamState state;
   GError *error;
 
@@ -152,6 +154,54 @@ pv_stream_set_property (GObject      *_object,
     }
 }
 
+static void
+stream_set_state (PvStream *stream, PvStreamState state)
+{
+  if (stream->priv->state != state) {
+    stream->priv->state = state;
+    g_object_notify (G_OBJECT (stream), "state");
+  }
+}
+
+static void
+subscription_cb (PvSubscribe         *subscribe,
+                 PvSubscriptionEvent  event,
+                 PvSubscriptionFlags  flags,
+                 GDBusProxy          *object,
+                 gpointer             user_data)
+{
+  PvStream *stream = PV_STREAM (user_data);
+  PvStreamPrivate *priv = stream->priv;
+
+  switch (flags) {
+    case PV_SUBSCRIPTION_FLAGS_SOURCE_OUTPUT:
+      if (event == PV_SUBSCRIPTION_EVENT_REMOVE) {
+        if (object == priv->source_output) {
+          priv->error = g_error_new_literal (G_IO_ERROR, G_IO_ERROR_CLOSED, "output disappeared");
+          stream_set_state (stream, PV_STREAM_STATE_ERROR);
+        }
+      }
+      break;
+
+    default:
+      break;
+  }
+}
+
+static void
+pv_stream_constructed (GObject * object)
+{
+  PvStream *stream = PV_STREAM (object);
+  PvStreamPrivate *priv = stream->priv;
+
+  priv->id = g_signal_connect (priv->context->priv->subscribe,
+                    "subscription-event",
+                    (GCallback) subscription_cb,
+                    stream);
+
+  G_OBJECT_CLASS (pv_stream_parent_class)->constructed (object);
+}
+
 static void
 pv_stream_finalize (GObject * object)
 {
@@ -173,6 +223,7 @@ pv_stream_finalize (GObject * object)
 
   if (priv->properties)
     g_variant_unref (priv->properties);
+  g_signal_handler_disconnect (priv->context->priv->subscribe, priv->id);
   g_clear_object (&priv->context);
   g_free (priv->name);
 
@@ -186,6 +237,7 @@ pv_stream_class_init (PvStreamClass * klass)
 
   g_type_class_add_private (klass, sizeof (PvStreamPrivate));
 
+  gobject_class->constructed = pv_stream_constructed;
   gobject_class->finalize = pv_stream_finalize;
   gobject_class->set_property = pv_stream_set_property;
   gobject_class->get_property = pv_stream_get_property;
@@ -338,15 +390,6 @@ pv_stream_new (PvContext * context, const gchar *name, GVariant *props)
   return g_object_new (PV_TYPE_STREAM, "context", context, "name", name, "properties", props, NULL);
 }
 
-static void
-stream_set_state (PvStream *stream, PvStreamState state)
-{
-  if (stream->priv->state != state) {
-    stream->priv->state = state;
-    g_object_notify (G_OBJECT (stream), "state");
-  }
-}
-
 /**
  * pv_stream_get_state:
  * @stream: a #PvStream
@@ -379,17 +422,6 @@ pv_stream_get_error (PvStream *stream)
   return stream->priv->error;
 }
 
-
-static void
-on_source_output_signal (GDBusProxy *proxy,
-                         gchar      *sender_name,
-                         gchar      *signal_name,
-                         GVariant   *parameters,
-                         gpointer    user_data)
-{
-  g_print ("on source output signal %s %s\n", sender_name, signal_name);
-}
-
 static void
 on_source_output_proxy (GObject *source_object,
                         GAsyncResult *res,
@@ -420,11 +452,6 @@ on_source_output_proxy (GObject *source_object,
     g_object_notify (G_OBJECT (stream), "possible-formats");
   }
 
-  g_signal_connect (priv->source_output,
-                    "g-signal",
-                    (GCallback) on_source_output_signal,
-                    stream);
-
   stream_set_state (stream, PV_STREAM_STATE_READY);
   g_object_unref (stream);
 

src/server/pv-client-source.c

@@ -33,6 +33,7 @@ struct _PvClientSourcePrivate
   GstElement *src;
   GstElement *filter;
   GstElement *sink;
+  guint id;
 
   GSocket *socket;
 
@@ -89,7 +90,7 @@ setup_pipeline (PvClientSource *source)
   priv->src = gst_bin_get_by_name (GST_BIN (priv->pipeline), "src");
 
   bus = gst_pipeline_get_bus (GST_PIPELINE (priv->pipeline));
-  gst_bus_add_watch (bus, bus_handler, source);
+  priv->id = gst_bus_add_watch (bus, bus_handler, source);
   gst_object_unref (bus);
 }
 
@@ -235,6 +236,7 @@ client_source_dispose (GObject * object)
 {
   PvClientSourcePrivate *priv = PV_CLIENT_SOURCE (object)->priv;
 
+  g_source_remove (priv->id);
   gst_element_set_state (priv->pipeline, GST_STATE_NULL);
 
   G_OBJECT_CLASS (pv_client_source_parent_class)->dispose (object);

src/server/pv-client.c

@@ -35,6 +35,8 @@ struct _PvClientPrivate
   GVariant *properties;
 
   PvClient1 *client1;
+
+  GList *outputs;
 };
 
 #define PV_CLIENT_GET_PRIVATE(obj)  \
@@ -129,8 +131,11 @@ static void
 handle_remove_source_output (PvSourceOutput *output,
                              gpointer        user_data)
 {
-  g_object_steal_data (G_OBJECT (user_data),
-                       pv_source_output_get_object_path (output));
+  PvClient *client = user_data;
+  PvClientPrivate *priv = client->priv;
+
+  priv->outputs = g_list_remove (priv->outputs, output);
+  g_object_unref (output);
 }
 
 static gboolean
@@ -172,16 +177,13 @@ handle_create_source_output (PvClient1              *interface,
 
   object_path = pv_source_output_get_object_path (output);
 
+  priv->outputs = g_list_prepend (priv->outputs, output);
+
   g_signal_connect (output,
                     "remove",
                     (GCallback) handle_remove_source_output,
                     client);
 
-  g_object_set_data_full (G_OBJECT (client),
-                          object_path,
-                          output,
-                          g_object_unref);
-
   g_dbus_method_invocation_return_value (invocation,
                                          g_variant_new ("(o)", object_path));
 
@@ -251,15 +253,12 @@ handle_create_source_input (PvClient1              *interface,
 
   source_input_path = pv_source_output_get_object_path (input);
 
+  priv->outputs = g_list_prepend (priv->outputs, input);
+
   g_signal_connect (input,
                     "remove",
                     (GCallback) handle_remove_source_output,
-                    source);
-
-  g_object_set_data_full (G_OBJECT (source),
-                          source_input_path,
-                          input,
-                          g_object_unref);
+                    client);
 
   g_dbus_method_invocation_return_value (invocation,
                                          g_variant_new ("(o)",
@@ -343,11 +342,19 @@ client_unregister_object (PvClient *client)
   g_free (priv->object_path);
 }
 
+static void
+do_remove_output (PvSourceOutput *output, PvClient *client)
+{
+  pv_source_output_remove (output);
+}
+
 static void
 pv_client_dispose (GObject * object)
 {
   PvClient *client = PV_CLIENT (object);
+  PvClientPrivate *priv = client->priv;
 
+  g_list_foreach (priv->outputs, (GFunc) do_remove_output, client);
   client_unregister_object (client);
 
   G_OBJECT_CLASS (pv_client_parent_class)->dispose (object);

src/server/pv-source-output.c

@@ -437,6 +437,14 @@ pv_source_output_init (PvSourceOutput * output)
   g_signal_connect (priv->iface, "handle-remove", (GCallback) handle_remove, output);
 }
 
+void
+pv_source_output_remove (PvSourceOutput *output)
+{
+  stop_transfer (output);
+
+  g_signal_emit (output, signals[SIGNAL_REMOVE], 0, NULL);
+}
+
 const gchar *
 pv_source_output_get_object_path (PvSourceOutput *output)
 {

src/server/pv-source-output.h

@@ -60,7 +60,8 @@ struct _PvSourceOutputClass {
 /* normal GObject stuff */
 GType              pv_source_output_get_type             (void);
 
-const gchar *      pv_source_output_get_sender           (PvSourceOutput *output);
+void               pv_source_output_remove               (PvSourceOutput *output);
+
 const gchar *      pv_source_output_get_object_path      (PvSourceOutput *output);
 
 G_END_DECLS

src/server/pv-source.c

@@ -38,10 +38,12 @@ struct _PvSourcePrivate
   gchar *object_path;
 
   gchar *name;
-  PvSourceState state;
   GVariant *properties;
 
+  PvSourceState state;
   GError *error;
+
+  GList *outputs;
 };
 
 G_DEFINE_ABSTRACT_TYPE (PvSource, pv_source, G_TYPE_OBJECT);
@@ -171,11 +173,20 @@ pv_source_constructed (GObject * object)
   G_OBJECT_CLASS (pv_source_parent_class)->constructed (object);
 }
 
+static void
+do_remove_output (PvSourceOutput *output,
+                  gpointer        user_data)
+{
+  pv_source_output_remove (output);
+}
+
 static void
 pv_source_dispose (GObject * object)
 {
   PvSource *source = PV_SOURCE (object);
+  PvSourcePrivate *priv = source->priv;
 
+  g_list_foreach (priv->outputs, (GFunc) do_remove_output, source);
   source_unregister_object (source);
 
   G_OBJECT_CLASS (pv_source_parent_class)->dispose (object);
@@ -229,14 +240,24 @@ default_create_source_output (PvSource    *source,
                                                 NULL);
 
   g_signal_connect (output, "remove", (GCallback) handle_remove_output, source);
+  priv->outputs = g_list_prepend (priv->outputs, output);
 
-  return output;
+  return g_object_ref (output);
 }
 
 static gboolean
 default_release_source_output (PvSource *source, PvSourceOutput *output)
 {
+  PvSourcePrivate *priv = source->priv;
+  GList *find;
+
+  find = g_list_find (priv->outputs, output);
+  if (find == NULL)
+    return FALSE;
+
+  priv->outputs = g_list_delete_link (priv->outputs, find);
   g_object_unref (output);
+
   return TRUE;
 }