Failover for func_odbc, allowing an INSERT query to be performed when the UPDATE...

affects 0 rows.
(closes issue #13083)
 Reported by: Corydon76
 Patches:
       20081031__bug13083.diff.txt uploaded by Corydon76 (license 14)

git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@153124 65c4cc65-6c06-0410-ace0-fbb531ad65f3

diff --git a/CHANGES b/CHANGES
index 42ea9db..9bb9ce2 100644 (file)
--- a/CHANGES
+++ b/CHANGES
@@ -26,6 +26,8 @@ Dialplan Functions
  * Permit the syntax and synopsis fields of the corresponding dialplan
    functions to be individually set from func_odbc.conf.
  * Added debugging CLI functions to func_odbc, 'odbc read' and 'odbc write'.
+ * func_odbc now may specify an insert query to execute, when the write query
+   affects 0 rows (usually indicating that no such row exists).
 
 Applications
 ------------

diff --git a/configs/func_odbc.conf.sample b/configs/func_odbc.conf.sample
index a9742c2..6b1276e 100644 (file)
--- a/configs/func_odbc.conf.sample
+++ b/configs/func_odbc.conf.sample
@@ -31,6 +31,8 @@
 ;              readhandle.  "dsn" is a synonym for "writehandle".
 ; readsql      The statement to execute when reading from the function class.
 ; writesql     The statement to execute when writing to the function class.
+; insertsql    The statement to execute when writing to the function class
+;              succeeds, but initially indicates that 0 rows were affected.
 ; prefix       Normally, all function classes are prefixed with "ODBC" to keep
 ;              them uniquely named.  You may choose to change this prefix, which
 ;              may be useful to segregate a collection of certain function

diff --git a/funcs/func_odbc.c b/funcs/func_odbc.c
index b25e993..85f7e23 100644 (file)
--- a/funcs/func_odbc.c
+++ b/funcs/func_odbc.c
@@ -60,6 +60,7 @@ struct acf_odbc_query {
        char writehandle[5][30];
        char sql_read[2048];
        char sql_write[2048];
+       char sql_insert[2048];
        unsigned int flags;
        int rowlimit;
        struct ast_custom_function *acf;
@@ -89,6 +90,7 @@ AST_RWLIST_HEAD_STATIC(queries, acf_odbc_query);
 static int resultcount = 0;
 
 AST_THREADSTORAGE(sql_buf);
+AST_THREADSTORAGE(sql2_buf);
 AST_THREADSTORAGE(coldata_buf);
 AST_THREADSTORAGE(colnames_buf);
 
@@ -113,15 +115,32 @@ static SQLHSTMT generic_execute(struct odbc_obj *obj, void *data)
 
        res = SQLAllocHandle (SQL_HANDLE_STMT, obj->con, &stmt);
        if ((res != SQL_SUCCESS) && (res != SQL_SUCCESS_WITH_INFO)) {
-               ast_log(LOG_WARNING, "SQL Alloc Handle failed!\n");
+               ast_log(LOG_WARNING, "SQL Alloc Handle failed (%d)!\n", res);
                return NULL;
        }
 
        res = SQLExecDirect(stmt, (unsigned char *)sql, SQL_NTS);
        if ((res != SQL_SUCCESS) && (res != SQL_SUCCESS_WITH_INFO)) {
-               ast_log(LOG_WARNING, "SQL Exec Direct failed![%s]\n", sql);
+               if (res == SQL_ERROR) {
+                       int i;
+                       SQLINTEGER nativeerror=0, numfields=0;
+                       SQLSMALLINT diagbytes=0;
+                       unsigned char state[10], diagnostic[256];
+
+                       SQLGetDiagField(SQL_HANDLE_STMT, stmt, 1, SQL_DIAG_NUMBER, &numfields, SQL_IS_INTEGER, &diagbytes);
+                       for (i = 0; i < numfields; i++) {
+                               SQLGetDiagRec(SQL_HANDLE_STMT, stmt, i + 1, state, &nativeerror, diagnostic, sizeof(diagnostic), &diagbytes);
+                               ast_log(LOG_WARNING, "SQL Execute returned an error %d: %s: %s (%d)\n", res, state, diagnostic, diagbytes);
+                               if (i > 10) {
+                                       ast_log(LOG_WARNING, "Oh, that was good.  There are really %d diagnostics?\n", (int)numfields);
+                                       break;
+                               }
+                       }
+               }
+
+               ast_log(LOG_WARNING, "SQL Exec Direct failed (%d)![%s]\n", res, sql);
                SQLCloseCursor(stmt);
-               SQLFreeHandle (SQL_HANDLE_STMT, stmt);
+               SQLFreeHandle(SQL_HANDLE_STMT, stmt);
                return NULL;
        }
 
@@ -146,6 +165,8 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
        SQLHSTMT stmt = NULL;
        SQLLEN rows=0;
        struct ast_str *buf = ast_str_thread_get(&sql_buf, 16);
+       struct ast_str *insertbuf = ast_str_thread_get(&sql2_buf, 16);
+       const char *status = "FAILURE";
 
        if (!buf) {
                return -1;
@@ -161,7 +182,7 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
        if (!query) {
                ast_log(LOG_ERROR, "No such function '%s'\n", cmd);
                AST_RWLIST_UNLOCK(&queries);
-               ast_free(buf);
+               pbx_builtin_setvar_helper(chan, "ODBCSTATUS", status);
                return -1;
        }
 
@@ -174,6 +195,7 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
                ast_autoservice_start(chan);
 
        ast_str_make_space(&buf, strlen(query->sql_write) * 2 + 300);
+       ast_str_make_space(&insertbuf, strlen(query->sql_insert) * 2 + 300);
 
        /* Parse our arguments */
        t = value ? ast_strdupa(value) : "";
@@ -183,9 +205,11 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
                AST_RWLIST_UNLOCK(&queries);
                if (chan)
                        ast_autoservice_stop(chan);
-               if (bogus_chan)
+               if (bogus_chan) {
                        ast_channel_free(chan);
-               ast_free(buf);
+               } else {
+                       pbx_builtin_setvar_helper(chan, "ODBCSTATUS", status);
+               }
                return -1;
        }
 
@@ -206,6 +230,7 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
        pbx_builtin_pushvar_helper(chan, "VALUE", value ? value : "");
 
        pbx_substitute_variables_helper(chan, query->sql_write, buf->str, buf->len - 1);
+       pbx_substitute_variables_helper(chan, query->sql_insert, insertbuf->str, insertbuf->len - 1);
 
        /* Restore prior values */
        for (i = 0; i < args.argc; i++) {
@@ -225,23 +250,40 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
                        if (obj)
                                stmt = ast_odbc_direct_execute(obj, generic_execute, buf);
                }
-               if (stmt)
+               if (stmt) {
+                       status = "SUCCESS";
+                       SQLRowCount(stmt, &rows);
                        break;
+               }
        }
 
-       AST_RWLIST_UNLOCK(&queries);
-
-       if (stmt) {
-               /* Rows affected */
-               SQLRowCount(stmt, &rows);
+       if (stmt && rows == 0 && !ast_strlen_zero(insertbuf->str)) {
+               SQLCloseCursor(stmt);
+               SQLFreeHandle(SQL_HANDLE_STMT, stmt);
+               for (dsn = 0; dsn < 5; dsn++) {
+                       if (!ast_strlen_zero(query->writehandle[dsn])) {
+                               obj = ast_odbc_request_obj(query->writehandle[dsn], 0);
+                               if (obj) {
+                                       stmt = ast_odbc_direct_execute(obj, generic_execute, insertbuf);
+                               }
+                       }
+                       if (stmt) {
+                               status = "FAILOVER";
+                               SQLRowCount(stmt, &rows);
+                               break;
+                       }
+               }
        }
 
+       AST_RWLIST_UNLOCK(&queries);
+
        /* Output the affected rows, for all cases.  In the event of failure, we
         * flag this as -1 rows.  Note that this is different from 0 affected rows
         * which would be the case if we succeeded in our query, but the values did
         * not change. */
        snprintf(varname, sizeof(varname), "%d", (int)rows);
        pbx_builtin_setvar_helper(chan, "ODBCROWS", varname);
+       pbx_builtin_setvar_helper(chan, "ODBCSTATUS", status);
 
        if (stmt) {
                SQLCloseCursor(stmt);
@@ -254,7 +296,6 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
                ast_autoservice_stop(chan);
        if (bogus_chan)
                ast_channel_free(chan);
-       ast_free(buf);
 
        return 0;
 }
@@ -276,8 +317,10 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
        struct odbc_datastore *resultset = NULL;
        struct odbc_datastore_row *row = NULL;
        struct ast_str *sql = ast_str_thread_get(&sql_buf, 16);
+       const char *status = "FAILURE";
 
        if (!sql) {
+               pbx_builtin_setvar_helper(chan, "ODBCSTATUS", status);
                return -1;
        }
 
@@ -292,7 +335,7 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
                ast_log(LOG_ERROR, "No such function '%s'\n", cmd);
                AST_RWLIST_UNLOCK(&queries);
                pbx_builtin_setvar_helper(chan, "ODBCROWS", rowcount);
-               ast_free(sql);
+               pbx_builtin_setvar_helper(chan, "ODBCSTATUS", status);
                return -1;
        }
 
@@ -358,7 +401,6 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
                if (bogus_chan) {
                        ast_channel_free(chan);
                }
-               ast_free(sql);
                return -1;
        }
 
@@ -375,7 +417,6 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
                if (bogus_chan) {
                        ast_channel_free(chan);
                }
-               ast_free(sql);
                return -1;
        }
 
@@ -387,21 +428,25 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
                        res1 = 0;
                        buf[0] = '\0';
                        ast_copy_string(rowcount, "0", sizeof(rowcount));
+                       status = "NODATA";
                } else {
                        ast_log(LOG_WARNING, "Error %d in FETCH [%s]\n", res, sql->str);
+                       status = "FETCHERROR";
                }
                SQLCloseCursor(stmt);
                SQLFreeHandle(SQL_HANDLE_STMT, stmt);
                ast_odbc_release_obj(obj);
                pbx_builtin_setvar_helper(chan, "ODBCROWS", rowcount);
+               pbx_builtin_setvar_helper(chan, "ODBCSTATUS", status);
                if (chan)
                        ast_autoservice_stop(chan);
                if (bogus_chan)
                        ast_channel_free(chan);
-               ast_free(sql);
                return res1;
        }
 
+       status = "SUCCESS";
+
        for (y = 0; y < rowlimit; y++) {
                for (x = 0; x < colcount; x++) {
                        int i;
@@ -447,11 +492,11 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
                                                SQLFreeHandle(SQL_HANDLE_STMT, stmt);
                                                ast_odbc_release_obj(obj);
                                                pbx_builtin_setvar_helper(chan, "ODBCROWS", rowcount);
+                                               pbx_builtin_setvar_helper(chan, "ODBCSTATUS", "MEMERROR");
                                                if (chan)
                                                        ast_autoservice_stop(chan);
                                                if (bogus_chan)
                                                        ast_channel_free(chan);
-                                               ast_free(sql);
                                                return -1;
                                        }
                                        resultset = tmp;
@@ -503,6 +548,7 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
                        row = ast_calloc(1, sizeof(*row) + buflen);
                        if (!row) {
                                ast_log(LOG_ERROR, "Unable to allocate space for more rows in this resultset.\n");
+                               status = "MEMERROR";
                                goto end_acf_read;
                        }
                        strcpy((char *)row + sizeof(*row), buf);
@@ -522,6 +568,7 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
 end_acf_read:
        snprintf(rowcount, sizeof(rowcount), "%d", y);
        pbx_builtin_setvar_helper(chan, "ODBCROWS", rowcount);
+       pbx_builtin_setvar_helper(chan, "ODBCSTATUS", status);
        pbx_builtin_setvar_helper(chan, "~ODBCFIELDS~", colnames->str);
        if (resultset) {
                int uid;
@@ -531,6 +578,7 @@ end_acf_read:
                odbc_store = ast_datastore_alloc(&odbc_info, buf);
                if (!odbc_store) {
                        ast_log(LOG_ERROR, "Rows retrieved, but unable to store it in the channel.  Results fail.\n");
+                       pbx_builtin_setvar_helper(chan, "ODBCSTATUS", "MEMERROR");
                        odbc_datastore_free(resultset);
                        SQLCloseCursor(stmt);
                        SQLFreeHandle(SQL_HANDLE_STMT, stmt);
@@ -539,7 +587,6 @@ end_acf_read:
                                ast_autoservice_stop(chan);
                        if (bogus_chan)
                                ast_channel_free(chan);
-                       ast_free(sql);
                        return -1;
                }
                odbc_store->data = resultset;
@@ -552,7 +599,6 @@ end_acf_read:
                ast_autoservice_stop(chan);
        if (bogus_chan)
                ast_channel_free(chan);
-       ast_free(sql);
        return 0;
 }
 
@@ -714,6 +760,10 @@ static int init_acf_query(struct ast_config *cfg, char *catg, struct acf_odbc_qu
                return EINVAL;
        }
 
+       if ((tmp = ast_variable_retrieve(cfg, catg, "insertsql"))) {
+               ast_copy_string((*query)->sql_insert, tmp, sizeof((*query)->sql_insert));
+       }
+
        /* Allow escaping of embedded commas in fields to be turned off */
        ast_set_flag((*query), OPT_ESCAPECOMMAS);
        if ((tmp = ast_variable_retrieve(cfg, catg, "escapecommas"))) {
@@ -783,9 +833,16 @@ static int init_acf_query(struct ast_config *cfg, char *catg, struct acf_odbc_qu
                                        "substitution of the arguments into the query as specified by ${ARG1},\n"
                                        "${ARG2}, ... ${ARGn}.  When setting the function, the values are provided\n"
                                        "either in whole as ${VALUE} or parsed as ${VAL1}, ${VAL2}, ... ${VALn}.\n"
-                                       "\nRead:\n%s\n\nWrite:\n%s\n",
+                                       "%s"
+                                       "\nRead:\n%s\n\nWrite:\n%s\n%s%s%s",
+                                       ast_strlen_zero((*query)->sql_insert) ? "" :
+                                               "If the write query affects no rows, the insert query will be\n"
+                                               "performed.\n",
                                        (*query)->sql_read,
-                                       (*query)->sql_write);
+                                       (*query)->sql_write,
+                                       ast_strlen_zero((*query)->sql_insert) ? "" : "Insert:\n",
+                                       ast_strlen_zero((*query)->sql_insert) ? "" : (*query)->sql_insert,
+                                       ast_strlen_zero((*query)->sql_insert) ? "" : "\n");
        } else if (!ast_strlen_zero((*query)->sql_read)) {
                asprintf((char **)&((*query)->acf->desc),
                                        "Runs the following query, as defined in func_odbc.conf, performing\n"
@@ -798,15 +855,21 @@ static int init_acf_query(struct ast_config *cfg, char *catg, struct acf_odbc_qu
                                        "substitution of the arguments into the query as specified by ${ARG1},\n"
                                        "${ARG2}, ... ${ARGn}.  The values are provided either in whole as\n"
                                        "${VALUE} or parsed as ${VAL1}, ${VAL2}, ... ${VALn}.\n"
-                                       "This function may only be set.\nSQL:\n%s\n",
-                                       (*query)->sql_write);
+                                       "This function may only be set.\n%sSQL:\n%s\n%s%s%s",
+                                       ast_strlen_zero((*query)->sql_insert) ? "" :
+                                               "If the write query affects no rows, the insert query will be\n"
+                                               "performed.\n",
+                                       (*query)->sql_write,
+                                       ast_strlen_zero((*query)->sql_insert) ? "" : "Insert:\n",
+                                       ast_strlen_zero((*query)->sql_insert) ? "" : (*query)->sql_insert,
+                                       ast_strlen_zero((*query)->sql_insert) ? "" : "\n");
        } else {
                ast_free((char *)(*query)->acf->synopsis);
                ast_free((char *)(*query)->acf->syntax);
                ast_free((char *)(*query)->acf->name);
                ast_free((*query)->acf);
                ast_free(*query);
-               ast_log(LOG_WARNING, "Section %s was found, but there was no SQL to execute.  Ignoring.\n", catg);
+               ast_log(LOG_WARNING, "Section '%s' was found, but there was no SQL to execute.  Ignoring.\n", catg);
                return EINVAL;
        }