hedera/db/db-conn.c

/*
 * Copyright (C) 2012 - Juan Ferrer Toribio
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "db-conn.h"
#include <glib/gstdio.h>
#include <gmodule.h>

#define IS_OPEN(obj)		(obj->status & DB_CONN_OPEN)
#define IS_CLOSING(obj)		(obj->status & DB_CONN_CLOSING)
#define IS_OPENING(obj)		(obj->status & DB_CONN_OPENING)
#define IS_TRANSACTION(obj)	(obj->status & DB_CONN_TRANSACTION)
#define IS_LOST(obj)		(obj->status & DB_CONN_LOST)

typedef struct
{
	DbConn * obj;
	gpointer data;
}
IdleData;

enum {
	 STATUS_CHANGED
	,ERROR
	,LAST_SIGNAL
};

static guint signals[LAST_SIGNAL] = {0};

/**
 * SECTION: db-conn
 * @Short_description: connection managing and wrapping class.
 * @Title: DbConn
 * @See_also: #DbPg
 * 
 * This class manages a connection to any kind of database (as long as it has
 * a plugin to acces to it). It uses the plugin set on the configuration 
 * parameters (not yet implemented) to connect, query and disconnect.
 * 
 * To do this use the methods db_conn_open(), db_conn_query() and
 * db_conn_close() respectively.
 * 
 * It is also possible to parse a SQL query and get an #SqlStmt object using the
 * db_conn_parse() method.
 *
 * This class is thread safe.
 **/
G_DEFINE_TYPE (DbConn, db_conn, G_TYPE_OBJECT);

/**
 * db_conn_new:
 *
 * Creates a new connection.
 *
 * Return value: the #DbConn
 **/
DbConn * db_conn_new ()
{
	return g_object_new (DB_TYPE_CONN, NULL);
}

//+++++++++++++++++++++++++++++++++++++++++++++++++++ Private

/*
 * Frees the #IdleData struct.
 */
static void idle_data_free (IdleData * idle_data)
{
	g_object_unref (idle_data->obj);
	g_free (idle_data);
}

/*
 * Notifies that an error happened.
 */
static gboolean db_conn_idle_error (IdleData * idle_data)
{
	GError * error = idle_data->data;
	g_signal_emit (idle_data->obj, signals[ERROR], 0, error);
	g_error_free (error);
	return G_SOURCE_REMOVE;
}

/*
 * Sets and error.
 */
static void db_conn_set_error (DbConn * obj, const GError * error)
{
	if (error)
	{
		IdleData * idle_data = g_new (IdleData, 1);
		idle_data->obj = g_object_ref (obj);
		idle_data->data = g_error_copy (error);
	
		g_idle_add_full (G_PRIORITY_DEFAULT,
			(GSourceFunc) db_conn_idle_error, idle_data, (GDestroyNotify) idle_data_free);
	}
}

/*
 * Notifies that connection status has changed.
 */
static gboolean db_conn_idle_status (IdleData * idle_data)
{
	DbConnStatus status = GPOINTER_TO_INT (idle_data->data);
	g_signal_emit (idle_data->obj, signals[STATUS_CHANGED], 0, status);
	return G_SOURCE_REMOVE;
}

/*
 * Sets the status of the connection.
 */
static void db_conn_set_status (DbConn * obj, DbConnStatus status)
{
	IdleData * idle_data;
	
	if (obj->status != status)
	{
		idle_data = g_new (IdleData, 1);
		idle_data->obj = g_object_ref (obj);
		idle_data->data = GINT_TO_POINTER (status);
	
		g_idle_add_full (G_PRIORITY_DEFAULT,
			(GSourceFunc) db_conn_idle_status, idle_data, (GDestroyNotify) idle_data_free);

		obj->status = status;
	}
}

/*
 * Cancels all outstanding requests.
 */
static void db_conn_cancel_requests (DbConn * obj)
{
	DbRequest * request;

	while ((request = g_queue_pop_head (obj->requests)))
	{
		db_request_cancel (request);
		db_request_complete (request);
	}
	
	db_plugin_kill_query (obj->plugin); 
}

/*
 * Executes asynchronous requests.
 */
static void db_conn_thread (DbConn * obj)
{
	gboolean exit = FALSE;
	gboolean exec_success;
	DbRequest * request;

	g_mutex_lock (obj->mutex);

	while (!exit)
	{	
		if (IS_OPEN (obj) && !IS_TRANSACTION (obj))
		{
			db_conn_set_status (obj, obj->status | DB_CONN_LOADING);

			while (TRUE)
			{
				if (!(request = g_queue_pop_head (obj->requests)))
					break;

				g_mutex_unlock (obj->mutex);
				exec_success = db_request_exec (request, NULL);
				g_mutex_lock (obj->mutex);

				if (!exec_success && IS_LOST (obj))
				{
					g_queue_push_head (obj->requests, request);
				}
				else
				{
					db_request_complete (request);
					g_object_unref (request);
				}
				
				if (IS_LOST (obj))
					break;
			}
			
			db_conn_set_status (obj, obj->status & ~DB_CONN_LOADING);
		}
		
		if (!IS_CLOSING (obj))
			g_cond_wait (obj->thread_cond, obj->mutex);
		else
			exit = TRUE;
	}

	g_mutex_unlock (obj->mutex);
}

/*
 * Adds a request to the queue.
 */
static void db_conn_add_request (DbConn * obj, DbRequest * request)
{
	g_mutex_lock (obj->mutex);
	
	if ((IS_OPEN (obj) || IS_LOST (obj)) && !IS_CLOSING (obj))
	{
		g_queue_push_tail (obj->requests, g_object_ref (request));
		
		if (!IS_TRANSACTION (obj))
			g_cond_signal (obj->thread_cond);
	}
	else
	{
		db_request_cancel (request);
		db_request_complete (request);
	}

	g_mutex_unlock (obj->mutex);
}

//+++++++++++++++++++++++++++++++++++++++++++++++++++ Public

/**
 * db_conn_load_plugin:
 * @obj: a #DbConn
 * @plugin: the plugin name
 * @err: return address of a #GError or #NULL
 *
 * Tries to load a plugin.
 *
 * Return vale: %TRUE if plugin was loaded successfully, %FALSE ortherwise
 **/
gboolean db_conn_load_plugin (DbConn * obj, const gchar * plugin, GError ** err)
{
	gchar * aux;
	gchar * path;
	gchar * plugin_name;
	GModule * module;
	DbPluginGetTypeFunc symbol;
	gboolean ret = FALSE;

	g_return_val_if_fail (DB_IS_CONN (obj), FALSE);
	g_return_val_if_fail (plugin, FALSE);
	
	g_mutex_lock (obj->mutex);
	
	if (!obj->plugin)
	{
		aux = g_strdup_printf (_PLUGIN_DIR, plugin);
		plugin_name = g_strdup_printf ("db%s", plugin);
		path = g_module_build_path (aux, plugin_name);
		g_free (aux);

		if ((module = g_module_open (path, 0)))
			g_module_make_resident (module);

		aux = g_strdup_printf ("db_%s_get_type", plugin);
	
		if (module && g_module_symbol (module, aux, (gpointer) &symbol))
		{
			obj->plugin_name = g_strdup (plugin);
			obj->plugin = g_object_new (symbol (), NULL);
			ret = TRUE;
		}
		else if (err)
			g_set_error (err
				,DB_CONN_LOG_DOMAIN
				,DB_CONN_ERROR_PLUGIN
				,_("Can't load DbPlugin '%s': %s")
				,plugin
				,g_module_error ()
			);
		else
			g_warning (_("Can't load DbPlugin '%s': %s")
				,plugin
				,g_module_error ()
			);

		g_free (plugin_name);
		g_free (path);
		g_free (aux);
	}
	else if (err)
		g_set_error (err
			,DB_CONN_LOG_DOMAIN
			,DB_CONN_ERROR_PLUGIN
			,_("Plugin can't be loaded")
		);

	g_mutex_unlock (obj->mutex);
	
	return ret;
}

/**
 * db_conn_set_query_path:
 * @obj: a #DbConn
 * @path: the path
 *
 * Sets the query search path.
 **/
void db_conn_set_query_path (DbConn * obj, const gchar * path)
{
	g_return_if_fail (DB_IS_CONN (obj));

	g_mutex_lock (obj->settings_mutex);

	g_free (obj->query_path);
	g_strfreev (obj->query_dirs);
	obj->query_path = g_strdup (path);
	obj->query_dirs = g_strsplit (obj->query_path, G_SEARCHPATH_SEPARATOR_S, 0);

	g_mutex_unlock (obj->settings_mutex);
}

/**
 * db_conn_get_query_path:
 * @obj: a #DbConn
 *
 * Sets the query search path.
 *
 * Return value: the path, should be freed with g_free()
 **/
gchar * db_conn_get_query_path (DbConn * obj)
{
	gchar * path;
	
	g_return_val_if_fail (DB_IS_CONN (obj), NULL);

	g_mutex_lock (obj->settings_mutex);
	path = g_strdup (obj->query_path);
	g_mutex_unlock (obj->settings_mutex);

	return path;
}

/**
 * db_conn_open:
 * @obj: a #DbConn
 * @host: Host name
 * @schema: Database name
 * @user: User name
 * @pass: Password
 *
 * Opens a new connection to the database and creates the thread that handles
 * all requests made to it asincronously.
 *
 * Return value: %TRUE if the connection was successfully open, %FALSE otherwise
 **/
gboolean db_conn_open (DbConn * obj, const gchar * host, const gchar * schema,
	const gchar * user, const gchar * pass, GError ** err)
{
	gboolean opened = FALSE;

	g_return_val_if_fail (DB_IS_CONN (obj), FALSE);

	g_mutex_lock (obj->mutex);
	
	if (!IS_OPEN (obj) && !IS_OPENING (obj) && !IS_CLOSING (obj))
	{
		DbConnStatus last_status = obj->status;

		db_conn_set_status (obj, obj->status | DB_CONN_OPENING | DB_CONN_LOADING);

		g_mutex_unlock (obj->mutex);
		opened = db_plugin_open (obj->plugin
			,host
			,schema
			,user
			,pass
			,err
		);
		g_mutex_lock (obj->mutex);

		if (opened)
		{
			g_mutex_lock (obj->settings_mutex);
			g_free (obj->host);
			g_free (obj->schema);
			g_free (obj->user);
			g_free (obj->pass);
			obj->host = g_strdup (host);
			obj->schema = g_strdup (schema);
			obj->user = g_strdup (user);
			obj->pass = g_strdup (pass);
			g_mutex_unlock (obj->settings_mutex);

			db_conn_set_status (obj, DB_CONN_OPEN);
			
			if (last_status & DB_CONN_LOST)
				g_cond_signal (obj->thread_cond);
			else
				obj->thread = g_thread_new ("DbConn",
					(GThreadFunc) db_conn_thread, obj);
		}
		else
			db_conn_set_status (obj, last_status);
	}
	else
		g_set_error (err
			,DB_CONN_LOG_DOMAIN
			,DB_CONN_ERROR_OPENING
			,_("Can't open a new connection, it's already open.")
		);
	
	g_mutex_unlock (obj->mutex);	
	return opened;
}

void db_conn_set_ssl (DbConn * obj, const gchar * ca)
{
	g_return_if_fail (DB_IS_CONN (obj));

	db_plugin_set_ssl (obj->plugin, ca);
}

/**
 * db_conn_reconnect:
 * @obj: a #DbConn
 *
 * Tries to reconnect to database.
 **/
gboolean db_conn_reconnect (DbConn * obj, GError ** err)
{
	gboolean opened;
	gchar * host;
	gchar * schema;
	gchar * user;
	gchar * pass;

	g_return_val_if_fail (DB_IS_CONN (obj), FALSE);
	
	g_mutex_lock (obj->settings_mutex);
	host = g_strdup (obj->host);
	schema = g_strdup (obj->schema);
	user = g_strdup (obj->user);
	pass = g_strdup (obj->pass);
	g_mutex_unlock (obj->settings_mutex);

	opened = db_conn_open (obj, host, schema, user, pass, err);

	g_free (host);
	g_free (schema);
	g_free (user);
	g_free (pass);
	
	return opened;
}

/**
 * db_conn_close:
 * @obj: a #DbConn
 * @wait: specifies whether to wait for pending requests before close the connection
 *
 * Closes the current connection to the database and the associated thread.
 **/
void db_conn_close (DbConn * obj, gboolean wait)
{
	g_return_if_fail (DB_IS_CONN (obj));

	g_mutex_lock (obj->mutex);
	
	if ((IS_OPEN (obj) || (IS_LOST (obj) && !IS_OPENING (obj))) && !IS_CLOSING (obj))
	{
		db_conn_set_status (obj, obj->status | DB_CONN_CLOSING | DB_CONN_LOADING);

		if (!wait)
			db_conn_cancel_requests (obj);

		g_cond_signal (obj->thread_cond);
		g_mutex_unlock (obj->mutex);

		g_thread_join (obj->thread);
		obj->thread = NULL;

		db_plugin_close (obj->plugin);

		g_mutex_lock (obj->mutex);
		db_conn_set_status (obj, DB_CONN_CLOSED);

		g_mutex_lock (obj->settings_mutex);
		g_free (obj->host);
		g_free (obj->schema);
		g_free (obj->user);
		g_free (obj->pass);
		obj->host = NULL;
		obj->schema = NULL;
		obj->user = NULL;
		obj->pass = NULL;
		g_mutex_unlock (obj->settings_mutex);
	}

	g_mutex_unlock (obj->mutex);
}

/**
 * db_conn_exec:
 * @obj: a #DbConn
 * @sql: the SQL statement
 * @err: (out) (allow-none): return location for a #GError or %NULL
 *
 * Sends a query to the database and waits for the resultset. The user is
 * responsible for releasing the result with db_result_set_free().
 *
 * Return value: (transfer full) (allow-none): the #DbResultSet on success, %NULL on failure
 **/
DbResultSet * db_conn_exec (DbConn * obj, const gchar * sql, GError ** err)
{
	GError * query_error = NULL;

	g_return_val_if_fail (DB_IS_CONN (obj), NULL);

	if (sql && g_strcmp0 (sql, ""))
	{
		DbResultSet * set;
	
		// XXX: Only for debug purposes.
		g_message ("DbConn: Query sent: %s", sql);

		set = db_plugin_query (obj->plugin, sql, &query_error);
		
		if (!set && query_error)
		{
			if (query_error->code == DB_CONN_ERROR_LOST)
			{
				g_mutex_lock (obj->mutex);
				
				if (IS_OPEN (obj) && !IS_CLOSING (obj))
					db_conn_set_status (obj, DB_CONN_CLOSED | DB_CONN_LOST);
			
				g_mutex_unlock (obj->mutex);
			}
			
			db_conn_set_error (obj, query_error);
			g_propagate_error (err, query_error);			
		}
		
		return set;		
	}
	else
	{
		query_error = g_error_new (
			 DB_CONN_LOG_DOMAIN
			,DB_CONN_ERROR_QUERY_EMPTY
			,_("The query is empty.")
		);
		
		db_conn_set_error (obj, query_error);
		g_propagate_error (err, query_error);
	}

	return NULL;
}

/**
 * db_conn_query:
 * @obj: the #DbConn
 * @sql: the SQL statement
 *
 * Sends a query to the database and waits for the result.
 * 
 * Return value: (transfer full): the new #DbRequest
 **/
DbRequest * db_conn_query (DbConn * obj, const gchar * sql)
{
	DbRequest * request;

	g_return_val_if_fail (DB_IS_CONN (obj), NULL);
	
	request = db_request_new (obj, sql);
	db_request_exec (request, NULL);
	return request;
}

/**
 * db_conn_query_with_stmt:
 * @obj: the #DbConn
 * @stmt: the #SqlStmt
 *
 * Sends a stmt to the database and waits for the result.
 *
 * Return value: (transfer full): the new #DbRequest
 **/
DbRequest * db_conn_query_with_stmt (DbConn * obj, SqlStmt * stmt)
{
	DbRequest * request;

	g_return_val_if_fail (DB_IS_CONN (obj), NULL);

	request = db_request_new_with_stmt (obj, stmt);
	db_request_exec (request, NULL);
	return request;
}

/**
 * db_conn_query_async:
 * @obj: the #DbConn
 * @sql: the SQL statement
 * @callback:
 * @user_data:
 * @notify:
 *
 * Return value: (transfer full): the new #DbRequest
 **/
DbRequest * db_conn_query_async (DbConn * obj, const gchar * sql,
	DbRequestDoneCallback callback, gpointer user_data, GDestroyNotify notify)
{
	DbRequest * request;

	g_return_val_if_fail (DB_IS_CONN (obj), NULL);

	request = db_request_new (obj, sql);
	db_request_set_callback (request, callback, user_data, notify);
	db_conn_add_request (obj, request);
	return request;
}

/**
 * db_conn_query_with_stmt_async:
 * @obj: the #DbConn
 * @stmt: the #SqlStmt
 * @callback:
 * @user_data:
 * @notify:
 *
 * Return value: (transfer full): the new #DbRequest
 **/
DbRequest * db_conn_query_with_stmt_async (DbConn * obj, SqlStmt * stmt,
	DbRequestDoneCallback callback, gpointer user_data, GDestroyNotify notify)
{
	DbRequest * request;

	g_return_val_if_fail (DB_IS_CONN (obj), NULL);

	request = db_request_new_with_stmt (obj, stmt);
	db_request_set_callback (request, callback, user_data, notify);
	db_conn_add_request (obj, request);
	return request;
}

/**
 * db_conn_query_value:
 * @obj: a #DbConn
 * @sql: a SQL statement
 * @value: (out): return location for a #GValue
 * @err: (out) (allow-none): return location for a #GError or %NULL
 *
 * Sends a sql to the database and takes the first row value of the result.
 *
 * Return value: %TRUE on success, %FALSE on failure
 **/
gboolean db_conn_query_value (DbConn * obj, const gchar * sql, GValue * value, GError ** err)
{
	gboolean success;
	DbRequest * request;

	g_return_val_if_fail (DB_IS_CONN (obj), FALSE);
	
	request = db_conn_query (obj, sql);
	success = db_request_fetch_value (request, value, err);
	g_object_unref (request);
	
	return success;
}

/**
 * db_conn_kill_query:
 * @obj: a #DbConn
 *
 * Attempts to kill the query that is running, if any.
 **/
void db_conn_kill_query (DbConn * obj)
{
	g_return_if_fail (DB_IS_CONN (obj));

	db_plugin_kill_query (obj->plugin); 
}

/**
 * db_conn_retry:
 * @obj: a #DbConn
 *
 * Tries rerunning pending requests.
 **/
void db_conn_retry (DbConn * obj)
{
	g_return_if_fail (DB_IS_CONN (obj));

	g_cond_signal (obj->thread_cond);
}

/**
 * db_conn_parse:
 * @obj: a #DbConn
 * @sql: an SQL string
 * 
 * Return value: (transfer full): a #SqlStmt
 **/
SqlStmt * db_conn_parse (DbConn * obj, gchar * sql)
{
	g_return_val_if_fail (DB_IS_CONN (obj), NULL);

	return db_plugin_parse (obj->plugin, sql);
}

/**
 * db_conn_render:
 * @obj: a #DbConn.
 * @object: the #GObject to render
 * @err: (out) (allow-none): the return location for #GError
 *
 * Renders a #GObject object as a SQL string to send it in a database
 * query. It takes the connection to know the codification in wich to escape
 * the data.
 * 
 * Return value: (transfer full): the rendered string, or %NULL if error.
 **/
gchar *	db_conn_render (DbConn * obj, gpointer object, GError ** err)
{
	g_return_val_if_fail (DB_IS_CONN (obj), NULL);
	g_return_val_if_fail (G_IS_OBJECT (object), NULL);

	return db_plugin_render (obj->plugin, object, err);
}

/**
 * db_conn_start_transaction:
 * @obj: a #DbConn
 *
 * Enter the transaction state, all requests made are retained until you call
 * the method db_conn_comit.
 **/
void db_conn_start_transaction (DbConn * obj)
{
	g_return_if_fail (DB_IS_CONN (obj));

	g_mutex_lock (obj->mutex);

	if (IS_OPEN (obj) && !IS_CLOSING (obj))
	{
		obj->transaction++;
		db_conn_set_status (obj, obj->status | DB_CONN_TRANSACTION);
	}

	g_mutex_unlock (obj->mutex);
}

/**
 * db_conn_commit:
 * @obj: a #DbConn
 *
 * Commits the current transaction. If transaction arrives to 0, all
 * outstanding requests in the queue are sent.
 **/
void db_conn_commit (DbConn * obj)
{
	g_return_if_fail (DB_IS_CONN (obj));

	g_mutex_lock (obj->mutex);

	if (IS_TRANSACTION (obj))
	{
		obj->transaction--;

		if (!obj->transaction)
		{
			db_conn_set_status (obj, obj->status & ~DB_CONN_TRANSACTION);
			g_mutex_unlock (obj->mutex);
			db_conn_retry (obj);
			return;
		}
	}

	g_mutex_unlock (obj->mutex);
}

/**
 * db_conn_rollback:
 * @obj: a #DbConn
 *
 * Deletes all pending requests from the input queue and rollbacks the current
 * transactions, if any.
 **/
void db_conn_rollback (DbConn * obj)
{
	g_return_if_fail (DB_IS_CONN (obj));

	g_mutex_lock (obj->mutex);

	if (IS_TRANSACTION (obj))
	{
		db_conn_cancel_requests (obj);
		obj->transaction = 0;
		db_conn_set_status (obj, obj->status & ~DB_CONN_TRANSACTION);
	}

	g_mutex_unlock (obj->mutex);
}

/**
 * db_conn_get_user:
 * @obj: a #DbConn
 *
 * Gets the connection user.
 *
 * Return value: the user, should be freed using g_free()
 **/
gchar * db_conn_get_user (DbConn * obj)
{
	gchar * user;
	
	g_return_val_if_fail (DB_IS_CONN (obj), NULL);

	g_mutex_lock (obj->settings_mutex);
	user = g_strdup (obj->user);
	g_mutex_unlock (obj->settings_mutex);

	return user;
}

/**
 * db_conn_get_host:
 * @obj: a #DbConn
 *
 * Gets the connection host.
 *
 * Return value: the host, should be freed using g_free()
 **/
gchar * db_conn_get_host (DbConn * obj)
{
	gchar * host;
	
	g_return_val_if_fail (DB_IS_CONN (obj), NULL);

	g_mutex_lock (obj->settings_mutex);
	host = g_strdup (obj->host);
	g_mutex_unlock (obj->settings_mutex);

	return host;
}

/**
 * db_conn_get_schema:
 * @obj: a #DbConn
 *
 * Gets the connection schema.
 *
 * Return value: the schema, should be freed using g_free()
 **/
gchar * db_conn_get_schema (DbConn * obj)
{
	gchar * schema;
	
	g_return_val_if_fail (DB_IS_CONN (obj), NULL);

	g_mutex_lock (obj->settings_mutex);
	schema = g_strdup (obj->schema);
	g_mutex_unlock (obj->settings_mutex);

	return schema;
}

/**
 * db_conn_create_stmt_from_file:
 * @obj: a #DbConn
 * @query_file: path to a file containing SQL statement
 * 
 * Reads an SQL query from the specified query_file and creates a new
 * #SqlString. The file path must be relative to any path in the "query-path"
 * property. 
 * 
 * Return value: (transfer full): an #SqlString
 **/
SqlString * db_conn_create_stmt_from_file (DbConn * obj, const gchar * query_file)
{
	gint i;
	gchar * file;
	SqlString * stmt = NULL;

	g_return_val_if_fail (DB_IS_CONN (obj), NULL);
	g_return_val_if_fail (query_file, NULL);

	if (g_str_has_suffix (query_file, ".sql"))
		file = g_strdup (query_file); 
	else
		file = g_strconcat (query_file, ".sql", NULL);

	if (obj->query_dirs)
	for (i = 0; obj->query_dirs[i] != NULL/* && !stmt*/; i++)
	{
		gchar * buffer;
		gchar * path = g_build_filename (obj->query_dirs[i], file, NULL);

		if (g_file_get_contents (path, &buffer, NULL, NULL))
		{
			stmt = sql_string_new (buffer);
			g_free (buffer);
		}

		g_free (path);
	}

	if (!stmt)
		g_warning ("DbConn: Can't create statement from file: %s", file);

	g_free (file);

	return stmt;
}

//+++++++++++++++++++++++++++++++++++++++++++++++++++ Properties

enum
{
	 PROP_PLUGIN = 1
	,PROP_QUERY_PATH
	,PROP_HOST
	,PROP_USER
	,PROP_DB
};

static void db_conn_set_property (DbConn * obj, guint id,
	const GValue * value, GParamSpec * pspec)
{
	switch (id)
	{
		case PROP_PLUGIN:
			db_conn_load_plugin (obj, g_value_get_string (value), NULL);
			break;
		case PROP_QUERY_PATH:
			db_conn_set_query_path (obj, g_value_get_string (value));
			break;
		default:
			G_OBJECT_WARN_INVALID_PROPERTY_ID (obj, id, pspec);	
	}
}

static void db_conn_get_property (DbConn * obj, guint id,
	GValue * value, GParamSpec * pspec)
{
	switch (id)
	{
		case PROP_PLUGIN:
			g_value_set_string (value, obj->plugin_name);
			break;
		case PROP_QUERY_PATH:
			g_value_set_string (value, db_conn_get_query_path (obj));
			break;
		case PROP_HOST:
			g_value_take_string (value, db_conn_get_host (obj));
			break;
		case PROP_USER:
			g_value_take_string (value, db_conn_get_user (obj));
			break;
		case PROP_DB:
			g_value_take_string (value, db_conn_get_schema (obj));
			break;
		default:		
			G_OBJECT_WARN_INVALID_PROPERTY_ID (obj, id, pspec);
	}
}

//+++++++++++++++++++++++++++++++++++++++++++++++++++ Class

static void db_conn_init (DbConn * obj)
{
	obj->plugin = NULL;
	obj->plugin_name = NULL;
	obj->query_path = NULL;
	obj->query_dirs = NULL;
	obj->schema = NULL;
	obj->host = NULL;
	obj->user = NULL;
	obj->pass = NULL;
	obj->thread = NULL;
	obj->thread_cond = g_new (GCond, 1);
	obj->transaction = 0;
	obj->status = DB_CONN_CLOSED;
	obj->requests = g_queue_new ();
	obj->mutex = g_new (GMutex, 1);
	obj->settings_mutex = g_new (GMutex, 1);
	g_mutex_init (obj->mutex);
	g_mutex_init (obj->settings_mutex);
	g_cond_init (obj->thread_cond);
}

static void db_conn_finalize (DbConn * obj)
{
	db_conn_close (obj, FALSE);

	if (obj->plugin)
		g_free (obj->plugin_name);
		
	g_mutex_clear (obj->settings_mutex);
	g_mutex_clear (obj->mutex);
	g_cond_clear (obj->thread_cond);
	g_free (obj->query_path);
	g_strfreev (obj->query_dirs);
	g_free (obj->settings_mutex);
	g_free (obj->mutex);
	g_free (obj->thread_cond);
	g_free (obj->host);
	g_free (obj->user);
	g_free (obj->pass);
	g_free (obj->schema);
	g_queue_free_full (obj->requests, (GDestroyNotify) g_object_unref);
	G_OBJECT_CLASS (db_conn_parent_class)->finalize (G_OBJECT (obj));
}

static void db_conn_class_init (DbConnClass * k)
{
	GObjectClass * klass = G_OBJECT_CLASS (k);
	klass->set_property = (GObjectSetPropertyFunc) db_conn_set_property;
	klass->get_property = (GObjectGetPropertyFunc) db_conn_get_property;
	klass->finalize = (GObjectFinalizeFunc) db_conn_finalize;

	signals[STATUS_CHANGED] = g_signal_new ("status-changed",
		DB_TYPE_CONN, G_SIGNAL_RUN_FIRST, 0, NULL, NULL,
		g_cclosure_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT
	);
	signals[ERROR] = g_signal_new ("error",
		DB_TYPE_CONN, G_SIGNAL_RUN_FIRST, 0, NULL, NULL,
		g_cclosure_marshal_VOID__POINTER, G_TYPE_NONE, 1, G_TYPE_POINTER
	);

	g_object_class_install_property (klass, PROP_PLUGIN,
		g_param_spec_string ("plugin"
			,_("Plugin")
			,_("The name of the plugin")
			,NULL
			,G_PARAM_READWRITE
	));
	g_object_class_install_property (klass, PROP_QUERY_PATH,
		g_param_spec_string ("query-path"
			,_("Query path")
			,_("The path where query files are located")
			,NULL
			,G_PARAM_READWRITE
	));
	g_object_class_install_property (klass, PROP_HOST,
		g_param_spec_string ("host"
			,_("Host")
			,_("The host name to connect to")
			,NULL
			,G_PARAM_READABLE
	));
	g_object_class_install_property (klass, PROP_USER,
		g_param_spec_string ("user"
			,_("User")
			,_("The user name")
			,NULL
			,G_PARAM_READABLE
	));
	g_object_class_install_property (klass, PROP_DB,
		g_param_spec_string ("db"
			,_("DB name")
			,_("The default schema")
			,NULL
			,G_PARAM_READABLE
	));
}