Merge pull request #1311 from grumvalski/async_unixodbc

db_unixodbc: added support for async queries

src/modules/db_unixodbc/db_unixodbc.c

@@ -99,7 +99,9 @@ int db_unixodbc_bind_api(db_func_t *dbb)
 		dbb->replace      = db_unixodbc_replace;
 	else
 		dbb->replace      = db_unixodbc_update_or_insert;
-
+	dbb->raw_query_async  = db_unixodbc_raw_query_async;
+	dbb->insert_async     = db_unixodbc_insert_async;
+
 	return 0;
 }
 

src/modules/db_unixodbc/dbase.c

@@ -36,6 +36,7 @@
 
 #include "../../core/mem/mem.h"
 #include "../../core/dprint.h"
+#include "../../core/async_task.h"
 #include "../../lib/srdb1/db_query.h"
 #include "val.h"
 #include "connection.h"
@@ -168,7 +169,68 @@ static int db_unixodbc_submit_query(const db1_con_t* _h, const str* _s)
 	return ret;
 }
 
-
+/**                                                                                                                                                                                                                                                                               
+ *
+ */
+void db_unixodbc_async_exec_task(void *param)
+{
+    str *p;
+    db1_con_t* dbc;
+
+    p = (str*)param;
+
+    dbc = db_unixodbc_init(&p[0]);
+
+    if(dbc==NULL) {
+        LM_ERR("failed to open connection for [%.*s]\n", p[0].len, p[0].s);
+        return;
+    }
+    if(db_unixodbc_submit_query(dbc, &p[1])<0) {
+		/* Sphere: we need the whole query for the reconciliation 
+		   LM_ERR("failed to execute query [%.*s] on async worker\n",
+		         (p[1].len>100)?100:p[1].len, p[1].s);
+		*/
+		LM_ERR("failed to execute query [%.*s] on async worker\n", p[1].len, p[1].s);
+    }
+    db_unixodbc_close(dbc);
+}
+/**
+ * Execute a raw SQL query via core async framework.
+ * \param _h handle for the database
+ * \param _s raw query string
+ * \return zero on success, negative value on failure
+ */
+int db_unixodbc_submit_query_async(const db1_con_t* _h, const str* _s)
+{
+    struct db_id* di;
+    async_task_t *atask;
+    int asize;
+    str *p;
+
+    di = ((struct pool_con*)_h->tail)->id;
+
+    asize = sizeof(async_task_t) + 2*sizeof(str) + di->url.len + _s->len + 2;
+    atask = shm_malloc(asize);
+    if(atask==NULL) {
+        LM_ERR("no more shared memory to allocate %d\n", asize);
+        return -1;
+    }
+
+    atask->exec = db_unixodbc_async_exec_task;
+    atask->param = (char*)atask + sizeof(async_task_t);
+
+    p = (str*)((char*)atask + sizeof(async_task_t));
+    p[0].s = (char*)p + 2*sizeof(str);
+    p[0].len = di->url.len;
+    strncpy(p[0].s, di->url.s, di->url.len);
+    p[1].s = p[0].s + p[0].len + 1;
+    p[1].len = _s->len;
+    strncpy(p[1].s, _s->s, _s->len);
+
+    async_task_push(atask);
+
+    return 0;
+}
 extern char *db_unixodbc_tquote;
 
 /*
@@ -437,6 +499,17 @@ int db_unixodbc_raw_query(const db1_con_t* _h, const str* _s, db1_res_t** _r)
 			db_unixodbc_store_result);
 }
 
+/**                                                                                                                                                                                                                                                                               
+ * Execute a raw SQL query via core async framework.
+ * \param _h handle for the database
+ * \param _s raw query string
+ * \return zero on success, negative value on failure
+ */
+int db_unixodbc_raw_query_async(const db1_con_t* _h, const str* _s)                                                                                                                                                                                                                  
+{
+	    return db_unixodbc_submit_query_async(_h, _s);
+
+}
 /*
  * Insert a row into specified table
  * _h: structure representing database connection
@@ -450,6 +523,19 @@ int db_unixodbc_insert(const db1_con_t* _h, const db_key_t* _k, const db_val_t*
 			db_unixodbc_submit_query);
 }
 
+/**
+ * Insert a row into a specified table via core async framework.
+ * \param _h structure representing database connection
+ * \param _k key names
+ * \param _v values of the keys
+ * \param _n number of key=value pairs
+ * \return zero on success, negative value on failure
+ */
+int db_unixodbc_insert_async(const db1_con_t* _h, const db_key_t* _k, const db_val_t* _v, const int _n)
+{
+	return db_do_insert(_h, _k, _v, _n, db_unixodbc_val2str,
+			db_unixodbc_submit_query_async);                                                                                                                                                                                                                                                 
+}
 /*
  * Delete a row from the specified table
  * _h: structure representing database connection

src/modules/db_unixodbc/dbase.h

@@ -69,11 +69,20 @@ int db_unixodbc_fetch_result(const db1_con_t* _h, db1_res_t** _r, const int nrow
  */
 int db_unixodbc_raw_query(const db1_con_t* _h, const str* _s, db1_res_t** _r);
 
+/*! \brief
+ *  * Raw SQL query via async framework
+ *   */
+int db_unixodbc_raw_query_async(const db1_con_t* _h, const str* _s);
+
 /*
  * Insert a row into table
  */
 int db_unixodbc_insert(const db1_con_t* _h, const db_key_t* _k, const db_val_t* _v, const int _n);
 
+/*! \brief
+ *  * Insert a row into table via async framework
+ *   */
+int db_unixodbc_insert_async(const db1_con_t* _h, const db_key_t* _k, const db_val_t* _v, const int _n);
 /*
  * Delete a row from table
  */