[tracker/wip/carlosg/bus-statements: 1/6] libtracker-sparql: Refactor code writing cursor to a fd

[Carlos Garnacho <carlosg src gnome org>]

commit 412a5fe4daeac76c1c15261cbcf5e7580c8665a5
Author: Carlos Garnacho <carlosg gnome org>
Date:   Sun Mar 1 14:09:32 2020 +0100

    libtracker-sparql: Refactor code writing cursor to a fd
    
    Just move it to a separate function, as we'll be using it in more
    than one place.

 src/libtracker-bus/tracker-bus.vala           | 46 ++++++++++--------
 src/libtracker-sparql/tracker-endpoint-dbus.c | 69 +++++++++++++++++++--------
 2 files changed, 73 insertions(+), 42 deletions(-)
---
diff --git a/src/libtracker-bus/tracker-bus.vala b/src/libtracker-bus/tracker-bus.vala
index 03b3030f1..5d2730c97 100644
--- a/src/libtracker-bus/tracker-bus.vala
+++ b/src/libtracker-bus/tracker-bus.vala
@@ -34,7 +34,7 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
                new Sparql.Error.INTERNAL ("");
        }
 
-       void pipe (out UnixInputStream input, out UnixOutputStream output) throws IOError {
+       static void pipe (out UnixInputStream input, out UnixOutputStream output) throws IOError {
                int pipefd[2];
                if (Posix.pipe (pipefd) < 0) {
                        throw new IOError.FAILED ("Pipe creation failed");
@@ -43,7 +43,7 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
                output = new UnixOutputStream (pipefd[1], true);
        }
 
-       void handle_error_reply (DBusMessage message) throws Sparql.Error, IOError, DBusError {
+       static void handle_error_reply (DBusMessage message) throws Sparql.Error, IOError, DBusError {
                try {
                        message.to_gerror ();
                } catch (IOError e_io) {
@@ -57,7 +57,7 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
                }
        }
 
-       void send_query (string sparql, UnixOutputStream output, Cancellable? cancellable, 
AsyncReadyCallback? callback) throws GLib.IOError, GLib.Error {
+       static void send_query (DBusConnection bus, string dbus_name, string object_path, string sparql, 
UnixOutputStream output, Cancellable? cancellable, AsyncReadyCallback? callback) throws GLib.IOError, 
GLib.Error {
                var message = new DBusMessage.method_call (dbus_name, object_path, ENDPOINT_IFACE, "Query");
                var fd_list = new UnixFDList ();
                message.set_body (new Variant ("(sh)", sparql, fd_list.append (output.fd)));
@@ -66,22 +66,7 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
                bus.send_message_with_reply.begin (message, DBusSendMessageFlags.NONE, int.MAX, null, 
cancellable, callback);
        }
 
-       public override Sparql.Cursor query (string sparql, Cancellable? cancellable) throws Sparql.Error, 
GLib.Error, GLib.IOError, DBusError {
-               // use separate main context for sync operation
-               var context = new MainContext ();
-               var loop = new MainLoop (context, false);
-               context.push_thread_default ();
-               AsyncResult async_res = null;
-               query_async.begin (sparql, cancellable, (o, res) => {
-                       async_res = res;
-                       loop.quit ();
-               });
-               loop.run ();
-               context.pop_thread_default ();
-               return query_async.end (async_res);
-       }
-
-       public async override Sparql.Cursor query_async (string sparql, Cancellable? cancellable = null) 
throws Sparql.Error, GLib.Error, GLib.IOError, DBusError {
+       public static async Sparql.Cursor perform_query_call (DBusConnection bus, string dbus_name, string 
object_path, string sparql, Cancellable? cancellable) throws GLib.IOError, GLib.Error {
                UnixInputStream input;
                UnixOutputStream output;
                pipe (out input, out output);
@@ -89,10 +74,10 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
                // send D-Bus request
                AsyncResult dbus_res = null;
                bool received_result = false;
-               send_query (sparql, output, cancellable, (o, res) => {
+               send_query (bus, dbus_name, object_path, sparql, output, cancellable, (o, res) => {
                        dbus_res = res;
                        if (received_result) {
-                               query_async.callback ();
+                               perform_query_call.callback ();
                        }
                });
 
@@ -119,6 +104,25 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
                return new FDCursor (mem_stream.steal_data (), mem_stream.data_size, variable_names);
        }
 
+       public override Sparql.Cursor query (string sparql, Cancellable? cancellable) throws Sparql.Error, 
GLib.Error, GLib.IOError, DBusError {
+               // use separate main context for sync operation
+               var context = new MainContext ();
+               var loop = new MainLoop (context, false);
+               context.push_thread_default ();
+               AsyncResult async_res = null;
+               query_async.begin (sparql, cancellable, (o, res) => {
+                       async_res = res;
+                       loop.quit ();
+               });
+               loop.run ();
+               context.pop_thread_default ();
+               return query_async.end (async_res);
+       }
+
+       public async override Sparql.Cursor query_async (string sparql, Cancellable? cancellable = null) 
throws Sparql.Error, GLib.Error, GLib.IOError, DBusError {
+               return yield perform_query_call (bus, dbus_name, object_path, sparql, cancellable);
+       }
+
        void send_update (string method, UnixInputStream input, Cancellable? cancellable, AsyncReadyCallback? 
callback) throws GLib.Error, GLib.IOError {
                var message = new DBusMessage.method_call (dbus_name, object_path, ENDPOINT_IFACE, method);
                var fd_list = new UnixFDList ();
diff --git a/src/libtracker-sparql/tracker-endpoint-dbus.c b/src/libtracker-sparql/tracker-endpoint-dbus.c
index c9ef380b0..89e12fa98 100644
--- a/src/libtracker-sparql/tracker-endpoint-dbus.c
+++ b/src/libtracker-sparql/tracker-endpoint-dbus.c
@@ -206,33 +206,21 @@ update_request_free (UpdateRequest *request)
        g_free (request);
 }
 
-static void
-query_cb (GObject      *object,
-          GAsyncResult *res,
-          gpointer      user_data)
+static gboolean
+write_cursor (QueryRequest          *request,
+              TrackerSparqlCursor   *cursor,
+              GError               **error)
 {
-       QueryRequest *request = user_data;
-       TrackerSparqlCursor *cursor;
-       GError *error = NULL;
        const gchar **values = NULL;
-       const gchar **variable_names = NULL;
        glong *offsets = NULL;
        gint i, n_columns = 0;
-
-       cursor = tracker_sparql_connection_query_finish (TRACKER_SPARQL_CONNECTION (object),
-                                                        res, &error);
-       if (!cursor)
-               goto error;
+       GError *inner_error = NULL;
 
        n_columns = tracker_sparql_cursor_get_n_columns (cursor);
-       variable_names = g_new0 (const gchar *, n_columns + 1);
        values = g_new0 (const char *, n_columns);
        offsets = g_new0 (glong, n_columns);
 
-       for (i = 0; i < n_columns; i++)
-               variable_names[i] = tracker_sparql_cursor_get_variable_name (cursor, i);
-
-       while (tracker_sparql_cursor_next (cursor, NULL, &error)) {
+       while (tracker_sparql_cursor_next (cursor, NULL, &inner_error)) {
                glong cur_offset = -1;
 
                g_data_output_stream_put_int32 (request->data_stream, n_columns, NULL, NULL);
@@ -262,17 +250,56 @@ query_cb (GObject      *object,
                }
        }
 
-error:
+       g_free (values);
+       g_free (offsets);
+
+       if (inner_error) {
+               g_propagate_error (error, inner_error);
+               return FALSE;
+       } else {
+               return TRUE;
+       }
+}
+
+/* Takes ownership on both cursor and error */
+static void
+handle_cursor_reply (QueryRequest        *request,
+                     TrackerSparqlCursor *cursor,
+                     GError              *error)
+{
+       const gchar **variable_names = NULL;
+       gint i, n_columns;
+
+       if (!error) {
+               n_columns = tracker_sparql_cursor_get_n_columns (cursor);
+               variable_names = g_new0 (const gchar *, n_columns + 1);
+               for (i = 0; i < n_columns; i++)
+                       variable_names[i] = tracker_sparql_cursor_get_variable_name (cursor, i);
+
+               write_cursor (request, cursor, &error);
+       }
+
        if (error)
                g_dbus_method_invocation_return_gerror (request->invocation, error);
        else
                g_dbus_method_invocation_return_value (request->invocation, g_variant_new ("(^as)", 
variable_names));
 
        g_free (variable_names);
-       g_free (values);
-       g_free (offsets);
        g_clear_object (&cursor);
+}
 
+static void
+query_cb (GObject      *object,
+          GAsyncResult *res,
+          gpointer      user_data)
+{
+       QueryRequest *request = user_data;
+       TrackerSparqlCursor *cursor;
+       GError *error = NULL;
+
+       cursor = tracker_sparql_connection_query_finish (TRACKER_SPARQL_CONNECTION (object),
+                                                        res, &error);
+       handle_cursor_reply (request, cursor, error);
        query_request_free (request);
 }