Merge pull request #1014 from lazedo/dbtext-fixes-master

db_text fixes & live query

src/modules/db_text/db_text.c

@@ -47,6 +47,7 @@ int db_mode = 0;  /* Database usage mode: 0 = cache, 1 = no cache */
 int empty_string = 0;  /* Treat empty string as "" = 0, 1 = NULL */
 int _db_text_read_buffer_size = DEFAULT_DB_TEXT_READ_BUFFER_SIZE;
 int _db_text_max_result_rows = DEFAULT_MAX_RESULT_ROWS;
+str dbt_default_connection = str_init("");
 
 int dbt_bind_api(db_func_t *dbb);
 
@@ -67,6 +68,7 @@ static param_export_t params[] = {
 	{"emptystring", INT_PARAM, &empty_string},
 	{"file_buffer_size", INT_PARAM, &_db_text_read_buffer_size},
 	{"max_result_rows", INT_PARAM, &_db_text_max_result_rows},
+	{"default_connection", PARAM_STR, &dbt_default_connection},
 	{0, 0, 0}
 };
 
@@ -156,8 +158,78 @@ static void rpc_dump(rpc_t *rpc, void *c) {
 	return;
 }
 
-static rpc_export_t rpc_methods[] = {
-	{"db_text.dump", rpc_dump, rpc_dump_doc, 0},
-	{0, 0, 0, 0}
+static const char *rpc_query_doc[2] = {
+        "Perform Live Query", 0
 };
 
+/* rpc function implementations */
+static void rpc_query(rpc_t *rpc, void *ctx) {
+        str sql;
+        db1_con_t *con;
+        db1_res_t* _r;
+        int res;
+        int n;
+        char *buf;
+        size_t len;
+        FILE *stream;
+        dbt_table_p tab;
+        dbt_row_p rowp;
+
+        rpc->scan(ctx, "S", &sql);
+
+        con = dbt_init(&dbt_default_connection);
+        if(con == NULL) {
+                rpc->rpl_printf(ctx, "invalid connection : %s", dbt_default_connection.s);
+                return;
+        }
+
+        res = dbt_raw_query(con, &sql, &_r);
+        if(res != 0) {
+                rpc->rpl_printf(ctx, "error executing sql statement");
+                goto end;
+        }
+
+
+        if(_r) {
+                tab = (dbt_table_p)_r->ptr;
+                if(_r->n == 0) {
+                        rpc->rpl_printf(ctx, "statement returned 0 rows");
+                } else {
+                        stream = open_memstream (&buf, &len);
+                        if (stream == NULL) {
+                                rpc->rpl_printf(ctx, "error opening stream");
+                                goto end;
+                        }
+                        dbt_print_table_header(tab, stream);
+                        fflush (stream);
+                        buf[len] = '\0';
+                        rpc->rpl_printf(ctx, "%s", buf);
+                        rowp = tab->rows;
+                        for(n=0; n < _r->n; n++) {
+                                fseeko (stream, 0, SEEK_SET);
+                                dbt_print_table_row_ex(tab, rowp, stream, 0);
+                                fflush (stream);
+                                buf[len] = '\0';
+                                rpc->rpl_printf(ctx, "%s", buf);
+                                rowp = rowp->next;
+                        }
+                        fclose (stream);
+                        free (buf);
+                        rpc->rpl_printf(ctx, "\ntotal rows %d / %d", _r->n, tab->nrrows);
+                }
+        } else {
+                rpc->rpl_printf(ctx, "%d affected rows", ((dbt_con_p)con->tail)->affected);
+        }
+
+        if(_r)
+                dbt_free_result(con, _r);
+
+end:
+        dbt_close(con);
+}
+
+static rpc_export_t rpc_methods[] = {
+        {"db_text.dump", rpc_dump, rpc_dump_doc, 0},
+        {"db_text.query", rpc_query, rpc_query_doc, 0},
+        {0, 0, 0, 0}
+};

src/modules/db_text/dbt_base.c

@@ -89,7 +89,6 @@ db1_con_t* dbt_init(const str* _sqlurl)
 	memset(_res, 0, sizeof(db1_con_t) + sizeof(dbt_con_t));
 	_res->tail = (unsigned long)((char*)_res+sizeof(db1_con_t));
 
-	LM_INFO("using database at: %.*s\n", _s.len, _s.s);
 	DBT_CON_CONNECTION(_res) = dbt_cache_get_db(&_s);
 	if (!DBT_CON_CONNECTION(_res))
 	{

src/modules/db_text/dbt_file.c

@@ -500,146 +500,178 @@ clean:
 /**
  *
  */
-int dbt_print_table(dbt_table_p _dtp, str *_dbn)
+int dbt_print_table_header(dbt_table_p _dtp, FILE *fout)
 {
-	dbt_column_p colp = NULL;
-	dbt_row_p rowp = NULL;
-	FILE *fout = NULL;
-	int ccol;
-	char *p, path[512];
-
-	if(!_dtp || !_dtp->name.s || _dtp->name.len <= 0)
-		return -1;
+        dbt_column_p colp = NULL;
+        colp = _dtp->cols;
+        while(colp)
+        {
+                switch(colp->type)
+                {
+                        case DB1_INT:
+                                fprintf(fout, "%.*s(int", colp->name.len, colp->name.s);
+                        break;
+                        case DB1_DOUBLE:
+                                fprintf(fout, "%.*s(double", colp->name.len, colp->name.s);
+                        break;
+                        case DB1_STR:
+                                fprintf(fout, "%.*s(str", colp->name.len, colp->name.s);
+                        break;
+                        case DB1_STRING:
+                                fprintf(fout, "%.*s(string", colp->name.len, colp->name.s);
+                        break;
+                        case DB1_BLOB:
+                                fprintf(fout, "%.*s(blob", colp->name.len, colp->name.s);
+                        break;
+                        case DB1_DATETIME:
+                                fprintf(fout, "%.*s(time", colp->name.len, colp->name.s);
+                        break;
+                        default:
+                                if(fout!=stdout)
+                                        fclose(fout);
+                                return -1;
+                }
+
+                if(colp->flag & DBT_FLAG_NULL)
+                                fprintf(fout,",null");
+                else if(colp->type==DB1_INT && colp->flag & DBT_FLAG_AUTO)
+                                        fprintf(fout,",auto");
+                fprintf(fout,")");
+
+                colp = colp->next;
+                if(colp)
+                        fprintf(fout,"%c", DBT_DELIM_C);
+        }
+        fprintf(fout, "%c", DBT_DELIM_R);
+        return 0;
+}
 
-	if(!_dbn || !_dbn->s || _dbn->len <= 0)
-	{
-		fout = stdout;
-		fprintf(fout, "\n Content of [%.*s::%.*s]\n",
-				_dtp->dbname.len, _dtp->dbname.s,
-				_dtp->name.len, _dtp->name.s);
-	}
-	else
-	{
-		if(_dtp->name.len+_dbn->len > 510)
-			return -1;
-		strncpy(path, _dbn->s, _dbn->len);
-		path[_dbn->len] = '/';
-		strncpy(path+_dbn->len+1, _dtp->name.s, _dtp->name.len);
-		path[_dbn->len+_dtp->name.len+1] = 0;
-		fout = fopen(path, "wt");
-		if(!fout)
-			return -1;
-	}
+int dbt_print_table_row_ex(dbt_table_p _dtp, dbt_row_p rowp, FILE *fout, int newline)
+{
+        int ccol;
+        char *p;
+        for(ccol=0; ccol<_dtp->nrcols; ccol++)
+        {
+                switch(_dtp->colv[ccol]->type)
+                {
+                case DB1_DATETIME:
+                case DB1_INT:
+                        if(!rowp->fields[ccol].nul)
+                                fprintf(fout,"%d",
+                                                rowp->fields[ccol].val.int_val);
+                        break;
+                case DB1_DOUBLE:
+                        if(!rowp->fields[ccol].nul)
+                                fprintf(fout, "%.2f",
+                                                rowp->fields[ccol].val.double_val);
+                        break;
+                case DB1_STR:
+                case DB1_STRING:
+                case DB1_BLOB:
+                        if(!rowp->fields[ccol].nul)
+                        {
+                                p = rowp->fields[ccol].val.str_val.s;
+                                while(p < rowp->fields[ccol].val.str_val.s
+                                                + rowp->fields[ccol].val.str_val.len)
+                                {
+                                        switch(*p)
+                                        {
+                                        case '\n':
+                                        fprintf(fout, "\\n");
+                                        break;
+                                        case '\r':
+                                                fprintf(fout, "\\r");
+                                                break;
+                                        case '\t':
+                                                fprintf(fout, "\\t");
+                                                break;
+                                        case '\\':
+                                                fprintf(fout, "\\\\");
+                                                break;
+                                        case DBT_DELIM:
+                                                fprintf(fout, "\\%c", DBT_DELIM);
+                                                break;
+                                        case '\0':
+                                                fprintf(fout, "\\0");
+                                                break;
+                                        default:
+                                                fprintf(fout, "%c", *p);
+                                        }
+                                        p++;
+                                }
+                        }
+                        break;
+                default:
+                        if(fout!=stdout)
+                                fclose(fout);
+                        return -1;
+                }
+                if(ccol<_dtp->nrcols-1)
+                        fprintf(fout, "%c",DBT_DELIM);
+        }
+        if(newline)
+                fprintf(fout, "%c", DBT_DELIM_R);
+
+        return 0;
+}
 
-	colp = _dtp->cols;
-	while(colp)
-	{
-		switch(colp->type)
-		{
-			case DB1_INT:
-				fprintf(fout, "%.*s(int", colp->name.len, colp->name.s);
-			break;
-			case DB1_DOUBLE:
-				fprintf(fout, "%.*s(double", colp->name.len, colp->name.s);
-			break;
-			case DB1_STR:
-				fprintf(fout, "%.*s(str", colp->name.len, colp->name.s);
-			break;
-			case DB1_STRING:
-				fprintf(fout, "%.*s(string", colp->name.len, colp->name.s);
-			break;
-			case DB1_BLOB:
-				fprintf(fout, "%.*s(blob", colp->name.len, colp->name.s);
-			break;
-			case DB1_DATETIME:
-				fprintf(fout, "%.*s(time", colp->name.len, colp->name.s);
-			break;
-			default:
-				if(fout!=stdout)
-					fclose(fout);
-				return -1;
-		}
+int dbt_print_table_row(dbt_table_p _dtp, dbt_row_p rowp, FILE *fout)
+{
+        return dbt_print_table_row_ex(_dtp, rowp, fout, 1);
+}
 
-		if(colp->flag & DBT_FLAG_NULL)
-				fprintf(fout,",null");
-		else if(colp->type==DB1_INT && colp->flag & DBT_FLAG_AUTO)
-					fprintf(fout,",auto");
-		fprintf(fout,")");
+int dbt_print_table_rows(dbt_table_p _dtp, FILE *fout)
+{
+        dbt_row_p rowp = _dtp->rows;
+        while(rowp) {
+                if(dbt_print_table_row(_dtp, rowp, fout))
+                        return -1;
+                rowp = rowp->next;
+        }
+
+        return 0;
+}
 
-		colp = colp->next;
-		if(colp)
-			fprintf(fout,"%c", DBT_DELIM_C);
-	}
-	fprintf(fout, "%c", DBT_DELIM_R);
-	rowp = _dtp->rows;
-	while(rowp)
-	{
-		for(ccol=0; ccol<_dtp->nrcols; ccol++)
-		{
-			switch(_dtp->colv[ccol]->type)
-			{
-				case DB1_DATETIME:
-				case DB1_INT:
-					if(!rowp->fields[ccol].nul)
-						fprintf(fout,"%d",
-								rowp->fields[ccol].val.int_val);
-				break;
-				case DB1_DOUBLE:
-					if(!rowp->fields[ccol].nul)
-						fprintf(fout, "%.2f",
-								rowp->fields[ccol].val.double_val);
-				break;
-				case DB1_STR:
-				case DB1_STRING:
-				case DB1_BLOB:
-					if(!rowp->fields[ccol].nul)
-					{
-						p = rowp->fields[ccol].val.str_val.s;
-						while(p < rowp->fields[ccol].val.str_val.s
-								+ rowp->fields[ccol].val.str_val.len)
-						{
-							switch(*p)
-							{
-								case '\n':
-									fprintf(fout, "\\n");
-								break;
-								case '\r':
-									fprintf(fout, "\\r");
-								break;
-								case '\t':
-									fprintf(fout, "\\t");
-								break;
-								case '\\':
-									fprintf(fout, "\\\\");
-								break;
-								case DBT_DELIM:
-									fprintf(fout, "\\%c", DBT_DELIM);
-								break;
-								case '\0':
-									fprintf(fout, "\\0");
-								break;
-								default:
-									fprintf(fout, "%c", *p);
-							}
-							p++;
-						}
-					}
-				break;
-				default:
-					if(fout!=stdout)
-						fclose(fout);
-					return -1;
-			}
-			if(ccol<_dtp->nrcols-1)
-				fprintf(fout, "%c",DBT_DELIM);
-		}
-		fprintf(fout, "%c", DBT_DELIM_R);
-		rowp = rowp->next;
-	}
+int dbt_print_table_content(dbt_table_p _dtp, FILE *fout)
+{
+        if(dbt_print_table_header(_dtp, fout))
+                return -1;
+        return dbt_print_table_rows(_dtp, fout);
+}
 
-	if(fout!=stdout)
-		fclose(fout);
+int dbt_print_table(dbt_table_p _dtp, str *_dbn)
+{
+        FILE *fout = NULL;
+        int res=0;
+        char path[512];
+
+        if(!_dtp || !_dtp->name.s || _dtp->name.len <= 0)
+                return -1;
+
+        if(!_dbn || !_dbn->s || _dbn->len <= 0)
+        {
+                fout = stdout;
+                fprintf(fout, "\n Content of [%.*s::%.*s]\n",
+                                _dtp->dbname.len, _dtp->dbname.s,
+                                _dtp->name.len, _dtp->name.s);
+        }
+        else
+        {
+                if(_dtp->name.len+_dbn->len > 510)
+                        return -1;
+                strncpy(path, _dbn->s, _dbn->len);
+                path[_dbn->len] = '/';
+                strncpy(path+_dbn->len+1, _dtp->name.s, _dtp->name.len);
+                path[_dbn->len+_dtp->name.len+1] = 0;
+                fout = fopen(path, "wt");
+                if(!fout)
+                        return -1;
+        }
+
+        res = dbt_print_table_content(_dtp, fout);
+        if(fout!=stdout)
+                fclose(fout);
+
+        return res;
 
-	return 0;
 }
-

src/modules/db_text/dbt_lib.c

@@ -143,7 +143,7 @@ dbt_cache_p dbt_cache_get_db(str *_s)
 		LM_ERR("database [%.*s] does not exists!\n", _s->len, _s->s);
 		goto done;
 	}
-	LM_DBG("new db!\n");
+	LM_INFO("using database at: %.*s\n", _s->len, _s->s);
 
 	_dcache = (dbt_cache_p)shm_malloc(sizeof(dbt_cache_t));
 	if(!_dcache)

src/modules/db_text/dbt_lib.h

@@ -141,6 +141,11 @@ int dbt_table_update_flags(dbt_table_p, int, int, int);
 int dbt_check_mtime(const str *, const str *, time_t *);
 dbt_table_p dbt_load_file(const str *, const str *);
 int dbt_print_table(dbt_table_p, str *);
+int dbt_print_table_header(dbt_table_p _dtp, FILE *fout);
+int dbt_print_table_row(dbt_table_p _dtp, dbt_row_p rowp, FILE *fout);
+int dbt_print_table_row_ex(dbt_table_p _dtp, dbt_row_p rowp, FILE *fout, int newline);
+int dbt_print_table_rows(dbt_table_p _dtp, FILE *fout);
+int dbt_print_table_content(dbt_table_p _dtp, FILE *fout);
 int dbt_is_neq_type(db_type_t _t0, db_type_t _t1);
 
 #endif

src/modules/db_text/dbt_raw_query.c

@@ -89,8 +89,8 @@ int dbt_raw_query_select(db1_con_t* _h, str* _s, db1_res_t** _r)
 	dbt_trim(table_ptr);
 
 	table.s = table_ptr;
-	table.len = len;
-	LM_DBG("using table '%.*s'\n", table.len, table.s);
+    table.len = strlen(table_ptr);
+    LM_DBG("using table '%.*s'\n", table.len, table.s);
 
 	if(dbt_use_table(_h, &table) != 0) {
 		LM_ERR("use table is invalid %.*s\n", table.len, table.s);
@@ -115,17 +115,48 @@ int dbt_raw_query_select(db1_con_t* _h, str* _s, db1_res_t** _r)
 	if(ncols == 1 && strncmp(*tokens, "*", 1) == 0) {
 		cols = _tbc->nrcols;
 		result_cols = pkg_malloc(sizeof(db_key_t) * cols);
+        if(result_cols == NULL) {
+            LM_ERR("no more memory allocating");
+            goto error;
+        }
 		memset(result_cols, 0, sizeof(db_key_t) * cols);
 		for(n=0; n < cols; n++) {
-			result_cols[n] = &_tbc->colv[n]->name;
+			result_cols[n] = pkg_malloc(sizeof(str));
+            if(result_cols[n] == NULL) {
+                LM_ERR("no more memory allocating");
+                goto error;
+            }
+			result_cols[n]->len = _tbc->colv[n]->name.len;
+			result_cols[n]->s = pkg_malloc((_tbc->colv[n]->name.len + 1) * sizeof(char));
+            if(result_cols[n]->s == NULL) {
+                LM_ERR("no more memory allocating");
+                goto error;
+            }
+			strncpy(result_cols[n]->s, _tbc->colv[n]->name.s, _tbc->colv[n]->name.len);
+			result_cols[n]->s[_tbc->colv[n]->name.len] = '\0';
 		}
 	} else {
 		cols = ncols;
 		result_cols = pkg_malloc(sizeof(db_key_t) * cols);
+        if(result_cols == NULL) {
+            LM_ERR("no more memory allocating");
+            goto error;
+        }
 		memset(result_cols, 0, sizeof(db_key_t) * cols);
 		for(n=0; *(tokens + n); n++) {
-			result_cols[n]->s = *(tokens + n);
+			result_cols[n] = pkg_malloc(sizeof(str));
+            if(result_cols[n] == NULL) {
+                LM_ERR("no more memory allocating");
+                goto error;
+            }
 			result_cols[n]->len = strlen(*(tokens + n));
+			result_cols[n]->s = pkg_malloc((strlen(*(tokens + n)) + 1) * sizeof(char));
+            if(result_cols[n]->s == NULL) {
+                LM_ERR("no more memory allocating");
+                goto error;
+            }
+			strncpy(result_cols[n]->s, *(tokens + n), strlen(*(tokens + n)));
+			result_cols[n]->s[strlen(*(tokens + n))] = '\0';
 		}
 	}
 
@@ -159,6 +190,12 @@ error:
 	dbt_clean_where(nc, _k, _op, _v);
 
 	if(result_cols) {
+		for(n=0; n < cols; n++) {
+            if(result_cols[n]->s)
+                pkg_free(result_cols[n]->s);
+            if(result_cols[n])
+                pkg_free(result_cols[n]);
+		}
 		pkg_free(result_cols);
 	}
 
@@ -200,9 +237,14 @@ int dbt_raw_query_update(db1_con_t* _h, str* _s, db1_res_t** _r)
 	table_ptr[len] = '\0';
 	dbt_trim(table_ptr);
 	table.s = table_ptr;
-	table.len = len;
+	table.len = strlen(table_ptr);
 
 	where_ptr = strcasestr(_s->s, " where ");
+	if(where_ptr == NULL) {
+		LM_ERR("specify where clause to determine keys\n");
+		goto error;
+	}
+	
 	fields_end_ptr = where_ptr;
 	len = fields_end_ptr - ( fields_start_ptr + 4) + 1;
 	fields_ptr = pkg_malloc(len);
@@ -216,11 +258,6 @@ int dbt_raw_query_update(db1_con_t* _h, str* _s, db1_res_t** _r)
 		goto error;
 	}
 
-
-	if(where_ptr == NULL) {
-		LM_ERR("specify where clause to determine keys\n");
-		goto error;
-	}
 	nkeys = dbt_build_where(where_ptr + 7, &_k, &_op1, &_kv);
 	if(nkeys < 1) {
 		LM_ERR("needsa at least one key\n");
@@ -383,9 +420,14 @@ int dbt_raw_query_replace(db1_con_t* _h, str* _s, db1_res_t** _r)
 	table_ptr[len] = '\0';
 	dbt_trim(table_ptr);
 	table.s = table_ptr;
-	table.len = len;
+	table.len = strlen(table_ptr);
 
 	where_ptr = strcasestr(_s->s, " where ");
+	if(where_ptr == NULL) {
+		LM_ERR("specify where clause to determine keys\n");
+		goto error;
+	}
+	
 	fields_end_ptr = where_ptr;
 	len = fields_end_ptr - ( fields_start_ptr + 4) + 1;
 	fields_ptr = pkg_malloc(len);
@@ -399,11 +441,6 @@ int dbt_raw_query_replace(db1_con_t* _h, str* _s, db1_res_t** _r)
 		goto error;
 	}
 
-
-	if(where_ptr == NULL) {
-		LM_ERR("specify where clause to determine keys\n");
-		goto error;
-	}
 	nkeys = dbt_build_where(where_ptr + 7, &_k, &_op1, &_kv);
 	if(nkeys < 1) {
 		LM_ERR("needsa at least one key\n");
@@ -492,6 +529,7 @@ int dbt_raw_query(db1_con_t* _h, str* _s, db1_res_t** _r)
 		return res;
 	}
 
+	((dbt_con_p)_h->tail)->affected = 0;
 	dbt_trim(_s->s);
 	_s->len = strlen(_s->s);
 

src/modules/db_text/doc/db_text_admin.xml

@@ -270,6 +270,26 @@ modparam("db_text", "db_mode", 1)
 </programlisting>
 			</example>
 		</section>
+        <section>
+            <title><varname>default_connection</varname> (string)</title>
+            <para>
+                connection for use with rpc query command.
+            </para>
+            
+            <para>
+                <emphasis>
+                    Default value is <quote>none</quote> (off).
+                </emphasis>
+            </para>
+            <example>
+                <title>Set <varname>default_connection</varname> parameter</title>
+                <programlisting format="linespecific">
+                    ...
+                    modparam("db_text", "default_connection", "text:///var/db/kamailio/dbtext")
+                    ...
+                </programlisting>
+            </example>
+        </section>
 		<section>
 			<title><varname>emptystring</varname> (integer)</title>
 			<para>
@@ -329,6 +349,18 @@ modparam("db_text", "file_buffer_size", 8192)
 	kamcmd db_text.dump
 		</programlisting>
 	</section>
+    <section>
+        <title>
+            <function moreinfo="none">db_text.query</function>
+        </title>
+        <para>run sql command</para>
+        <para>Name: <emphasis>db_text.query</emphasis></para>
+        <para>Parameters: <emphasis>sqlcmd</emphasis></para>
+        <para>RPC Command Format:</para>
+        <programlisting  format="linespecific">
+            kamcmd db_text.query 'select * from location where username="xxx"'
+        </programlisting>
+    </section>
 	</section>
 	<section>
 		<title>Installation and Running</title>