November 2008 - Freeswitch-svn - FreeSWITCH Mailing Lists

[commit] r10281 - freeswitch/trunk/src/mod/event_handlers/mod_erlang_event
by Freeswitch SVN 07 Nov '08

07 Nov '08

Author: andrew Date: Thu Nov 6 22:21:19 2008 New Revision: 10281 Modified: freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c Log: Don't create a new memory pool until it's clear we actually need it. Modified: freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c ============================================================================== --- freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c (original) +++ freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c Thu Nov 6 22:21:19 2008 @@ -1191,11 +1191,6 @@ switch_log_bind_logger(socket_logger, SWITCH_LOG_DEBUG, SWITCH_FALSE); for (;;) { - if (switch_core_new_memory_pool(&listener_pool) != SWITCH_STATUS_SUCCESS) { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "OH OH no pool\n"); - goto fail; - } - /* zero out errno because ei_accept doesn't differentiate between a * failed authentication or a socket failure, or a client version * mismatch or a godzilla attack */ @@ -1216,6 +1211,11 @@ break; } + if (switch_core_new_memory_pool(&listener_pool) != SWITCH_STATUS_SUCCESS) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "OH OH no pool\n"); + goto fail; + } + if (!(listener = switch_core_alloc(listener_pool, sizeof(*listener)))) { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Memory Error\n"); break;

1 0

[commit] r10280 - freeswitch/trunk/src/mod/event_handlers/mod_erlang_event
by Freeswitch SVN 07 Nov '08

07 Nov '08

Author: andrew Date: Thu Nov 6 20:29:44 2008 New Revision: 10280 Modified: freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c Log: Add initial support for sendevent to mod_erlang_event Modified: freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c ============================================================================== --- freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c (original) +++ freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c Thu Nov 6 20:29:44 2008 @@ -687,6 +687,60 @@ ei_x_encode_string(rbuf, acs->uuid_str); break; + } else if (!strncmp(tupletag, "sendevent", MAXATOMLEN)) { + char ename[MAXATOMLEN]; + + if (ei_decode_atom(buf->buff, &buf->index, ename)) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + int headerlength; + + if (ei_decode_list_header(buf->buff, &buf->index, &headerlength)) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + switch_event_types_t etype; + if (switch_name_event(ename, &etype) == SWITCH_STATUS_SUCCESS) { + switch_event_t *event; + + if (switch_event_create(&event, etype) == SWITCH_STATUS_SUCCESS) { + + char key[1024]; + char value[1024]; + int i = 0; + while(!ei_decode_tuple_header(buf->buff, &buf->index, &arity) && arity == 2) { + i++; + if (ei_decode_string(buf->buff, &buf->index, key)) + goto sendmsg_fail; + if (ei_decode_string(buf->buff, &buf->index, value)) + goto sendmsg_fail; + + switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, key, value); + } + + if (headerlength != i) + goto sendmsg_fail; + + + switch_event_fire(&event); + ei_x_encode_atom(rbuf, "ok"); + break; + +sendmsg_fail: + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + } + } else { ei_x_encode_tuple_header(rbuf, 2); ei_x_encode_atom(rbuf, "error"); @@ -740,7 +794,7 @@ } else if (!strncmp(atom, "exit", MAXATOMLEN)) { switch_clear_flag_locked(listener, LFLAG_RUNNING); ei_x_encode_atom(rbuf, "ok"); - goto done; + goto event_done; } else { ei_x_encode_tuple_header(rbuf, 2); ei_x_encode_atom(rbuf, "error"); @@ -762,7 +816,7 @@ noreply: return 0; -done: +event_done: ei_send(listener->sockfd, &msg->from, rbuf->buff, rbuf->index); return 1; }

1 0

[commit] r10279 - freeswitch/trunk/src/mod/applications/mod_fifo
by Freeswitch SVN 07 Nov '08

07 Nov '08

Author: anthm Date: Thu Nov 6 19:57:47 2008 New Revision: 10279 Modified: freeswitch/trunk/src/mod/applications/mod_fifo/mod_fifo.c Log: silly compiler warning 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:57:47 2008 @@ -1376,7 +1376,8 @@ char exp_buf[128] = ""; switch_time_exp_t tm; switch_time_t etime = 0; - + char *expires = exp_buf; + if (argv[7]) { if ((etime = atol(argv[7]))) { switch_size_t retsize; @@ -1396,7 +1397,7 @@ 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_attr_d(x_out, "next-available", expires); switch_xml_set_txt_d(x_out, argv[2]);

1 0

[commit] r10278 - freeswitch/trunk/conf/autoload_configs
by Freeswitch SVN 07 Nov '08

07 Nov '08

Author: anthm Date: Thu Nov 6 19:26:16 2008 New Revision: 10278 Added: freeswitch/trunk/conf/autoload_configs/fifo.conf.xml Log: add some stuff to mod_fifo Added: freeswitch/trunk/conf/autoload_configs/fifo.conf.xml ============================================================================== --- (empty file) +++ freeswitch/trunk/conf/autoload_configs/fifo.conf.xml Thu Nov 6 19:26:16 2008 @@ -0,0 +1,7 @@ +<configuration name="fifo.conf" description="FIFO Configuration"> + <fifos> + <fifo name="cool_fifo@$${domain}" importance="0"> +  + </fifo> + </fifos> +</configuration>

1 0

[commit] r10277 - freeswitch/trunk/src/mod/applications/mod_fifo
by Freeswitch SVN 07 Nov '08

07 Nov '08

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));

1 0

[commit] r10276 - in freeswitch/trunk/src: include mod/endpoints/mod_sofia
by Freeswitch SVN 07 Nov '08

07 Nov '08

Author: anthm Date: Thu Nov 6 19:24:31 2008 New Revision: 10276 Modified: freeswitch/trunk/src/include/switch_utils.h freeswitch/trunk/src/mod/endpoints/mod_sofia/mod_sofia.c Log: move macro to the core Modified: freeswitch/trunk/src/include/switch_utils.h ============================================================================== --- freeswitch/trunk/src/include/switch_utils.h (original) +++ freeswitch/trunk/src/include/switch_utils.h Thu Nov 6 19:24:31 2008 @@ -404,6 +404,7 @@ \return the epoch time in usec */ SWITCH_DECLARE(switch_time_t) switch_str_time(const char *in); +#define switch_time_from_sec(sec) ((switch_time_t)(sec) * 1000000) /*! \brief Declares a function designed to set a dymaic global string Modified: freeswitch/trunk/src/mod/endpoints/mod_sofia/mod_sofia.c ============================================================================== --- freeswitch/trunk/src/mod/endpoints/mod_sofia/mod_sofia.c (original) +++ freeswitch/trunk/src/mod/endpoints/mod_sofia/mod_sofia.c Thu Nov 6 19:24:31 2008 @@ -1276,7 +1276,7 @@ switch_stream_handle_t *stream; }; -#define switch_time_from_sec(sec) ((switch_time_t)(sec) * 1000000) + static int show_reg_callback(void *pArg, int argc, char **argv, char **columnNames) {

1 0

[commit] r10275 - freeswitch/branches/gmaruzz/src/mod/endpoints/mod_skypiax
by Freeswitch SVN 07 Nov '08

07 Nov '08

Author: gmaruzz Date: Thu Nov 6 17:55:42 2008 New Revision: 10275 Modified: freeswitch/branches/gmaruzz/src/mod/endpoints/mod_skypiax/mod_skypiax.c Log: skypiax: ask the Skype client for API connection Modified: freeswitch/branches/gmaruzz/src/mod/endpoints/mod_skypiax/mod_skypiax.c ============================================================================== --- freeswitch/branches/gmaruzz/src/mod/endpoints/mod_skypiax/mod_skypiax.c (original) +++ freeswitch/branches/gmaruzz/src/mod/endpoints/mod_skypiax/mod_skypiax.c Thu Nov 6 17:55:42 2008 @@ -2210,6 +2210,53 @@ #include <X11/Xatom.h> #include <switch.h> +#define SKYPIAX_SVN_VERSION "SVN 123456" + +#define DEBUGA_SKYPE(...) switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_SKYPE %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define DEBUGA_CALL(...) switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_CALL %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define DEBUGA_PBX(...) switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_PBX %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define ERRORA(...) switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][ERRORA %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define WARNINGA(...) switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][WARNINGA %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define NOTICA(...) switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][NOTICA %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); + + +#define SKYPIAX_P_LOG NULL, (unsigned long)55, __LINE__, "none", -1, -2, -3 + +#ifdef ASTER +/* debugging bitmask */ +#define DEBUG_SOUND 1 +#define DEBUG_SERIAL 2 +#define DEBUG_SKYPE 4 +#define DEBUG_AT 8 +#define DEBUG_FBUS2 16 +#define DEBUG_CALL 32 +#define DEBUG_LOCKS 64 +#define DEBUG_PBX 128 +#define DEBUG_MONITORLOCKS 256 +#define DEBUG_ALL DEBUG_SOUND|DEBUG_SERIAL|DEBUG_SKYPE|DEBUG_AT|DEBUG_FBUS2|DEBUG_CALL|DEBUG_PBX|DEBUG_LOCKS|DEBUG_MONITORLOCKS +/* wrappers for ast_log */ +#define DEBUGA_SOUND(...) if (skypiax_debug & DEBUG_SOUND) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_SOUND %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define DEBUGA_SERIAL(...) if (skypiax_debug & DEBUG_SERIAL) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_SERIAL %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define DEBUGA_SKYPE(...) if (skypiax_debug & DEBUG_SKYPE) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_SKYPE %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define DEBUGA_AT(...) if (skypiax_debug & DEBUG_AT) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_AT %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define DEBUGA_FBUS2(...) if (skypiax_debug & DEBUG_FBUS2) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_FBUS2 %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define DEBUGA_CALL(...) if (skypiax_debug & DEBUG_CALL) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_CALL %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define DEBUGA_PBX(...) if (skypiax_debug & DEBUG_PBX) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_PBX %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define ERRORA(...) ast_log(LOG_ERROR, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][ERROR %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define NOTICA(...) ast_log(LOG_NOTICE, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][NOTICE %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +#define WARNINGA(...) ast_log(LOG_WARNING, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][WARNING %-5d][%-10s][%2d,%2d,%2d] " __VA_ARGS__ ); +/* macros for logging */ +#define SKYPIAX_P_LOG p ? p->owner : NULL, (unsigned long)pthread_self(), __LINE__, p ? p->name ? p->name : "none" : "none", p ? p->owner ? p->owner->_state : -1 : -1, p ? p->interface_state : -1, p ? p->skype_callflow : -1 +#define SKYPIAX_TMP_LOG tmp ? tmp->owner : NULL, (unsigned long)pthread_self(), __LINE__, tmp ? tmp->name ? tmp->name : "none" : "none", tmp ? tmp->owner ? tmp->owner->_state : -1 : -1, tmp ? tmp->interface_state : -1, tmp ? tmp->skype_callflow : -1 +/* logging wrappers for ast_mutex_lock and ast_mutex_unlock */ +#define LOKKA(x) if (skypiax_debug & DEBUG_LOCKS) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_LOCKS %-5d][%-10s][%2d,%2d,%2d] going to lock %p (%s)\n", SKYPIAX_P_LOG, x, x == &skypiax_monlock ? "MONLOCK" : x == &skypiax_iflock ? "IFLOCK" : x == &skypiax_usecnt_lock ? "USECNT_LOCK" : "?????"); if (ast_mutex_lock(x)) ast_log(LOG_ERROR, "ast_mutex_lock failed, BAD\n"); if (skypiax_debug & DEBUG_LOCKS) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_LOCKS %-5d][%-10s][%2d,%2d,%2d] locked %p (%s)\n", SKYPIAX_P_LOG, x, x == &skypiax_monlock ? "MONLOCK" : x == &skypiax_iflock ? "IFLOCK" : x == &skypiax_usecnt_lock ? "USECNT_LOCK" : "?????"); +#define UNLOCKA(x) if (skypiax_debug & DEBUG_LOCKS) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_LOCKS %-5d][%-10s][%2d,%2d,%2d] going to unlock %p (%s)\n", SKYPIAX_P_LOG, x, x == &skypiax_monlock ? "MONLOCK" : x == &skypiax_iflock ? "IFLOCK" : x == &skypiax_usecnt_lock ? "USECNT_LOCK" : "?????"); if (ast_mutex_unlock(x)) ast_log(LOG_ERROR, "ast_mutex_lock failed, BAD\n"); if (skypiax_debug & DEBUG_LOCKS) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_LOCKS %-5d][%-10s][%2d,%2d,%2d] unlocked %p (%s)\n", SKYPIAX_P_LOG, x, x == &skypiax_monlock ? "MONLOCK" : x == &skypiax_iflock ? "IFLOCK" : x == &skypiax_usecnt_lock ? "USECNT_LOCK" : "?????"); +#define PUSHA_UNLOCKA(x) pthread_cleanup_push(skypiax_unlocka_log, (void *) x); +#define POPPA_UNLOCKA(x) pthread_cleanup_pop(0); +#define MONITORLOKKA(x) if (skypiax_debug & DEBUG_MONITORLOCKS) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_MONITORLOCKS %-5d][%-10s][%2d,%2d,%2d] going to lock %p (%s)\n", SKYPIAX_P_LOG, x, x == &skypiax_monlock ? "MONLOCK" : x == &skypiax_iflock ? "IFLOCK" : x == &skypiax_usecnt_lock ? "USECNT_LOCK" : "?????"); if (ast_mutex_lock(x)) ast_log(LOG_ERROR, "ast_mutex_lock failed, BAD\n"); if (skypiax_debug & DEBUG_MONITORLOCKS) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_MONITORLOCKS %-5d][%-10s][%2d,%2d,%2d] locked %p (%s)\n", SKYPIAX_P_LOG, x, x == &skypiax_monlock ? "MONLOCK" : x == &skypiax_iflock ? "IFLOCK" : x == &skypiax_usecnt_lock ? "USECNT_LOCK" : "?????"); +#define MONITORUNLOCKA(x) if (skypiax_debug & DEBUG_MONITORLOCKS) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_MONITORLOCKS %-5d][%-10s][%2d,%2d,%2d] going to unlock %p (%s)\n", SKYPIAX_P_LOG, x, x == &skypiax_monlock ? "MONLOCK" : x == &skypiax_iflock ? "IFLOCK" : x == &skypiax_usecnt_lock ? "USECNT_LOCK" : "?????"); if (ast_mutex_unlock(x)) ast_log(LOG_ERROR, "ast_mutex_lock failed, BAD\n"); if (skypiax_debug & DEBUG_MONITORLOCKS) ast_log(LOG_DEBUG, "rev "SKYPIAX_SVN_VERSION "[%p|%-7lx][DEBUG_MONITORLOCKS %-5d][%-10s][%2d,%2d,%2d] unlocked %p (%s)\n", SKYPIAX_P_LOG, x, x == &skypiax_monlock ? "MONLOCK" : x == &skypiax_iflock ? "IFLOCK" : x == &skypiax_usecnt_lock ? "USECNT_LOCK" : "?????"); +#endif /* ASTER */ + #define SKYPIAX_MAX_INTERFACES 64 SWITCH_MODULE_LOAD_FUNCTION(mod_skypiax_load); @@ -2222,6 +2269,13 @@ static switch_memory_pool_t *module_pool = NULL; static int running = 1; +struct AsteriskHandles { + Window skype_win; + Display *disp; + Window win; + int fdesc[2]; +}; + typedef enum { TFLAG_IO = (1 << 0), @@ -2676,16 +2730,147 @@ /*.receive_event */ channel_receive_event }; #define SKYPE_THREAD + struct AsteriskHandles ciapa; #ifdef SKYPE_THREAD +#define SKYPE_X11_BUF_SIZE 512 +Window skype_win = (Window) - 1; +static XErrorHandler old_handler = 0; +static int xerror = 0; + +int X11_errors_handler(Display * dpy, XErrorEvent * err) +{ + (void) dpy; + //struct skypiax_pvt *p = NULL; + + xerror = err->error_code; + DEBUGA_SKYPE("Received error code %d from X Server\n", SKYPIAX_P_LOG, xerror); + return 0; /* ignore the error */ +} + +static void X11_errors_trap(void) +{ + xerror = 0; + old_handler = XSetErrorHandler(X11_errors_handler); +} + +static int X11_errors_untrap(void) +{ + XSetErrorHandler(old_handler); + return (xerror != BadValue) && (xerror != BadWindow); +} + +int skypiax_skype_send_message(struct AsteriskHandles *AsteriskHandlesAst, + const char *message_P) +{ + + Window w_P; + Display *disp; + Window handle_P; + //struct skypiax_pvt *p = NULL; + + w_P = AsteriskHandlesAst->skype_win; + disp = AsteriskHandlesAst->disp; + handle_P = AsteriskHandlesAst->win; + + Atom atom1 = XInternAtom(disp, "SKYPECONTROLAPI_MESSAGE_BEGIN", False); + Atom atom2 = XInternAtom(disp, "SKYPECONTROLAPI_MESSAGE", False); + unsigned int pos = 0; + unsigned int len = strlen(message_P); + XEvent e; + int ok; + + memset(&e, 0, sizeof(e)); + e.xclient.type = ClientMessage; + e.xclient.message_type = atom1; /* leading message */ + e.xclient.display = disp; + e.xclient.window = handle_P; + e.xclient.format = 8; + + X11_errors_trap(); + //XLockDisplay(disp); + do { + unsigned int i; + for (i = 0; i < 20 && i + pos <= len; ++i) + e.xclient.data.b[i] = message_P[i + pos]; + XSendEvent(disp, w_P, False, 0, &e); + + e.xclient.message_type = atom2; /* following messages */ + pos += i; + } while (pos <= len); + + XSync(disp, False); + //XUnlockDisplay(disp); + ok = X11_errors_untrap(); + + if (!ok) + DEBUGA_SKYPE("Sending message failed with status %d\n", SKYPIAX_P_LOG, xerror); + + return ok; +} + +int skypiax_skype_present(Display * disp) +{ + Atom skype_inst = XInternAtom(disp, "_SKYPE_INSTANCE", True); + + Atom type_ret; + int format_ret; + unsigned long nitems_ret; + unsigned long bytes_after_ret; + unsigned char *prop; + int status; + //struct skypiax_pvt *p = NULL; + + X11_errors_trap(); + //XLockDisplay(disp); + status = + XGetWindowProperty(disp, DefaultRootWindow(disp), skype_inst, 0, 1, False, XA_WINDOW, + &type_ret, &format_ret, &nitems_ret, &bytes_after_ret, &prop); + //XUnlockDisplay(disp); + X11_errors_untrap(); + + /* sanity check */ + if (status != Success || format_ret != 32 || nitems_ret != 1) { + skype_win = (Window) - 1; + DEBUGA_SKYPE("Skype instance not found\n", SKYPIAX_P_LOG); + return 0; + } + + skype_win = *(const unsigned long *) prop & 0xffffffff; + //DEBUGA_SKYPE("Skype instance found with id #%x\n", SKYPIAX_P_LOG, + DEBUGA_SKYPE("Skype instance found with id #%d\n", SKYPIAX_P_LOG, + (unsigned int) skype_win); + return 1; +} + +void skypiax_skype_clean_disp(void *data) +{ + + int *dispptr; + int disp; + //struct skypiax_pvt *p = NULL; + + dispptr = data; + disp = *dispptr; + + if (disp) { + DEBUGA_SKYPE("to be destroyed disp %d\n", SKYPIAX_P_LOG, disp); + close(disp); + DEBUGA_SKYPE("destroyed disp\n", SKYPIAX_P_LOG); + } else { + DEBUGA_SKYPE("NOT destroyed disp\n", SKYPIAX_P_LOG); + } + DEBUGA_SKYPE("OUT destroyed disp\n", SKYPIAX_P_LOG); + usleep(1000); +} void *do_skype_thread(void *data) { // struct skypiax_pvt *p; - // struct AsteriskHandles *AsteriskHandlesAst; - //FIXME char buf[SKYPE_X11_BUF_SIZE]; + struct AsteriskHandles *AsteriskHandlesAst; + char buf[SKYPE_X11_BUF_SIZE]; Display *disp = NULL; - //FIXME Window root = -1; - //FIXME Window win = -1; + Window root = -1; + Window win = -1; /* p = data; @@ -2703,6 +2888,7 @@ AsteriskHandlesAst = &p->AsteriskHandlesAst; */ + AsteriskHandlesAst = &ciapa; disp = XOpenDisplay(getenv("DISPLAY")); if (!disp) { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Cannot open X Display, exiting skype thread\n"); @@ -2715,8 +2901,8 @@ xfd = XConnectionNumber(disp); fcntl(xfd, F_SETFD, FD_CLOEXEC); -#ifdef NOTDEF - pthread_cleanup_push(skypiax_skype_clean_disp, &xfd); +#ifndef NOTDEF + //FIXME pthread_cleanup_push(skypiax_skype_clean_disp, &xfd); DEBUGA_SKYPE("PUSH disp %d\n", SKYPIAX_P_LOG, xfd); if (skypiax_skype_present(disp)) { @@ -2734,17 +2920,15 @@ AsteriskHandlesAst->disp = disp; AsteriskHandlesAst->win = win; - snprintf(buf, SKYPE_X11_BUF_SIZE, "NAME skypiax"); + snprintf(buf, SKYPE_X11_BUF_SIZE, "NAME FS_skypiax"); if (!skypiax_skype_send_message(AsteriskHandlesAst, buf)) { ERRORA ("Sending message failed - probably Skype crashed. Please run/restart Skype manually and launch Skypiax again\n", SKYPIAX_P_LOG); - p->skype = 0; - p->skype_thread = AST_PTHREADT_NULL; - if (option_debug > 10) { + //p->skype = 0; + //p->skype_thread = AST_PTHREADT_NULL; DEBUGA_PBX("EXITING FUNC\n", SKYPIAX_P_LOG); - } return NULL; } @@ -2753,23 +2937,22 @@ ERRORA ("Sending message failed - probably Skype crashed. Please run/restart Skype manually and launch Skypiax again\n", SKYPIAX_P_LOG); - p->skype = 0; - p->skype_thread = AST_PTHREADT_NULL; - if (option_debug > 10) { + //p->skype = 0; + //p->skype_thread = AST_PTHREADT_NULL; DEBUGA_PBX("EXITING FUNC\n", SKYPIAX_P_LOG); - } return NULL; } /* perform an events loop */ - XEvent an_event; - char buf[21]; /* can't be longer */ + //FIXME XEvent an_event; + //FIXME char buf[21]; /* can't be longer */ char buffer[17000]; char *b; - int i; + //FIXME int i; b = buffer; +#ifdef LOOPI while (1) { pthread_testcancel(); XNextEvent(disp, &an_event); @@ -2804,20 +2987,19 @@ break; } } +#endif // LOOPI } else { ERRORA ("Skype is not running, maybe crashed. Please run/restart Skype and relaunch Skypiax\n", SKYPIAX_P_LOG); - p->skype = 0; - p->skype_thread = AST_PTHREADT_NULL; + //p->skype = 0; + //p->skype_thread = AST_PTHREADT_NULL; return NULL; } - p->skype = 0; - p->skype_thread = AST_PTHREADT_NULL; - pthread_cleanup_pop(1); - if (option_debug > 10) { + //p->skype = 0; + //p->skype_thread = AST_PTHREADT_NULL; + //pthread_cleanup_pop(1); DEBUGA_PBX("EXITING FUNC\n", SKYPIAX_P_LOG); - } #endif /* NOTDEF */ return NULL; @@ -3011,8 +3193,11 @@ #endif /* _WINDOWS_ */ + load_config(); + DEBUGA_PBX("CIAPALO FUNC\n", SKYPIAX_P_LOG); + *module_interface = switch_loadable_module_create_module_interface(pool, modname); skypiax_endpoint_interface = switch_loadable_module_create_interface(*module_interface, SWITCH_ENDPOINT_INTERFACE); skypiax_endpoint_interface->interface_name = "skypiax";

1 0

[commit] r10274 - freeswitch/trunk/src/mod/event_handlers/mod_erlang_event
by Freeswitch SVN 07 Nov '08

07 Nov '08

Author: andrew Date: Thu Nov 6 16:03:30 2008 New Revision: 10274 Added: freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/Makefile freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c Log: Add mod_erlang_event to tree - a module that makes FreeSWITCH act like an erlang c-node. Added: freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/Makefile ============================================================================== --- (empty file) +++ freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/Makefile Thu Nov 6 16:03:30 2008 @@ -0,0 +1,5 @@ +BASE=../../../.. +include $(BASE)/build/modmake.rules + +LOCAL_CFLAGS=-I/usr/local/lib/erlang/lib/erl_interface-3.5.8/include -L/usr/local/lib/erlang/lib/erl_interface-3.5.8/lib/ -D_REENTRANT +LOCAL_LDFLAGS=-lei Added: freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c ============================================================================== --- (empty file) +++ freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c Thu Nov 6 16:03:30 2008 @@ -0,0 +1,1216 @@ +/* + * FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application + * Copyright (C) 2005/2006, Anthony Minessale II <anthmct(a)yahoo.com> + * + * 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 <anthmct(a)yahoo.com> + * Portions created by the Initial Developer are Copyright (C) + * the Initial Developer. All Rights Reserved. + * + * Contributor(s): + * + * Anthony Minessale II <anthmct(a)yahoo.com> + * Andrew Thompson <andrew(a)hijacked.us> + * + * + * mod_erlang_event.c -- Erlang Event Handler derived from mod_event_socket + * + */ +#include <switch.h> + +#include <ei.h> + +SWITCH_MODULE_LOAD_FUNCTION(mod_erlang_event_load); +SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_erlang_event_shutdown); +SWITCH_MODULE_RUNTIME_FUNCTION(mod_erlang_event_runtime); +SWITCH_MODULE_DEFINITION(mod_erlang_event, mod_erlang_event_load, mod_erlang_event_shutdown, mod_erlang_event_runtime); + +static char *MARKER = "1"; + +typedef enum { + LFLAG_AUTHED = (1 << 0), + LFLAG_RUNNING = (1 << 1), + LFLAG_EVENTS = (1 << 2), + LFLAG_LOG = (1 << 3), + LFLAG_FULL = (1 << 4), + LFLAG_MYEVENTS = (1 << 5), + LFLAG_SESSION = (1 << 6), + LFLAG_ASYNC = (1 << 7), + LFLAG_STATEFUL = (1 << 8) +} event_flag_t; + + +/* TODO - support multiple event handlers per erlang connection each with their own event filters? */ +struct event_handler { + erlang_pid pid; + switch_hash_t *event_hash; + struct event_handler *next; +}; + +struct listener { + int sockfd; + const struct ei_cnode_s *ec; + erlang_pid log_pid; + erlang_pid event_pid; + switch_queue_t *event_queue; + switch_queue_t *log_queue; + switch_memory_pool_t *pool; + switch_mutex_t *flag_mutex; + char *ebuf; + uint32_t flags; + switch_log_level_t level; + uint8_t event_list[SWITCH_EVENT_ALL + 1]; + switch_hash_t *event_hash; + switch_thread_rwlock_t *rwlock; + switch_core_session_t *session; + int lost_events; + int lost_logs; + time_t last_flush; + uint32_t timeout; + uint32_t id; + char remote_ip[50]; + /*switch_port_t remote_port;*/ + struct listener *next; +}; + +typedef struct listener listener_t; + +static struct { + int sockfd; + switch_mutex_t *mutex; + switch_mutex_t *sock_mutex; + listener_t *listeners; + uint8_t ready; +} listen_list; + +#define MAX_ACL 100 + +static struct { + switch_mutex_t *mutex; + char *ip; + char *nodename; + switch_bool_t shortname; + uint16_t port; + char *cookie; + int done; + int threads; + char *acl[MAX_ACL]; + uint32_t acl_count; + uint32_t id; +} prefs; + + +static void remove_listener(listener_t *listener); + +SWITCH_DECLARE_GLOBAL_STRING_FUNC(set_pref_ip, prefs.ip); +SWITCH_DECLARE_GLOBAL_STRING_FUNC(set_pref_cookie, prefs.cookie); +SWITCH_DECLARE_GLOBAL_STRING_FUNC(set_pref_nodename, prefs.nodename); + +static void *SWITCH_THREAD_FUNC listener_run(switch_thread_t *thread, void *obj); +static void launch_listener_thread(listener_t *listener); + +static struct { + switch_event_node_t *node; +} globals; + + +static switch_status_t socket_logger(const switch_log_node_t *node, switch_log_level_t level) +{ + listener_t *l; + + switch_mutex_lock(listen_list.mutex); + for (l = listen_list.listeners; l; l = l->next) { + if (switch_test_flag(l, LFLAG_LOG) && l->level >= node->level) { + char *data = strdup(node->data); + if (data) { + if (switch_queue_trypush(l->log_queue, data) == SWITCH_STATUS_SUCCESS) { + if (l->lost_logs) { + int ll = l->lost_logs; + switch_event_t *event; + l->lost_logs = 0; + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Lost %d log lines!\n", ll); + if (switch_event_create(&event, SWITCH_EVENT_TRAP) == SWITCH_STATUS_SUCCESS) { + switch_event_add_header(event, SWITCH_STACK_BOTTOM, "info", "lost %d log lines", ll); + switch_event_fire(&event); + } + } + } else { + switch_safe_free(data); + l->lost_logs++; + } + } else { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Memory Error!\n"); + } + } + } + switch_mutex_unlock(listen_list.mutex); + + return SWITCH_STATUS_SUCCESS; +} + + +static void expire_listener(listener_t **listener) +{ + void *pop; + + switch_thread_rwlock_unlock((*listener)->rwlock); + switch_core_hash_destroy(&(*listener)->event_hash); + switch_core_destroy_memory_pool(&(*listener)->pool); + + while (switch_queue_trypop((*listener)->event_queue, &pop) == SWITCH_STATUS_SUCCESS) { + switch_event_t *pevent = (switch_event_t *) pop; + switch_event_destroy(&pevent); + } + + *listener = NULL; +} + + +static void ei_encode_switch_event(ei_x_buff *ebuf, switch_event_t *event) +{ + int i; + switch_event_header_t *hp; + for (i = 0, hp = event->headers; hp; hp = hp->next, i++) { + } + + if (event->body) + i++; + + ei_x_encode_tuple_header(ebuf, 2); + ei_x_encode_atom(ebuf, "event"); + ei_x_encode_list_header(ebuf, i); + + for (hp = event->headers; hp; hp = hp->next) { + ei_x_encode_tuple_header(ebuf, 2); + ei_x_encode_string(ebuf, hp->name); + ei_x_encode_string(ebuf, hp->value); + } + + if (event->body) { + ei_x_encode_tuple_header(ebuf, 2); + ei_x_encode_string(ebuf, "body"); + ei_x_encode_string(ebuf, event->body); + } + + ei_x_encode_empty_list(ebuf); +} + + +static void event_handler(switch_event_t *event) +{ + switch_event_t *clone = NULL; + listener_t *l, *lp; + + switch_assert(event != NULL); + + if (!listen_list.ready) { + return; + } + + lp = listen_list.listeners; + + switch_mutex_lock(listen_list.mutex); + while(lp) { + uint8_t send = 0; + + l = lp; + lp = lp->next; + + if (!switch_test_flag(l, LFLAG_EVENTS)) { + continue; + } + + if (switch_test_flag(l, LFLAG_STATEFUL) && l->timeout && switch_timestamp(NULL) - l->last_flush > l->timeout) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "Stateful Listener %u has expired\n", l->id); + remove_listener(l); + expire_listener(&l); + continue; + } + + if (l->event_list[SWITCH_EVENT_ALL]) { + send = 1; + } else if ((l->event_list[event->event_id])) { + if (event->event_id != SWITCH_EVENT_CUSTOM || !event->subclass_name || (switch_core_hash_find(l->event_hash, event->subclass_name))) { + send = 1; + } + } + + if (send && switch_test_flag(l, LFLAG_MYEVENTS)) { + char *uuid = switch_event_get_header(event, "unique-id"); + if (!uuid || strcmp(uuid, switch_core_session_get_uuid(l->session))) { + send = 0; + } + } + + if (send) { + if (switch_event_dup(&clone, event) == SWITCH_STATUS_SUCCESS) { + if (switch_queue_trypush(l->event_queue, clone) == SWITCH_STATUS_SUCCESS) { + if (l->lost_events) { + int le = l->lost_events; + l->lost_events = 0; + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Lost %d events!\n", le); + clone = NULL; + if (switch_event_create(&clone, SWITCH_EVENT_TRAP) == SWITCH_STATUS_SUCCESS) { + switch_event_add_header(clone, SWITCH_STACK_BOTTOM, "info", "lost %d events", le); + switch_event_fire(&clone); + } + } + } else { + l->lost_events++; + switch_event_destroy(&clone); + } + } else { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Memory Error!\n"); + } + } + + } + switch_mutex_unlock(listen_list.mutex); +} + + +SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_erlang_event_shutdown) +{ + listener_t *l; + int sanity = 0; + + prefs.done = 1; + + switch_log_unbind_logger(socket_logger); + + close(listen_list.sockfd); + + switch_yield(500); + + while (prefs.threads) { + switch_yield(10000); + if (++sanity == 1000) { + break; + } + } + switch_event_unbind(&globals.node); + + switch_mutex_lock(listen_list.mutex); + for (l = listen_list.listeners; l; l = l->next) { + close(l->sockfd); + } + switch_mutex_unlock(listen_list.mutex); + + return SWITCH_STATUS_SUCCESS; +} + + +static void add_listener(listener_t *listener) +{ + /* add me to the listeners so I get events */ + switch_mutex_lock(listen_list.mutex); + listener->next = listen_list.listeners; + listen_list.listeners = listener; + switch_mutex_unlock(listen_list.mutex); +} + + +static void remove_listener(listener_t *listener) +{ + listener_t *l, *last = NULL; + + switch_mutex_lock(listen_list.mutex); + for (l = listen_list.listeners; l; l = l->next) { + if (l == listener) { + if (last) { + last->next = l->next; + } else { + listen_list.listeners = l->next; + } + } + last = l; + } + switch_mutex_unlock(listen_list.mutex); +} + +SWITCH_MODULE_LOAD_FUNCTION(mod_erlang_event_load) +{ + /* connect my internal structure to the blank pointer passed to me */ + *module_interface = switch_loadable_module_create_module_interface(pool, modname); + + /* indicate that the module should continue to be loaded */ + return SWITCH_STATUS_SUCCESS; +} + + +struct api_command_struct { + char *api_cmd; + char *arg; + listener_t *listener; + char uuid_str[SWITCH_UUID_FORMATTED_LENGTH + 1]; + uint8_t bg; + erlang_pid pid; + switch_memory_pool_t *pool; +}; + + +static void *SWITCH_THREAD_FUNC api_exec(switch_thread_t *thread, void *obj) +{ + switch_bool_t r = SWITCH_TRUE; + struct api_command_struct *acs = (struct api_command_struct *) obj; + switch_stream_handle_t stream = { 0 }; + char *reply, *freply = NULL; + switch_status_t status; + + if (!acs) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Internal error.\n"); + return NULL; + } + + if (!acs->listener || !acs->listener->rwlock || switch_thread_rwlock_tryrdlock(acs->listener->rwlock) != SWITCH_STATUS_SUCCESS) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error! cannot get read lock.\n"); + goto done; + } + + + SWITCH_STANDARD_STREAM(stream); + + if ((status = switch_api_execute(acs->api_cmd, acs->arg, NULL, &stream)) == SWITCH_STATUS_SUCCESS) { + reply = stream.data; + } else { + freply = switch_mprintf("%s: Command not found!\n", acs->api_cmd); + reply = freply; + r = SWITCH_FALSE; + } + + if (!reply) { + reply = "Command returned no output!"; + r = SWITCH_FALSE; + } + + if (*reply == '-') + r = SWITCH_FALSE; + + if (acs->bg) { + switch_event_t *event; + + if (switch_event_create(&event, SWITCH_EVENT_BACKGROUND_JOB) == SWITCH_STATUS_SUCCESS) { + switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Job-UUID", acs->uuid_str); + switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Job-Command", acs->api_cmd); + + ei_x_buff ebuf; + ei_x_new_with_version(&ebuf); + + if (acs->arg) { + switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Job-Command-Arg", acs->arg); + } + + switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Job-Successful", r ? "true" : "false"); + switch_event_add_body(event, "%s", reply); + + switch_event_fire(&event); + + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Sending bgapi reply to %s\n", acs->pid.node); + + ei_x_encode_tuple_header(&ebuf, 3); + + if (r) + ei_x_encode_atom(&ebuf, "bgok"); + else + ei_x_encode_atom(&ebuf, "bgerror"); + + ei_x_encode_string(&ebuf, acs->uuid_str); + ei_x_encode_string(&ebuf, reply); + + ei_send(acs->listener->sockfd, &acs->pid, ebuf.buff, ebuf.index); + + ei_x_free(&ebuf); + } + } else { + ei_x_buff rbuf; + ei_x_new_with_version(&rbuf); + ei_x_encode_tuple_header(&rbuf, 2); + + if (!strlen(reply)) { + reply = "Command returned no output!"; + r = SWITCH_FALSE; + } + + if (r) { + ei_x_encode_atom(&rbuf, "ok"); + } else { + ei_x_encode_atom(&rbuf, "error"); + } + + ei_x_encode_string(&rbuf, reply); + + ei_send(acs->listener->sockfd, &acs->pid, rbuf.buff, rbuf.index); + ei_x_free(&rbuf); + } + + switch_safe_free(stream.data); + switch_safe_free(freply); + + if (acs->listener->rwlock) { + switch_thread_rwlock_unlock(acs->listener->rwlock); + } + + done: + if (acs->bg) { + switch_memory_pool_t *pool = acs->pool; + acs = NULL; + switch_core_destroy_memory_pool(&pool); + pool = NULL; + } + return NULL; + +} + + +static int handle_msg(listener_t *listener, erlang_msg *msg, ei_x_buff *buf, ei_x_buff *rbuf) +{ + int type, size, version, arity; + char tupletag[MAXATOMLEN]; + char atom[MAXATOMLEN]; + + buf->index = 0; + ei_decode_version(buf->buff, &buf->index, &version); + ei_get_type(buf->buff, &buf->index, &type, &size); + + switch(type) { + case ERL_SMALL_TUPLE_EXT : + case ERL_LARGE_TUPLE_EXT : + ei_decode_tuple_header(buf->buff, &buf->index, &arity); + if (ei_decode_atom(buf->buff, &buf->index, tupletag)) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + if (!strncmp(tupletag, "set_log_level", MAXATOMLEN)) { + if (arity == 2) { + switch_log_level_t ltype = SWITCH_LOG_DEBUG; + char loglevelstr[MAXATOMLEN]; + if (ei_decode_atom(buf->buff, &buf->index, loglevelstr)) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + ltype = switch_log_str2level(loglevelstr); + + if (ltype && ltype != SWITCH_LOG_INVALID) { + listener->level = ltype; + } else { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + } else { + /* tuple too long */ + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + } else if (!strncmp(tupletag, "event", MAXATOMLEN)) { + if (arity == 1) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + int custom = 0; + switch_event_types_t type; + + if (!switch_test_flag(listener, LFLAG_EVENTS)) { + switch_set_flag_locked(listener, LFLAG_EVENTS); + } + + for (int i = 1; i < arity; i++) { + if (!ei_decode_atom(buf->buff, &buf->index, atom)) { + + if (custom) { + switch_core_hash_insert(listener->event_hash, atom, MARKER); + } else if (switch_name_event(atom, &type) == SWITCH_STATUS_SUCCESS) { + if (type == SWITCH_EVENT_ALL) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "ALL events enabled\n"); + uint32_t x = 0; + for (x = 0; x < SWITCH_EVENT_ALL; x++) { + listener->event_list[x] = 1; + } + } + if (type <= SWITCH_EVENT_ALL) { + listener->event_list[type] = 1; + } + if (type == SWITCH_EVENT_CUSTOM) { + custom++; + } + + } + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "enable event %s\n", atom); + } + } + } else if (!strncmp(tupletag, "nixevent", MAXATOMLEN)) { + if (arity == 1) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + int custom = 0; + switch_event_types_t type; + + for (int i = 1; i < arity; i++) { + if (!ei_decode_atom(buf->buff, &buf->index, atom)) { + + if (custom) { + switch_core_hash_delete(listener->event_hash, atom); + } else if (switch_name_event(atom, &type) == SWITCH_STATUS_SUCCESS) { + uint32_t x = 0; + + if (type == SWITCH_EVENT_CUSTOM) { + custom++; + } else if (type == SWITCH_EVENT_ALL) { + for (x = 0; x <= SWITCH_EVENT_ALL; x++) { + listener->event_list[x] = 0; + } + } else { + if (listener->event_list[SWITCH_EVENT_ALL]) { + listener->event_list[SWITCH_EVENT_ALL] = 0; + for (x = 0; x < SWITCH_EVENT_ALL; x++) { + listener->event_list[x] = 1; + } + } + listener->event_list[type] = 0; + } + } + } + } + } else if (!strncmp(tupletag, "api", MAXATOMLEN)) { + if (arity < 3) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + char api_cmd[MAXATOMLEN]; + char arg[1024]; + + if (ei_decode_atom(buf->buff, &buf->index, api_cmd)) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + if (ei_decode_string(buf->buff, &buf->index, arg)) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + struct api_command_struct acs = { 0 }; + acs.listener = listener; + acs.api_cmd = api_cmd; + acs.arg = arg; + acs.bg = 0; + acs.pid = msg->from; + api_exec(NULL, (void *) &acs); + goto noreply; + + } else if (!strncmp(tupletag, "bgapi", MAXATOMLEN)) { + if (arity < 3) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + char api_cmd[MAXATOMLEN]; + char arg[1024]; + + if (ei_decode_atom(buf->buff, &buf->index, api_cmd)) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + if (ei_decode_string(buf->buff, &buf->index, arg)) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + struct api_command_struct *acs = NULL; + switch_memory_pool_t *pool; + switch_thread_t *thread; + switch_threadattr_t *thd_attr = NULL; + switch_uuid_t uuid; + + switch_core_new_memory_pool(&pool); + acs = switch_core_alloc(pool, sizeof(*acs)); + switch_assert(acs); + acs->pool = pool; + acs->listener = listener; + acs->api_cmd = switch_core_strdup(acs->pool, api_cmd); + acs->arg = switch_core_strdup(acs->pool, arg); + acs->bg = 1; + acs->pid = msg->from; + + switch_threadattr_create(&thd_attr, acs->pool); + switch_threadattr_detach_set(thd_attr, 1); + switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE); + + switch_uuid_get(&uuid); + switch_uuid_format(acs->uuid_str, &uuid); + switch_thread_create(&thread, thd_attr, api_exec, acs, acs->pool); + + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "ok"); + ei_x_encode_string(rbuf, acs->uuid_str); + + break; + } else { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "undef"); + break; + } + + ei_x_encode_atom(rbuf, "ok"); + break; + case ERL_ATOM_EXT : + if(ei_decode_atom(buf->buff, &buf->index, atom)) { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "badarg"); + break; + } + + if (!strncmp(atom, "nolog", MAXATOMLEN)) { + if (switch_test_flag(listener, LFLAG_LOG)) { + switch_clear_flag_locked(listener, LFLAG_LOG); + } + } else if (!strncmp(atom, "register_log_handler", MAXATOMLEN)) { + listener->log_pid = msg->from; + listener->level = SWITCH_LOG_DEBUG; + switch_set_flag(listener, LFLAG_LOG); + } else if (!strncmp(atom, "register_event_handler", MAXATOMLEN)) { + listener->event_pid = msg->from; + if (!switch_test_flag(listener, LFLAG_EVENTS)) { + switch_set_flag_locked(listener, LFLAG_EVENTS); + } + } else if (!strncmp(atom, "noevents", MAXATOMLEN)) { + void *pop; + /*purge the event queue */ + while (switch_queue_trypop(listener->event_queue, &pop) == SWITCH_STATUS_SUCCESS); + + if (switch_test_flag(listener, LFLAG_EVENTS)) { + uint8_t x = 0; + switch_clear_flag_locked(listener, LFLAG_EVENTS); + for (x = 0; x <= SWITCH_EVENT_ALL; x++) { + listener->event_list[x] = 0; + } + /* wipe the hash */ + switch_core_hash_destroy(&listener->event_hash); + switch_core_hash_init(&listener->event_hash, listener->pool); + } else { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "notlistening"); + break; + } + } else if (!strncmp(atom, "exit", MAXATOMLEN)) { + switch_clear_flag_locked(listener, LFLAG_RUNNING); + ei_x_encode_atom(rbuf, "ok"); + goto done; + } else { + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "undef"); + break; + } + + ei_x_encode_atom(rbuf, "ok"); + break; + default : + /* some other kind of erlang term */ + ei_x_encode_tuple_header(rbuf, 2); + ei_x_encode_atom(rbuf, "error"); + ei_x_encode_atom(rbuf, "undef"); + break; + } + + ei_send(listener->sockfd, &msg->from, rbuf->buff, rbuf->index); +noreply: + return 0; + +done: + ei_send(listener->sockfd, &msg->from, rbuf->buff, rbuf->index); + return 1; +} + + +static void *SWITCH_THREAD_FUNC listener_run(switch_thread_t *thread, void *obj) +{ + listener_t *listener = (listener_t *) obj; + switch_core_session_t *session = NULL; + switch_channel_t *channel = NULL; + int status = 1; + void *pop; + + switch_mutex_lock(listen_list.mutex); + prefs.threads++; + switch_mutex_unlock(listen_list.mutex); + + switch_assert(listener != NULL); + + if (prefs.acl_count && !switch_strlen_zero(listener->remote_ip)) { + uint32_t x = 0; + for (x = 0; x < prefs.acl_count; x++) { + if (!switch_check_network_list_ip(listener->remote_ip, prefs.acl[x])) { + erlang_msg msg; + + ei_x_buff buf; + ei_x_new(&buf); + + status = ei_xreceive_msg(listener->sockfd, &msg, &buf); + /* get data off the socket, just so we can get the pid on the other end */ + if (status == ERL_MSG) { + /* if we got a message, return an ACL error. */ + ei_x_buff rbuf; + ei_x_new_with_version(&rbuf); + + ei_x_encode_tuple_header(&rbuf, 2); + ei_x_encode_atom(&rbuf, "error"); + ei_x_encode_atom(&rbuf, "acldeny"); + + ei_send(listener->sockfd, &msg.from, rbuf.buff, rbuf.index); + + ei_x_free(&rbuf); + } + + ei_x_free(&buf); + + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Connection from %s denied by acl %s\n", listener->remote_ip, prefs.acl[x]); + goto done; + } + } + } + + if ((session = listener->session)) { + channel = switch_core_session_get_channel(session); + switch_core_session_read_lock(session); + } + + if (switch_strlen_zero(listener->remote_ip)) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Connection Open\n"); + } else { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Connection Open from %s\n", listener->remote_ip);/*, listener->remote_port);*/ + } + + switch_set_flag_locked(listener, LFLAG_RUNNING); + add_listener(listener); + + while ((status >= 0 || erl_errno == ETIMEDOUT || erl_errno == EAGAIN) && !prefs.done) { + erlang_msg msg; + + ei_x_buff buf; + ei_x_new(&buf); + + ei_x_buff rbuf; + ei_x_new_with_version(&rbuf); + + status = ei_xreceive_msg_tmo(listener->sockfd, &msg, &buf, 100); + + switch(status) { + case ERL_TICK : + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Tick\n"); + break; + case ERL_MSG : + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Message\n"); + switch(msg.msgtype) { + case ERL_SEND : + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "erl_send\n"); + if (handle_msg(listener, &msg, &buf, &rbuf)) { + goto done; + } + break; + case ERL_REG_SEND : + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "erl_reg_send\n"); + if (handle_msg(listener, &msg, &buf, &rbuf)) { + goto done; + } + break; + case ERL_LINK : + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "erl_link\n"); + break; + case ERL_UNLINK : + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "erl_unlink\n"); + break; + case ERL_EXIT : + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "erl_exit\n"); + break; + default : + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "unexpected msg type %d\n", (int)(msg.msgtype)); + break; + } + break; + case ERL_ERROR : + if (erl_errno != ETIMEDOUT && erl_errno != EAGAIN) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "erl_error\n"); + } + break; + default : + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "unexpected status %d \n", status); + break; + } + + ei_x_free(&buf); + ei_x_free(&rbuf); + + /* send out any pending crap in the log queue */ + if (switch_test_flag(listener, LFLAG_LOG)) { + if (switch_queue_trypop(listener->log_queue, &pop) == SWITCH_STATUS_SUCCESS) { + char *data = (char *) pop; + + if (data) { + ei_x_buff lbuf; + ei_x_new_with_version(&lbuf); + ei_x_encode_tuple_header(&lbuf, 2); + ei_x_encode_atom(&lbuf, "log"); + ei_x_encode_string(&lbuf, data); + ei_send(listener->sockfd, &listener->log_pid, lbuf.buff, lbuf.index); + ei_x_free(&lbuf); + } + } + } + + /* ditto with the event queue */ + if (switch_test_flag(listener, LFLAG_EVENTS)) { + if (switch_queue_trypop(listener->event_queue, &pop) == SWITCH_STATUS_SUCCESS) { + + switch_event_t *pevent = (switch_event_t *) pop; + + ei_x_buff ebuf; + ei_x_new_with_version(&ebuf); + + ei_encode_switch_event(&ebuf, pevent); + + ei_send(listener->sockfd, &listener->event_pid, ebuf.buff, ebuf.index); + + ei_x_free(&ebuf); + switch_event_destroy(&pevent); + } + } + } + +done: + remove_listener(listener); + + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Session complete, waiting for children\n"); + + switch_thread_rwlock_wrlock(listener->rwlock); + + if (listener->sockfd) { + close(listener->sockfd); + } + + switch_thread_rwlock_unlock(listener->rwlock); + + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Connection Closed\n"); + switch_core_hash_destroy(&listener->event_hash); + + if (listener->session) { + switch_channel_clear_flag(switch_core_session_get_channel(listener->session), CF_CONTROLLED); + switch_clear_flag_locked(listener, LFLAG_SESSION); + switch_core_session_rwunlock(listener->session); + } else if (listener->pool) { + switch_memory_pool_t *pool = listener->pool; + switch_core_destroy_memory_pool(&pool); + } + + switch_mutex_lock(listen_list.mutex); + prefs.threads--; + switch_mutex_unlock(listen_list.mutex); + + return NULL; +} + + +/* Create a thread for the socket and launch it */ +static void launch_listener_thread(listener_t *listener) +{ + switch_thread_t *thread; + switch_threadattr_t *thd_attr = NULL; + + switch_threadattr_create(&thd_attr, listener->pool); + switch_threadattr_detach_set(thd_attr, 1); + switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE); + switch_thread_create(&thread, thd_attr, listener_run, listener, listener->pool); +} + + +static int config(void) +{ + char *cf = "erlang_event.conf"; + switch_xml_t cfg, xml, settings, param; + + memset(&prefs, 0, sizeof(prefs)); + + prefs.shortname = SWITCH_TRUE; + + if (!(xml = switch_xml_open_cfg(cf, &cfg, NULL))) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Open of %s failed\n", cf); + } else { + if ((settings = switch_xml_child(cfg, "settings"))) { + for (param = switch_xml_child(settings, "param"); param; param = param->next) { + char *var = (char *) switch_xml_attr_soft(param, "name"); + char *val = (char *) switch_xml_attr_soft(param, "value"); + + if (!strcmp(var, "listen-ip")) { + set_pref_ip(val); + } else if (!strcmp(var, "listen-port")) { + prefs.port = (uint16_t) atoi(val); + } else if (!strcmp(var, "cookie")) { + set_pref_cookie(val); + } else if (!strcmp(var, "nodename")) { + set_pref_nodename(val); + } else if (!strcmp(var, "shortname") && strcmp(val, "yes")) { + prefs.shortname = SWITCH_FALSE; + } else if (!strcasecmp(var, "apply-inbound-acl")) { + if (prefs.acl_count < MAX_ACL) { + prefs.acl[prefs.acl_count++] = strdup(val); + } else { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Max acl records of %d reached\n", MAX_ACL); + } + } + } + } + switch_xml_free(xml); + } + + if (switch_strlen_zero(prefs.ip)) { + set_pref_ip("127.0.0.1"); + } + + if (switch_strlen_zero(prefs.cookie)) { + set_pref_cookie("ClueCon"); + } + + if (!prefs.port) { + prefs.port = 8031; + } + + if (!prefs.nodename) { + prefs.nodename = "freeswitch"; + } + + return 0; +} + + +SWITCH_MODULE_RUNTIME_FUNCTION(mod_erlang_event_runtime) +{ + switch_memory_pool_t *pool = NULL, *listener_pool = NULL; + switch_status_t rv; + listener_t *listener; + uint32_t x = 0; + struct ei_cnode_s ec; + ErlConnect conn; + struct sockaddr_in server_addr; + int on = 1; + int clientfd; + + memset(&listen_list, 0, sizeof(listen_list)); + config(); + + if (switch_core_new_memory_pool(&pool) != SWITCH_STATUS_SUCCESS) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "OH OH no pool\n"); + return SWITCH_STATUS_TERM; + } + + switch_mutex_init(&listen_list.mutex, SWITCH_MUTEX_NESTED, pool); + switch_mutex_init(&listen_list.sock_mutex, SWITCH_MUTEX_NESTED, pool); + + /* zero out the struct before we use it */ + memset(&server_addr, 0, sizeof(server_addr)); + + /* convert the configured IP to network byte order, handing errors */ + rv = inet_pton(AF_INET, prefs.ip, &server_addr.sin_addr.s_addr); + if (rv == 0) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Could not parse invalid ip address: %s\n", prefs.ip); + return SWITCH_STATUS_GENERR; + } else if (rv == -1) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error when parsing ip address %s : %s\n", prefs.ip, strerror(errno)); + return SWITCH_STATUS_GENERR; + } + + /* set the address family and port */ + server_addr.sin_family = AF_INET; + server_addr.sin_port = htons(prefs.port); + + /* do the socket setup ei is too lazy to do for us */ + for (;;) { + + if ((listen_list.sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to allocate socket on %s:%u\n", prefs.ip, prefs.port); + goto sock_fail; + } + + if (setsockopt(listen_list.sockfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) != 0) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to enable SO_REUSEADDR for socket on %s:%u : %s\n", prefs.ip, prefs.port, strerror(errno)); + goto sock_fail; + } + + if (bind(listen_list.sockfd, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to bind to %s:%u\n", prefs.ip, prefs.port); + goto sock_fail; + } + + if (listen(listen_list.sockfd, 5) < 0) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to listen on %s:%u\n", prefs.ip, prefs.port); + goto sock_fail; + } + + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Socket up listening on %s:%u\n", prefs.ip, prefs.port); + break; + sock_fail: + switch_yield(100000); + } + + struct hostent *nodehost = gethostbyaddr(&server_addr.sin_addr.s_addr, sizeof(server_addr.sin_addr.s_addr), AF_INET); + + char *thishostname = nodehost->h_name; + char thisnodename[MAXNODELEN+1]; + + if (!strcmp(thishostname, "localhost")) + gethostname(thishostname, EI_MAXHOSTNAMELEN); + + if (prefs.shortname) { + char *off; + if ((off = strchr(thishostname, '.'))) { + *off = '\0'; + } + } + + snprintf(thisnodename, MAXNODELEN+1, "%s@%s", prefs.nodename, thishostname); + + /* init the ei stuff */ + if (ei_connect_xinit(&ec, thishostname, prefs.nodename, thisnodename, (Erl_IpAddr)(&server_addr.sin_addr.s_addr), prefs.cookie, 0) < 0) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to init ei connection\n"); + return SWITCH_STATUS_GENERR; + } + + /* return value is -1 for error, a descriptor pointing to epmd otherwise */ + if (ei_publish(&ec, prefs.port) == -1) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to publish port to empd\n"); + return SWITCH_STATUS_GENERR; + } + + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Connected and published erlang cnode at %s port %u\n", thisnodename, prefs.port); + + listen_list.ready = 1; + + if (switch_event_bind_removable(modname, SWITCH_EVENT_ALL, SWITCH_EVENT_SUBCLASS_ANY, event_handler, NULL, &globals.node) != SWITCH_STATUS_SUCCESS) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't bind!\n"); + return SWITCH_STATUS_GENERR; + } + + switch_log_bind_logger(socket_logger, SWITCH_LOG_DEBUG, SWITCH_FALSE); + + for (;;) { + if (switch_core_new_memory_pool(&listener_pool) != SWITCH_STATUS_SUCCESS) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "OH OH no pool\n"); + goto fail; + } + + /* zero out errno because ei_accept doesn't differentiate between a + * failed authentication or a socket failure, or a client version + * mismatch or a godzilla attack */ + errno = 0; + if ((clientfd = ei_accept_tmo(&ec, listen_list.sockfd, &conn, 100)) == ERL_ERROR) { + if (prefs.done) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Shutting Down\n"); + } else if (erl_errno == ETIMEDOUT) { + continue; + } else if (errno) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Socket Error %d %d\n", erl_errno, errno); + } else { + /* if errno didn't get set, assume nothing *too* horrible occured */ + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, + "Ignorable error in ei_accept - probable bad client version, bad cookie or bad nodename\n"); + continue; + } + break; + } + + if (!(listener = switch_core_alloc(listener_pool, sizeof(*listener)))) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Memory Error\n"); + break; + } + + switch_thread_rwlock_create(&listener->rwlock, listener_pool); + switch_queue_create(&listener->event_queue, SWITCH_CORE_QUEUE_LEN, listener_pool); + switch_queue_create(&listener->log_queue, SWITCH_CORE_QUEUE_LEN, listener_pool); + + inet_ntop(AF_INET, conn.ipadr, listener->remote_ip, sizeof(listener->remote_ip)); + + listener->ec = &ec; + listener->sockfd = clientfd; + listener->pool = listener_pool; + listener_pool = NULL; + listener->level = SWITCH_LOG_DEBUG; + switch_mutex_init(&listener->flag_mutex, SWITCH_MUTEX_NESTED, listener->pool); + switch_core_hash_init(&listener->event_hash, listener->pool); + + launch_listener_thread(listener); + + } + + close(listen_list.sockfd); + + if (pool) { + switch_core_destroy_memory_pool(&pool); + } + + if (listener_pool) { + switch_core_destroy_memory_pool(&listener_pool); + } + + + for (x = 0; x < prefs.acl_count; x++) { + switch_safe_free(prefs.acl[x]); + } + + fail: + return SWITCH_STATUS_TERM; +} + +/* 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: + */

1 0

[commit] r10273 - freeswitch/trunk/src/mod/event_handlers/mod_erlang_event
by Freeswitch SVN 07 Nov '08

07 Nov '08

Author: mikej Date: Thu Nov 6 15:56:05 2008 New Revision: 10273 Added: freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/ Log: add folder for new user contributed module.

1 0

[commit] r10272 - in freeswitch/trunk: . src src/include
by Freeswitch SVN 07 Nov '08

07 Nov '08

Author: mikej Date: Thu Nov 6 12:29:50 2008 New Revision: 10272 Modified: freeswitch/trunk/Makefile.am freeswitch/trunk/src/include/switch.h freeswitch/trunk/src/include/switch_dso.h freeswitch/trunk/src/switch_dso.c freeswitch/trunk/src/switch_loadable_module.c Log: add our own module load code (LBAPR-1) Modified: freeswitch/trunk/Makefile.am ============================================================================== --- freeswitch/trunk/Makefile.am (original) +++ freeswitch/trunk/Makefile.am Thu Nov 6 12:29:50 2008 @@ -70,6 +70,7 @@ src/switch_core.c \ src/switch_scheduler.c \ src/switch_core_db.c\ +src/switch_dso.c\ src/switch_loadable_module.c \ src/switch_utils.c \ src/switch_event.c \ @@ -114,6 +115,7 @@ src/include/switch_event.h\ src/include/switch_frame.h\ src/include/switch_ivr.h\ +src/include/switch_dso.h\ src/include/switch_loadable_module.h\ src/include/switch_module_interfaces.h\ src/include/switch_platform.h\ Modified: freeswitch/trunk/src/include/switch.h ============================================================================== --- freeswitch/trunk/src/include/switch.h (original) +++ freeswitch/trunk/src/include/switch.h Thu Nov 6 12:29:50 2008 @@ -98,6 +98,7 @@ #include "switch_apr.h" #include "switch_core_db.h" +#include "switch_dso.h" #include "switch_regex.h" #include "switch_core.h" #include "switch_loadable_module.h" Modified: freeswitch/trunk/src/include/switch_dso.h ============================================================================== --- freeswitch/trunk/src/include/switch_dso.h (original) +++ freeswitch/trunk/src/include/switch_dso.h Thu Nov 6 12:29:50 2008 @@ -21,21 +21,19 @@ #ifndef _SWITCH_DSO_H #define _SWITCH_DSO_H -typedef void (*switch_func_ptr_t) (void); +typedef int (*switch_dso_func_t) (void); #ifdef WIN32 typedef HINSTANCE switch_dso_lib_t; #else typedef void * switch_dso_lib_t; #endif -#ifdef WIN32 -typedef FARPROC switch_dso_func_t; -#else -typedef void * switch_dso_func_t; -#endif + +typedef void * switch_dso_data_t; void switch_dso_destroy(switch_dso_lib_t *lib); switch_dso_lib_t switch_dso_open(const char *path, int global, char **err); switch_dso_func_t switch_dso_func_sym(switch_dso_lib_t lib, const char *sym, char **err); +void *switch_dso_data_sym(switch_dso_lib_t lib, const char *sym, char **err); #endif Modified: freeswitch/trunk/src/switch_dso.c ============================================================================== --- freeswitch/trunk/src/switch_dso.c (original) +++ freeswitch/trunk/src/switch_dso.c Thu Nov 6 12:29:50 2008 @@ -54,9 +54,19 @@ DWORD error = GetLastError(); *err = switch_mprintf("dll sym error [%ul]\n", error); } - return func; + return (switch_dso_func_t)func; } +void *switch_dso_data_sym(switch_dso_lib_t lib, const char *sym, char **err) { + FARPROC addr = GetProcAddress(lib, sym); + if (!addr) { + DWORD error = GetLastError(); + *err = switch_mprintf("dll sym error [%ul]\n", error); + } + return (void *)(intptr_t)addr; +} + + #else /* ** {======================================================================== @@ -92,13 +102,22 @@ return lib; } -void *switch_dso_func_sym(switch_dso_lib_t lib, const char *sym, char **err) { +switch_dso_func_t switch_dso_func_sym(switch_dso_lib_t lib, const char *sym, char **err) { void *func = dlsym(lib, sym); if (!func) { *err = strdup(dlerror()); } - return func; + return (switch_dso_func_t)(intptr_t)func; } + +void *switch_dso_data_sym(switch_dso_lib_t lib, const char *sym, char **err) { + void *addr = dlsym(lib, sym); + if (!addr) { + *err = strdup(dlerror()); + } + return addr; +} + #endif /* }====================================================== */ Modified: freeswitch/trunk/src/switch_loadable_module.c ============================================================================== --- freeswitch/trunk/src/switch_loadable_module.c (original) +++ freeswitch/trunk/src/switch_loadable_module.c Thu Nov 6 12:29:50 2008 @@ -46,7 +46,7 @@ char *filename; int perm; switch_loadable_module_interface_t *module_interface; - void *lib; + switch_dso_lib_t lib; switch_module_load_t switch_module_load; switch_module_runtime_t switch_module_runtime; switch_module_shutdown_t switch_module_shutdown; @@ -75,6 +75,7 @@ static struct switch_loadable_module_container loadable_modules; static void do_shutdown(switch_loadable_module_t *module, switch_bool_t shutdown, switch_bool_t unload); +static switch_status_t switch_loadable_module_load_module_ex(char *dir, char *fname, switch_bool_t runtime, switch_bool_t global, const char **err); static void *switch_loadable_module_exec(switch_thread_t *thread, void *obj) { @@ -676,22 +677,19 @@ } -static switch_status_t switch_loadable_module_load_file(char *path, char *filename, switch_loadable_module_t **new_module) +static switch_status_t switch_loadable_module_load_file(char *path, char *filename, switch_bool_t global, switch_loadable_module_t **new_module) { switch_loadable_module_t *module = NULL; - switch_dso_handle_t *dso = NULL; + switch_dso_lib_t dso = NULL; apr_status_t status = SWITCH_STATUS_SUCCESS; - switch_dso_handle_sym_t interface_struct_handle = NULL; + switch_loadable_module_function_table_t *interface_struct_handle = NULL; switch_loadable_module_function_table_t *mod_interface_functions = NULL; char *struct_name = NULL; - switch_dso_handle_sym_t load_function_handle = NULL; - switch_dso_handle_sym_t shutdown_function_handle = NULL; - switch_dso_handle_sym_t runtime_function_handle = NULL; switch_module_load_t load_func_ptr = NULL; int loading = 1; - const char *err = NULL; switch_loadable_module_interface_t *module_interface = NULL; - char derr[512] = ""; + char *derr = NULL; + const char *err = NULL; switch_memory_pool_t *pool; switch_assert(path != NULL); @@ -702,36 +700,40 @@ struct_name = switch_core_sprintf(pool, "%s_module_interface", filename); #ifdef WIN32 - status = switch_dso_load(&dso, "FreeSwitch.dll", loadable_modules.pool); + dso = switch_dso_open("FreeSwitch.dll", global, &derr); #elif defined (MACOSX) || defined(DARWIN) - status = switch_dso_load(&dso, SWITCH_PREFIX_DIR "/lib/libfreeswitch.dylib", loadable_modules.pool); + dso = switch_dso_open(SWITCH_PREFIX_DIR "/lib/libfreeswitch.dylib", global, &derr); #else - status = switch_dso_load(&dso, NULL, loadable_modules.pool); + dso = switch_dso_open(NULL, global, &derr); #endif - status = switch_dso_sym(&interface_struct_handle, dso, struct_name); + if (!derr && dso) { + interface_struct_handle = switch_dso_data_sym(dso, struct_name, &derr); + } + + switch_safe_free(derr) if (!interface_struct_handle) { - status = switch_dso_load(&dso, path, loadable_modules.pool); + dso = switch_dso_open(path, global, &derr); } - while (loading) { - if (status != APR_SUCCESS) { - switch_dso_error(dso, derr, sizeof(derr)); + if (derr) { err = derr; break; } if (!interface_struct_handle) { - status = switch_dso_sym(&interface_struct_handle, dso, struct_name); + interface_struct_handle = switch_dso_data_sym(dso, struct_name, &derr); + } + + if (derr) { + err = derr; + break; } if (interface_struct_handle) { mod_interface_functions = interface_struct_handle; load_func_ptr = mod_interface_functions->load; - } else { - status = switch_dso_sym(&load_function_handle, dso, "switch_module_load"); - load_func_ptr = (switch_module_load_t) (intptr_t) load_function_handle; } if (load_func_ptr == NULL) { @@ -765,6 +767,7 @@ switch_core_destroy_memory_pool(&pool); } switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Error Loading module %s\n**%s**\n", path, err); + switch_safe_free(derr); return SWITCH_STATUS_GENERR; } @@ -776,11 +779,6 @@ if (mod_interface_functions) { module->switch_module_shutdown = mod_interface_functions->shutdown; module->switch_module_runtime = mod_interface_functions->runtime; - } else { - switch_dso_sym(&shutdown_function_handle, dso, "switch_module_shutdown"); - module->switch_module_shutdown = (switch_module_shutdown_t) (intptr_t) shutdown_function_handle; - switch_dso_sym(&runtime_function_handle, dso, "switch_module_runtime"); - module->switch_module_runtime = (switch_module_runtime_t) (intptr_t) runtime_function_handle; } module->lib = dso; @@ -791,9 +789,13 @@ return SWITCH_STATUS_SUCCESS; } - SWITCH_DECLARE(switch_status_t) switch_loadable_module_load_module(char *dir, char *fname, switch_bool_t runtime, const char **err) { + return switch_loadable_module_load_module_ex(dir, fname, runtime, SWITCH_FALSE, err); +} + +static switch_status_t switch_loadable_module_load_module_ex(char *dir, char *fname, switch_bool_t runtime, switch_bool_t global, const char **err) +{ switch_size_t len = 0; char *path; char *file, *dot; @@ -833,7 +835,7 @@ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "Module %s Already Loaded!\n", file); *err = "Module already loaded"; status = SWITCH_STATUS_FALSE; - } else if ((status = switch_loadable_module_load_file(path, file, &new_module)) == SWITCH_STATUS_SUCCESS) { + } else if ((status = switch_loadable_module_load_file(path, file, global, &new_module)) == SWITCH_STATUS_SUCCESS) { if ((status = switch_loadable_module_process(file, new_module)) == SWITCH_STATUS_SUCCESS && runtime) { if (new_module->switch_module_runtime) { switch_core_launch_thread(switch_loadable_module_exec, new_module, new_module->pool); @@ -1042,12 +1044,15 @@ switch_xml_t mods, ld; if ((mods = switch_xml_child(cfg, "modules"))) { for (ld = switch_xml_child(mods, "load"); ld; ld = ld->next) { + switch_bool_t global = SWITCH_FALSE; const char *val = switch_xml_attr_soft(ld, "module"); + const char *sglobal = switch_xml_attr_soft(ld, "global"); if (switch_strlen_zero(val) || (strchr(val, '.') && !strstr(val, ext) && !strstr(val, EXT))) { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "Invalid extension for %s\n", val); continue; } - switch_loadable_module_load_module((char *) SWITCH_GLOBAL_dirs.mod_dir, (char *) val, SWITCH_FALSE, &err); + global = switch_true(sglobal); + switch_loadable_module_load_module_ex((char *) SWITCH_GLOBAL_dirs.mod_dir, (char *) val, SWITCH_FALSE, global, &err); count++; } } @@ -1062,12 +1067,15 @@ if ((mods = switch_xml_child(cfg, "modules"))) { for (ld = switch_xml_child(mods, "load"); ld; ld = ld->next) { + switch_bool_t global = SWITCH_FALSE; const char *val = switch_xml_attr_soft(ld, "module"); + const char *sglobal = switch_xml_attr_soft(ld, "global"); if (switch_strlen_zero(val) || (strchr(val, '.') && !strstr(val, ext) && !strstr(val, EXT))) { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "Invalid extension for %s\n", val); continue; } - switch_loadable_module_load_module((char *) SWITCH_GLOBAL_dirs.mod_dir, (char *) val, SWITCH_FALSE, &err); + global = switch_true(sglobal); + switch_loadable_module_load_module_ex((char *) SWITCH_GLOBAL_dirs.mod_dir, (char *) val, SWITCH_FALSE, global, &err); count++; } } @@ -1135,8 +1143,7 @@ if (unload && module->status != SWITCH_STATUS_NOUNLOAD && !(flags & SCF_VG)) { switch_memory_pool_t *pool; switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "%s unloaded.\n", module->module_interface->module_name); - switch_dso_unload(module->lib); - module->lib = NULL; + switch_dso_destroy(&module->lib); if ((pool = module->pool)) { module = NULL; switch_core_destroy_memory_pool(&pool);

1 0