[tracker/writeback-refactor: 6/6] miners/fs: Implement listener class

[Philip Van Hoof <pvanhoof src gnome org>]

commit 275d9c388bda81664b0bd82aca18f8985ef3df9d
Author: Philip Van Hoof <philip codeminded be>
Date:   Fri Jul 8 17:05:33 2011 +0200

    miners/fs: Implement listener class

 src/miners/fs/tracker-writeback-listener.c |  273 +++++++++++++++++++++++++++-
 1 files changed, 269 insertions(+), 4 deletions(-)
---
diff --git a/src/miners/fs/tracker-writeback-listener.c b/src/miners/fs/tracker-writeback-listener.c
index 8f67d9b..ad1bbde 100644
--- a/src/miners/fs/tracker-writeback-listener.c
+++ b/src/miners/fs/tracker-writeback-listener.c
@@ -25,10 +25,15 @@
 
 #include "tracker-writeback-listener.h"
 
+#define TRACKER_SERVICE                 "org.freedesktop.Tracker1"
+#define TRACKER_RESOURCES_OBJECT        "/org/freedesktop/Tracker1/Resources"
+#define TRACKER_INTERFACE_RESOURCES     "org.freedesktop.Tracker1.Resources"
 
 typedef struct {
 	TrackerMinerFiles *files_miner;
-	GDBusConnection *connection;
+	GDBusConnection *d_connection;
+	TrackerSparqlConnection *connection;
+	guint d_signal;
 } TrackerWritebackListenerPrivate;
 
 enum {
@@ -50,6 +55,13 @@ static void     writeback_listener_finalize        (GObject              *object
 static gboolean writeback_listener_initable_init   (GInitable            *initable,
                                                     GCancellable         *cancellable,
                                                     GError              **error);
+static void     on_writeback_cb                    (GDBusConnection      *connection,
+                                                    const gchar          *sender_name,
+                                                    const gchar          *object_path,
+                                                    const gchar          *interface_name,
+                                                    const gchar          *signal_name,
+                                                    GVariant             *parameters,
+                                                    gpointer              user_data);
 
 static void
 writeback_listener_initable_iface_init (GInitableIface *iface)
@@ -129,11 +141,21 @@ writeback_listener_finalize (GObject *object)
 {
 	TrackerWritebackListenerPrivate *priv = TRACKER_WRITEBACK_LISTENER_GET_PRIVATE (object);
 
+	if (priv->connection && priv->d_signal) {
+		g_dbus_connection_signal_unsubscribe (priv->d_connection, priv->d_signal);
+	}
+
 	if (priv->connection) {
 		g_object_unref (priv->connection);
 	}
 
-	g_object_unref (priv->files_miner);
+	if (priv->d_connection) {
+		g_object_unref (priv->d_connection);
+	}
+
+	if (priv->files_miner) {
+		g_object_unref (priv->files_miner);
+	}
 }
 
 
@@ -152,13 +174,31 @@ writeback_listener_initable_init (GInitable    *initable,
 
 	priv = TRACKER_WRITEBACK_LISTENER_GET_PRIVATE (initable);
 
-	priv->connection = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &internal_error);
+	priv->connection = tracker_sparql_connection_get (NULL, &internal_error);
 
 	if (internal_error) {
 		g_propagate_error (error, internal_error);
 		return FALSE;
 	}
 
+	priv->d_connection = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &internal_error);
+
+	if (internal_error) {
+		g_propagate_error (error, internal_error);
+		return FALSE;
+	}
+
+	priv->d_signal = g_dbus_connection_signal_subscribe (priv->d_connection,
+	                                                     TRACKER_SERVICE,
+	                                                     TRACKER_INTERFACE_RESOURCES,
+	                                                     "Writeback",
+	                                                     TRACKER_RESOURCES_OBJECT,
+	                                                     NULL,
+	                                                     G_DBUS_SIGNAL_FLAGS_NONE,
+	                                                     on_writeback_cb,
+	                                                     initable,
+	                                                     NULL);
+
 	return TRUE;
 }
 
@@ -183,5 +223,230 @@ tracker_writeback_listener_new (TrackerMinerFiles  *miner_files,
 	return (TrackerWritebackListener *) miner;
 }
 
-	// tracker_miner_fs_writeback_file (TRACKER_MINER_FS (miner_files), file, ...);
+typedef struct {
+	TrackerWritebackListener *self;
+	gint subject_id;
+	GStrv rdf_types;
+} QueryData;
+
+static QueryData*
+query_data_new (TrackerWritebackListener *self,
+                gint subject_id)
+{
+	QueryData *data = g_slice_new0 (QueryData);
+
+	data->self = g_object_ref (self);
+	data->subject_id = subject_id;
+
+	return data;
+}
+
+static void
+query_data_free (QueryData *data)
+{
+	g_object_unref (data->self);
+	if (data->rdf_types) {
+		g_strfreev (data->rdf_types);
+	}
+	g_slice_free (QueryData, data);
+}
+
+static void
+sparql_query_cb (GObject      *object,
+                 GAsyncResult *result,
+                 gpointer      user_data)
+{
+	QueryData *data = user_data;
+	TrackerWritebackListener *self = TRACKER_WRITEBACK_LISTENER (data->self);
+	TrackerWritebackListenerPrivate *priv;
+	TrackerSparqlCursor *cursor;
+	GError *error = NULL;
+
+	priv = TRACKER_WRITEBACK_LISTENER_GET_PRIVATE (self);
+
+	cursor = tracker_sparql_connection_query_finish (TRACKER_SPARQL_CONNECTION (object), result, &error);
+
+	if (!error) {
+		guint cols = tracker_sparql_cursor_get_n_columns (cursor);
+		GPtrArray *results = NULL;
+		GFile *file = NULL;
+
+		while (tracker_sparql_cursor_next (cursor, NULL, NULL)) {
+			GStrv row = NULL;
+			guint i;
+
+			if (file == NULL) {
+				file = g_file_new_for_uri (tracker_sparql_cursor_get_string (cursor, 0, NULL));
+				if (!g_file_query_exists (file, NULL)) {
+					g_object_unref (file);
+					g_message ("  No files qualify for updates");
+					query_data_free (data);
+					g_object_unref (cursor);
+					return;
+				}
+			}
+
+			for (i = 0; i < cols; i++) {
+				if (!row) {
+					row = g_new0 (gchar*, cols);
+				}
+				row[i] = g_strdup (tracker_sparql_cursor_get_string (cursor, i, NULL));
+			}
+
+			if (!results) {
+				results = g_ptr_array_new_with_free_func ((GDestroyNotify) g_strfreev);
+			}
+
+			g_ptr_array_add (results, row);
+		}
+
+		if (results->len > 0) {
+			tracker_miner_fs_writeback_file (TRACKER_MINER_FS (priv->files_miner),
+			                                 file,
+			                                 results);
+		} else {
+			g_message ("  No files qualify for updates");
+		}
+
+		g_object_unref (file);
+		g_ptr_array_unref (results);
+		g_object_unref (cursor);
+	} else {
+		g_message ("  No files qualify for updates (%s)", error->message);
+		g_error_free (error);
+	}
+
+	query_data_free (data);
+}
+
+static void
+rdf_types_to_uris_cb (GObject      *object,
+                      GAsyncResult *result,
+                      gpointer      user_data)
+{
+	QueryData *data = user_data;
+	TrackerWritebackListener *self = TRACKER_WRITEBACK_LISTENER (data->self);
+	TrackerWritebackListenerPrivate *priv;
+	TrackerSparqlCursor *cursor;
+	TrackerSparqlConnection *connection;
+	GError *error = NULL;
+
+	priv = TRACKER_WRITEBACK_LISTENER_GET_PRIVATE (self);
+	connection = priv->connection;
+
+	cursor = tracker_sparql_connection_query_finish (connection, result, &error);
+
+	if (!error) {
+		gchar *query;
+		const gchar *subject;
+		GArray *rdf_types;
+
+		rdf_types = g_array_new (TRUE, TRUE, sizeof (gchar *));
+
+		while (tracker_sparql_cursor_next (cursor, NULL, NULL)) {
+			gchar *uri = g_strdup (tracker_sparql_cursor_get_string (cursor, 0, NULL));
+			g_array_append_val (rdf_types, uri);
+		}
+
+		data->rdf_types = (GStrv) rdf_types->data;
+		g_array_free (rdf_types, FALSE);
+
+		g_object_unref (cursor);
+
+		query = g_strdup_printf ("SELECT tracker:uri (%d) {}", data->subject_id);
+		cursor = tracker_sparql_connection_query (connection, query, NULL, NULL);
+		g_free (query);
+
+		if (cursor && tracker_sparql_cursor_next (cursor, NULL, NULL)) {
+			subject = tracker_sparql_cursor_get_string (cursor, 0, NULL);
+			if (!subject) {
+				g_object_unref (cursor);
+				goto trouble;
+			}
+		} else {
+			if (cursor)
+				g_object_unref (cursor);
+			goto trouble;
+		}
+
+		query = g_strdup_printf ("SELECT ?url '%s' ?predicate ?object { "
+		                            "<%s> a nfo:FileDataObject . "
+		                            "<%s> ?predicate ?object ; nie:url ?url . "
+		                            "?predicate tracker:writeback true . "
+		                            "FILTER (NOT EXISTS { GRAPH <"TRACKER_MINER_FS_GRAPH_URN"> "
+		                              "{ <%s> ?predicate ?object } }) } ",
+		                         subject, subject, subject, subject);
+
+		tracker_sparql_connection_query_async (priv->connection,
+		                                       query,
+		                                       NULL,
+		                                       sparql_query_cb,
+		                                       data);
+
+		g_free (query);
+		g_object_unref (cursor);
+
+	} else {
+		goto trouble;
+	}
+
+	return;
+
+trouble:
+	if (error) {
+		g_message ("  No files qualify for updates (%s)", error->message);
+		g_error_free (error);
+	}
+	query_data_free (data);
+}
+
+static void
+on_writeback_cb (GDBusConnection      *connection,
+                 const gchar          *sender_name,
+                 const gchar          *object_path,
+                 const gchar          *interface_name,
+                 const gchar          *signal_name,
+                 GVariant             *parameters,
+                 gpointer              user_data)
+{
+	TrackerWritebackListener *self = TRACKER_WRITEBACK_LISTENER (user_data);
+	TrackerWritebackListenerPrivate *priv;
+	GVariantIter *iter1, *iter2;
+	gint subject_id = 0, rdf_type = 0;
+
+	priv = TRACKER_WRITEBACK_LISTENER_GET_PRIVATE (self);
+
+	g_variant_get (parameters, "a{iai}", &iter1);
+	while (g_variant_iter_loop (iter1, "{iai}", &subject_id, &iter2)) {
+		GString *query;
+		gboolean comma = FALSE;
+		QueryData *data;
+
+		data = query_data_new (self, subject_id);
+
+		query = g_string_new ("SELECT ?resource { ?resource a rdfs:Class . "
+		                      "FILTER (tracker:id (?resource) IN (");
 
+		while (g_variant_iter_loop (iter2, "i", &rdf_type)) {
+			if (comma) {
+				g_string_append_printf (query, ", %d", rdf_type);
+			} else {
+				g_string_append_printf (query, "%d", rdf_type);
+				comma = TRUE;
+			}
+		}
+
+		g_string_append (query, ")) }");
+
+		tracker_sparql_connection_query_async (priv->connection,
+		                                       query->str,
+		                                       NULL,
+		                                       rdf_types_to_uris_cb,
+		                                       data);
+
+		g_string_free (query, TRUE);
+		g_variant_iter_free (iter2);
+	}
+
+	g_variant_iter_free (iter1);
+}