[Freeswitch-svn] [commit] r10277 - freeswitch/trunk/src/mod/applications/mod_fifo
Freeswitch SVN
anthm at freeswitch.org
Fri Nov 7 11:41:50 PST 2008
Author: anthm
Date: Thu Nov 6 19:25:51 2008
New Revision: 10277
Modified:
freeswitch/trunk/src/mod/applications/mod_fifo/mod_fifo.c
Log:
add some stuff to mod_fifo
Modified: freeswitch/trunk/src/mod/applications/mod_fifo/mod_fifo.c
==============================================================================
--- freeswitch/trunk/src/mod/applications/mod_fifo/mod_fifo.c (original)
+++ freeswitch/trunk/src/mod/applications/mod_fifo/mod_fifo.c Thu Nov 6 19:25:51 2008
@@ -36,8 +36,10 @@
#define FIFO_EVENT "fifo::info"
+static switch_status_t load_config(int reload, int del_all);
#define MAX_PRI 10
+
struct fifo_node {
char *name;
switch_mutex_t *mutex;
@@ -48,6 +50,9 @@
int consumer_count;
switch_time_t start_waiting;
uint32_t importance;
+ switch_thread_rwlock_t *rwlock;
+ switch_memory_pool_t *pool;
+ int has_outbound;
};
typedef struct fifo_node fifo_node_t;
@@ -240,11 +245,116 @@
static struct {
switch_hash_t *fifo_hash;
switch_mutex_t *mutex;
+ switch_mutex_t *sql_mutex;
switch_memory_pool_t *pool;
int running;
switch_event_node_t *node;
+ char hostname[256];
+ char *dbname;
+ char *odbc_dsn;
+ int node_thread_running;
+
+#ifdef SWITCH_HAVE_ODBC
+ switch_odbc_handle_t *master_odbc;
+#else
+ void *filler1;
+#endif
} globals;
+
+static switch_status_t fifo_execute_sql(char *sql, switch_mutex_t *mutex)
+{
+ switch_core_db_t *db;
+ switch_status_t status = SWITCH_STATUS_SUCCESS;
+ char *err_str;
+
+ if (mutex) {
+ switch_mutex_lock(mutex);
+ }
+#ifdef SWITCH_HAVE_ODBC
+ if (globals.odbc_dsn) {
+ SQLHSTMT stmt;
+ if (switch_odbc_handle_exec(globals.master_odbc, sql, &stmt) != SWITCH_ODBC_SUCCESS) {
+
+ err_str = switch_odbc_handle_get_error(globals.master_odbc, stmt);
+ if (!switch_strlen_zero(err_str)) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "ERR: [%s]\n[%s]\n", sql, switch_str_nil(err_str));
+ }
+ switch_safe_free(err_str);
+ status = SWITCH_STATUS_FALSE;
+ }
+ SQLFreeHandle(SQL_HANDLE_STMT, stmt);
+ } else {
+#endif
+ if (!(db = switch_core_db_open_file(globals.dbname))) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error Opening DB %s\n", globals.dbname);
+ status = SWITCH_STATUS_FALSE;
+ goto end;
+ }
+
+ err_str = NULL;
+ switch_core_db_exec(db, sql, NULL, NULL, &err_str);
+ if (err_str) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error [%s]\n[%s]\n", sql, err_str);
+ free(err_str);
+ }
+
+ switch_core_db_close(db);
+
+#ifdef SWITCH_HAVE_ODBC
+ }
+#endif
+
+ end:
+ if (mutex) {
+ switch_mutex_unlock(mutex);
+ }
+
+ return status;
+}
+
+static switch_bool_t fifo_execute_sql_callback(switch_mutex_t *mutex, char *sql, switch_core_db_callback_func_t callback, void *pdata)
+{
+ switch_bool_t ret = SWITCH_FALSE;
+ switch_core_db_t *db;
+ char *errmsg = NULL;
+
+ if (mutex) {
+ switch_mutex_lock(mutex);
+ }
+
+#ifdef SWITCH_HAVE_ODBC
+ if (globals.odbc_dsn) {
+ switch_odbc_handle_callback_exec(globals.master_odbc, sql, callback, pdata);
+ } else {
+#endif
+ if (!(db = switch_core_db_open_file(globals.dbname))) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error Opening DB %s\n", globals.dbname);
+ goto end;
+ }
+
+ switch_core_db_exec(db, sql, callback, pdata, &errmsg);
+
+ if (errmsg) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "SQL ERR: [%s] %s\n", sql, errmsg);
+ free(errmsg);
+ }
+
+ if (db) {
+ switch_core_db_close(db);
+ }
+#ifdef SWITCH_HAVE_ODBC
+ }
+#endif
+
+ end:
+ if (mutex) {
+ switch_mutex_unlock(mutex);
+ }
+
+ return ret;
+}
+
static fifo_node_t *create_node(const char *name, uint32_t importance)
{
fifo_node_t *node;
@@ -255,20 +365,221 @@
}
node = switch_core_alloc(globals.pool, sizeof(*node));
- node->name = switch_core_strdup(globals.pool, name);
+ node->pool = globals.pool;
+
+ node->name = switch_core_strdup(node->pool, name);
for (x = 0; x < MAX_PRI; x++) {
- switch_queue_create(&node->fifo_list[x], SWITCH_CORE_QUEUE_LEN, globals.pool);
+ switch_queue_create(&node->fifo_list[x], SWITCH_CORE_QUEUE_LEN, node->pool);
switch_assert(node->fifo_list[x]);
}
- switch_core_hash_init(&node->caller_hash, globals.pool);
- switch_core_hash_init(&node->consumer_hash, globals.pool);
-
- switch_mutex_init(&node->mutex, SWITCH_MUTEX_NESTED, globals.pool);
+ switch_core_hash_init(&node->caller_hash, node->pool);
+ switch_core_hash_init(&node->consumer_hash, node->pool);
+ switch_thread_rwlock_create(&node->rwlock, node->pool);
+ switch_mutex_init(&node->mutex, SWITCH_MUTEX_NESTED, node->pool);
+ node->importance = importance;
switch_core_hash_insert(globals.fifo_hash, name, node);
+ return node;
+}
- node->importance = importance;
+static int node_idle_consumers(fifo_node_t *node)
+{
+ switch_hash_index_t *hi;
+ void *val;
+ const void *var;
+ switch_core_session_t *session;
+ switch_channel_t *channel;
+ int total = 0;
- return node;
+ switch_mutex_lock(node->mutex);
+ for (hi = switch_hash_first(NULL, node->consumer_hash); hi; hi = switch_hash_next(hi)) {
+ switch_hash_this(hi, &var, NULL, &val);
+ session = (switch_core_session_t *) val;
+ channel = switch_core_session_get_channel(session);
+ if (!switch_channel_test_flag(channel, CF_BRIDGED)) {
+ total++;
+ }
+ }
+ switch_mutex_unlock(node->mutex);
+
+ return total;
+
+}
+
+static switch_status_t hanguphook(switch_core_session_t *session)
+{
+ switch_channel_t *channel = switch_core_session_get_channel(session);
+ switch_channel_state_t state = switch_channel_get_state(channel);
+ const char *uuid = NULL;
+ char sql[256] = "";
+
+ if (state == CS_HANGUP || state == CS_ROUTING) {
+ if ((uuid = switch_channel_get_variable(channel, "fifo_outbound_uuid"))) {
+ switch_snprintf(sql, sizeof(sql), "update fifo_outbound set use_count=use_count-1, outbound_call_count=outbound_call_count+1, next_avail=%ld + lag where uuid='%s'",
+ (long)switch_timestamp(NULL), uuid);
+
+ fifo_execute_sql(sql, globals.sql_mutex);
+ }
+ switch_core_event_hook_remove_state_change(session, hanguphook);
+ }
+ return SWITCH_STATUS_SUCCESS;
+}
+
+struct call_helper {
+ char *uuid;
+ char *node_name;
+ char *originate_string;
+ int timeout;
+ switch_memory_pool_t *pool;
+};
+
+static void *SWITCH_THREAD_FUNC o_thread_run(switch_thread_t *thread, void *obj)
+{
+ struct call_helper *h = (struct call_helper *) obj;
+
+ switch_core_session_t *session = NULL;
+ switch_channel_t *channel;
+ switch_call_cause_t cause = SWITCH_CAUSE_NONE;
+ switch_caller_extension_t *extension = NULL;
+ char *app_name, *arg = NULL;
+ char sql[256] = "";
+
+ switch_snprintf(sql, sizeof(sql), "update fifo_outbound set use_count=use_count+1 where uuid='%s'", h->uuid);
+ fifo_execute_sql(sql, globals.sql_mutex);
+
+ if (switch_ivr_originate(NULL, &session, &cause, h->originate_string, h->timeout, NULL, NULL, NULL, NULL, SOF_NONE) != SWITCH_STATUS_SUCCESS) {
+ switch_snprintf(sql, sizeof(sql), "update fifo_outbound set use_count=use_count-1, outbound_fail_count=outbound_fail_count+1, next_avail=%ld + lag where uuid='%s'",
+ (long)switch_timestamp(NULL), h->uuid);
+ fifo_execute_sql(sql, globals.sql_mutex);
+ goto end;
+ }
+
+
+ channel = switch_core_session_get_channel(session);
+ switch_channel_set_variable(channel, "fifo_outbound_uuid", h->uuid);
+ switch_core_event_hook_add_state_change(session, hanguphook);
+ app_name = "fifo";
+ arg = switch_core_session_sprintf(session, "%s out wait", h->node_name);
+ extension = switch_caller_extension_new(session, app_name, arg);
+ switch_caller_extension_add_application(session, extension, app_name, arg);
+ switch_channel_set_caller_extension(channel, extension);
+ switch_channel_set_state(channel, CS_EXECUTE);
+ switch_core_session_rwunlock(session);
+
+ end:
+
+ switch_core_destroy_memory_pool(&h->pool);
+
+ return NULL;
+}
+
+static int place_call_callback(void *pArg, int argc, char **argv, char **columnNames)
+{
+
+ int *need = (int *) pArg;
+
+ switch_thread_t *thread;
+ switch_threadattr_t *thd_attr = NULL;
+ switch_memory_pool_t *pool;
+ struct call_helper *h;
+
+ switch_core_new_memory_pool(&pool);
+ h = switch_core_alloc(pool, sizeof(*h));
+ h->pool = pool;
+ h->uuid = switch_core_strdup(h->pool, argv[0]);
+ h->node_name = switch_core_strdup(h->pool, argv[1]);
+ h->originate_string = switch_core_strdup(h->pool, argv[2]);
+ h->timeout = atoi(argv[5]);
+
+
+ switch_threadattr_create(&thd_attr, h->pool);
+ switch_threadattr_detach_set(thd_attr, 1);
+ switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
+ switch_thread_create(&thread, thd_attr, o_thread_run, h, h->pool);
+
+ (*need)--;
+
+ return *need ? 0 : -1;
+}
+
+static void find_consumers(fifo_node_t *node)
+{
+ int need = node_consumer_wait_count(node);
+ char *sql;
+
+ sql = switch_mprintf("select uuid, fifo_name, originate_string, simo_count, use_count, timeout, lag, next_avail, expires, static, outbound_call_count, outbound_fail_count, hostname "
+ "from fifo_outbound where (use_count < simo_count) and (next_avail = 0 or next_avail <= %ld) order by outbound_call_count",
+ (long) switch_timestamp(NULL));
+ switch_assert(sql);
+ fifo_execute_sql_callback(globals.sql_mutex, sql, place_call_callback, &need);
+ free(sql);
+}
+
+static void *SWITCH_THREAD_FUNC node_thread_run(switch_thread_t *thread, void *obj)
+{
+ fifo_node_t *node;
+
+ globals.node_thread_running = 1;
+
+ while(globals.node_thread_running == 1) {
+ switch_hash_index_t *hi;
+ void *val;
+ const void *var;
+ int ppl_waiting, consumer_total, idle_consumers;
+
+ switch_mutex_lock(globals.mutex);
+ for (hi = switch_hash_first(NULL, globals.fifo_hash); hi; hi = switch_hash_next(hi)) {
+ switch_hash_this(hi, &var, NULL, &val);
+ if ((node = (fifo_node_t *) val) && node->has_outbound) {
+ switch_mutex_lock(node->mutex);
+ ppl_waiting = node_consumer_wait_count(node);
+ consumer_total = node->consumer_count;
+ idle_consumers = node_idle_consumers(node);
+
+ //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,
+ //"%s waiting %d consumer_total %d idle_consumers %d\n", node->name, ppl_waiting, consumer_total, idle_consumers);
+
+ if (ppl_waiting && (!consumer_total || !idle_consumers)) {
+ find_consumers(node);
+ }
+ switch_mutex_unlock(node->mutex);
+ }
+ }
+ switch_mutex_unlock(globals.mutex);
+
+ switch_yield(1000000);
+ }
+
+ globals.node_thread_running = 0;
+
+ return NULL;
+}
+
+static void start_node_thread(switch_memory_pool_t *pool)
+{
+ switch_thread_t *thread;
+ switch_threadattr_t *thd_attr = NULL;
+
+ switch_threadattr_create(&thd_attr, pool);
+ switch_threadattr_detach_set(thd_attr, 1);
+ switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
+ switch_thread_create(&thread, thd_attr, node_thread_run, pool, pool);
+}
+
+static int stop_node_thread(void)
+{
+ int sanity = 20;
+
+ if (globals.node_thread_running) {
+ globals.node_thread_running = -1;
+ while(globals.node_thread_running) {
+ switch_yield(500000);
+ if (!--sanity) {
+ return -1;
+ }
+ }
+ }
+
+ return 0;
}
static void send_presence(fifo_node_t *node)
@@ -509,6 +820,14 @@
node->start_waiting = switch_timestamp_now();
}
+ if (switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, FIFO_EVENT) == SWITCH_STATUS_SUCCESS) {
+ switch_channel_event_set_data(channel, event);
+ switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "FIFO-Name", argv[0]);
+ switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "FIFO-Action", "push");
+ switch_event_add_header(event, SWITCH_STACK_BOTTOM, "FIFO-Slot", "%d", p);
+ switch_event_fire(&event);
+ }
+
switch_queue_push(node->fifo_list[p], (void *) strdup(uuid));
if (!pri) {
@@ -524,14 +843,6 @@
switch_channel_set_variable(channel, "fifo_status", "WAITING");
switch_channel_set_variable(channel, "fifo_timestamp", date);
- if (switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, FIFO_EVENT) == SWITCH_STATUS_SUCCESS) {
- switch_channel_event_set_data(channel, event);
- switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "FIFO-Name", argv[0]);
- switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "FIFO-Action", "push");
- switch_event_add_header(event, SWITCH_STACK_BOTTOM, "FIFO-Slot", "%d", p);
- switch_event_fire(&event);
- }
-
switch_channel_set_flag(channel, CF_TAGGED);
if (chime_list) {
@@ -610,6 +921,7 @@
if (!aborted && switch_channel_ready(channel)) {
switch_channel_set_state(channel, CS_HIBERNATE);
+ return;
} else {
ts = switch_timestamp_now();
switch_time_exp_lt(&tm, ts);
@@ -866,7 +1178,26 @@
}
if (!(switch_channel_ready(channel))) {
- switch_channel_hangup(other_channel, SWITCH_CAUSE_NORMAL_CLEARING);
+ const char *app = switch_channel_get_variable(other_channel, "current_application");
+ const char *arg = switch_channel_get_variable(other_channel, "current_application_data");
+ switch_caller_extension_t *extension = NULL;
+
+ switch_mutex_lock(node->mutex);
+ node->caller_count--;
+ switch_core_hash_delete(node->caller_hash, uuid);
+ switch_mutex_unlock(node->mutex);
+ send_presence(node);
+
+
+ if (app) {
+ extension = switch_caller_extension_new(other_session, app, arg);
+ switch_caller_extension_add_application(other_session, extension, app, arg);
+ switch_channel_set_caller_extension(other_channel, extension);
+ switch_channel_set_state(other_channel, CS_EXECUTE);
+ } else {
+ switch_channel_hangup(other_channel, SWITCH_CAUSE_NORMAL_CLEARING);
+ }
+
switch_core_session_rwunlock(other_session);
break;
}
@@ -933,7 +1264,10 @@
sfifo_consumer_wrapup_time = switch_channel_get_variable(channel, "fifo_consumer_wrapup_time");
if (!switch_strlen_zero(sfifo_consumer_wrapup_time)){
fifo_consumer_wrapup_time = atoi(sfifo_consumer_wrapup_time);
+ } else {
+ fifo_consumer_wrapup_time = 5000;
}
+
memset(buf, 0, sizeof(buf));
if (fifo_consumer_wrapup_time || !switch_strlen_zero(fifo_consumer_wrapup_key)) {
@@ -976,7 +1310,7 @@
}
}
- } else if (fifo_consumer_wrapup_time && !strcmp(buf, fifo_consumer_wrapup_key)) {
+ } else if (fifo_consumer_wrapup_time && (switch_strlen_zero(fifo_consumer_wrapup_key) || !strcmp(buf, fifo_consumer_wrapup_key))) {
while(switch_channel_ready(channel)) {
wrapup_time_elapsed = (switch_timestamp_now() - wrapup_time_started) / 1000;
if (wrapup_time_elapsed > fifo_consumer_wrapup_time) {
@@ -1025,6 +1359,67 @@
}
}
+struct xml_helper {
+ switch_xml_t xml;
+ fifo_node_t *node;
+ char *container;
+ char *tag;
+ int cc_off;
+ int verbose;
+};
+
+static int xml_callback(void *pArg, int argc, char **argv, char **columnNames)
+{
+ struct xml_helper *h = (struct xml_helper *) pArg;
+ switch_xml_t x_tmp, x_out;
+ int c_off = 0;
+ char exp_buf[128] = "";
+ switch_time_exp_t tm;
+ switch_time_t etime = 0;
+
+ if (argv[7]) {
+ if ((etime = atol(argv[7]))) {
+ switch_size_t retsize;
+
+ switch_time_exp_lt(&tm, switch_time_from_sec(etime));
+ switch_strftime_nocheck(exp_buf, &retsize, sizeof(exp_buf), "%Y-%m-%d %T", &tm);
+ } else {
+ switch_set_string(exp_buf, "now");
+ }
+ }
+
+ x_tmp = switch_xml_add_child_d(h->xml, h->container, h->cc_off++);
+
+ x_out = switch_xml_add_child_d(x_tmp, h->tag, c_off++);
+ switch_xml_set_attr_d(x_out, "timeout", argv[5]);
+ switch_xml_set_attr_d(x_out, "simo", argv[3]);
+ switch_xml_set_attr_d(x_out, "lag", argv[6]);
+ switch_xml_set_attr_d(x_out, "outbound-call-count", argv[10]);
+ switch_xml_set_attr_d(x_out, "outbound-fail-count", argv[11]);
+ switch_xml_set_attr_d(x_out, "next-available", exp_buf);
+
+ switch_xml_set_txt_d(x_out, argv[2]);
+
+ return 0;
+}
+
+static int xml_outbound(switch_xml_t xml, fifo_node_t *node, char *container, char *tag, int cc_off, int verbose)
+{
+ struct xml_helper h;
+ char *sql = "select uuid, fifo_name, originate_string, simo_count, use_count, timeout, lag, next_avail, expires, static, outbound_call_count, outbound_fail_count, hostname from fifo_outbound";
+
+ h.xml = xml;
+ h.node = node;
+ h.container = container;
+ h.tag = tag;
+ h.cc_off = cc_off;
+ h.verbose = verbose;
+
+ fifo_execute_sql_callback(globals.sql_mutex, sql, xml_callback, &h);
+
+ return h.cc_off;
+}
+
static int xml_hash(switch_xml_t xml, switch_hash_t *hash, char *container, char *tag, int cc_off, int verbose)
{
switch_xml_t x_tmp, x_caller, x_cp, variables;
@@ -1098,7 +1493,8 @@
switch_xml_set_attr_d(x_fifo, "waiting_count", tmp);
switch_snprintf(tmp, sizeof(buffer), "%u", node->importance);
switch_xml_set_attr_d(x_fifo, "importance", tmp);
-
+
+ cc_off = xml_outbound(x_fifo, node, "outbound", "member", cc_off, verbose);
cc_off = xml_hash(x_fifo, node->caller_hash, "callers", "caller", cc_off, verbose);
cc_off = xml_hash(x_fifo, node->consumer_hash, "consumers", "consumer", cc_off, verbose);
}
@@ -1133,6 +1529,10 @@
switch_mutex_lock(globals.mutex);
verbose = !strcasecmp(argv[0], "list_verbose");
+ if (!strcasecmp(argv[0], "reparse")) {
+ load_config(1, argv[1] && !strcasecmp(argv[1], "del_all"));
+ }
+
if (!strcasecmp(argv[0], "list") || verbose) {
char *xml_text = NULL;
switch_xml_t x_report = switch_xml_new("fifo_report");
@@ -1202,6 +1602,310 @@
return SWITCH_STATUS_SUCCESS;
}
+
+const char outbound_sql[] =
+ "create table fifo_outbound (\n"
+ " uuid varchar(255),\n"
+ " fifo_name varchar(255),\n"
+ " originate_string varchar(255),\n"
+ " simo_count integer,\n"
+ " use_count integer,\n"
+ " timeout integer,\n"
+ " lag integer,\n"
+ " next_avail integer,\n"
+ " expires integer,\n"
+ " static integer,\n"
+ " outbound_call_count integer,"
+ " outbound_fail_count integer,"
+ " hostname varchar(255)\n"
+ ");\n";
+
+
+static switch_status_t load_config(int reload, int del_all)
+{
+ char *cf = "fifo.conf";
+ switch_xml_t cfg, xml, fifo, fifos, member, settings, param;
+ char *odbc_user = NULL;
+ char *odbc_pass = NULL;
+ switch_core_db_t *db;
+ switch_status_t status = SWITCH_STATUS_SUCCESS;
+
+ gethostname(globals.hostname, sizeof(globals.hostname));
+
+ if (!(xml = switch_xml_open_cfg(cf, &cfg, NULL))) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Open of %s failed\n", cf);
+ return SWITCH_STATUS_TERM;
+ }
+
+ if ((settings = switch_xml_child(cfg, "settings"))) {
+ for (param = switch_xml_child(settings, "param"); param; param = param->next) {
+ char *var = NULL;
+ char *val = NULL;
+
+ var = (char *) switch_xml_attr_soft(param, "name");
+ val = (char *) switch_xml_attr_soft(param, "value");
+
+ if (!strcasecmp(var, "odbc-dsn") && !switch_strlen_zero(val)) {
+#ifdef SWITCH_HAVE_ODBC
+ globals.odbc_dsn = switch_core_strdup(globals.pool, val);
+ if ((odbc_user = strchr(globals.odbc_dsn, ':'))) {
+ *odbc_user++ = '\0';
+ if ((odbc_pass = strchr(odbc_user, ':'))) {
+ *odbc_pass++ = '\0';
+ }
+ }
+#else
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "ODBC IS NOT AVAILABLE!\n");
+#endif
+ }
+ }
+ }
+
+
+ if (switch_strlen_zero(globals.odbc_dsn) || switch_strlen_zero(odbc_user) || switch_strlen_zero(odbc_pass)) {
+ globals.dbname = "fifo";
+ }
+
+#ifdef SWITCH_HAVE_ODBC
+ if (globals.odbc_dsn) {
+ if (!(globals.master_odbc = switch_odbc_handle_new(globals.odbc_dsn, odbc_user, odbc_pass))) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Cannot Open ODBC Database!\n");
+ status = SWITCH_STATUS_FALSE;
+ goto done;
+ }
+ if (switch_odbc_handle_connect(globals.master_odbc) != SWITCH_ODBC_SUCCESS) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Cannot Open ODBC Database!\n");
+ status = SWITCH_STATUS_FALSE;
+ goto done;
+ }
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Connected ODBC DSN: %s\n", globals.odbc_dsn);
+ if (switch_odbc_handle_exec(globals.master_odbc, "delete from fifo_outbound", NULL) != SWITCH_STATUS_SUCCESS) {
+ if (switch_odbc_handle_exec(globals.master_odbc, (char *)outbound_sql, NULL) != SWITCH_STATUS_SUCCESS) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Cannot Create SQL Database!\n");
+ }
+ }
+ } else {
+#endif
+ if ((db = switch_core_db_open_file(globals.dbname))) {
+ switch_core_db_test_reactive(db, "delete from fifo_outbound", NULL, (char *)outbound_sql);
+ } else {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Cannot Open SQL Database!\n");
+ status = SWITCH_STATUS_FALSE;
+ goto done;
+ }
+ switch_core_db_close(db);
+#ifdef SWITCH_HAVE_ODBC
+ }
+#endif
+
+ if ((fifos = switch_xml_child(cfg, "fifos"))) {
+ for (fifo = switch_xml_child(fifos, "fifo"); fifo; fifo = fifo->next) {
+ const char *name;
+ const char *importance;
+ int imp = 0;
+ int simo_i = 1;
+ int timeout_i = 60;
+ int lag_i = 10;
+ fifo_node_t *node;
+ char *sql;
+
+ name = switch_xml_attr(fifo, "name");
+
+ if ((importance = switch_xml_attr(fifo, "importance"))) {
+ if ((imp = atoi(importance)) < 0) {
+ imp = 0;
+ }
+ }
+
+ if (!name) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "fifo has no name!\n");
+ continue;
+ }
+
+ switch_mutex_lock(globals.mutex);
+ if (!(node = switch_core_hash_find(globals.fifo_hash, name))) {
+ node = create_node(name, imp);
+ }
+ switch_mutex_unlock(globals.mutex);
+
+ if (del_all) {
+ sql = switch_mprintf("delete from fifo_outbound where fifo_name='%q' and hostname='%q'", node->name, globals.hostname);
+ } else {
+ sql = switch_mprintf("delete from fifo_outbound where static=1 and fifo_name='%q' and hostname='%q'", node->name, globals.hostname);
+ }
+
+ fifo_execute_sql(sql, globals.sql_mutex);
+ switch_safe_free(sql);
+
+ switch_mutex_lock(node->mutex);
+
+ for (member = switch_xml_child(fifo, "member"); member; member = member->next) {
+ const char *simo = switch_xml_attr_soft(member, "simo");
+ const char *lag = switch_xml_attr_soft(member, "lag");
+ const char *timeout = switch_xml_attr_soft(member, "timeout");
+
+ char *sql;
+ char digest[SWITCH_MD5_DIGEST_STRING_SIZE] = { 0 };
+ switch_md5_string(digest, (void *) member->txt, strlen(member->txt));
+
+ if (simo) {
+ simo_i = atoi(simo);
+ }
+
+ if (timeout) {
+ if ((timeout_i = atoi(timeout)) < 10) {
+ timeout_i = 60;
+ }
+
+ }
+
+ if (lag) {
+ if ((lag_i = atoi(lag)) < 0) {
+ lag_i = 10;
+ }
+ }
+
+ sql = switch_mprintf("insert into fifo_outbound "
+ "(uuid, fifo_name, originate_string, simo_count, use_count, timeout, lag, next_avail, expires, static, outbound_call_count, outbound_fail_count, hostname) "
+ "values ('%q','%q','%q',%d,%d,%d,%d,0,0,1,0,0,'%q')",
+ digest, node->name, member->txt, simo_i, 0, timeout_i, lag_i, 0, 0, globals.hostname
+ );
+ switch_assert(sql);
+ fifo_execute_sql(sql, globals.sql_mutex);
+ free(sql);
+
+ node->has_outbound = 1;
+
+ }
+ switch_mutex_unlock(node->mutex);
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "%s started\n", node->name);
+
+ }
+ }
+ switch_xml_free(xml);
+
+ done:
+
+ return status;
+}
+
+
+static void fifo_member_add(char *fifo_name, char *originate_string, int simo_count, int timeout, int lag, time_t expires)
+{
+ char digest[SWITCH_MD5_DIGEST_STRING_SIZE] = { 0 };
+ char *sql;
+ fifo_node_t *node = NULL;
+
+ switch_md5_string(digest, (void *) originate_string, strlen(originate_string));
+
+ sql = switch_mprintf("delete from fifo_outbound where fifo_name='%q' and uuid = '%q'", fifo_name, digest);
+ switch_assert(sql);
+ fifo_execute_sql(sql, globals.sql_mutex);
+ free(sql);
+
+
+ switch_mutex_lock(globals.mutex);
+ if (!(node = switch_core_hash_find(globals.fifo_hash, fifo_name))) {
+ node = create_node(fifo_name, 0);
+ }
+ switch_mutex_unlock(globals.mutex);
+
+ node->has_outbound = 1;
+
+ sql = switch_mprintf("insert into fifo_outbound "
+ "(uuid, fifo_name, originate_string, simo_count, use_count, timeout, lag, next_avail, expires, static, outbound_call_count, outbound_fail_count, hostname) "
+ "values ('%q','%q','%q',%d,%d,%d,%d,%d,%ld,0,0,0,'%q')",
+ digest, fifo_name, originate_string, simo_count, 0, timeout, lag, 0, (long)expires, globals.hostname
+ );
+ switch_assert(sql);
+ fifo_execute_sql(sql, globals.sql_mutex);
+ free(sql);
+
+}
+
+static void fifo_member_del(char *fifo_name, char *originate_string)
+{
+ char digest[SWITCH_MD5_DIGEST_STRING_SIZE] = { 0 };
+ char *sql;
+
+ switch_md5_string(digest, (void *) originate_string, strlen(originate_string));
+
+ sql = switch_mprintf("delete from fifo_outbound where fifo_name='%q' and uuid = '%q' and hostname='%q'", fifo_name, digest, globals.hostname);
+ switch_assert(sql);
+ fifo_execute_sql(sql, globals.sql_mutex);
+ free(sql);
+}
+
+#define FIFO_MEMBER_API_SYNTAX "[add <fifo_name> <originate_string> [<simo_count>] [<timeout>] [<lag>] | del <fifo_name> <originate_string>]"
+SWITCH_STANDARD_API(fifo_member_api_function)
+{
+ char *fifo_name;
+ char *originate_string;
+ int simo_count = 1;
+ int timeout = 60;
+ int lag = 5;
+ char *action;
+ char *mydata = NULL, *argv[8] = { 0 };
+ int argc;
+ time_t expires = 0;
+
+ mydata = strdup(cmd);
+ switch_assert(mydata);
+
+ argc = switch_separate_string(mydata, ' ', argv, (sizeof(argv) / sizeof(argv[0])));
+
+ if (argc < 3) {
+ stream->write_function(stream, "%s", "-ERR Invalid!\n");
+ goto done;
+ }
+
+ action = argv[0];
+ fifo_name = argv[1];
+ originate_string = argv[2];
+
+ if (!strcasecmp(action, "add")) {
+ if (argc > 3) {
+ simo_count = atoi(argv[3]);
+ }
+ if (argc > 4) {
+ timeout = atoi(argv[4]);
+ }
+ if (argc > 5) {
+ lag = atoi(argv[5]);
+ }
+ if (argc > 6) {
+ expires = switch_timestamp(NULL) + atoi(argv[6]);
+ }
+ if (simo_count < 0) {
+ simo_count = 1;
+ }
+ if (timeout < 0) {
+ timeout = 60;
+ }
+ if (lag < 0) {
+ lag = 5;
+ }
+
+ fifo_member_add(fifo_name, originate_string, simo_count, timeout, lag, expires);
+ stream->write_function(stream, "%s", "+OK\n");
+ } else if (!strcasecmp(action, "del")) {
+ fifo_member_del(fifo_name, originate_string);
+ stream->write_function(stream, "%s", "+OK\n");
+ } else {
+ stream->write_function(stream, "%s", "-ERR Invalid!\n");
+ goto done;
+ }
+
+ done:
+
+ free(mydata);
+
+ return SWITCH_STATUS_SUCCESS;
+
+}
+
SWITCH_MODULE_LOAD_FUNCTION(mod_fifo_load)
{
switch_application_interface_t *app_interface;
@@ -1223,11 +1927,13 @@
switch_core_new_memory_pool(&globals.pool);
switch_core_hash_init(&globals.fifo_hash, globals.pool);
switch_mutex_init(&globals.mutex, SWITCH_MUTEX_NESTED, globals.pool);
+ switch_mutex_init(&globals.sql_mutex, SWITCH_MUTEX_NESTED, globals.pool);
/* connect my internal structure to the blank pointer passed to me */
*module_interface = switch_loadable_module_create_module_interface(pool, modname);
SWITCH_ADD_APP(app_interface, "fifo", "Park with FIFO", FIFO_DESC, fifo_function, FIFO_USAGE, SAF_NONE);
SWITCH_ADD_API(commands_api_interface, "fifo", "Return data about a fifo", fifo_api_function, FIFO_API_SYNTAX);
+ SWITCH_ADD_API(commands_api_interface, "fifo_member", "Add members to a fifo", fifo_member_api_function, FIFO_MEMBER_API_SYNTAX);
switch_console_set_complete("add fifo list");
switch_console_set_complete("add fifo list_verbose");
switch_console_set_complete("add fifo count");
@@ -1235,6 +1941,9 @@
globals.running = 1;
+ load_config(0,1);
+ start_node_thread(globals.pool);
+
return SWITCH_STATUS_SUCCESS;
}
@@ -1256,10 +1965,18 @@
globals.running = 0;
/* Cleanup */
+
+ if (globals.node_thread_running) {
+ stop_node_thread();
+ }
+
+
for (hi = switch_hash_first(NULL, globals.fifo_hash); hi; hi = switch_hash_next(hi)) {
int x = 0;
switch_hash_this(hi, NULL, NULL, &val);
node = (fifo_node_t *) val;
+
+ switch_thread_rwlock_wrlock(node->rwlock);
for (x = 0; x < MAX_PRI; x++) {
while (switch_queue_trypop(node->fifo_list[x], &pop) == SWITCH_STATUS_SUCCESS) {
free(pop);
@@ -1268,6 +1985,7 @@
switch_core_hash_destroy(&node->caller_hash);
switch_core_hash_destroy(&node->consumer_hash);
+ switch_thread_rwlock_unlock(node->rwlock);
}
switch_core_hash_destroy(&globals.fifo_hash);
memset(&globals, 0, sizeof(globals));
More information about the Freeswitch-svn
mailing list