retronetworking
/
freeswitch
13
0
Fork 0
Code Pull Requests Releases Activity
freeswitch/src/switch_pgsql.c

926 lines
26 KiB
C
Raw Blame History

/*
* FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
* Copyright (C) 2005-2014, Anthony Minessale II <anthm@freeswitch.org>
*
* Version: MPL 1.1
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
*
* The Initial Developer of the Original Code is
* Anthony Minessale II <anthm@freeswitch.org>
* Portions created by the Initial Developer are Copyright (C)
* the Initial Developer. All Rights Reserved.
*
* Contributor(s):
*
* Anthony Minessale II <anthm@freeswitch.org>
* Eliot Gable <egable@gmail.com>
* Seven Du <dujinfang@gmail.com>
*
* switch_pgsql.c -- PGSQL Driver
*
*/
#include <switch.h>
#ifndef WIN32
#include <switch_private.h>
#endif
#ifdef SWITCH_HAVE_PGSQL
#include <libpq-fe.h>
#ifndef _WIN32
#include <poll.h>
#else
#include <winsock2.h>
#endif
struct switch_pgsql_handle {
char *dsn;
char *sql;
PGconn* con;
int sock;
switch_pgsql_state_t state;
int affected_rows;
int num_retries;
switch_bool_t auto_commit;
switch_bool_t in_txn;
};
struct switch_pgsql_result {
PGresult *result;
ExecStatusType status;
char *err;
int rows;
int cols;
};
#endif
SWITCH_DECLARE(switch_pgsql_handle_t *) switch_pgsql_handle_new(const char *dsn)
{
#ifdef SWITCH_HAVE_PGSQL
switch_pgsql_handle_t *new_handle;
if (!(new_handle = malloc(sizeof(*new_handle)))) {
goto err;
}
memset(new_handle, 0, sizeof(*new_handle));
if (!(new_handle->dsn = strdup(dsn))) {
goto err;
}
new_handle->sock = 0;
new_handle->state = SWITCH_PGSQL_STATE_INIT;
new_handle->con = NULL;
new_handle->affected_rows = 0;
new_handle->num_retries = DEFAULT_PGSQL_RETRIES;
new_handle->auto_commit = SWITCH_TRUE;
new_handle->in_txn = SWITCH_FALSE;
return new_handle;
err:
if (new_handle) {
switch_safe_free(new_handle->dsn);
switch_safe_free(new_handle);
}
#endif
return NULL;
}
#ifdef SWITCH_HAVE_PGSQL
static int db_is_up(switch_pgsql_handle_t *handle)
{
int ret = 0;
switch_event_t *event;
char *err_str = NULL;
int max_tries = DEFAULT_PGSQL_RETRIES;
int code = 0, recon = 0;
if (handle) {
max_tries = handle->num_retries;
if (max_tries < 1)
max_tries = DEFAULT_PGSQL_RETRIES;
}
top:
if (!handle) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "No DB Handle\n");
goto done;
}
if (!handle->con) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "No DB Connection\n");
goto done;
}
/* Try a non-blocking read on the connection to gobble up any EOF from a closed connection and mark the connection BAD if it is closed. */
PQconsumeInput(handle->con);
if (PQstatus(handle->con) == CONNECTION_BAD) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "PQstatus returned bad connection; reconnecting...\n");
handle->state = SWITCH_PGSQL_STATE_ERROR;
PQreset(handle->con);
if (PQstatus(handle->con) == CONNECTION_BAD) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "PQstatus returned bad connection -- reconnection failed!\n");
goto error;
}
handle->state = SWITCH_PGSQL_STATE_CONNECTED;
handle->sock = PQsocket(handle->con);
}
/* if (!PQsendQuery(handle->con, "SELECT 1")) {
code = __LINE__;
goto error;
}
if(switch_pgsql_next_result(handle, &result) == SWITCH_PGSQL_FAIL) {
code = __LINE__;
goto error;
}
if (!result || result->status != PGRES_COMMAND_OK) {
code = __LINE__;
goto error;
}
switch_pgsql_free_result(&result);
switch_pgsql_finish_results(handle);
*/
ret = 1;
goto done;
error:
err_str = switch_pgsql_handle_get_error(handle);
if (PQstatus(handle->con) == CONNECTION_BAD) {
handle->state = SWITCH_PGSQL_STATE_ERROR;
PQreset(handle->con);
if (PQstatus(handle->con) == CONNECTION_OK) {
handle->state = SWITCH_PGSQL_STATE_CONNECTED;
recon = SWITCH_PGSQL_SUCCESS;
handle->sock = PQsocket(handle->con);
}
}
max_tries--;
if (switch_event_create(&event, SWITCH_EVENT_TRAP) == SWITCH_STATUS_SUCCESS) {
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Failure-Message", "The sql server is not responding for DSN %s [%s][%d]",
switch_str_nil(handle->dsn), switch_str_nil(err_str), code);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "The sql server is not responding for DSN %s [%s][%d]\n",
switch_str_nil(handle->dsn), switch_str_nil(err_str), code);
if (recon == SWITCH_PGSQL_SUCCESS) {
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Additional-Info", "The connection has been re-established");
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "The connection has been re-established\n");
} else {
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Additional-Info", "The connection could not be re-established");
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "The connection could not be re-established\n");
}
if (!max_tries) {
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Additional-Info", "Giving up!");
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Giving up!\n");
}
switch_event_fire(&event);
}
if (!max_tries) {
goto done;
}
switch_safe_free(err_str);
switch_yield(1000000);
goto top;
done:
switch_safe_free(err_str);
return ret;
}
#endif
SWITCH_DECLARE(void) switch_pgsql_set_num_retries(switch_pgsql_handle_t *handle, int num_retries)
{
#ifdef SWITCH_HAVE_PGSQL
if (handle) {
handle->num_retries = num_retries;
}
#endif
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_handle_disconnect(switch_pgsql_handle_t *handle)
{
#ifdef SWITCH_HAVE_PGSQL
if (!handle) {
return SWITCH_PGSQL_FAIL;
}
if (handle->state == SWITCH_PGSQL_STATE_CONNECTED) {
PQfinish(handle->con);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG10, "Disconnected from [%s]\n", handle->dsn);
}
switch_safe_free(handle->sql);
handle->state = SWITCH_PGSQL_STATE_DOWN;
return SWITCH_PGSQL_SUCCESS;
#else
return SWITCH_PGSQL_FAIL;
#endif
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_send_query(switch_pgsql_handle_t *handle, const char* sql)
{
#ifdef SWITCH_HAVE_PGSQL
char *err_str;
switch_safe_free(handle->sql);
handle->sql = strdup(sql);
if (!PQsendQuery(handle->con, sql)) {
err_str = switch_pgsql_handle_get_error(handle);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Failed to send query (%s) to database: %s\n", sql, err_str);
switch_safe_free(err_str);
switch_pgsql_finish_results(handle);
goto error;
}
return SWITCH_PGSQL_SUCCESS;
error:
#endif
return SWITCH_PGSQL_FAIL;
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_cancel_real(const char *file, const char *func, int line, switch_pgsql_handle_t *handle)
{
switch_pgsql_status_t ret = SWITCH_PGSQL_SUCCESS;
#ifdef SWITCH_HAVE_PGSQL
char err_buf[256];
PGcancel *cancel = NULL;
memset(err_buf, 0, 256);
cancel = PQgetCancel(handle->con);
if(!PQcancel(cancel, err_buf, 256)) {
switch_log_printf(SWITCH_CHANNEL_ID_LOG, file, func, line, NULL, SWITCH_LOG_CRIT, "Failed to cancel long-running query (%s): %s\n", handle->sql, err_buf);
ret = SWITCH_PGSQL_FAIL;
}
PQfreeCancel(cancel);
switch_pgsql_flush(handle);
#endif
return ret;
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_next_result_timed(switch_pgsql_handle_t *handle, switch_pgsql_result_t **result_out, int msec)
{
#ifdef SWITCH_HAVE_PGSQL
switch_pgsql_result_t *res;
switch_time_t start;
switch_time_t ctime;
unsigned int usec = msec * 1000;
char *err_str;
#ifndef _WIN32
struct pollfd fds[2] = { { 0 } };
#else
fd_set rs, es;
#endif
int poll_res = 0;
if(!handle) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "**BUG** Null handle passed to switch_pgsql_next_result.\n");
return SWITCH_PGSQL_FAIL;
}
/* Try to consume input that might be waiting right away */
if (PQconsumeInput(handle->con)) {
/* And check to see if we have a full result ready for reading */
if (PQisBusy(handle->con)) {
/* Wait for a result to become available, up to msec milliseconds */
start = switch_micro_time_now();
while((ctime = switch_micro_time_now()) - start <= usec) {
int wait_time = (usec - (ctime - start)) / 1000;
/* Wait for the PostgreSQL socket to be ready for data reads. */
#ifndef _WIN32
fds[0].fd = handle->sock;
fds[0].events |= POLLIN;
fds[0].events |= POLLERR;
fds[0].events |= POLLNVAL;
fds[0].events |= POLLHUP;
fds[0].events |= POLLPRI;
fds[0].events |= POLLRDNORM;
fds[0].events |= POLLRDBAND;
poll_res = poll(&fds[0], 1, wait_time);
#else
struct timeval wait = { wait_time * 1000, 0};
FD_ZERO(&rs);
FD_SET(handle->sock, &rs);
FD_ZERO(&es);
FD_SET(handle->sock, &es);
poll_res = select(0, &rs, 0, &es, &wait);
#endif
if (poll_res > 0 ) {
#ifndef _WIN32
if (fds[0].revents & POLLHUP || fds[0].revents & POLLNVAL) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "PGSQL socket closed or invalid while waiting for result for query (%s)\n", handle->sql);
goto error;
} else if (fds[0].revents & POLLERR) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Poll error trying to read PGSQL socket for query (%s)\n", handle->sql);
goto error;
} else if (fds[0].revents & POLLIN || fds[0].revents & POLLPRI || fds[0].revents & POLLRDNORM || fds[0].revents & POLLRDBAND) {
#else
if (FD_ISSET(handle->sock, &rs)) {
#endif
/* Then try to consume any input waiting. */
if (PQconsumeInput(handle->con)) {
if (PQstatus(handle->con) == CONNECTION_BAD) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Connection terminated while waiting for result.\n");
handle->state = SWITCH_PGSQL_STATE_ERROR;
goto error;
}
/* And check to see if we have a full result ready for reading */
if (!PQisBusy(handle->con)) {
/* If we can pull a full result without blocking, then break this loop */
break;
}
} else {
/* If we had an error trying to consume input, report it and cancel the query. */
err_str = switch_pgsql_handle_get_error(handle);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "An error occurred trying to consume input for query (%s): %s\n", handle->sql, err_str);
switch_safe_free(err_str);
switch_pgsql_cancel(handle);
goto error;
}
}
} else if (poll_res == -1) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Poll failed trying to read PGSQL socket for query (%s)\n", handle->sql);
goto error;
}
}
/* If we broke the loop above because of a timeout, report that and cancel the query. */
if (ctime - start > usec) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Query (%s) took too long to complete or database not responding.\n", handle->sql);
switch_pgsql_cancel(handle);
goto error;
}
}
} else {
/* If we had an error trying to consume input, report it and cancel the query. */
err_str = switch_pgsql_handle_get_error(handle);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "An error occurred trying to consume input for query (%s): %s\n", handle->sql, err_str);
switch_safe_free(err_str);
/* switch_pgsql_cancel(handle); */
goto error;
}
/* At this point, we know we can read a full result without blocking. */
if(!(res = malloc(sizeof(switch_pgsql_result_t)))) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Malloc failed!\n");
goto error;
}
memset(res, 0, sizeof(switch_pgsql_result_t));
res->result = PQgetResult(handle->con);
if (res->result) {
*result_out = res;
res->status = PQresultStatus(res->result);
switch(res->status) {
#if (POSTGRESQL_MAJOR_VERSION == 9 && POSTGRESQL_MINOR_VERSION >= 2) || POSTGRESQL_MAJOR_VERSION > 9
case PGRES_SINGLE_TUPLE:
/* Added in PostgreSQL 9.2 */
#endif
case PGRES_TUPLES_OK:
{
res->rows = PQntuples(res->result);
handle->affected_rows = res->rows;
res->cols = PQnfields(res->result);
}
break;
#if (POSTGRESQL_MAJOR_VERSION == 9 && POSTGRESQL_MINOR_VERSION >= 1) || POSTGRESQL_MAJOR_VERSION > 9
case PGRES_COPY_BOTH:
/* Added in PostgreSQL 9.1 */
#endif
case PGRES_COPY_OUT:
case PGRES_COPY_IN:
case PGRES_COMMAND_OK:
break;
case PGRES_EMPTY_QUERY:
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Query (%s) returned PGRES_EMPTY_QUERY\n", handle->sql);
case PGRES_BAD_RESPONSE:
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Query (%s) returned PGRES_BAD_RESPONSE\n", handle->sql);
case PGRES_NONFATAL_ERROR:
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Query (%s) returned PGRES_NONFATAL_ERROR\n", handle->sql);
case PGRES_FATAL_ERROR:
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Query (%s) returned PGRES_FATAL_ERROR\n", handle->sql);
res->err = PQresultErrorMessage(res->result);
goto error;
break;
}
} else {
free(res);
res = NULL;
*result_out = NULL;
}
return SWITCH_PGSQL_SUCCESS;
error:
/* Make sure the failed connection does not have any transactions marked as in progress */
switch_pgsql_flush(handle);
/* Try to reconnect to the DB if we were dropped */
db_is_up(handle);
#endif
return SWITCH_PGSQL_FAIL;
}
SWITCH_DECLARE(void) switch_pgsql_free_result(switch_pgsql_result_t **result)
{
#ifdef SWITCH_HAVE_PGSQL
if (!*result) {
return;
}
if ((*result)->result) {
PQclear((*result)->result);
}
free(*result);
*result = NULL;
#else
return;
#endif
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_finish_results_real(const char* file, const char* func, int line, switch_pgsql_handle_t *handle)
{
#ifdef SWITCH_HAVE_PGSQL
switch_pgsql_result_t *res = NULL;
switch_pgsql_status_t final_status = SWITCH_PGSQL_SUCCESS;
int done = 0;
do {
switch_pgsql_next_result(handle, &res);
if (res && res->err && !switch_stristr("already exists", res->err) && !switch_stristr("duplicate key name", res->err)) {
switch_log_printf(SWITCH_CHANNEL_ID_LOG, file, func, line, NULL, SWITCH_LOG_ERROR, "Error executing query:\n%s\n", res->err);
final_status = SWITCH_PGSQL_FAIL;
}
if (!res) {
done = 1;
} else if (res->result) {
char *affected_rows = PQcmdTuples(res->result);
if (!zstr(affected_rows)) {
handle->affected_rows = atoi(affected_rows);
}
}
switch_pgsql_free_result(&res);
} while (!done);
return final_status;
#else
return SWITCH_PGSQL_FAIL;
#endif
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_handle_connect(switch_pgsql_handle_t *handle)
{
#ifdef SWITCH_HAVE_PGSQL
if (handle->state == SWITCH_PGSQL_STATE_CONNECTED) {
switch_pgsql_handle_disconnect(handle);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG1, "Re-connecting %s\n", handle->dsn);
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG1, "Connecting %s\n", handle->dsn);
PQinitSSL(0);
handle->con = PQconnectdb(handle->dsn);
if (PQstatus(handle->con) != CONNECTION_OK) {
char *err_str;
if ((err_str = switch_pgsql_handle_get_error(handle))) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "%s\n", err_str);
switch_safe_free(err_str);
} else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to connect to the database [%s]\n", handle->dsn);
switch_pgsql_handle_disconnect(handle);
}
return SWITCH_PGSQL_FAIL;
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG1, "Connected to [%s]\n", handle->dsn);
handle->state = SWITCH_PGSQL_STATE_CONNECTED;
handle->sock = PQsocket(handle->con);
return SWITCH_PGSQL_SUCCESS;
#else
return SWITCH_PGSQL_FAIL;
#endif
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_handle_exec_string_detailed(const char *file, const char *func, int line,
switch_pgsql_handle_t *handle, const char *sql, char *resbuf, size_t len, char **err)
{
#ifdef SWITCH_HAVE_PGSQL
switch_pgsql_status_t sstatus = SWITCH_PGSQL_SUCCESS;
char *val = NULL;
switch_pgsql_result_t *result = NULL;
handle->affected_rows = 0;
if (switch_pgsql_handle_exec_base_detailed(file, func, line, handle, sql, err) == SWITCH_PGSQL_FAIL) {
goto error;
}
if(switch_pgsql_next_result(handle, &result) == SWITCH_PGSQL_FAIL) {
goto error;
}
if (result) {
switch (result->status) {
#if POSTGRESQL_MAJOR_VERSION >= 9 && POSTGRESQL_MINOR_VERSION >= 2
case PGRES_SINGLE_TUPLE:
/* Added in PostgreSQL 9.2 */
#endif
case PGRES_COMMAND_OK:
case PGRES_TUPLES_OK:
break;
default:
sstatus = SWITCH_PGSQL_FAIL;
goto done;
}
}
if (handle->affected_rows <= 0) {
goto done;
}
val = PQgetvalue(result->result, 0, 0);
strncpy(resbuf, val, len);
done:
switch_pgsql_free_result(&result);
if (switch_pgsql_finish_results(handle) != SWITCH_PGSQL_SUCCESS) {
sstatus = SWITCH_PGSQL_FAIL;
}
return sstatus;
error:
return SWITCH_PGSQL_FAIL;
#else
return SWITCH_PGSQL_FAIL;
#endif
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_handle_exec_base_detailed(const char *file, const char *func, int line,
switch_pgsql_handle_t *handle, const char *sql, char **err)
{
#ifdef SWITCH_HAVE_PGSQL
char *err_str = NULL, *er = NULL;
switch_pgsql_flush(handle);
handle->affected_rows = 0;
if (!db_is_up(handle)) {
er = strdup("Database is not up!");
goto error;
}
if (handle->auto_commit == SWITCH_FALSE && handle->in_txn == SWITCH_FALSE) {
if (switch_pgsql_send_query(handle, "BEGIN") != SWITCH_PGSQL_SUCCESS) {
er = strdup("Error sending BEGIN!");
if (switch_pgsql_finish_results(handle) != SWITCH_PGSQL_SUCCESS) {
db_is_up(handle); /* If finish_results failed, maybe the db went dead */
}
goto error;
}
if (switch_pgsql_finish_results(handle) != SWITCH_PGSQL_SUCCESS) {
db_is_up(handle);
er = strdup("Error sending BEGIN!");
goto error;
}
handle->in_txn = SWITCH_TRUE;
}
if (switch_pgsql_send_query(handle, sql) != SWITCH_PGSQL_SUCCESS) {
er = strdup("Error sending query!");
if (switch_pgsql_finish_results(handle) != SWITCH_PGSQL_SUCCESS) {
db_is_up(handle);
}
goto error;
}
return SWITCH_PGSQL_SUCCESS;
error:
err_str = switch_pgsql_handle_get_error(handle);
if (zstr(err_str)) {
if (zstr(er)) {
err_str = strdup((char *)"SQL ERROR!");
} else {
err_str = er;
}
} else {
if (!zstr(er)) {
free(er);
}
}
if (err_str) {
if (!switch_stristr("already exists", err_str) && !switch_stristr("duplicate key name", err_str)) {
switch_log_printf(SWITCH_CHANNEL_ID_LOG, file, func, line, NULL, SWITCH_LOG_ERROR, "ERR: [%s]\n[%s]\n", sql, switch_str_nil(err_str));
}
if (err) {
*err = err_str;
} else {
free(err_str);
}
}
#endif
return SWITCH_PGSQL_FAIL;
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_handle_exec_detailed(const char *file, const char *func, int line,
switch_pgsql_handle_t *handle, const char *sql, char **err)
{
#ifdef SWITCH_HAVE_PGSQL
if (switch_pgsql_handle_exec_base_detailed(file, func, line, handle, sql, err) == SWITCH_PGSQL_FAIL) {
goto error;
}
return switch_pgsql_finish_results(handle);
error:
#endif
return SWITCH_PGSQL_FAIL;
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_handle_callback_exec_detailed(const char *file, const char *func, int line,
switch_pgsql_handle_t *handle,
const char *sql, switch_core_db_callback_func_t callback, void *pdata,
char **err)
{
#ifdef SWITCH_HAVE_PGSQL
char *err_str = NULL;
int row = 0, col = 0, err_cnt = 0;
switch_pgsql_result_t *result = NULL;
handle->affected_rows = 0;
switch_assert(callback != NULL);
if (switch_pgsql_handle_exec_base(handle, sql, err) == SWITCH_PGSQL_FAIL) {
goto error;
}
if (switch_pgsql_next_result(handle, &result) == SWITCH_PGSQL_FAIL) {
err_cnt++;
err_str = switch_pgsql_handle_get_error(handle);
if (result && !zstr(result->err)) {
switch_log_printf(SWITCH_CHANNEL_ID_LOG, file, func, line, NULL, SWITCH_LOG_ERROR, "ERR: [%s]\n[%s]\n", sql, switch_str_nil(result->err));
}
if (!zstr(err_str)) {
switch_log_printf(SWITCH_CHANNEL_ID_LOG, file, func, line, NULL, SWITCH_LOG_ERROR, "ERR: [%s]\n[%s]\n", sql, switch_str_nil(err_str));
}
switch_safe_free(err_str);
err_str = NULL;
}
while (result != NULL) {
/*switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Processing result with %d rows and %d columns.\n", result->rows, result->cols);*/
for (row = 0; row < result->rows; ++row) {
char **names;
char **vals;
names = calloc(result->cols, sizeof(*names));
vals = calloc(result->cols, sizeof(*vals));
switch_assert(names && vals);
for (col = 0; col < result->cols; ++col) {
char * tmp;
int len;
tmp = PQfname(result->result, col);
if (tmp) {
len = strlen(tmp);
names[col] = malloc(len+1);
names[col][len] = '\0';
strncpy(names[col], tmp, len);
len = PQgetlength(result->result, row, col);
vals[col] = malloc(len+1);
vals[col][len] = '\0';
tmp = PQgetvalue(result->result, row, col);
strncpy(vals[col], tmp, len);
/*switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Processing result row %d, col %d: %s => %s\n", row, col, names[col], vals[col]);*/
} else {
/*switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Processing result row %d, col %d.\n", row, col);*/
switch_log_printf(SWITCH_CHANNEL_ID_LOG, file, func, line, NULL, SWITCH_LOG_ERROR, "ERR: Column number %d out of range\n", col);
}
}
/*switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Executing callback for row %d...\n", row);*/
if (callback(pdata, result->cols, vals, names)) {
switch_pgsql_finish_results(handle); /* Makes sure next call to switch_pgsql_next_result will return NULL */
row = result->rows; /* Makes us exit the for loop */
}
for (col = 0; col < result->cols; ++col) {
free(names[col]);
free(vals[col]);
}
free(names);
free(vals);
}
switch_pgsql_free_result(&result);
if (switch_pgsql_next_result(handle, &result) == SWITCH_PGSQL_FAIL) {
err_cnt++;
err_str = switch_pgsql_handle_get_error(handle);
if (result && !zstr(result->err)) {
switch_log_printf(SWITCH_CHANNEL_ID_LOG, file, func, line, NULL, SWITCH_LOG_ERROR, "ERR: [%s]\n[%s]\n", sql, switch_str_nil(result->err));
}
if (!zstr(err_str)) {
switch_log_printf(SWITCH_CHANNEL_ID_LOG, file, func, line, NULL, SWITCH_LOG_ERROR, "ERR: [%s]\n[%s]\n", sql, switch_str_nil(err_str));
}
switch_safe_free(err_str);
err_str = NULL;
}
}
if (err_cnt) {
goto error;
}
return SWITCH_PGSQL_SUCCESS;
error:
#endif
return SWITCH_PGSQL_FAIL;
}
SWITCH_DECLARE(void) switch_pgsql_handle_destroy(switch_pgsql_handle_t **handlep)
{
#ifdef SWITCH_HAVE_PGSQL
switch_pgsql_handle_t *handle = NULL;
if (!handlep) {
return;
}
handle = *handlep;
if (handle) {
switch_pgsql_handle_disconnect(handle);
switch_safe_free(handle->dsn);
free(handle);
}
*handlep = NULL;
#else
return;
#endif
}
SWITCH_DECLARE(switch_pgsql_state_t) switch_pgsql_handle_get_state(switch_pgsql_handle_t *handle)
{
#ifdef SWITCH_HAVE_PGSQL
return handle ? handle->state : SWITCH_PGSQL_STATE_INIT;
#else
return SWITCH_PGSQL_STATE_ERROR;
#endif
}
SWITCH_DECLARE(char *) switch_pgsql_handle_get_error(switch_pgsql_handle_t *handle)
{
#ifdef SWITCH_HAVE_PGSQL
char * err_str;
if (!handle) {
return NULL;
};
switch_strdup(err_str, PQerrorMessage(handle->con));
return err_str;
#else
return NULL;
#endif
}
SWITCH_DECLARE(int) switch_pgsql_handle_affected_rows(switch_pgsql_handle_t *handle)
{
#ifdef SWITCH_HAVE_PGSQL
return handle->affected_rows;
#else
return 0;
#endif
}
SWITCH_DECLARE(switch_bool_t) switch_pgsql_available(void)
{
#ifdef SWITCH_HAVE_PGSQL
return SWITCH_TRUE;
#else
return SWITCH_FALSE;
#endif
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_SQLSetAutoCommitAttr(switch_pgsql_handle_t *handle, switch_bool_t on)
{
#ifdef SWITCH_HAVE_PGSQL
if (on) {
handle->auto_commit = SWITCH_TRUE;
} else {
handle->auto_commit = SWITCH_FALSE;
}
return SWITCH_PGSQL_SUCCESS;
#else
return (switch_pgsql_status_t) SWITCH_FALSE;
#endif
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_flush(switch_pgsql_handle_t *handle)
{
#ifdef SWITCH_HAVE_PGSQL
PGresult *tmp = NULL;
int x = 0;
/* Make sure the query is fully cleared */
while ((tmp = PQgetResult(handle->con)) != NULL) {
PQclear(tmp);
x++;
}
if (x) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG10, "Flushing %d results\n", x);
}
return SWITCH_PGSQL_SUCCESS;
#else
return (switch_pgsql_status_t) SWITCH_FALSE;
#endif
}
SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_SQLEndTran(switch_pgsql_handle_t *handle, switch_bool_t commit)
{
#ifdef SWITCH_HAVE_PGSQL
char * err_str = NULL;
if (commit) {
if(!PQsendQuery(handle->con, "COMMIT")) {
err_str = switch_pgsql_handle_get_error(handle);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Could not commit transaction: %s\n", err_str);
switch_safe_free(err_str);
return SWITCH_PGSQL_FAIL;
}
} else {
if(!PQsendQuery(handle->con, "ROLLBACK")) {
err_str = switch_pgsql_handle_get_error(handle);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Could not rollback transaction: %s\n", err_str);
switch_safe_free(err_str);
return SWITCH_PGSQL_FAIL;
}
}
handle->in_txn = SWITCH_FALSE;
return SWITCH_PGSQL_SUCCESS;
#else
return (switch_pgsql_status_t) SWITCH_FALSE;
#endif
}
/* For Emacs:
* Local Variables:
* mode:c
* indent-tabs-mode:t
* tab-width:4
* c-basic-offset:4
* End:
* For VIM:
* vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
*/
View Git Blame Copy Permalink