Merge pull request #2359 from kamailio/jchavanton/mqueue_db_persistent

mqueue: add support for db persistency

src/modules/mqueue/Makefile

@@ -6,4 +6,6 @@ auto_gen=
 NAME=mqueue.so
 LIBS=
 
+SERLIBPATH=../../lib
+SER_LIBS+=$(SERLIBPATH)/srdb1/srdb1
 include ../../Makefile.modules

src/modules/mqueue/doc/mqueue.xml

@@ -46,7 +46,7 @@
 	    <holder>Elena-Ramona Modroiu (asipto.com)</holder>
 	</copyright>
 	<copyright>
-	    <year>2018</year>
+	    <year>2018-2020</year>
 	    <holder>Julien chavanton, Flowroute</holder>
 	</copyright>
     </bookinfo>

src/modules/mqueue/doc/mqueue_admin.xml

@@ -61,6 +61,32 @@
     <section>
 	<title>Parameters</title>
 
+	<section id="mqueue.p.db_url">
+		<title><varname>db_url</varname> (str)</title>
+		<para>
+			The <acronym>URL</acronym> to connect to database for loading values in mqueue table at start up and/or saving values at shutdown.
+		</para>
+		<para>
+		<emphasis>
+			Default value is NULL (do not connect).
+		</emphasis>
+		</para>
+		<example>
+		<title>Set <varname>db_url</varname> parameter</title>
+		<programlisting format="linespecific">
+...
+modparam("mqueue", "db_url", "&defaultdb;")
+
+# Example of table in sqlite, you have the set the fields to support the length according to the data that will be present in the mqueue
+CREATE TABLE mqueue_name (
+id INTEGER PRIMARY KEY AUTOINCREMENT,
+key character varying(64) DEFAULT "" NOT NULL,
+val character varying(4096) DEFAULT "" NOT NULL
+);
+...
+</programlisting>
+		</example>
+	</section>
 	<section id="mqueue.p.mqueue">
 	    <title><varname>mqueue</varname> (string)</title>
 	    <para>
@@ -95,6 +121,16 @@
 				If not set the queue will be limitless.
 				</para>
 			</listitem>
+			<listitem>
+				<para>
+				<emphasis>dbmode</emphasis>: If set to 1, the content of the queue
+				is written to database table when the SIP server is stopped
+				(i.e., ensure persistency over restarts).
+				If set to 2, it is written at shutdown but not read at startup.
+				If set to 3, it is read at sartup but not written at shutdown.
+				Default value is 0 (no db table interaction).
+				</para>
+			</listitem>
 			</itemizedlist>
 		</listitem>
 		</itemizedlist>

src/modules/mqueue/mqueue_api.c

@@ -34,7 +34,7 @@
 #include "../../core/fmsg.h"
 
 #include "mqueue_api.h"
-
+#include "mqueue_db.h"
 
 /**
  *
@@ -71,6 +71,11 @@ void mq_destroy(void)
 	mh = _mq_head_list;
 	while(mh!=NULL)
 	{
+		if(mh->dbmode == 1 || mh->dbmode == 3)
+		{
+			LM_INFO("mqueue[%.*s] dbmode[%d]\n", mh->name.len, mh->name.s, mh->dbmode);
+			mqueue_db_save_queue(&mh->name);
+		}
 		mi = mh->ifirst;
 		while(mi!=NULL)
 		{
@@ -180,6 +185,27 @@ mq_head_t *mq_head_get(str *name)
 	return NULL;
 }
 
+/**
+ *
+ */
+int mq_set_dbmode(str *name, int dbmode)
+{
+	mq_head_t *mh = NULL;
+
+	mh = _mq_head_list;
+	while(mh!=NULL)
+	{
+		if(name->len == mh->name.len
+				&& strncmp(mh->name.s, name->s, name->len)==0)
+		{
+			mh->dbmode = dbmode;
+			return 0;
+		}
+		mh = mh->next;
+	}
+	return -1;
+}
+
 /**
  *
  */

src/modules/mqueue/mqueue_api.h

@@ -44,6 +44,7 @@ typedef struct _mq_head
 	str name;
 	int msize;
 	int csize;
+	int dbmode;
 	gen_lock_t lock;
 	mq_item_t *ifirst;
 	mq_item_t *ilast;
@@ -61,7 +62,6 @@ typedef struct _mq_pv
 } mq_pv_t;
 
 mq_pv_t *mq_pv_get(str *name);
-
 int pv_parse_mq_name(pv_spec_p sp, str *in);
 int pv_get_mqk(struct sip_msg *msg, pv_param_t *param,
 		pv_value_t *res);
@@ -79,6 +79,7 @@ void mq_pv_free(str *name);
 int mq_item_add(str *qname, str *key, str *val);
 
 int _mq_get_csize(str *);
+int mq_set_dbmode(str *, int dbmode);
 
 #endif
 

src/modules/mqueue/mqueue_db.c

@@ -0,0 +1,324 @@
+/**
+ * Copyright (C) 2020 Julien Chavanton
+ *
+ * This file is part of Kamailio, a free SIP server.
+ *
+ * This file 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 2 of the License, or
+ * (at your option) any later version
+ *
+ *
+ * This file 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, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ *
+ */
+
+#include "../../lib/srdb1/db.h"
+#include "mqueue_api.h"
+
+/** database connection */
+db1_con_t *mqueue_db_con = NULL;
+db_func_t mq_dbf;
+
+/** db parameters */
+str mqueue_db_url = {0, 0};
+str mq_db_key_column = str_init("key");
+str mq_db_val_column = str_init("val");
+str mq_db_id_column  = str_init("id");
+
+/**
+ * initialize database connection
+ */
+int mqueue_db_init_con(void)
+{
+	if (mqueue_db_url.len<=0) {
+		LM_ERR("failed to connect to the database, no db url\n");
+		return -1;
+	}
+	/* binding to DB module */
+	if (db_bind_mod(&mqueue_db_url, &mq_dbf))
+	{
+		LM_ERR("database module not found\n");
+		return -1;
+	}
+
+	if (!DB_CAPABILITY(mq_dbf, DB_CAP_ALL))
+	{
+		LM_ERR("database module does not "
+				"implement all functions needed by the module\n");
+		return -1;
+	}
+	return 0;
+}
+
+/**
+ * open database connection
+ */
+int mqueue_db_open_con(void) {
+	if (mqueue_db_init_con()==0) {
+		mqueue_db_con = mq_dbf.init(&mqueue_db_url);
+		if (mqueue_db_con==NULL) {
+			LM_ERR("failed to connect to the database\n");
+			return -1;
+		}
+		LM_DBG("database connection opened successfully\n");
+		return 0;
+	}
+	return 0;
+}
+
+/**
+ * close database connection
+ */
+int mqueue_db_close_con(void)
+{
+	if (mqueue_db_con!=NULL && mq_dbf.close!=NULL)
+		mq_dbf.close(mqueue_db_con);
+	mqueue_db_con=NULL;
+	return 0;
+}
+
+int mqueue_db_load_queue(str *name)
+{
+	int ncols=2;
+	db1_res_t* db_res = NULL;
+	db_key_t db_cols[2] = {&mq_db_key_column, &mq_db_val_column};
+	db_key_t db_ord = &mq_db_id_column;
+	int mq_fetch_rows = 100;
+	int ret = 0;
+	str val = str_init("");
+	str key = str_init("");
+	int i;
+	int cnt=0;
+
+	if (mqueue_db_open_con() != 0) {
+		LM_ERR("no db connection\n");
+		return -1;
+	}
+
+	if (mq_dbf.use_table(mqueue_db_con, name) < 0)
+	{
+		LM_ERR("failed to use_table\n");
+		goto error;
+	}
+
+	LM_INFO("=============== loading queue table [%.*s] from database\n",
+			name->len, name->s);
+
+	if (DB_CAPABILITY(mq_dbf, DB_CAP_FETCH)) {
+		if(mq_dbf.query(mqueue_db_con,0,0,0,db_cols,0,ncols,db_ord,0) < 0)
+		{
+			LM_ERR("Error while querying db\n");
+			goto error;
+		}
+		if(mq_dbf.fetch_result(mqueue_db_con, &db_res, mq_fetch_rows)<0)
+		{
+			LM_ERR("Error while fetching result\n");
+			if (db_res)
+				mq_dbf.free_result(mqueue_db_con, db_res);
+			goto error;
+		} else {
+			if(RES_ROW_N(db_res)==0)
+			{
+				mq_dbf.free_result(mqueue_db_con, db_res);
+				LM_DBG("Nothing to be loaded in queue\n");
+				mqueue_db_close_con();
+				return 0;
+			}
+		}
+	} else {
+		if((ret=mq_dbf.query(mqueue_db_con, NULL, NULL, NULL, db_cols,
+				0, ncols, 0, &db_res))!=0
+			|| RES_ROW_N(db_res)<=0 )
+		{
+			if(ret==0)
+			{
+				mq_dbf.free_result(mqueue_db_con, db_res);
+				mqueue_db_close_con();
+				return 0;
+			} else {
+				goto error;
+			}
+		}
+	}
+
+	do {
+		for(i=0; i<RES_ROW_N(db_res); i++)
+		{
+			if(VAL_NULL(&RES_ROWS(db_res)[i].values[0])) {
+				LM_ERR("mqueue [%.*s] row [%d] has NULL key string\n",
+					name->len, name->s, i);
+				goto error;
+			}
+			if(VAL_NULL(&RES_ROWS(db_res)[i].values[1])) {
+				LM_ERR("mqueue [%.*s] row [%d] has NULL value string\n",
+					name->len, name->s, i);
+				goto error;
+			}
+			switch(RES_ROWS(db_res)[i].values[0].type) {
+			case DB1_STR:
+				key.s = (RES_ROWS(db_res)[i].values[0].val.str_val.s);
+				if(key.s==NULL) {
+					LM_ERR("mqueue [%.*s] row [%d] has NULL key\n",
+						name->len, name->s, i);
+					goto error;
+				}
+				key.len = (RES_ROWS(db_res)[i].values[0].val.str_val.len);
+				break;
+			case DB1_BLOB:
+				key.s = (RES_ROWS(db_res)[i].values[0].val.blob_val.s);
+				if(key.s==NULL) {
+					LM_ERR("mqueue [%.*s] row [%d] has NULL key\n",
+						name->len, name->s, i);
+					goto error;
+				}
+				key.len = (RES_ROWS(db_res)[i].values[0].val.blob_val.len);
+				break;
+			case DB1_STRING:
+				key.s = (char*)(RES_ROWS(db_res)[i].values[0].val.string_val);
+				if(key.s==NULL) {
+					LM_ERR("mqueue [%.*s] row [%d] has NULL key\n",
+						name->len, name->s, i);
+					goto error;
+				}
+				key.len = strlen(key.s);
+				break;
+			default:
+				LM_ERR("key type must be string (type=%d)\n",
+						RES_ROWS(db_res)[i].values[0].type);
+				goto error;
+			}
+			switch(RES_ROWS(db_res)[i].values[1].type) {
+			case DB1_STR:
+				val.s = (RES_ROWS(db_res)[i].values[1].val.str_val.s);
+				if(val.s==NULL) {
+					LM_ERR("mqueue [%.*s] row [%d] has NULL value\n",
+						name->len, name->s, i);
+					goto error;
+				}
+				val.len = (RES_ROWS(db_res)[i].values[1].val.str_val.len);
+				break;
+			case DB1_BLOB:
+				val.s = (RES_ROWS(db_res)[i].values[1].val.blob_val.s);
+				if(val.s==NULL) {
+					LM_ERR("mqueue [%.*s] row [%d] has NULL value\n",
+						name->len, name->s, i);
+					goto error;
+				}
+				val.len = (RES_ROWS(db_res)[i].values[1].val.blob_val.len);
+				break;
+			case DB1_STRING:
+				val.s = (char*)(RES_ROWS(db_res)[i].values[1].val.string_val);
+				if(val.s==NULL) {
+					LM_ERR("mqueue [%.*s] row [%d] has NULL value\n",
+						name->len, name->s, i);
+					goto error;
+				}
+				val.len = strlen(val.s);
+				break;
+			default:
+				LM_ERR("key type must be string (type=%d)\n",
+						RES_ROWS(db_res)[i].values[1].type);
+				goto error;
+			}
+			cnt++;
+			LM_DBG("adding item[%d] key[%.*s] value[%.*s]\n", cnt, key.len, key.s, val.len, val.s);
+			mq_item_add(name, &key, &val);
+		}
+
+		if (DB_CAPABILITY(mq_dbf, DB_CAP_FETCH)) {
+			if(mq_dbf.fetch_result(mqueue_db_con, &db_res, mq_fetch_rows)<0) {
+				LM_ERR("Error while fetching!\n");
+				goto error;
+			}
+		} else {
+			break;
+		}
+	}  while(RES_ROW_N(db_res)>0);
+
+	mq_dbf.free_result(mqueue_db_con, db_res);
+
+	if (mq_dbf.delete(mqueue_db_con, 0, 0, 0, 0) < 0) {
+		LM_ERR("failed to clear table\n");
+		goto error;
+	}
+
+	LM_DBG("loaded %d values in queue\n", cnt);
+	mqueue_db_close_con();
+	return 0;
+error:
+	mqueue_db_close_con();
+	return -1;
+}
+
+int mqueue_db_save_queue(str *name)
+{
+	int ncols=2;
+	db_key_t db_cols[2] = {&mq_db_key_column, &mq_db_val_column};
+	db_val_t db_vals[2];
+	int i;
+	int mqueue_sz = 0;
+	int ret = 0;
+
+	if (mqueue_db_open_con() != 0) {
+		LM_ERR("no db connection\n");
+		return -1;
+	}
+
+	if (mq_dbf.use_table(mqueue_db_con, name) < 0)
+	{
+		LM_ERR("failed to use_table\n");
+		goto error;
+	}
+
+	if (name->len <= 0 || name->s == NULL) {
+		LM_ERR("bad mqueue name\n");
+		goto error;
+	}
+
+	mqueue_sz = _mq_get_csize(name);
+
+	if (mqueue_sz < 0) {
+		LM_ERR("no such mqueue\n");
+		goto error;
+	}
+	for(i=0;i<mqueue_sz;i++) {
+		ret = mq_head_fetch(name);
+		if (ret != 0) break;
+		str *key = NULL;
+		str *val = NULL;
+		key = get_mqk(name);
+		val = get_mqv(name);
+		LM_DBG("inserting mqueue[%.*s] name[%.*s] value[%.*s]\n",
+				name->len, name->s, key->len, key->s, val->len, val->s);
+		db_vals[0].type = DB1_STR;
+		db_vals[0].nul  = 0;
+		db_vals[0].val.str_val.s   = key->s;
+		db_vals[0].val.str_val.len = key->len;
+		db_vals[1].type = DB1_STR;
+		db_vals[1].nul  = 0;
+		db_vals[1].val.str_val.s   = val->s;
+		db_vals[1].val.str_val.len = val->len;
+		if(mq_dbf.insert(mqueue_db_con, db_cols, db_vals, ncols) < 0)
+		{
+			LM_ERR("failed to store key [%.*s] val [%.*s]\n",
+					key->len, key->s,
+					val->len, val->s);
+		}
+	}
+
+	LM_INFO("queue [%.*s] saved in db\n",
+			name->len, name->s);
+	mqueue_db_close_con();
+	return 0;
+error:
+	mqueue_db_close_con();
+	return -1;
+}

src/modules/mqueue/mqueue_db.h

@@ -0,0 +1,33 @@
+/**
+ * Copyright (C) 2020 Julien Chavanton
+ *
+ * This file is part of Kamailio, a free SIP server.
+ *
+ * This file 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 2 of the License, or
+ * (at your option) any later version
+ *
+ *
+ * This file 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, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ *
+ */
+
+#ifndef _MQUEUE_DB_H_
+#define _MQUEUE_DB_H_
+
+#include "../../lib/srdb1/db.h"
+#include "mqueue_api.h"
+
+extern str mqueue_db_url;
+
+int mqueue_db_load_queue(str *name);
+int mqueue_db_save_queue(str *name);
+#endif

src/modules/mqueue/mqueue_mod.c

@@ -37,6 +37,7 @@
 #include "../../core/kemi.h"
 
 #include "mqueue_api.h"
+#include "mqueue_db.h"
 #include "api.h"
 
 MODULE_VERSION
@@ -54,6 +55,7 @@ static int bind_mq(mq_api_t* api);
 
 static int mqueue_rpc_init(void);
 
+
 static pv_export_t mod_pvs[] = {
 	{ {"mqk", sizeof("mqk")-1}, PVT_OTHER, pv_get_mqk, 0,
 		pv_parse_mq_name, 0, 0, 0 },
@@ -80,6 +82,7 @@ static cmd_export_t cmds[]={
 };
 
 static param_export_t params[]={
+	{"db_url",          PARAM_STR, &mqueue_db_url},
 	{"mqueue",          PARAM_STRING|USE_FUNC_PARAM, (void*)mq_param},
 	{0, 0, 0}
 };
@@ -205,6 +208,7 @@ int mq_param(modparam_t type, void *val)
 	param_t *pit=NULL;
 	str qname = {0, 0};
 	int msize = 0;
+	int dbmode = 0;
 
 	if(val==NULL)
 		return -1;
@@ -229,6 +233,9 @@ int mq_param(modparam_t type, void *val)
 		} else if(pit->name.len==4
 				&& strncasecmp(pit->name.s, "size", 4)==0) {
 			str2sint(&pit->body, &msize);
+		} else if(pit->name.len==6
+				&& strncasecmp(pit->name.s, "dbmode", 6)==0) {
+			str2sint(&pit->body, &dbmode);
 		}  else {
 			LM_ERR("unknown param: %.*s\n", pit->name.len, pit->name.s);
 			free_params(params_list);
@@ -247,6 +254,16 @@ int mq_param(modparam_t type, void *val)
 		free_params(params_list);
 		return -1;
 	}
+	LM_INFO("mqueue param: [%.*s|%d]\n", qname.len, qname.s, dbmode);
+	if(dbmode == 1 || dbmode == 2) {
+		if(mqueue_db_load_queue(&qname)<0)
+		{
+			LM_ERR("error loading mqueue: %.*s from DB\n", qname.len, qname.s);
+			free_params(params_list);
+			return -1;
+		}
+	}
+	mq_set_dbmode(&qname, dbmode);
 	free_params(params_list);
 	return 0;
 }