Re: Multi record sending patch

Index: sqlbox-cfg.def
===================================================================
--- sqlbox-cfg.def	(revision 52)
+++ sqlbox-cfg.def	(working copy)
@@ -21,4 +21,6 @@
     OCTSTR(ssl-server-cert-file)
     OCTSTR(ssl-server-key-file)
     OCTSTR(ssl-trusted-ca-file)
+    OCTSTR(limit-per-cycle)
+    OCTSTR(sql-log-table-split)
 )
Index: sqlbox.c
===================================================================
--- sqlbox.c	(revision 52)
+++ sqlbox.c	(working copy)
@@ -85,6 +85,7 @@
 static Octstr *bearerbox_host;
 static int bearerbox_port_ssl = 0;
 static Octstr *global_sender;
+static long limit_per_cycle;
 
 #ifndef HAVE_MSSQL
 #ifndef HAVE_MYSQL
@@ -103,8 +104,15 @@
 #endif
 Octstr *sqlbox_id;
 
-#define SLEEP_BETWEEN_SELECTS 1.0
+/* changed name from SLEEP_BETWEEN_SELECTS to SLEEP_BETWEEN_EMPTY_SELECTS
+ * to reflect what this does.
+ * If needed we can make this configurable from config.
+ */
+#define SLEEP_BETWEEN_EMPTY_SELECTS 1.0
 
+/* this defines how many msg will be fetched from database on each select cycle */
+#define DEFAULT_LIMIT_PER_CYCLE 10
+
 typedef struct _boxc {
     Connection    *smsbox_connection;
     Connection    *bearerbox_connection;
@@ -420,7 +428,6 @@
                 break;
             }
         }
-
         if (msg_type(msg) == heartbeat) {
         // todo
             debug("sqlbox", 0, "bearerbox_to_smsbox: catch an heartbeat - we are alive");
@@ -522,6 +529,7 @@
 {
     Boxc *conn = (Boxc *)arg;
     Msg *msg, *mack;
+    Octstr *sql;
 
     while (sqlbox_status == SQL_RUNNING && conn->alive) {
         msg = read_from_box(conn->bearerbox_connection, conn);
@@ -560,8 +568,11 @@
 	    uuid_copy(mack->ack.id, msg->sms.id);
 	    send_msg(conn->bearerbox_connection, conn, mack);
 	    msg_destroy(mack);
-
         }
+		/* we need the ack here to handle actions on sql-insert-table */
+        else if (msg_type(msg) == ack) {
+            gw_sql_save_msg(msg, octstr_imm("MT"));
+        }
 
         msg_destroy(msg);
     }
@@ -571,6 +582,7 @@
 {
     Boxc *boxc;
     Msg *msg;
+    List *qlist;
 
     boxc = gw_malloc(sizeof(Boxc));
     boxc->bearerbox_connection = connect_to_bearerbox_real(bearerbox_host, bearerbox_port, bearerbox_port_ssl, NULL /* bb_our_host */);
@@ -588,8 +600,12 @@
 
     identify_to_bearerbox(boxc);
 
+    qlist = gwlist_create();
+    gwlist_add_producer(qlist);
+
     while (sqlbox_status == SQL_RUNNING && boxc->alive) {
-        if ((msg = gw_sql_fetch_msg()) != NULL) {
+        if ( gw_sql_fetch_msg(qlist, limit_per_cycle) > 0 ) {
+            while((gwlist_len(qlist)>0) && ((msg = gwlist_consume(qlist)) != NULL )) {
             if (charset_processing(msg) == -1) {
                 error(0, "Could not charset process message, dropping it!");
                 msg_destroy(msg);
@@ -598,28 +614,50 @@
             if (global_sender != NULL && (msg->sms.sender == NULL || octstr_len(msg->sms.sender) == 0)) {
                 msg->sms.sender = octstr_duplicate(global_sender);
             }
-            /* convert validity and deferred to unix timestamp */
+                /* convert validity and deferred to unix timestamp 
+				 * REMEMBER - we have those data in MINUTES in databases
+				 */
             if (msg->sms.validity != SMS_PARAM_UNDEFINED)
                 msg->sms.validity = time(NULL) + msg->sms.validity * 60;
             if (msg->sms.deferred != SMS_PARAM_UNDEFINED)
                 msg->sms.deferred = time(NULL) + msg->sms.deferred * 60;
-            send_msg(boxc->bearerbox_connection, boxc, msg);
 
+				/* send_msg - returns: 0 on success, -1 on failure */
+                if (send_msg(boxc->bearerbox_connection, boxc, msg) == -1) {
+					/* temp error message for now
+					 * TODO - FIX: handle this case also on database x_status
+					 */
+					error(0, "FAILED sql_to_bearerbox:send_msg, dropping it!");
+				}
+
             /* convert validity & deferred back to minutes
              * TODO clarify why we fetched message from DB and then insert it back here???
+                 *
+				 * WE do not need this anymore
+				 * DO NOT DELETE the msg from sql-insert-table
+				 * DO NOT INSERT the msg in sql-log-table
+				 * WE WAIT for ACK from bearerbox (bearerbox_to_sql) to perform those actions
              */
+				/* 
             if (msg->sms.validity != SMS_PARAM_UNDEFINED)
                 msg->sms.validity = (msg->sms.validity - time(NULL))/60;
             if (msg->sms.deferred != SMS_PARAM_UNDEFINED)
                 msg->sms.deferred = (msg->sms.deferred - time(NULL))/60;
             gw_sql_save_msg(msg, octstr_imm("MT"));
+				*/
+				
+				/* I think we need this here - CHECK and CONFIRM ??? */
+				msg_destroy(msg);
         }
-        else {
-            gwthread_sleep(SLEEP_BETWEEN_SELECTS);
+        } else {
+            gwthread_sleep(SLEEP_BETWEEN_EMPTY_SELECTS);
         }
-        msg_destroy(msg);
+		/* Remove from here - CHECK and CONFIRM - mem leak??? */
+        //msg_destroy(msg);
     }
 
+    gwlist_remove_producer(qlist);
+    gwlist_destroy(qlist,NULL); /* is this really needed ??? */
     boxc_destroy(boxc);
 }
 
@@ -745,6 +783,11 @@
 
     if (cfg_get_integer(&sqlbox_port, grp, octstr_imm("smsbox-port")) == -1)
         sqlbox_port = 13005;
+
+    /* setup limit per cycle */
+    if (cfg_get_integer(&limit_per_cycle, grp, octstr_imm("limit-per-cycle")) == -1)
+        limit_per_cycle = DEFAULT_LIMIT_PER_CYCLE;
+
     /* setup logfile stuff */
     logfile = cfg_get(grp, octstr_imm("log-file"));
 
Index: sqlbox_mysql.c
===================================================================
--- sqlbox_mysql.c	(revision 52)
+++ sqlbox_mysql.c	(working copy)
@@ -10,6 +10,7 @@
 
 static Octstr *sqlbox_logtable;
 static Octstr *sqlbox_insert_table;
+static int sqlbox_logtable_split = 0;
 
 /*
  * Our connection pool to mysql.
@@ -17,9 +18,10 @@
 
 static DBPool *pool = NULL;
 
-static void mysql_update(const Octstr *sql)
+/* Returns 0 if mysql_query FAILS - 1 otherwise */
+static int mysql_update(const Octstr *sql)
 {
-    int state;
+    int state = 0;
     DBPoolConn *pc;
 
 #if defined(SQLBOX_TRACE)
@@ -29,14 +31,21 @@
     pc = dbpool_conn_consume(pool);
     if (pc == NULL) {
         error(0, "MYSQL: Database pool got no connection! DB update failed!");
-        return;
+        return state;
     }
 
     state = mysql_query(pc->conn, octstr_get_cstr(sql));
-    if (state != 0)
+    if (state != 0) {
         error(0, "MYSQL: %s", mysql_error(pc->conn));
+        state = 0;
+    }
+    else {
+        state = 1;
+    }
 
     dbpool_conn_produce(pc);
+
+    return state;
 }
 
 static MYSQL_RES* mysql_select(const Octstr *sql)
@@ -83,11 +92,26 @@
     if (sqlbox_insert_table == NULL) {
         panic(0, "No 'sql-insert-table' not configured.");
     }
-
-    /* create send_sms && sent_sms tables if they do not exist */
+    cfg_get_bool(&sqlbox_logtable_split, grp, octstr_imm("sql-log-table-split"));
+    if (sqlbox_logtable_split < 0)
+        sqlbox_logtable_split = 0;    
+    /* create send_sms && sent_sms[MO MT DLR] tables if they do not exist */
+    if (sqlbox_logtable_split) {
+        sql = octstr_format(SQLBOX_MYSQL_CREATE_LOG_TABLE, octstr_cat(sqlbox_logtable, octstr_imm("MO")));
+        sql_update(sql);
+        octstr_destroy(sql);
+        sql = octstr_format(SQLBOX_MYSQL_CREATE_LOG_TABLE, octstr_cat(sqlbox_logtable, octstr_imm("MT")));
+        sql_update(sql);
+        octstr_destroy(sql);
+        sql = octstr_format(SQLBOX_MYSQL_CREATE_LOG_TABLE, octstr_cat(sqlbox_logtable, octstr_imm("DLR")));
+        sql_update(sql);
+        octstr_destroy(sql);
+    }
+    else {
     sql = octstr_format(SQLBOX_MYSQL_CREATE_LOG_TABLE, sqlbox_logtable);
     sql_update(sql);
     octstr_destroy(sql);
+    }
     sql = octstr_format(SQLBOX_MYSQL_CREATE_INSERT_TABLE, sqlbox_insert_table);
     sql_update(sql);
     octstr_destroy(sql);
@@ -96,14 +120,19 @@
 
 #define octstr_null_create(x) ((x != NULL) ? octstr_create(x) : octstr_create(""))
 #define atol_null(x) ((x != NULL) ? atol(x) : -1)
-Msg *mysql_fetch_msg()
+int mysql_fetch_msg(List *qlist, long limit)
 {
     Msg *msg = NULL;
-    Octstr *sql, *delet, *id;
     MYSQL_RES *res;
     MYSQL_ROW row;
+    Octstr *sql;
+    Octstr *x_update;
+    Octstr *stored_local_uuid;
+    Octstr *id;
+    char local_uuid[UUID_STR_LEN + 1];
+    int ret = 0;
 
-    sql = octstr_format(SQLBOX_MYSQL_SELECT_QUERY, sqlbox_insert_table);
+    sql = octstr_format(SQLBOX_MYSQL_SELECT_QUERY, sqlbox_insert_table, limit);
 #if defined(SQLBOX_TRACE)
      debug("SQLBOX", 0, "sql: %s", octstr_get_cstr(sql));
 #endif
@@ -112,8 +141,10 @@
         debug("sqlbox", 0, "SQL statement failed: %s", octstr_get_cstr(sql));
     }
     else {
-        if (mysql_num_rows(res) >= 1) {
-            row = mysql_fetch_row(res);
+        ret = mysql_num_rows(res);
+        if (ret >= 1) {
+			//debug("sqlbox", 0, "Rows found: %i", ret);
+            while (row = mysql_fetch_row(res)) {
             id = octstr_null_create(row[0]);
             /* save fields in this row as msg struct */
             msg = msg_create(sms);
@@ -147,19 +178,32 @@
             else {
                 msg->sms.boxc_id= octstr_null_create(row[24]);
             }
-            /* delete current row */
-            delet = octstr_format(SQLBOX_MYSQL_DELETE_QUERY, sqlbox_insert_table, id);
+                /* we need sms.id for later ack - match from bearerbox (bearerbox_to_sql) */
+                if (!uuid_is_null(msg->sms.id)) {
+                    uuid_unparse(msg->sms.id, local_uuid);
+                    stored_local_uuid = octstr_create(local_uuid);
+                    x_update = octstr_format(X_SQLBOX_MYSQL_UPDATE_ID_QUERY, sqlbox_insert_table, stored_local_uuid, id);
 #if defined(SQLBOX_TRACE)
-            debug("SQLBOX", 0, "sql: %s", octstr_get_cstr(delet));
+                    debug("SQLBOX", 0, "sql: %s", octstr_get_cstr(x_update));
 #endif
-            mysql_update(delet);
+                    /* insert msg into qlist only if update succeed */
+                    if (sql_update(x_update)) {
+                        gwlist_produce(qlist, msg);
+                    }
+                    /* we could use "else { ret--; }" here */
+					//octstr_destroy(stored_local_uuid);
+					//octstr_destroy(x_update);
+                }
+                /*  we could use "else { ret--; }" also here */
             octstr_destroy(id);
-            octstr_destroy(delet);
+                octstr_destroy(stored_local_uuid);
+                octstr_destroy(x_update);
         }
+        }
         mysql_free_result(res);
     }
     octstr_destroy(sql);
-    return msg;
+    return ret;
 }
 
 static Octstr *get_numeric_value_or_return_null(long int num)
@@ -186,15 +230,31 @@
 #define st_num(x) (stuffer[stuffcount++] = get_numeric_value_or_return_null(x))
 #define st_str(x) (stuffer[stuffcount++] = get_string_value_or_return_null(x))
 
-void mysql_save_msg(Msg *msg, Octstr *momt /*, Octstr smsbox_id */)
+void mysql_save_msg(Msg *msg, Octstr *momt)
 {
-    Octstr *sql;
+    Octstr *sql, *sqld;
     Octstr *stuffer[30];
     int stuffcount = 0;
 
-    sql = octstr_format(SQLBOX_MYSQL_INSERT_QUERY, sqlbox_logtable, st_str(momt), st_str(msg->sms.sender),
+    Octstr *sqlbox_logtable_tmp = NULL;
+    char local_uuid[UUID_STR_LEN + 1];
+    Octstr *stored_local_uuid = NULL;
+
+    if (sqlbox_logtable_split) {
+        sqlbox_logtable_tmp = octstr_cat(sqlbox_logtable, momt);
+    }
+    else {
+        sqlbox_logtable_tmp = octstr_duplicate(sqlbox_logtable);
+    }
+/* SMS Type */
+    if (msg_type(msg) == sms) {
+        if (!uuid_is_null(msg->sms.id)) {
+            uuid_unparse(msg->sms.id, local_uuid);
+            stored_local_uuid = octstr_create(local_uuid);
+        }
+        sql = octstr_format(SQLBOX_MYSQL_INSERT_QUERY, sqlbox_logtable_tmp, st_str(momt), st_str(msg->sms.sender),
         st_str(msg->sms.receiver), st_str(msg->sms.udhdata), st_str(msg->sms.msgdata), st_num(msg->sms.time),
-        st_str(msg->sms.smsc_id), st_str(msg->sms.service), st_str(msg->sms.account), st_num(msg->sms.sms_type),
+            st_str(msg->sms.smsc_id), st_str(msg->sms.service), st_str(msg->sms.account), st_str(stored_local_uuid), st_num(msg->sms.sms_type),
         st_num(msg->sms.mclass), st_num(msg->sms.mwi), st_num(msg->sms.coding), st_num(msg->sms.compress),
         st_num(msg->sms.validity), st_num(msg->sms.deferred), st_num(msg->sms.dlr_mask), st_str(msg->sms.dlr_url),
         st_num(msg->sms.pid), st_num(msg->sms.alt_dcs), st_num(msg->sms.rpi), st_str(msg->sms.charset),
@@ -203,7 +263,49 @@
     while (stuffcount > 0) {
         octstr_destroy(stuffer[--stuffcount]);
     }
+    }
+/* ACK Type */
+    else if ( msg_type(msg) == ack /* && momt == octstr_imm("MT") */  && !uuid_is_null(msg->ack.id) ) {
+		uuid_unparse(msg->ack.id, local_uuid);
+		stored_local_uuid = octstr_create(local_uuid);
+		//debug("sqlbox", 0, "mysql_save_msg:ACK received. %i - %s", msg->ack.nack,octstr_get_cstr(stored_local_uuid));
+        switch (msg->ack.nack) {
+          /* 0: Accepted for delivery - HTTP_ACCEPTED - delete msg from sqlbox_insert_table and insert into sqlbox_logtable with x_status=10 */
+          case ack_success:
+            sql = octstr_format(X_SQLBOX_MYSQL_MOVE_QUERY, sqlbox_logtable_tmp, octstr_imm("10"), sqlbox_insert_table, stored_local_uuid);
+            sqld= octstr_format(X_SQLBOX_MYSQL_DELETE_QUERY, sqlbox_insert_table, stored_local_uuid);
+            sql_update(sql);
+            sql_update(sqld);
+            break;
+          /* 3: Queued for later delivery - HTTP_ACCEPTED - delete msg from sqlbox_insert_table and insert into sqlbox_logtable with x_status=13 */
+          case ack_buffered:
+            sql = octstr_format(X_SQLBOX_MYSQL_MOVE_QUERY, sqlbox_logtable_tmp, octstr_imm("13"), sqlbox_insert_table, stored_local_uuid);
+            sqld= octstr_format(X_SQLBOX_MYSQL_DELETE_QUERY, sqlbox_insert_table, stored_local_uuid);
+            sql_update(sql);
+            sql_update(sqld);
+            break;
+          /* 1: Not routable. Do not try again. - HTTP_FORBIDDEN - leave msg on sqlbox_insert_table and update x_status=11 */
+          case ack_failed:
+            sql = octstr_format(X_SQLBOX_MYSQL_UPDATE_STATUS_QUERY, sqlbox_insert_table, octstr_imm("11"), stored_local_uuid);
+            sql_update(sql);
+            break;
+          /* 2: Temporal failure, try again later. - HTTP_SERVICE_UNAVAILABLE - leave msg on sqlbox_insert_table and update x_status=12 (OR??? set x_status=1 to resend) */
+          case ack_failed_tmp:
+            sql = octstr_format(X_SQLBOX_MYSQL_UPDATE_STATUS_QUERY, sqlbox_insert_table, octstr_imm("12"), stored_local_uuid);
+            sql_update(sql);
+            break;
+          /* Strange reply from bearerbox! - HTTP_SERVICE_UNAVAILABLE - leave msg on sqlbox_insert_table and update x_status=14 */
+          default:
+            error(0, "Strange reply from bearerbox!");
+            sql = octstr_format(X_SQLBOX_MYSQL_UPDATE_STATUS_QUERY, sqlbox_insert_table, octstr_imm("14"), stored_local_uuid);
+            sql_update(sql);
+            break;
+        }    
+    }
     octstr_destroy(sql);
+	octstr_destroy(sqld);
+    octstr_destroy(sqlbox_logtable_tmp);
+    octstr_destroy(stored_local_uuid);
 }
 
 void mysql_leave()
Index: sqlbox_mysql.h
===================================================================
--- sqlbox_mysql.h	(revision 52)
+++ sqlbox_mysql.h	(working copy)
@@ -7,46 +7,58 @@
 momt ENUM('MO', 'MT', 'DLR') NULL, sender VARCHAR(20) NULL, \
 receiver VARCHAR(20) NULL, udhdata BLOB NULL, msgdata TEXT NULL, \
 time BIGINT(20) NULL, smsc_id VARCHAR(255) NULL, service VARCHAR(255) NULL, \
-account VARCHAR(255) NULL, id BIGINT(20) NULL, sms_type BIGINT(20) NULL, \
+account VARCHAR(255) NULL, id VARCHAR(40) NULL, sms_type BIGINT(20) NULL, \
 mclass BIGINT(20) NULL, mwi BIGINT(20) NULL, coding BIGINT(20) NULL, \
 compress BIGINT(20) NULL, validity BIGINT(20) NULL, deferred BIGINT(20) NULL, \
 dlr_mask BIGINT(20) NULL, dlr_url VARCHAR(255) NULL, pid BIGINT(20) NULL, \
 alt_dcs BIGINT(20) NULL, rpi BIGINT(20) NULL, charset VARCHAR(255) NULL, \
-boxc_id VARCHAR(255) NULL, binfo VARCHAR(255) NULL, meta_data TEXT)"
+boxc_id VARCHAR(255) NULL, binfo VARCHAR(255) NULL, meta_data TEXT, x_status tinyint(4) NULL)"
 
 #define SQLBOX_MYSQL_CREATE_INSERT_TABLE "CREATE TABLE IF NOT EXISTS %S ( \
 sql_id BIGINT(20) NOT NULL AUTO_INCREMENT PRIMARY KEY, \
 momt ENUM('MO', 'MT') NULL, sender VARCHAR(20) NULL, \
 receiver VARCHAR(20) NULL, udhdata BLOB NULL, msgdata TEXT NULL, \
 time BIGINT(20) NULL, smsc_id VARCHAR(255) NULL, service VARCHAR(255) NULL, \
-account VARCHAR(255) NULL, id BIGINT(20) NULL, sms_type BIGINT(20) NULL, \
+account VARCHAR(255) NULL, id VARCHAR(40) NOT NULL, sms_type BIGINT(20) NULL, \
 mclass BIGINT(20) NULL, mwi BIGINT(20) NULL, coding BIGINT(20) NULL, \
 compress BIGINT(20) NULL, validity BIGINT(20) NULL, deferred BIGINT(20) NULL, \
 dlr_mask BIGINT(20) NULL, dlr_url VARCHAR(255) NULL, pid BIGINT(20) NULL, \
 alt_dcs BIGINT(20) NULL, rpi BIGINT(20) NULL, charset VARCHAR(255) NULL, \
-boxc_id VARCHAR(255) NULL, binfo VARCHAR(255) NULL, meta_data TEXT)"
+boxc_id VARCHAR(255) NULL, binfo VARCHAR(255) NULL, meta_data TEXT, x_status tinyint(4) NOT NULL DEFAULT '1')"
 
 #define SQLBOX_MYSQL_SELECT_QUERY "SELECT sql_id, momt, sender, receiver, udhdata, \
 msgdata, time, smsc_id, service, account, id, sms_type, mclass, mwi, coding, \
 compress, validity, deferred, dlr_mask, dlr_url, pid, alt_dcs, rpi, \
-charset, boxc_id, binfo, meta_data FROM %S LIMIT 0,1"
+charset, boxc_id, binfo, meta_data FROM %S WHERE x_status = 1 LIMIT 0,%i"
 
 #define SQLBOX_MYSQL_INSERT_QUERY "INSERT INTO %S ( sql_id, momt, sender, \
-receiver, udhdata, msgdata, time, smsc_id, service, account, sms_type, \
+receiver, udhdata, msgdata, time, smsc_id, service, account, id, sms_type, \
 mclass, mwi, coding, compress, validity, deferred, dlr_mask, dlr_url, \
 pid, alt_dcs, rpi, charset, boxc_id, binfo, meta_data ) VALUES ( \
-NULL, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, \
+NULL, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, %S, \
 %S, %S, %S, %S, %S, %S, %S, %S, %S)"
 
 #define SQLBOX_MYSQL_DELETE_QUERY "DELETE FROM %S WHERE sql_id = %S"
 
+#define X_SQLBOX_MYSQL_UPDATE_ID_QUERY "UPDATE %S SET id = '%S', x_status = 2 WHERE sql_id = %S"
+
+#define X_SQLBOX_MYSQL_UPDATE_STATUS_QUERY "UPDATE %S SET x_status = %S WHERE id = '%S'"
+
+#define X_SQLBOX_MYSQL_DELETE_QUERY "DELETE FROM %S WHERE id = '%S'"
+
+#define X_SQLBOX_MYSQL_MOVE_QUERY "INSERT INTO %S (SELECT sql_id, momt, sender, receiver, udhdata, \
+msgdata, time, smsc_id, service, account, id, sms_type, mclass, mwi, coding, \
+compress, validity, deferred, dlr_mask, dlr_url, pid, alt_dcs, rpi, \
+charset, boxc_id, binfo, meta_data, %S FROM %S WHERE id = '%S' LIMIT 0,1)"
+
+
 #endif /* HAVE_MYSQL || HAVE_SDB */
 
 #ifdef HAVE_MYSQL
 #include "gw/msg.h"
 #include "sqlbox_sql.h"
 void sql_save_msg(Msg *msg, Octstr *momt);
-Msg *mysql_fetch_msg();
+int mysql_fetch_msg(List *list, long limit);
 void sql_shutdown();
 struct server_type *sqlbox_init_mysql(Cfg* cfg);
 #ifndef sqlbox_mysql_c
Index: sqlbox_sql.h
===================================================================
--- sqlbox_sql.h	(revision 52)
+++ sqlbox_sql.h	(working copy)
@@ -14,7 +14,7 @@
     Octstr *type;
     void (*sql_enter) (Cfg *);
     void (*sql_leave) ();
-    Msg *(*sql_fetch_msg) ();
+    int  (*sql_fetch_msg) (List *, long);
     void (*sql_save_msg) (Msg *, Octstr *);
 };