[tracker/direct-access-update-fd-passing] Unfinished work on support for update in fd-passing
- From: Philip Van Hoof <pvanhoof src gnome org>
- To: commits-list gnome org
- Cc:
- Subject: [tracker/direct-access-update-fd-passing] Unfinished work on support for update in fd-passing
- Date: Tue, 20 Jul 2010 12:35:14 +0000 (UTC)
commit 581782882ad83de6050c667ff4a41fc65401d480
Author: Philip Van Hoof <philip codeminded be>
Date: Tue Jul 20 14:32:55 2010 +0200
Unfinished work on support for update in fd-passing
src/Makefile.am | 1 -
src/libtracker-bus/Makefile.am | 13 +-
src/libtracker-bus/tracker-bus-fd-update.c | 497 ++++++++++++++++++++++++++++
src/libtracker-bus/tracker-bus.vala | 27 ++-
4 files changed, 526 insertions(+), 12 deletions(-)
---
diff --git a/src/Makefile.am b/src/Makefile.am
index 640e0ec..ebbaac6 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -24,7 +24,6 @@ SUBDIRS = \
plugins \
tracker-store \
tracker-control \
- tracker-utils \
tracker-extract \
tracker-writeback \
vapi
diff --git a/src/libtracker-bus/Makefile.am b/src/libtracker-bus/Makefile.am
index cd345dc..baeb939 100644
--- a/src/libtracker-bus/Makefile.am
+++ b/src/libtracker-bus/Makefile.am
@@ -9,7 +9,8 @@ INCLUDES = \
$(WARN_CFLAGS) \
$(DBUS_CFLAGS) \
$(GLIB2_CFLAGS) \
- $(GCOV_CFLAGS)
+ $(GCOV_CFLAGS) \
+ $(GIO_CFLAGS)
lib_LTLIBRARIES = libtracker-bus- TRACKER_API_VERSION@.la
@@ -20,9 +21,10 @@ libtracker_bus_la_VALASOURCES = \
tracker-array-cursor.vala
libtracker_bus_ TRACKER_API_VERSION@_la_SOURCES = \
- libtracker-bus.vala.stamp \
- $(libtracker_bus_la_VALASOURCES:.vala=.c) \
- tracker-bus-fd-cursor.c
+ libtracker-bus.vala.stamp \
+ $(libtracker_bus_la_VALASOURCES:.vala=.c) \
+ tracker-bus-fd-cursor.c \
+ tracker-bus-fd-update.c
libtracker_bus_ TRACKER_API_VERSION@_la_LDFLAGS = \
-shared -fPIC -module -avoid-version
@@ -50,7 +52,8 @@ libtracker-bus.vala.stamp: $(libtracker_bus_la_VALASOURCES) $(vapi_sources)
libtracker_bus_ TRACKER_API_VERSION@_la_LIBADD = \
$(top_builddir)/src/libtracker-common/libtracker-common.la \
$(GLIB2_LIBS) \
- $(DBUS_LIBS)
+ $(DBUS_LIBS) \
+ $(GIO_LIBS)
BUILT_SOURCES = \
libtracker-bus.vala.stamp
diff --git a/src/libtracker-bus/tracker-bus-fd-update.c b/src/libtracker-bus/tracker-bus-fd-update.c
new file mode 100644
index 0000000..a4cd57f
--- /dev/null
+++ b/src/libtracker-bus/tracker-bus-fd-update.c
@@ -0,0 +1,497 @@
+/*
+ * Copyright (C) 2006, Jamie McCracken <jamiemcc gnome org>
+ * Copyright (C) 2008-2010, Nokia <ivan frade nokia com>
+ * Copyright (C) 2010, Codeminded BVBA <abustany gnome org>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#include "config.h"
+
+#include <string.h>
+#include <errno.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <libtracker-common/tracker-common.h>
+
+#include <libtracker-sparql/tracker-sparql.h>
+
+#include "tracker-bus.h"
+
+#ifdef HAVE_DBUS_FD_PASSING
+
+/* Are also defined in src/tracker-store/tracker-steroids.h */
+#define TRACKER_STEROIDS_BUFFER_SIZE 65536
+
+typedef enum {
+ FAST_UPDATE,
+ FAST_UPDATE_BLANK
+} FastOperationType;
+
+typedef struct {
+ DBusConnection *connection;
+ FastOperationType operation_type;
+ GCancellable *cancellable;
+ DBusPendingCall *dbus_call;
+ GAsyncReadyCallback callback;
+ gpointer user_data;
+} FastAsyncData;
+
+static FastAsyncData *
+fast_async_data_new (DBusConnection *connection,
+ FastOperationType operation_type,
+ GCancellable *cancellable,
+ gpointer user_data)
+{
+ FastAsyncData *data;
+
+ data = g_slice_new0 (FastAsyncData);
+
+ data->connection = dbus_connection_ref (connection);
+ data->operation_type = operation_type;
+ data->cancellable = cancellable;
+ data->user_data = user_data;
+
+ return data;
+}
+
+static void
+fast_async_data_free (gpointer data)
+{
+ FastAsyncData *fad = data;
+
+ if (fad) {
+ if (fad->cancellable) {
+ g_object_unref (fad->cancellable);
+ }
+
+ if (fad->connection) {
+ dbus_connection_unref (fad->connection);
+ }
+
+ g_slice_free (FastAsyncData, fad);
+ }
+}
+
+static GHashTable *
+unmarshal_hash_table (DBusMessageIter *iter)
+{
+ GHashTable *result;
+ DBusMessageIter subiter, subsubiter;
+
+ result = g_hash_table_new_full (g_str_hash,
+ g_str_equal,
+ (GDestroyNotify) g_free,
+ (GDestroyNotify) g_free);
+
+ dbus_message_iter_recurse (iter, &subiter);
+
+ while (dbus_message_iter_get_arg_type (&subiter) != DBUS_TYPE_INVALID) {
+ const gchar *key, *value;
+
+ dbus_message_iter_recurse (&subiter, &subsubiter);
+ dbus_message_iter_get_basic (&subsubiter, &key);
+ dbus_message_iter_next (&subsubiter);
+ dbus_message_iter_get_basic (&subsubiter, &value);
+ g_hash_table_insert (result, g_strdup (key), g_strdup (value));
+
+ dbus_message_iter_next (&subiter);
+ }
+
+ return result;
+}
+
+static void
+sparql_update_fast_callback (DBusPendingCall *call,
+ void *user_data)
+{
+ FastAsyncData *fad = user_data;
+ DBusMessage *reply;
+ GError *error = NULL;
+ DBusMessageIter iter, subiter, subsubiter;
+ GPtrArray *result;
+
+ /* Check for errors */
+ reply = dbus_pending_call_steal_reply (call);
+
+ if (dbus_message_get_type (reply) == DBUS_MESSAGE_TYPE_ERROR) {
+ DBusError dbus_error;
+
+ dbus_error_init (&dbus_error);
+ dbus_set_error_from_message (&dbus_error, reply);
+ dbus_set_g_error (&error, &dbus_error);
+ dbus_error_free (&dbus_error);
+
+ g_print ("TODO: callback error !\n");
+
+ /*
+ switch (fad->operation_type) {
+ case FAST_UPDATE:
+ case FAST_UPDATE_BATCH:
+ (* fad->void_callback) (error, fad->user_data);
+ break;
+ case FAST_UPDATE_BLANK:
+ (* fad->gptrarray_callback) (NULL, error, fad->user_data);
+ break;
+ default:
+ g_assert_not_reached ();
+ break;
+ }
+ */
+
+ dbus_message_unref (reply);
+
+ fast_async_data_free (fad);
+
+ return;
+ }
+
+ /* Call iterator callback */
+ switch (fad->operation_type) {
+ case FAST_UPDATE:
+ case FAST_UPDATE_BLANK:
+ result = g_ptr_array_new ();
+ dbus_message_iter_init (reply, &iter);
+ dbus_message_iter_recurse (&iter, &subiter);
+
+ while (dbus_message_iter_get_arg_type (&subiter) != DBUS_TYPE_INVALID) {
+ GPtrArray *inner_array;
+
+ inner_array = g_ptr_array_new ();
+ g_ptr_array_add (result, inner_array);
+ dbus_message_iter_recurse (&subiter, &subsubiter);
+
+ while (dbus_message_iter_get_arg_type (&subsubiter) != DBUS_TYPE_INVALID) {
+ g_ptr_array_add (inner_array, unmarshal_hash_table (&subsubiter));
+ dbus_message_iter_next (&subsubiter);
+ }
+
+ dbus_message_iter_next (&subiter);
+ }
+
+ g_print ("TODO: callback !\n");
+
+// (* fad->gptrarray_callback) (result, error, fad->user_data);
+
+ break;
+ default:
+ g_assert_not_reached ();
+ break;
+ }
+
+ /* Clean up */
+ dbus_message_unref (reply);
+
+ fast_async_data_free (fad);
+}
+
+static DBusPendingCall *
+sparql_update_fast_send (DBusConnection *connection,
+ const gchar *query,
+ FastOperationType type,
+ GError **error)
+{
+ const gchar *dbus_method;
+ DBusMessage *message;
+ DBusMessageIter iter;
+ DBusPendingCall *call;
+ int pipefd[2];
+ GOutputStream *output_stream;
+ GOutputStream *buffered_output_stream;
+ GDataOutputStream *data_output_stream;
+ GError *inner_error = NULL;
+
+ g_return_val_if_fail (query != NULL, NULL);
+
+ if (pipe (pipefd) < 0) {
+ g_set_error (error,
+ TRACKER_BUS_ERROR,
+ TRACKER_BUS_ERROR_UNSUPPORTED,
+ "Cannot open pipe");
+ return NULL;
+ }
+
+ switch (type) {
+ case FAST_UPDATE:
+ dbus_method = "Update";
+ break;
+ case FAST_UPDATE_BLANK:
+ dbus_method = "UpdateBlank";
+ break;
+ default:
+ g_assert_not_reached ();
+ }
+
+ message = dbus_message_new_method_call (TRACKER_DBUS_SERVICE,
+ TRACKER_DBUS_OBJECT_STEROIDS,
+ TRACKER_DBUS_INTERFACE_STEROIDS,
+ dbus_method);
+ dbus_message_iter_init_append (message, &iter);
+ dbus_message_iter_append_basic (&iter, DBUS_TYPE_UNIX_FD, &pipefd[0]);
+ dbus_connection_send_with_reply (connection, message, &call, -1);
+ dbus_message_unref (message);
+ close (pipefd[0]);
+
+ if (!call) {
+ g_set_error (error,
+ TRACKER_BUS_ERROR,
+ TRACKER_BUS_ERROR_UNSUPPORTED,
+ "FD passing unsupported or connection disconnected");
+ return NULL;
+ }
+
+ output_stream = g_unix_output_stream_new (pipefd[1], TRUE);
+ buffered_output_stream = g_buffered_output_stream_new_sized (output_stream,
+ TRACKER_STEROIDS_BUFFER_SIZE);
+ data_output_stream = g_data_output_stream_new (buffered_output_stream);
+
+ g_data_output_stream_put_int32 (data_output_stream, strlen (query),
+ NULL, &inner_error);
+
+ if (inner_error) {
+ g_propagate_error (error, inner_error);
+ g_object_unref (data_output_stream);
+ g_object_unref (buffered_output_stream);
+ g_object_unref (output_stream);
+ return NULL;
+ }
+
+ g_data_output_stream_put_string (data_output_stream,
+ query,
+ NULL,
+ &inner_error);
+
+ if (inner_error) {
+ g_propagate_error (error, inner_error);
+ g_object_unref (data_output_stream);
+ g_object_unref (buffered_output_stream);
+ g_object_unref (output_stream);
+ return NULL;
+ }
+
+ g_object_unref (data_output_stream);
+ g_object_unref (buffered_output_stream);
+ g_object_unref (output_stream);
+
+ return call;
+}
+
+static DBusMessage *
+sparql_update_fast (DBusConnection *connection,
+ const gchar *query,
+ FastOperationType type,
+ GError **error)
+{
+ DBusPendingCall *call;
+ DBusMessage *reply;
+
+ call = sparql_update_fast_send (connection, query, type, error);
+ if (!call) {
+ return NULL;
+ }
+
+ dbus_pending_call_block (call);
+
+ reply = dbus_pending_call_steal_reply (call);
+
+ if (dbus_message_get_type (reply) == DBUS_MESSAGE_TYPE_ERROR) {
+ DBusError dbus_error;
+
+ dbus_error_init (&dbus_error);
+ dbus_set_error_from_message (&dbus_error, reply);
+ dbus_set_g_error (error, &dbus_error);
+ dbus_pending_call_unref (call);
+ dbus_error_free (&dbus_error);
+
+ return NULL;
+ }
+
+ dbus_pending_call_unref (call);
+
+ return reply;
+}
+
+static void
+sparql_update_fast_async (DBusConnection *connection,
+ const gchar *query,
+ FastAsyncData *fad,
+ GError **error)
+{
+ DBusPendingCall *call;
+
+ call = sparql_update_fast_send (connection, query, fad->operation_type, error);
+ if (!call) {
+ /* Do some clean up ?*/
+ return;
+ }
+
+ fad->dbus_call = call;
+
+ dbus_pending_call_set_notify (call, sparql_update_fast_callback, fad, NULL);
+}
+
+#endif /* HAVE_DBUS_FD_PASSING */
+
+/* Public API */
+
+void
+tracker_bus_fd_sparql_update (DBusGConnection *connection,
+ const char *query,
+ GError **error)
+{
+#ifdef HAVE_DBUS_FD_PASSING
+ DBusMessage *reply;
+
+ g_return_if_fail (query != NULL);
+
+ reply = sparql_update_fast (dbus_g_connection_get_connection (connection),
+ query, FAST_UPDATE, error);
+
+ if (!reply) {
+ return;
+ }
+
+ dbus_message_unref (reply);
+#else
+ g_assert_not_reached ();
+#endif /* HAVE_DBUS_FD_PASSING */
+}
+
+void
+tracker_bus_fd_sparql_update_async (DBusGConnection *connection,
+ const char *query,
+ GCancellable *cancellable,
+ GAsyncReadyCallback callback,
+ gpointer user_data)
+{
+#ifdef HAVE_DBUS_FD_PASSING
+ FastAsyncData *fad;
+ GError *error = NULL;
+
+ g_return_val_if_fail (query != NULL, 0);
+ g_return_val_if_fail (callback != NULL, 0);
+
+ fad = fast_async_data_new (dbus_g_connection_get_connection (connection),
+ FAST_UPDATE, cancellable, user_data);
+ fad->callback = callback;
+
+ sparql_update_fast_async (dbus_g_connection_get_connection (connection),
+ query, fad, &error);
+
+ if (error) {
+ g_critical ("Could not initiate update: %s", error->message);
+ g_error_free (error);
+ fast_async_data_free (fad);
+ }
+
+#else
+ g_assert_not_reached ();
+#endif /* HAVE_DBUS_FD_PASSING */
+}
+
+/*
+GPtrArray *
+tracker_bus_fd_sparql_update_blank (DBusGConnection *connection,
+ const gchar *query,
+ GError **error)
+{
+#ifdef HAVE_DBUS_FD_PASSING
+ DBusMessage *reply;
+ DBusMessageIter iter, subiter, subsubiter;
+ GPtrArray *result;
+
+ g_return_val_if_fail (query != NULL, NULL);
+
+ reply = sparql_update_fast (dbus_g_connection_get_connection (connection),
+ query, FAST_UPDATE_BLANK, error);
+
+ if (!reply) {
+ return NULL;
+ }
+
+ if (g_strcmp0 (dbus_message_get_signature (reply), "aaa{ss}")) {
+ g_set_error (error,
+ TRACKER_BUS_ERROR,
+ TRACKER_BUS_ERROR_UNSUPPORTED,
+ "Server returned invalid results");
+ dbus_message_unref (reply);
+ return NULL;
+ }
+
+ result = g_ptr_array_new ();
+ dbus_message_iter_init (reply, &iter);
+ dbus_message_iter_recurse (&iter, &subiter);
+
+ while (dbus_message_iter_get_arg_type (&subiter) != DBUS_TYPE_INVALID) {
+ GPtrArray *inner_array;
+
+ inner_array = g_ptr_array_new ();
+ g_ptr_array_add (result, inner_array);
+ dbus_message_iter_recurse (&subiter, &subsubiter);
+
+ while (dbus_message_iter_get_arg_type (&subsubiter) != DBUS_TYPE_INVALID) {
+ g_ptr_array_add (inner_array, unmarshal_hash_table (&subsubiter));
+ dbus_message_iter_next (&subsubiter);
+ }
+
+ dbus_message_iter_next (&subiter);
+ }
+
+ dbus_message_unref (reply);
+
+ return result;
+#else * HAVE_DBUS_FD_PASSING *
+ g_assert_not_reached ();
+ return NULL;
+#endif * HAVE_DBUS_FD_PASSING *
+}
+
+void
+tracker_bus_fd_sparql_update_blank_async (DBusGConnection *connection,
+ const gchar *query,
+ GCancellable *cancellable,
+ GAsyncReadyCallback callback,
+ gpointer user_data)
+{
+#ifdef HAVE_DBUS_FD_PASSING
+ FastAsyncData *fad;
+ GError *error = NULL;
+
+ g_return_val_if_fail (query != NULL, 0);
+ g_return_val_if_fail (callback != NULL, 0);
+
+ fad = fast_async_data_new (dbus_g_connection_get_connection (connection),
+ FAST_UPDATE_BLANK,
+ cancellable,
+ user_data);
+ fad->gptrarray_callback = callback;
+
+ sparql_update_fast_async (dbus_g_connection_get_connection (connection),
+ query, fad, &error);
+
+ if (error) {
+ g_critical ("Could not initiate update: %s", error->message);
+ g_error_free (error);
+
+ fast_async_data_free (fad);
+ }
+
+#else * HAVE_DBUS_FD_PASSING *
+ g_assert_not_reached ();
+#endif * HAVE_DBUS_FD_PASSING *
+}
+*/
diff --git a/src/libtracker-bus/tracker-bus.vala b/src/libtracker-bus/tracker-bus.vala
index cf1d75a..94935dd 100644
--- a/src/libtracker-bus/tracker-bus.vala
+++ b/src/libtracker-bus/tracker-bus.vala
@@ -31,8 +31,16 @@ private interface Tracker.Bus.Resources : GLib.Object {
public abstract async void sparql_update_async (string query) throws DBus.Error;
}
-// Imported DBus FD API until we have support with Vala
+// Imported DBus FD APIs until we have support with Vala
public extern Tracker.Sparql.Cursor tracker_bus_fd_query (DBus.Connection connection, string query) throws GLib.Error;
+public extern void tracker_bus_fd_sparql_update (DBus.Connection connection, string query) throws GLib.Error;
+// public extern GLib.PtrArray tracker_bus_fd_sparql_update_blank (DBus.Connection connection, string query) throws GLib.Error;
+public extern async void tracker_bus_fd_sparql_update_async (DBus.Connection connection, string query, Cancellable? cancellable = null);
+// public extern async uint tracker_bus_fd_sparql_update_blank_async (DBus.Connection connection, string query, Cancellable? cancellable = null);
+
+public errordomain Tracker.BusError {
+ UNSUPPORTED
+}
// Actual class definition
public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
@@ -40,7 +48,7 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
static Resources resources;
static bool initialized;
static bool use_steroids;
-
+
public Connection ()
requires (!initialized) {
initialized = true;
@@ -48,7 +56,7 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
try {
connection = DBus.Bus.get (DBus.BusType.SESSION);
- // FIXME: Test for steroids and resources interfaces?
+ // FIXME: Test for steroids and resources interfaces?
use_steroids = false;
resources = (Resources) connection.get_object (TRACKER_DBUS_SERVICE,
@@ -75,7 +83,6 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
//
// Really we need #ifdef here, unsupported in vala AFAIK
-
if (use_steroids) {
return tracker_bus_fd_query (connection, sparql);
} else {
@@ -90,11 +97,19 @@ public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
}
public override void update (string sparql, Cancellable? cancellable = null) throws GLib.Error {
- resources.sparql_update (sparql);
+ if (use_steroids) {
+ tracker_bus_fd_sparql_update (connection, sparql);
+ } else {
+ resources.sparql_update (sparql);
+ }
}
public async override void update_async (string sparql, int? priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws GLib.Error {
- yield resources.sparql_update_async (sparql);
+ if (use_steroids) {
+ yield tracker_bus_fd_sparql_update_async (connection, sparql, cancellable);
+ } else {
+ yield resources.sparql_update_async (sparql);
+ }
}
}
[
Date Prev][
Date Next] [
Thread Prev][
Thread Next]
[
Thread Index]
[
Date Index]
[
Author Index]
