kenmirrors/asterisk
1
0
Fork 0
mirror of https://github.com/asterisk/asterisk.git synced 2025-10-18 02:32:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge the adaptive realtime branch, which will make adding new required fields

Browse Source 
to realtime less painful in the future.


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@120789 65c4cc65-6c06-0410-ace0-fbb531ad65f3
This commit is contained in:
Tilghman Lesher
2008-06-05 19:07:27 +00:00
parent 2347ebf436
commit 9471b87d27
13 changed files with 892 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
res/res_config_curl.c
View File

@@ -406,6 +406,42 @@ static int destroy_curl(const char *url, const char *unused, const char *keyfiel
return -1;
}
static int require_curl(const char *url, const char *unused, va_list ap)
{
struct ast_str *query;
char *elm, field[256], buffer[128];
int type, size;
const int EncodeSpecialChars = 1;
if (!ast_custom_function_find("CURL")) {
ast_log(LOG_ERROR, "func_curl.so must be loaded in order to use res_config_curl.so!!\n");
return -1;
}
if (!(query = ast_str_create(100))) {
return -1;
}
ast_str_set(&query, 0, "${CURL(%s/require,", url);
while ((elm = va_arg(ap, char *))) {
type = va_arg(ap, require_type);
size = va_arg(ap, int);
ast_uri_encode(elm, field, sizeof(field), EncodeSpecialChars);
ast_str_append(&query, 0, "%s=%s%%3A%d", field,
type == RQ_CHAR ? "char" :
type == RQ_INTEGER ? "integer" :
type == RQ_DATE ? "date" :
type == RQ_DATETIME ? "datetime" :
type == RQ_FLOAT ? "float" :
"unknown", size);
}
va_end(ap);
ast_str_append(&query, 0, ")}");
pbx_substitute_variables_helper(NULL, query->str, buffer, sizeof(buffer));
return atoi(buffer);
}
static struct ast_config *config_curl(const char *url, const char *unused, const char *file, struct ast_config *cfg, struct ast_flags flags, const char *sugg_incl, const char *who_asked)
{
@@ -489,7 +525,8 @@ static struct ast_config_engine curl_engine = {
.realtime_multi_func = realtime_multi_curl,
.store_func = store_curl,
.destroy_func = destroy_curl,
.update_func = update_curl
.update_func = update_curl,
.require_func = require_curl,
};
static int unload_module (void)

316
res/res_config_odbc.c
View File

@@ -51,11 +51,159 @@ struct custom_prepare_struct {
const char *sql;
const char *extra;
va_list ap;
unsigned long long skip;
};
/*!\brief The structures referenced are in include/asterisk/res_odbc.h */
static AST_RWLIST_HEAD_STATIC(odbc_tables, odbc_cache_tables);
static void destroy_table_cache(struct odbc_cache_tables *table) {
struct odbc_cache_columns *col;
ast_debug(1, "Destroying table cache for %s\n", table->table);
AST_RWLIST_WRLOCK(&table->columns);
while ((col = AST_RWLIST_REMOVE_HEAD(&table->columns, list))) {
ast_free(col);
}
AST_RWLIST_UNLOCK(&table->columns);
AST_RWLIST_HEAD_DESTROY(&table->columns);
ast_free(table);
}
#define release_table(ptr) if (ptr) { AST_RWLIST_UNLOCK(&(ptr)->columns); }
/*!
* \brief Find or create an entry describing the table specified.
* \param obj An active ODBC handle on which to query the table
* \param table Tablename to describe
* \retval A structure describing the table layout, or NULL, if the table is not found or another error occurs.
* When a structure is returned, the contained columns list will be
* rdlock'ed, to ensure that it will be retained in memory.
*/
static struct odbc_cache_tables *find_table(const char *database, const char *tablename)
{
struct odbc_cache_tables *tableptr;
struct odbc_cache_columns *entry;
char columnname[80];
SQLLEN sqlptr;
SQLHSTMT stmt = NULL;
int res = 0, error = 0, try = 0;
struct odbc_obj *obj = ast_odbc_request_obj(database, 0);
AST_RWLIST_RDLOCK(&odbc_tables);
AST_RWLIST_TRAVERSE(&odbc_tables, tableptr, list) {
if (strcmp(tableptr->connection, database) == 0 && strcmp(tableptr->table, tablename) == 0) {
break;
}
}
if (tableptr) {
AST_RWLIST_RDLOCK(&tableptr->columns);
AST_RWLIST_UNLOCK(&odbc_tables);
return tableptr;
}
if (!obj) {
ast_log(LOG_WARNING, "Unable to retrieve database handle for table description '%s@%s'\n", tablename, database);
return NULL;
}
/* Table structure not already cached; build it now. */
do {
retry:
res = SQLAllocHandle(SQL_HANDLE_STMT, obj->con, &stmt);
if ((res != SQL_SUCCESS) && (res != SQL_SUCCESS_WITH_INFO)) {
if (try == 0) {
try = 1;
ast_odbc_sanity_check(obj);
goto retry;
}
ast_log(LOG_WARNING, "SQL Alloc Handle failed on connection '%s'!\n", database);
break;
}
res = SQLColumns(stmt, NULL, 0, NULL, 0, (unsigned char *)tablename, SQL_NTS, (unsigned char *)"%", SQL_NTS);
if ((res != SQL_SUCCESS) && (res != SQL_SUCCESS_WITH_INFO)) {
if (try == 0) {
try = 1;
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_sanity_check(obj);
goto retry;
}
ast_log(LOG_ERROR, "Unable to query database columns on connection '%s'.\n", database);
break;
}
if (!(tableptr = ast_calloc(sizeof(char), sizeof(*tableptr) + strlen(database) + 1 + strlen(tablename) + 1))) {
ast_log(LOG_ERROR, "Out of memory creating entry for table '%s' on connection '%s'\n", tablename, database);
break;
}
tableptr->connection = (char *)tableptr + sizeof(*tableptr);
tableptr->table = (char *)tableptr + sizeof(*tableptr) + strlen(database) + 1;
strcpy(tableptr->connection, database); /* SAFE */
strcpy(tableptr->table, tablename); /* SAFE */
AST_RWLIST_HEAD_INIT(&(tableptr->columns));
while ((res = SQLFetch(stmt)) != SQL_NO_DATA && res != SQL_ERROR) {
SQLGetData(stmt, 4, SQL_C_CHAR, columnname, sizeof(columnname), &sqlptr);
if (!(entry = ast_calloc(sizeof(char), sizeof(*entry) + strlen(columnname) + 1))) {
ast_log(LOG_ERROR, "Out of memory creating entry for column '%s' in table '%s' on connection '%s'\n", columnname, tablename, database);
error = 1;
break;
}
entry->name = (char *)entry + sizeof(*entry);
strcpy(entry->name, columnname);
SQLGetData(stmt, 5, SQL_C_SHORT, &entry->type, sizeof(entry->type), NULL);
SQLGetData(stmt, 7, SQL_C_LONG, &entry->size, sizeof(entry->size), NULL);
SQLGetData(stmt, 9, SQL_C_SHORT, &entry->decimals, sizeof(entry->decimals), NULL);
SQLGetData(stmt, 10, SQL_C_SHORT, &entry->radix, sizeof(entry->radix), NULL);
SQLGetData(stmt, 11, SQL_C_SHORT, &entry->nullable, sizeof(entry->nullable), NULL);
SQLGetData(stmt, 16, SQL_C_LONG, &entry->octetlen, sizeof(entry->octetlen), NULL);
/* Specification states that the octenlen should be the maximum number of bytes
* returned in a char or binary column, but it seems that some drivers just set
* it to NULL. (Bad Postgres! No biscuit!) */
if (entry->octetlen == 0) {
entry->octetlen = entry->size;
}
ast_verb(10, "Found %s column with type %hd with len %ld, octetlen %ld, and numlen (%hd,%hd)\n", entry->name, entry->type, (long) entry->size, (long) entry->octetlen, entry->decimals, entry->radix);
/* Insert column info into column list */
AST_LIST_INSERT_TAIL(&(tableptr->columns), entry, list);
}
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
AST_RWLIST_INSERT_TAIL(&odbc_tables, tableptr, list);
AST_RWLIST_RDLOCK(&(tableptr->columns));
} while (0);
AST_RWLIST_UNLOCK(&odbc_tables);
if (error) {
destroy_table_cache(tableptr);
tableptr = NULL;
}
if (obj) {
ast_odbc_release_obj(obj);
}
return tableptr;
}
static struct odbc_cache_columns *find_column(struct odbc_cache_tables *table, const char *colname)
{
struct odbc_cache_columns *col;
AST_RWLIST_TRAVERSE(&table->columns, col, list) {
if (strcasecmp(col->name, colname) == 0) {
return col;
}
}
return NULL;
}
static SQLHSTMT custom_prepare(struct odbc_obj *obj, void *data)
{
int res, x = 1;
int res, x = 1, count = 0;
struct custom_prepare_struct *cps = data;
const char *newparam, *newval;
SQLHSTMT stmt;
@@ -78,6 +226,9 @@ static SQLHSTMT custom_prepare(struct odbc_obj *obj, void *data)
while ((newparam = va_arg(ap, const char *))) {
newval = va_arg(ap, const char *);
if ((1 << count) & cps->skip) {
continue;
}
SQLBindParameter(stmt, x++, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_CHAR, strlen(newval), 0, (void *)newval, 0, NULL);
}
va_end(ap);
@@ -398,33 +549,51 @@ static int update_odbc(const char *database, const char *table, const char *keyf
char sql[256];
SQLLEN rowcount=0;
const char *newparam, *newval;
int res;
int res, count = 0;
va_list aq;
struct custom_prepare_struct cps = { .sql = sql, .extra = lookup };
struct odbc_cache_tables *tableptr = find_table(database, table);
struct odbc_cache_columns *column;
va_copy(cps.ap, ap);
va_copy(aq, ap);
if (!table)
if (!table) {
release_table(tableptr);
return -1;
}
obj = ast_odbc_request_obj(database, 0);
if (!obj)
if (!obj) {
release_table(tableptr);
return -1;
}
newparam = va_arg(aq, const char *);
if (!newparam) {
ast_odbc_release_obj(obj);
release_table(tableptr);
return -1;
}
newval = va_arg(aq, const char *);
if (tableptr && !(column = find_column(tableptr, newparam))) {
ast_log(LOG_WARNING, "Key field '%s' does not exist in table '%s@%s'. Update will fail\n", newparam, table, database);
}
snprintf(sql, sizeof(sql), "UPDATE %s SET %s=?", table, newparam);
while((newparam = va_arg(aq, const char *))) {
snprintf(sql + strlen(sql), sizeof(sql) - strlen(sql), ", %s=?", newparam);
newval = va_arg(aq, const char *);
if ((tableptr && (column = find_column(tableptr, newparam))) || count > 63) {
snprintf(sql + strlen(sql), sizeof(sql) - strlen(sql), ", %s=?", newparam);
newval = va_arg(aq, const char *);
} else { /* the column does not exist in the table OR we've exceeded the space in our flag field */
cps.skip |= (((long long)1) << count);
}
count++;
}
va_end(aq);
snprintf(sql + strlen(sql), sizeof(sql) - strlen(sql), " WHERE %s=?", keyfield);
release_table(tableptr);
stmt = ast_odbc_prepare_and_execute(obj, custom_prepare, &cps);
@@ -709,6 +878,112 @@ static struct ast_config *config_odbc(const char *database, const char *table, c
return cfg;
}
#define warn_length(col, size) ast_log(LOG_WARNING, "Column %s is not long enough to contain realtime data (needs %d)\n", col->name, size)
#define warn_type(col, type) ast_log(LOG_WARNING, "Column %s is of the incorrect type to contain realtime data\n", col->name)
static int require_odbc(const char *database, const char *table, va_list ap)
{
struct odbc_cache_tables *tableptr = find_table(database, table);
struct odbc_cache_columns *col;
char *elm;
int type, size;
if (!tableptr) {
return -1;
}
while ((elm = va_arg(ap, char *))) {
type = va_arg(ap, require_type);
size = va_arg(ap, int);
/* Check if the field matches the criteria */
AST_RWLIST_TRAVERSE(&tableptr->columns, col, list) {
if (strcmp(col->name, elm) == 0) {
/* Type check, first. Some fields are more particular than others */
switch (col->type) {
case SQL_CHAR:
case SQL_VARCHAR:
case SQL_LONGVARCHAR:
case SQL_BINARY:
case SQL_VARBINARY:
case SQL_LONGVARBINARY:
case SQL_GUID:
if ((type == RQ_INTEGER && size > 10) || (type == RQ_CHAR && col->size < size)) {
warn_length(col, size);
} else if (type == RQ_DATE && col->size < 10) {
warn_length(col, 10);
} else if (type == RQ_DATETIME && col->size < 19) {
warn_length(col, 19);
} else if (type == RQ_FLOAT && col->size < 10) {
warn_length(col, 10);
}
break;
case SQL_TYPE_DATE:
if (type != RQ_DATE) {
warn_type(col, type);
}
break;
case SQL_TYPE_TIMESTAMP:
case SQL_TIMESTAMP:
if (type != RQ_DATE && type != RQ_DATETIME) {
warn_type(col, type);
}
break;
case SQL_INTEGER:
case SQL_BIGINT:
case SQL_SMALLINT:
case SQL_TINYINT:
case SQL_BIT:
if (type != RQ_INTEGER) {
warn_type(col, type);
}
if ((col->type == SQL_BIT && size > 1) ||
(col->type == SQL_TINYINT && size > 2) ||
(col->type == SQL_SMALLINT && size > 4) ||
(col->type == SQL_INTEGER && size > 10)) {
warn_length(col, size);
}
break;
case SQL_NUMERIC:
case SQL_DECIMAL:
case SQL_FLOAT:
case SQL_REAL:
case SQL_DOUBLE:
if (type != RQ_INTEGER && type != RQ_FLOAT) {
warn_type(col, type);
}
break;
default:
ast_log(LOG_WARNING, "Column type (%d) unrecognized for field '%s' in %s@%s\n", col->type, elm, table, database);
}
break;
}
}
if (!col) {
ast_log(LOG_WARNING, "Table %s@%s requires column '%s', but that column does not exist!\n", table, database, elm);
}
}
va_end(ap);
AST_RWLIST_UNLOCK(&tableptr->columns);
return 0;
}
static int unload_odbc(const char *database, const char *tablename)
{
struct odbc_cache_tables *tableptr;
AST_RWLIST_RDLOCK(&odbc_tables);
AST_RWLIST_TRAVERSE_SAFE_BEGIN(&odbc_tables, tableptr, list) {
if (strcmp(tableptr->connection, database) == 0 && strcmp(tableptr->table, tablename) == 0) {
AST_LIST_REMOVE_CURRENT(list);
destroy_table_cache(tableptr);
break;
}
}
AST_RWLIST_TRAVERSE_SAFE_END
AST_RWLIST_UNLOCK(&odbc_tables);
return tableptr ? 0 : -1;
}
static struct ast_config_engine odbc_engine = {
.name = "odbc",
.load_func = config_odbc,
@@ -716,12 +991,24 @@ static struct ast_config_engine odbc_engine = {
.realtime_multi_func = realtime_multi_odbc,
.store_func = store_odbc,
.destroy_func = destroy_odbc,
.update_func = update_odbc
.update_func = update_odbc,
.require_func = require_odbc,
.unload_func = unload_odbc,
};
static int unload_module (void)
{
struct odbc_cache_tables *table;
ast_config_engine_deregister(&odbc_engine);
/* Empty the cache */
AST_RWLIST_WRLOCK(&odbc_tables);
while ((table = AST_RWLIST_REMOVE_HEAD(&odbc_tables, list))) {
destroy_table_cache(table);
}
AST_RWLIST_UNLOCK(&odbc_tables);
ast_verb(1, "res_config_odbc unloaded.\n");
return 0;
}
@@ -733,7 +1020,22 @@ static int load_module (void)
return 0;
}
static int reload_module(void)
{
struct odbc_cache_tables *table;
/* Empty the cache; it will get rebuilt the next time the tables are needed. */
AST_RWLIST_WRLOCK(&odbc_tables);
while ((table = AST_RWLIST_REMOVE_HEAD(&odbc_tables, list))) {
destroy_table_cache(table);
}
AST_RWLIST_UNLOCK(&odbc_tables);
return 0;
}
AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS, "Realtime ODBC configuration",
.load = load_module,
.unload = unload_module,
.reload = reload_module,
);

386
res/res_config_pgsql.c
View File

@@ -49,6 +49,24 @@ PGconn *pgsqlConn = NULL;
#define MAX_DB_OPTION_SIZE 64
struct columns {
char *name;
char *type;
int len;
unsigned int notnull:1;
unsigned int hasdefault:1;
AST_LIST_ENTRY(columns) list;
};
struct tables {
ast_mutex_t lock;
AST_LIST_HEAD_NOLOCK(psql_columns, columns) columns;
AST_LIST_ENTRY(tables) list;
char name[0];
};
static AST_LIST_HEAD_STATIC(psql_tables, tables);
static char dbhost[MAX_DB_OPTION_SIZE] = "";
static char dbuser[MAX_DB_OPTION_SIZE] = "";
static char dbpass[MAX_DB_OPTION_SIZE] = "";
@@ -60,11 +78,112 @@ static time_t connect_time = 0;
static int parse_config(int reload);
static int pgsql_reconnect(const char *database);
static char *handle_cli_realtime_pgsql_status(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
static char *handle_cli_realtime_pgsql_cache(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
enum { RQ_WARN, RQ_CREATECLOSE, RQ_CREATECHAR } requirements;
static struct ast_cli_entry cli_realtime[] = {
AST_CLI_DEFINE(handle_cli_realtime_pgsql_status, "Shows connection information for the PostgreSQL RealTime driver"),
AST_CLI_DEFINE(handle_cli_realtime_pgsql_cache, "Shows cached tables within the PostgreSQL realtime driver"),
};
static void destroy_table(struct tables *table)
{
struct columns *column;
ast_mutex_lock(&table->lock);
while ((column = AST_LIST_REMOVE_HEAD(&table->columns, list))) {
ast_free(column);
}
ast_mutex_unlock(&table->lock);
ast_mutex_destroy(&table->lock);
ast_free(table);
}
static struct tables *find_table(const char *tablename)
{
struct columns *column;
struct tables *table;
struct ast_str *sql = ast_str_create(330);
char *pgerror;
PGresult *result;
char *fname, *ftype, *flen, *fnotnull, *fdef;
int i, rows;
AST_LIST_LOCK(&psql_tables);
AST_LIST_TRAVERSE(&psql_tables, table, list) {
if (!strcasecmp(table->name, tablename)) {
ast_debug(1, "Found table in cache; now locking\n");
ast_mutex_lock(&table->lock);
ast_debug(1, "Lock cached table; now returning\n");
AST_LIST_UNLOCK(&psql_tables);
return table;
}
}
ast_debug(1, "Table '%s' not found in cache, querying now\n", tablename);
/* Not found, scan the table */
ast_str_set(&sql, 0, "SELECT a.attname, t.typname, a.attlen, a.attnotnull, d.adsrc FROM pg_class c, pg_type t, pg_attribute a LEFT OUTER JOIN pg_attrdef d ON a.atthasdef AND d.adrelid = a.attrelid AND d.adnum = a.attnum WHERE c.oid = a.attrelid AND a.atttypid = t.oid AND (a.attnum > 0) AND c.relname = '%s' ORDER BY c.relname, attnum", tablename);
result = PQexec(pgsqlConn, sql->str);
ast_debug(1, "Query of table structure complete. Now retrieving results.\n");
if (PQresultStatus(result) != PGRES_TUPLES_OK) {
pgerror = PQresultErrorMessage(result);
ast_log(LOG_ERROR, "Failed to query database columns: %s\n", pgerror);
PQclear(result);
AST_LIST_UNLOCK(&psql_tables);
return NULL;
}
if (!(table = ast_calloc(1, sizeof(*table) + strlen(tablename) + 1))) {
ast_log(LOG_ERROR, "Unable to allocate memory for new table structure\n");
AST_LIST_UNLOCK(&psql_tables);
return NULL;
}
strcpy(table->name, tablename); /* SAFE */
ast_mutex_init(&table->lock);
AST_LIST_HEAD_INIT_NOLOCK(&table->columns);
rows = PQntuples(result);
for (i = 0; i < rows; i++) {
fname = PQgetvalue(result, i, 0);
ftype = PQgetvalue(result, i, 1);
flen = PQgetvalue(result, i, 2);
fnotnull = PQgetvalue(result, i, 3);
fdef = PQgetvalue(result, i, 4);
ast_verb(4, "Found column '%s' of type '%s'\n", fname, ftype);
if (!(column = ast_calloc(1, sizeof(*column) + strlen(fname) + strlen(ftype) + 2))) {
ast_log(LOG_ERROR, "Unable to allocate column element for %s, %s\n", tablename, fname);
destroy_table(table);
AST_LIST_UNLOCK(&psql_tables);
return NULL;
}
sscanf(flen, "%d", &column->len);
column->name = (char *)column + sizeof(*column);
column->type = (char *)column + sizeof(*column) + strlen(fname) + 1;
strcpy(column->name, fname);
strcpy(column->type, ftype);
if (*fnotnull == 't') {
column->notnull = 1;
} else {
column->notnull = 0;
}
if (!ast_strlen_zero(fdef)) {
column->hasdefault = 1;
} else {
column->hasdefault = 0;
}
AST_LIST_INSERT_TAIL(&table->columns, column, list);
}
PQclear(result);
AST_LIST_INSERT_TAIL(&psql_tables, table, list);
ast_mutex_lock(&table->lock);
AST_LIST_UNLOCK(&psql_tables);
return table;
}
static struct ast_variable *realtime_pgsql(const char *database, const char *table, va_list ap)
{
PGresult *result = NULL;
@@ -360,16 +479,26 @@ static struct ast_config *realtime_multi_pgsql(const char *database, const char
return cfg;
}
static int update_pgsql(const char *database, const char *table, const char *keyfield,
static int update_pgsql(const char *database, const char *tablename, const char *keyfield,
const char *lookup, va_list ap)
{
PGresult *result = NULL;
int numrows = 0, pgerror;
char sql[256], escapebuf[513];
char escapebuf[513];
const char *newparam, *newval;
struct ast_str *sql = ast_str_create(100);
struct tables *table;
struct columns *column = NULL;
if (!table) {
if (!tablename) {
ast_log(LOG_WARNING, "PostgreSQL RealTime: No table specified.\n");
ast_free(sql);
return -1;
}
if (!(table = find_table(tablename))) {
ast_log(LOG_ERROR, "Table '%s' does not exist!!\n", tablename);
ast_free(sql);
return -1;
}
@@ -383,6 +512,22 @@ static int update_pgsql(const char *database, const char *table, const char *key
PQfinish(pgsqlConn);
pgsqlConn = NULL;
};
ast_mutex_unlock(&table->lock);
ast_free(sql);
return -1;
}
/* Check that the column exists in the table */
AST_LIST_TRAVERSE(&table->columns, column, list) {
if (strcmp(column->name, newparam) == 0) {
break;
}
}
if (!column) {
ast_log(LOG_ERROR, "PostgreSQL RealTime: Updating on column '%s', but that column does not exist within the table '%s'!\n", newparam, tablename);
ast_mutex_unlock(&table->lock);
ast_free(sql);
return -1;
}
@@ -393,50 +538,68 @@ static int update_pgsql(const char *database, const char *table, const char *key
if (pgerror) {
ast_log(LOG_ERROR, "Postgres detected invalid input: '%s'\n", newval);
va_end(ap);
ast_mutex_unlock(&table->lock);
ast_free(sql);
return -1;
}
snprintf(sql, sizeof(sql), "UPDATE %s SET %s = '%s'", table, newparam, escapebuf);
ast_str_set(&sql, 0, "UPDATE %s SET %s = '%s'", tablename, newparam, escapebuf);
while ((newparam = va_arg(ap, const char *))) {
newval = va_arg(ap, const char *);
/* If the column is not within the table, then skip it */
AST_LIST_TRAVERSE(&table->columns, column, list) {
if (strcmp(column->name, newparam) == 0) {
break;
}
}
if (!column) {
ast_log(LOG_WARNING, "Attempted to update column '%s' in table '%s', but column does not exist!\n", newparam, tablename);
continue;
}
PQescapeStringConn(pgsqlConn, escapebuf, newval, (sizeof(escapebuf) - 1) / 2, &pgerror);
if (pgerror) {
ast_log(LOG_ERROR, "Postgres detected invalid input: '%s'\n", newval);
va_end(ap);
ast_mutex_unlock(&table->lock);
ast_free(sql);
return -1;
}
snprintf(sql + strlen(sql), sizeof(sql) - strlen(sql), ", %s = '%s'", newparam,
escapebuf);
ast_str_append(&sql, 0, ", %s = '%s'", newparam, escapebuf);
}
va_end(ap);
ast_mutex_unlock(&table->lock);
PQescapeStringConn(pgsqlConn, escapebuf, lookup, (sizeof(escapebuf) - 1) / 2, &pgerror);
if (pgerror) {
ast_log(LOG_ERROR, "Postgres detected invalid input: '%s'\n", lookup);
va_end(ap);
ast_free(sql);
return -1;
}
snprintf(sql + strlen(sql), sizeof(sql) - strlen(sql), " WHERE %s = '%s'", keyfield,
escapebuf);
ast_str_append(&sql, 0, " WHERE %s = '%s'", keyfield, escapebuf);
ast_debug(1, "PostgreSQL RealTime: Update SQL: %s\n", sql);
ast_debug(1, "PostgreSQL RealTime: Update SQL: %s\n", sql->str);
/* We now have our complete statement; Lets connect to the server and execute it. */
ast_mutex_lock(&pgsql_lock);
if (!pgsql_reconnect(database)) {
ast_mutex_unlock(&pgsql_lock);
ast_free(sql);
return -1;
}
if (!(result = PQexec(pgsqlConn, sql))) {
if (!(result = PQexec(pgsqlConn, sql->str))) {
ast_log(LOG_WARNING,
"PostgreSQL RealTime: Failed to query database. Check debug for more info.\n");
ast_debug(1, "PostgreSQL RealTime: Query: %s\n", sql);
ast_debug(1, "PostgreSQL RealTime: Query: %s\n", sql->str);
ast_debug(1, "PostgreSQL RealTime: Query Failed because: %s\n", PQerrorMessage(pgsqlConn));
ast_mutex_unlock(&pgsql_lock);
ast_free(sql);
return -1;
} else {
ExecStatusType result_status = PQresultStatus(result);
@@ -445,18 +608,20 @@ static int update_pgsql(const char *database, const char *table, const char *key
&& result_status != PGRES_NONFATAL_ERROR) {
ast_log(LOG_WARNING,
"PostgreSQL RealTime: Failed to query database. Check debug for more info.\n");
ast_debug(1, "PostgreSQL RealTime: Query: %s\n", sql);
ast_debug(1, "PostgreSQL RealTime: Query: %s\n", sql->str);
ast_debug(1, "PostgreSQL RealTime: Query Failed because: %s (%s)\n",
PQresultErrorMessage(result), PQresStatus(result_status));
ast_mutex_unlock(&pgsql_lock);
ast_free(sql);
return -1;
}
}
numrows = atoi(PQcmdTuples(result));
ast_mutex_unlock(&pgsql_lock);
ast_free(sql);
ast_debug(1, "PostgreSQL RealTime: Updated %d rows on table: %s\n", numrows, table);
ast_debug(1, "PostgreSQL RealTime: Updated %d rows on table: %s\n", numrows, tablename);
/* From http://dev.pgsql.com/doc/pgsql/en/pgsql-affected-rows.html
* An integer greater than zero indicates the number of rows affected
@@ -752,6 +917,124 @@ static struct ast_config *config_pgsql(const char *database, const char *table,
return cfg;
}
static int require_pgsql(const char *database, const char *tablename, va_list ap)
{
struct columns *column;
struct tables *table = find_table(tablename);
char *elm;
int type, size, res = 0;
if (!table) {
ast_log(LOG_WARNING, "Table %s not found in database. This table should exist if you're using realtime.\n", tablename);
return -1;
}
while ((elm = va_arg(ap, char *))) {
type = va_arg(ap, require_type);
size = va_arg(ap, int);
AST_LIST_TRAVERSE(&table->columns, column, list) {
if (strcmp(column->name, elm) == 0) {
/* Char can hold anything, as long as it is large enough */
if ((strncmp(column->type, "char", 4) == 0 || strncmp(column->type, "varchar", 7) == 0 || strcmp(column->type, "bpchar") == 0)) {
if ((size > column->len) && column->len != -1) {
ast_log(LOG_WARNING, "Column '%s' should be at least %d long, but is only %d long.\n", column->name, size, column->len);
res = -1;
}
} else if (strncmp(column->type, "int", 3) == 0) {
int typesize = atoi(column->type + 3);
/* Integers can hold only other integers */
if (type == RQ_INTEGER && ((typesize == 2 && size > 4) || (typesize == 4 && size > 10))) {
ast_log(LOG_WARNING, "Column '%s' may not be large enough for the required data length: %d\n", column->name, size);
res = -1;
} else if (type != RQ_INTEGER) {
ast_log(LOG_WARNING, "Column '%s' is of the incorrect type: (need %s(%d) but saw %s)\n", column->name, type == RQ_CHAR ? "char" : "something else ", size, column->type);
res = -1;
}
} else if (strncmp(column->type, "float", 5) == 0 && type != RQ_INTEGER && type != RQ_FLOAT) {
ast_log(LOG_WARNING, "Column %s cannot be a %s\n", column->name, column->type);
res = -1;
} else { /* There are other types that no module implements yet */
ast_log(LOG_WARNING, "Possibly unsupported column type '%s' on column '%s'\n", column->type, column->name);
res = -1;
}
break;
}
}
if (!column) {
if (requirements == RQ_WARN) {
ast_log(LOG_WARNING, "Table %s requires a column '%s' of size '%d', but no such column exists.\n", tablename, elm, size);
} else {
struct ast_str *sql = ast_str_create(100), *fieldtype = ast_str_create(16);
PGresult *res;
if (requirements == RQ_CREATECHAR || type == RQ_CHAR) {
ast_str_set(&fieldtype, 0, "CHAR(%d)", size);
} else if (type == RQ_INTEGER) {
ast_str_set(&fieldtype, 0, "INT%d", size < 5 ? 2 : (size < 11 ? 4 : 8));
} else if (type == RQ_FLOAT) {
ast_str_set(&fieldtype, 0, "FLOAT8");
} else if (type == RQ_DATE) {
ast_str_set(&fieldtype, 0, "DATE");
} else if (type == RQ_DATETIME) {
ast_str_set(&fieldtype, 0, "TIMESTAMP");
} else {
ast_free(sql);
ast_free(fieldtype);
continue;
}
ast_str_set(&sql, 0, "ALTER TABLE %s ADD COLUMN %s %s", tablename, elm, fieldtype->str);
ast_debug(1, "About to lock pgsql_lock (running alter on table '%s' to add column '%s')\n", tablename, elm);
ast_mutex_lock(&pgsql_lock);
if (!pgsql_reconnect(database)) {
ast_mutex_unlock(&pgsql_lock);
ast_log(LOG_ERROR, "Unable to add column: %s\n", sql->str);
ast_free(sql);
ast_free(fieldtype);
continue;
}
ast_debug(1, "About to run ALTER query on table '%s' to add column '%s'\n", tablename, elm);
res = PQexec(pgsqlConn, sql->str);
ast_debug(1, "Finished running ALTER query on table '%s'\n", tablename);
if (PQresultStatus(res) != PGRES_COMMAND_OK) {
ast_log(LOG_ERROR, "Unable to add column: %s\n", sql->str);
}
PQclear(res);
ast_mutex_unlock(&pgsql_lock);
ast_free(sql);
ast_free(fieldtype);
}
}
}
ast_mutex_unlock(&table->lock);
return res;
}
static int unload_pgsql(const char *database, const char *tablename)
{
struct tables *cur;
ast_debug(1, "About to lock table cache list\n");
AST_LIST_LOCK(&psql_tables);
ast_debug(1, "About to traverse table cache list\n");
AST_LIST_TRAVERSE_SAFE_BEGIN(&psql_tables, cur, list) {
if (strcmp(cur->name, tablename) == 0) {
ast_debug(1, "About to remove matching cache entry\n");
AST_LIST_REMOVE_CURRENT(list);
ast_debug(1, "About to destroy matching cache entry\n");
destroy_table(cur);
ast_debug(1, "Cache entry destroyed\n");
break;
}
}
AST_LIST_TRAVERSE_SAFE_END
AST_LIST_UNLOCK(&psql_tables);
ast_debug(1, "About to return\n");
return cur ? 0 : -1;
}
static struct ast_config_engine pgsql_engine = {
.name = "pgsql",
.load_func = config_pgsql,
@@ -759,7 +1042,9 @@ static struct ast_config_engine pgsql_engine = {
.realtime_multi_func = realtime_multi_pgsql,
.store_func = store_pgsql,
.destroy_func = destroy_pgsql,
.update_func = update_pgsql
.update_func = update_pgsql,
.require_func = require_pgsql,
.unload_func = unload_pgsql,
};
static int load_module(void)
@@ -776,6 +1061,7 @@ static int load_module(void)
static int unload_module(void)
{
struct tables *table;
/* Acquire control before doing anything to the module itself. */
ast_mutex_lock(&pgsql_lock);
@@ -787,6 +1073,13 @@ static int unload_module(void)
ast_config_engine_deregister(&pgsql_engine);
ast_verb(1, "PostgreSQL RealTime unloaded.\n");
/* Destroy cached table info */
AST_LIST_LOCK(&psql_tables);
while ((table = AST_LIST_REMOVE_HEAD(&psql_tables, list))) {
destroy_table(table);
}
AST_LIST_UNLOCK(&psql_tables);
/* Unlock so something else can destroy the lock. */
ast_mutex_unlock(&pgsql_lock);
@@ -870,6 +1163,17 @@ static int parse_config(int reload)
} else {
ast_copy_string(dbsock, s, sizeof(dbsock));
}
if (!(s = ast_variable_retrieve(config, "general", "requirements"))) {
ast_log(LOG_WARNING,
"PostgreSQL RealTime: no requirements setting found, using 'warn' as default.\n");
requirements = RQ_WARN;
} else if (!strcasecmp(s, "createclose")) {
requirements = RQ_CREATECLOSE;
} else if (!strcasecmp(s, "createchar")) {
requirements = RQ_CREATECHAR;
}
ast_config_destroy(config);
if (option_debug) {
@@ -943,6 +1247,60 @@ static int pgsql_reconnect(const char *database)
}
}
static char *handle_cli_realtime_pgsql_cache(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
{
struct tables *cur;
int l, which;
char *ret = NULL;
switch (cmd) {
case CLI_INIT:
e->command = "realtime pgsql cache";
e->usage =
"Usage: realtime pgsql cache [<table>]\n"
" Shows table cache for the PostgreSQL RealTime driver\n";
return NULL;
case CLI_GENERATE:
if (a->argc != 3) {
return NULL;
}
l = strlen(a->word);
which = 0;
AST_LIST_LOCK(&psql_tables);
AST_LIST_TRAVERSE(&psql_tables, cur, list) {
if (!strncasecmp(a->word, cur->name, l) && ++which > a->n) {
ret = ast_strdup(cur->name);
break;
}
}
AST_LIST_UNLOCK(&psql_tables);
return ret;
}
if (a->argc == 3) {
/* List of tables */
AST_LIST_LOCK(&psql_tables);
AST_LIST_TRAVERSE(&psql_tables, cur, list) {
ast_cli(a->fd, "%s\n", cur->name);
}
AST_LIST_UNLOCK(&psql_tables);
} else if (a->argc == 4) {
/* List of columns */
if ((cur = find_table(a->argv[3]))) {
struct columns *col;
ast_cli(a->fd, "Columns for Table Cache '%s':\n", a->argv[3]);
ast_cli(a->fd, "%-20.20s %-20.20s %-3.3s %-8.8s\n", "Name", "Type", "Len", "Nullable");
AST_LIST_TRAVERSE(&cur->columns, col, list) {
ast_cli(a->fd, "%-20.20s %-20.20s %3d %-8.8s\n", col->name, col->type, col->len, col->notnull ? "NOT NULL" : "");
}
ast_mutex_unlock(&cur->lock);
} else {
ast_cli(a->fd, "No such table '%s'\n", a->argv[3]);
}
}
return 0;
}
static char *handle_cli_realtime_pgsql_status(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
{
char status[256], status2[100] = "";