<commit_before>/*
 * Copyright (C) 2010, Nokia <<EMAIL>>
 *
 * 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.
 */

[DBus (name = "org.freedesktop.Tracker1.Resources")]
private interface Tracker.Bus.Resources : DBusProxy {
	public abstract void load (string uri, Cancellable? cancellable) throws Sparql.Error, DBusError;
	[DBus (name = "Load")]
	public abstract async void load_async (string uri, Cancellable? cancellable) throws Sparql.Error, DBusError;
}

[DBus (name = "org.freedesktop.Tracker1.Steroids")]
private interface Tracker.Bus.Steroids : DBusProxy {
	public abstract async string[] query (string query, UnixOutputStream result_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	public abstract async void update (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	[DBus (signature = "aaa{ss}")]
	public abstract async Variant update_blank (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	public abstract async void batch_update (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	[DBus (signature = "as")]
	public abstract async Variant update_array (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	[DBus (signature = "as")]
	public abstract async Variant batch_update_array (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;

	[DBus (visible = false)]
	public void update_begin (UnixInputStream sparql_stream, int priority, Cancellable? cancellable, AsyncReadyCallback callback) {
		if (priority >= GLib.Priority.DEFAULT) {
			update.begin (sparql_stream, cancellable, callback);
		} else {
			batch_update.begin (sparql_stream, cancellable, callback);
		}
	}

	[DBus (visible = false)]
	public void update_array_begin (UnixInputStream sparql_stream, int priority, Cancellable? cancellable, AsyncReadyCallback callback) {
		if (priority >= GLib.Priority.DEFAULT) {
			update_array.begin (sparql_stream, cancellable, callback);
		} else {
			batch_update_array.begin (sparql_stream, cancellable, callback);
		}
	}
}

[DBus (name = "org.freedesktop.Tracker1.Statistics")]
private interface Tracker.Bus.Statistics : DBusProxy {
	public abstract string[,] Get (Cancellable? cancellable) throws DBusError;
	public async abstract string[,] Get_async (Cancellable? cancellable) throws DBusError;
}

// Actual class definition
public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
	static Resources resources_object;
	static Steroids steroids_object;
	static Statistics statistics_object;
	static bool initialized;

	public Connection ()
	requires (!initialized) {
		initialized = true;
		
		try {
			// FIXME: Ideally we would just get these as and when we need them
			resources_object = GLib.Bus.get_proxy_sync (BusType.SESSION,
			                                            TRACKER_DBUS_SERVICE,
			                                            TRACKER_DBUS_OBJECT_RESOURCES);
			steroids_object = GLib.Bus.get_proxy_sync (BusType.SESSION,
			                                           TRACKER_DBUS_SERVICE,
			                                           TRACKER_DBUS_OBJECT_STEROIDS);
			statistics_object = GLib.Bus.get_proxy_sync (BusType.SESSION,
			                                             TRACKER_DBUS_SERVICE,
			                                             TRACKER_DBUS_OBJECT_STATISTICS);
		} catch (DBusError e) {
			warning ("Could not connect to D-Bus service:'%s': %s", TRACKER_DBUS_INTERFACE_RESOURCES, e.message);
			initialized = false;
			return;
		}
		
		initialized = true;
	}
 
	~Connection () {
		initialized = false;
	}

	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");
		}
		input = new UnixInputStream (pipefd[0], true);
		output = new UnixOutputStream (pipefd[1], true);
	}

	public override Sparql.Cursor query (string sparql, Cancellable? cancellable) throws Sparql.Error, IOError {
		// 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, IOError {
		try {
			UnixInputStream input;
			UnixOutputStream output;
			pipe (out input, out output);

			// send D-Bus request
			AsyncResult dbus_res = null;
			bool received_result = false;
			steroids_object.query.begin (sparql, output, cancellable, (o, res) => {
				dbus_res = res;
				if (received_result) {
					query_async.callback ();
				}
			});

			output = null;

			// receive query results via FD
			var mem_stream = new MemoryOutputStream (null, 0, GLib.realloc, GLib.free);
			yield mem_stream.splice_async (input, OutputStreamSpliceFlags.CLOSE_SOURCE | OutputStreamSpliceFlags.CLOSE_TARGET, Priority.DEFAULT, cancellable);

			// wait for D-Bus reply
			received_result = true;
			if (dbus_res == null) {
				yield;
			}
			string[] variable_names = steroids_object.query.end (dbus_res);
			mem_stream.close ();
			return new FDCursor (mem_stream.steal_data (), mem_stream.data_size, variable_names);
		} catch (DBusError e) {
			throw new Sparql.Error.INTERNAL (e.message);
		}
	}

	public override void update (string sparql, int priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws Sparql.Error, IOError {
		// 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;
		update_async.begin (sparql, priority, cancellable, (o, res) => {
			async_res = res;
			loop.quit ();
		});
		loop.run ();
		context.pop_thread_default ();<commit_msg>libtracker-bus: Update to VAPI changes<commit_after>/*
 * Copyright (C) 2010, Nokia <<EMAIL>>
 *
 * 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.
 */

[DBus (name = "org.freedesktop.Tracker1.Resources")]
private interface Tracker.Bus.Resources : DBusProxy {
	public abstract void load (string uri, Cancellable? cancellable) throws Sparql.Error, DBusError;
	[DBus (name = "Load")]
	public abstract async void load_async (string uri, Cancellable? cancellable) throws Sparql.Error, DBusError;
}

[DBus (name = "org.freedesktop.Tracker1.Steroids")]
private interface Tracker.Bus.Steroids : DBusProxy {
	public abstract async string[] query (string query, UnixOutputStream result_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	public abstract async void update (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	[DBus (signature = "aaa{ss}")]
	public abstract async Variant update_blank (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	public abstract async void batch_update (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	[DBus (signature = "as")]
	public abstract async Variant update_array (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;
	[DBus (signature = "as")]
	public abstract async Variant batch_update_array (UnixInputStream sparql_stream, Cancellable? cancellable) throws Sparql.Error, DBusError;

	[DBus (visible = false)]
	public void update_begin (UnixInputStream sparql_stream, int priority, Cancellable? cancellable, AsyncReadyCallback callback) {
		if (priority >= GLib.Priority.DEFAULT) {
			update.begin (sparql_stream, cancellable, callback);
		} else {
			batch_update.begin (sparql_stream, cancellable, callback);
		}
	}

	[DBus (visible = false)]
	public void update_array_begin (UnixInputStream sparql_stream, int priority, Cancellable? cancellable, AsyncReadyCallback callback) {
		if (priority >= GLib.Priority.DEFAULT) {
			update_array.begin (sparql_stream, cancellable, callback);
		} else {
			batch_update_array.begin (sparql_stream, cancellable, callback);
		}
	}
}

[DBus (name = "org.freedesktop.Tracker1.Statistics")]
private interface Tracker.Bus.Statistics : DBusProxy {
	public abstract string[,] Get (Cancellable? cancellable) throws DBusError;
	public async abstract string[,] Get_async (Cancellable? cancellable) throws DBusError;
}

// Actual class definition
public class Tracker.Bus.Connection : Tracker.Sparql.Connection {
	static Resources resources_object;
	static Steroids steroids_object;
	static Statistics statistics_object;
	static bool initialized;

	public Connection ()
	requires (!initialized) {
		initialized = true;
		
		try {
			// FIXME: Ideally we would just get these as and when we need them
			resources_object = GLib.Bus.get_proxy_sync (BusType.SESSION,
			                                            TRACKER_DBUS_SERVICE,
			                                            TRACKER_DBUS_OBJECT_RESOURCES);
			steroids_object = GLib.Bus.get_proxy_sync (BusType.SESSION,
			                                           TRACKER_DBUS_SERVICE,
			                                           TRACKER_DBUS_OBJECT_STEROIDS);
			statistics_object = GLib.Bus.get_proxy_sync (BusType.SESSION,
			                                             TRACKER_DBUS_SERVICE,
			                                             TRACKER_DBUS_OBJECT_STATISTICS);
		} catch (DBusError e) {
			warning ("Could not connect to D-Bus service:'%s': %s", TRACKER_DBUS_INTERFACE_RESOURCES, e.message);
			initialized = false;
			return;
		}
		
		initialized = true;
	}
 
	~Connection () {
		initialized = false;
	}

	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");
		}
		input = new UnixInputStream (pipefd[0], true);
		output = new UnixOutputStream (pipefd[1], true);
	}

	public override Sparql.Cursor query (string sparql, Cancellable? cancellable) throws Sparql.Error, IOError {
		// 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, IOError {
		try {
			UnixInputStream input;
			UnixOutputStream output;
			pipe (out input, out output);

			// send D-Bus request
			AsyncResult dbus_res = null;
			bool received_result = false;
			steroids_object.query.begin (sparql, output, cancellable, (o, res) => {
				dbus_res = res;
				if (received_result) {
					query_async.callback ();
				}
			});

			output = null;

			// receive query results via FD
			var mem_stream = new MemoryOutputStream (null, GLib.realloc, GLib.free);
			yield mem_stream.splice_async (input, OutputStreamSpliceFlags.CLOSE_SOURCE | OutputStreamSpliceFlags.CLOSE_TARGET, Priority.DEFAULT, cancellable);

			// wait for D-Bus reply
			received_result = true;
			if (dbus_res == null) {
				yield;
			}
			string[] variable_names = steroids_object.query.end (dbus_res);
			mem_stream.close ();
			return new FDCursor (mem_stream.steal_data (), mem_stream.data_size, variable_names);
		} catch (DBusError e) {
			throw new Sparql.Error.INTERNAL (e.message);
		}
	}

	public override void update (string sparql, int priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws Sparql.Error, IOError {
		// 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;
		update_async.begin (sparql, priority, cancellable, (o, res) => {
			async_res = res;
			loop.quit ();
		});
		loop.run ();
		context.pop_thread_default ();
{'fraction_non_alphanumeric': 0.09131582807686665, 'fraction_numerical': 0.003593956285756198, 'mean_word_length': 3.775831873905429, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5198641', 'n_tokens_mistral': 4396, 'n_tokens_neox': 3962, 'n_words': 1229}