diff --git a/apps/db_reg_agent/DBRegAgent.cpp b/apps/db_reg_agent/DBRegAgent.cpp new file mode 100644 index 00000000..9eeece56 --- /dev/null +++ b/apps/db_reg_agent/DBRegAgent.cpp @@ -0,0 +1,988 @@ +/* + * Copyright (C) 2011 Stefan Sayer + * + * This file is part of SEMS, a free SIP media server. + * + * SEMS is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * For a license to use the sems software under conditions + * other than those described here, or to purchase support for this + * software, please contact iptel.org by e-mail at the following addresses: + * info@iptel.org + * + * SEMS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +#include "DBRegAgent.h" +#include "AmSession.h" +#include "AmEventDispatcher.h" + +#include +#include + +EXPORT_MODULE_FACTORY(DBRegAgent); +DEFINE_MODULE_INSTANCE(DBRegAgent, MOD_NAME); + +mysqlpp::Connection DBRegAgent::DBConnection(mysqlpp::use_exceptions); +mysqlpp::Connection DBRegAgent::StatusDBConnection(mysqlpp::use_exceptions); + +string DBRegAgent::joined_query; +string DBRegAgent::registrations_table = "registrations"; + +double DBRegAgent::reregister_interval = 0.5; +double DBRegAgent::minimum_reregister_interval = -1; + +bool DBRegAgent::enable_ratelimiting = false; +unsigned int DBRegAgent::ratelimit_rate = 0; +unsigned int DBRegAgent::ratelimit_per = 0; + +static void _timer_cb(RegTimer* timer, long subscriber_id, void* data2) { + DBRegAgent::instance()->timer_cb(timer, subscriber_id, data2); +} + +DBRegAgent::DBRegAgent(const string& _app_name) + : AmDynInvokeFactory(_app_name), + AmEventQueue(this), + uac_auth_i(NULL) +{ +} + +DBRegAgent::~DBRegAgent() { +} + +int DBRegAgent::onLoad() +{ + + DBG("loading db_reg_agent....\n"); + + AmDynInvokeFactory* uac_auth_f = AmPlugIn::instance()->getFactory4Di("uac_auth"); + if (uac_auth_f == NULL) { + WARN("unable to get a uac_auth factory. " + "registrations will not be authenticated.\n"); + WARN("(do you want to load uac_auth module?)\n"); + } else { + uac_auth_i = uac_auth_f->getInstance(); + } + + AmConfigReader cfg; + if(cfg.loadFile(add2path(AmConfig::ModConfigPath,1, MOD_NAME ".conf"))) + return -1; + + expires = cfg.getParameterInt("expires", 7200); + DBG("requesting registration expires of %u seconds\n", expires); + + if (cfg.hasParameter("reregister_interval")) { + reregister_interval = -1; + reregister_interval = atof(cfg.getParameter("reregister_interval").c_str()); + if (reregister_interval <= 0 || reregister_interval > 1) { + ERROR("configuration value 'reregister_interval' could not be read. " + "needs to be 0 .. 1.0 (recommended: 0.5)\n"); + return -1; + } + } + + if (cfg.hasParameter("minimum_reregister_interval")) { + minimum_reregister_interval = -1; + minimum_reregister_interval = atof(cfg.getParameter("minimum_reregister_interval").c_str()); + if (minimum_reregister_interval <= 0 || minimum_reregister_interval > 1) { + ERROR("configuration value 'minimum_reregister_interval' could not be read. " + "needs to be 0 .. reregister_interval (recommended: 0.4)\n"); + return -1; + } + + if (minimum_reregister_interval >= reregister_interval) { + ERROR("configuration value 'minimum_reregister_interval' must be smaller " + "than reregister_interval (recommended: 0.4)\n"); + return -1; + } + } + + enable_ratelimiting = cfg.getParameter("enable_ratelimiting") == "yes"; + if (enable_ratelimiting) { + if (!cfg.hasParameter("ratelimit_rate") || !cfg.hasParameter("ratelimit_per")) { + ERROR("if ratelimiting is enabled, ratelimit_rate and ratelimit_per must be set\n"); + return -1; + } + ratelimit_rate = cfg.getParameterInt("ratelimit_rate", 0); + ratelimit_per = cfg.getParameterInt("ratelimit_per", 0); + if (!ratelimit_rate || !ratelimit_per) { + ERROR("ratelimit_rate and ratelimit_per must be > 0\n"); + return -1; + } + } + + string mysql_server, mysql_user, mysql_passwd, mysql_db; + + mysql_server = cfg.getParameter("mysql_server", "localhost"); + + mysql_user = cfg.getParameter("mysql_user"); + if (mysql_user.empty()) { + ERROR(MOD_NAME ".conf parameter 'mysql_user' is missing.\n"); + return -1; + } + + mysql_passwd = cfg.getParameter("mysql_passwd"); + if (mysql_passwd.empty()) { + ERROR(MOD_NAME ".conf parameter 'mysql_passwd' is missing.\n"); + return -1; + } + + mysql_db = cfg.getParameter("mysql_db", "sems"); + + try { + + DBConnection.set_option(new mysqlpp::ReconnectOption(true)); + DBConnection.connect(mysql_db.c_str(), mysql_server.c_str(), + mysql_user.c_str(), mysql_passwd.c_str()); + if (!DBConnection) { + ERROR("Database connection failed: %s\n", DBConnection.error()); + return -1; + } + + StatusDBConnection.set_option(new mysqlpp::ReconnectOption(true)); + StatusDBConnection.connect(mysql_db.c_str(), mysql_server.c_str(), + mysql_user.c_str(), mysql_passwd.c_str()); + if (!StatusDBConnection) { + ERROR("Database connection failed: %s\n", StatusDBConnection.error()); + return -1; + } + + } catch (const mysqlpp::Exception& er) { + // Catch-all for any MySQL++ exceptions + ERROR("MySQL++ error: %s\n", er.what()); + return -1; + } + + // register us as SIP event receiver for MOD_NAME + AmEventDispatcher::instance()->addEventQueue(MOD_NAME,this); + + if (!AmPlugIn::registerDIInterface(MOD_NAME, this)) { + ERROR("registering "MOD_NAME" DI interface\n"); + return -1; + } + + joined_query = cfg.getParameter("joined_query"); + if (joined_query.empty()) { + // todo: name! + ERROR("joined_query must be set\n"); + return -1; + } + + if (cfg.hasParameter("registrations_table")) { + registrations_table = cfg.getParameter("registrations_table"); + } + DBG("using registrations table '%s'\n", registrations_table.c_str()); + + if (!loadRegistrations()) { + ERROR("loading registrations from DB\n"); + return -1; + } + + DBG("starting registration timer thread...\n"); + registration_timer.start(); + + // run_tests(); + + start(); + + return 0; +} + +bool DBRegAgent::loadRegistrations() { + try { + time_t now_time = time(NULL); + + mysqlpp::Query query = DBRegAgent::DBConnection.query(); + + string query_string, table; + + query_string = joined_query + ";"; + + DBG("querying all registrations with : '%s'\n", + query_string.c_str()); + + query << query_string; + mysqlpp::UseQueryResult res = query.use(); + + // mysqlpp::Row::size_type row_count = res.num_rows(); + // DBG("got %zd subscriptions\n", row_count); + + while (mysqlpp::Row row = res.fetch_row()) { + int status = 0; + long subscriber_id = row[COLNAME_SUBSCRIBER_ID]; + + if (row[COLNAME_STATUS] != mysqlpp::null) + status = row[COLNAME_STATUS]; + else { + DBG("registration status entry for id %ld does not exist, creating...\n", + subscriber_id); + createDBRegistration(subscriber_id, StatusDBConnection); + } + + DBG("got subscriber '%s@%s' status %i\n", + string(row[COLNAME_USER]).c_str(), string(row[COLNAME_REALM]).c_str(), + status); + + switch (status) { + case REG_STATUS_INACTIVE: + case REG_STATUS_PENDING: // try again + case REG_STATUS_FAILED: // try again + { + createRegistration(subscriber_id, + (string)row[COLNAME_USER], + (string)row[COLNAME_PASS], + (string)row[COLNAME_REALM] + ); + scheduleRegistration(subscriber_id); + }; break; + + case REG_STATUS_ACTIVE: + { + createRegistration(subscriber_id, + (string)row[COLNAME_USER], + (string)row[COLNAME_PASS], + (string)row[COLNAME_REALM] + ); + + time_t dt_expiry = now_time; + if (row[COLNAME_EXPIRY] != mysqlpp::null) { + dt_expiry = (time_t)((mysqlpp::DateTime)row[COLNAME_EXPIRY]); + } + + time_t dt_registration_ts = now_time; + if (row[COLNAME_REGISTRATION_TS] != mysqlpp::null) { + dt_registration_ts = (time_t)((mysqlpp::DateTime)row[COLNAME_REGISTRATION_TS]); + } + + DBG("got expiry '%ld, registration_ts %ld, now %ld'\n", + dt_expiry, dt_registration_ts, now_time); + + if (dt_registration_ts > now_time) { + WARN("needed to sanitize last_registration timestamp TS from the %ld (now %ld) - " + "DB host time mismatch?\n", dt_registration_ts, now_time); + dt_registration_ts = now_time; + } + + // if expired add to pending registrations, else schedule re-regstration + if (dt_expiry <= now_time) { + DBG("scheduling imminent re-registration for subscriber %ld\n", subscriber_id); + scheduleRegistration(subscriber_id); + } else { + setRegistrationTimer(subscriber_id, dt_expiry, dt_registration_ts); + } + + }; break; + case REG_STATUS_REMOVED: + { + DBG("ignoring removed registration %ld %s@%s", + subscriber_id, + ((string)row[COLNAME_USER]).c_str(), ((string)row[COLNAME_REALM]).c_str()); + } break; + } + } + + + } catch (const mysqlpp::Exception& er) { + // Catch-all for any MySQL++ exceptions + ERROR("MySQL++ error: %s\n", er.what()); + return false; + } + + return true; +} + +/** create registration in our list */ +void DBRegAgent::createRegistration(long subscriber_id, + const string& user, + const string& pass, + const string& realm) { + string handle = AmSession::getNewId(); + SIPRegistrationInfo reg_info(realm, user, + user, // name + user, // auth_user + pass, + "", // proxy + "" // contact + ); + + registrations_mut.lock(); + try { + if (registrations.find(subscriber_id) != registrations.end()) { + registrations_mut.unlock(); + WARN("registration with ID %ld already exists, removing\n", subscriber_id); + removeRegistration(subscriber_id); + registrations_mut.lock(); + } + + AmSIPRegistration* reg = new AmSIPRegistration(handle, reg_info, "" /*MOD_NAME*/); + reg->setExpiresInterval(expires); + + registrations[subscriber_id] = reg; + registration_ltags[handle] = subscriber_id; + + if (NULL != uac_auth_i) { + DBG("enabling UAC Auth for new registration.\n"); + + // get a sessionEventHandler from uac_auth + AmArg di_args,ret; + AmArg a; + a.setBorrowedPointer(reg); + di_args.push(a); + di_args.push(a); + + uac_auth_i->invoke("getHandler", di_args, ret); + if (!ret.size()) { + ERROR("Can not add auth handler to new registration!\n"); + } else { + ArgObject* p = ret.get(0).asObject(); + if (p != NULL) { + AmSessionEventHandler* h = dynamic_cast(p); + if (h != NULL) + reg->setSessionEventHandler(h); + } + } + } + } catch (const AmArg::OutOfBoundsException& e) { + ERROR("OutOfBoundsException"); + } catch (const AmArg::TypeMismatchException& e) { + ERROR("TypeMismatchException"); + } catch (...) { + ERROR("unknown exception occured\n"); + } + + registrations_mut.unlock(); + + // register us as SIP event receiver for this ltag + AmEventDispatcher::instance()->addEventQueue(handle,this); + + DBG("created new registration with ID %ld and ltag '%s'\n", + subscriber_id, handle.c_str()); +} + +void DBRegAgent::updateRegistration(long subscriber_id, + const string& user, + const string& pass, + const string& realm) { + + registrations_mut.lock(); + map::iterator it=registrations.find(subscriber_id); + if (it == registrations.end()) { + registrations_mut.unlock(); + WARN("updateRegistration - registration %ld %s@%s unknown, creating\n", + subscriber_id, user.c_str(), realm.c_str()); + createRegistration(subscriber_id, user, pass, realm); + scheduleRegistration(subscriber_id); + return; + } + + it->second->setRegistrationInfo(SIPRegistrationInfo(realm, user, + user, // name + user, // auth_user + pass, + "", // proxy + "")); // contact + registrations_mut.unlock(); +} + +/** remove registration from our list */ +void DBRegAgent::removeRegistration(long subscriber_id) { + bool res = false; + string handle; + registrations_mut.lock(); + map::iterator it = registrations.find(subscriber_id); + if (it != registrations.end()) { + handle = it->second->getHandle(); + registration_ltags.erase(handle); + delete it->second; + registrations.erase(it); + res = true; + } + registrations_mut.unlock(); + + if (res) { + // deregister us as SIP event receiver for this ltag + AmEventDispatcher::instance()->delEventQueue(handle); + + DBG("removed registration with ID %ld\n", subscriber_id); + } else { + DBG("registration with ID %ld not found for removing\n", subscriber_id); + } +} + +/** schedule this registration to REGISTER (immediately) */ +void DBRegAgent::scheduleRegistration(long subscriber_id) { + if (enable_ratelimiting) { + registration_processor. + postEvent(new RegistrationActionEvent(RegistrationActionEvent::Register, + subscriber_id)); + } else { + // use our own thread + postEvent(new RegistrationActionEvent(RegistrationActionEvent::Register, + subscriber_id)); + } + DBG("added to pending actions: REGISTER of %ld\n", subscriber_id); +} + +/** schedule this registration to de-REGISTER (immediately) */ +void DBRegAgent::scheduleDeregistration(long subscriber_id) { + if (enable_ratelimiting) { + registration_processor. + postEvent(new RegistrationActionEvent(RegistrationActionEvent::Deregister, + subscriber_id)); + } else { + // use our own thread + postEvent(new RegistrationActionEvent(RegistrationActionEvent::Deregister, + subscriber_id)); + } + DBG("added to pending actions: DEREGISTER of %ld\n", subscriber_id); +} + +void DBRegAgent::process(AmEvent* ev) { + + if (ev->event_id == RegistrationActionEventID) { + RegistrationActionEvent* reg_action_ev = + dynamic_cast(ev); + if (reg_action_ev) { + onRegistrationActionEvent(reg_action_ev); + return; + } + } + + AmSipReplyEvent* sip_rep = dynamic_cast(ev); + if (sip_rep) { + onSipReplyEvent(sip_rep); + return; + } + + if (ev->event_id == E_SYSTEM) { + AmSystemEvent* sys_ev = dynamic_cast(ev); + if(sys_ev){ + DBG("Session received system Event\n"); + if (sys_ev->sys_event == AmSystemEvent::ServerShutdown) { + running = false; + } + return; + } + } + + ERROR("unknown event received!\n"); +} + +void DBRegAgent::onRegistrationActionEvent(RegistrationActionEvent* reg_action_ev) { + switch (reg_action_ev->action) { + case RegistrationActionEvent::Register: + { + DBG("REGISTER of registration %i\n", reg_action_ev->subscriber_id); + registrations_mut.lock(); + map::iterator it= + registrations.find(reg_action_ev->subscriber_id); + if (it==registrations.end()) { + DBG("ignoring scheduled REGISTER of unknown registration %i\n", + reg_action_ev->subscriber_id); + } else { + if (!it->second->doRegistration()) { + updateDBRegistration(reg_action_ev->subscriber_id, + 480, "unable to send request", + true, REG_STATUS_FAILED); + } + } + registrations_mut.unlock(); + } break; + case RegistrationActionEvent::Deregister: + { + DBG("De-REGISTER of registration %i\n", reg_action_ev->subscriber_id); + registrations_mut.lock(); + map::iterator it= + registrations.find(reg_action_ev->subscriber_id); + if (it==registrations.end()) { + DBG("ignoring scheduled De-REGISTER of unknown registration %i\n", + reg_action_ev->subscriber_id); + } else { + if (!it->second->doUnregister()) { + updateDBRegistration(reg_action_ev->subscriber_id, + 480, "unable to send request", + true, REG_STATUS_FAILED); + } + } + registrations_mut.unlock(); + } break; + } +} + +void DBRegAgent::createDBRegistration(long subscriber_id, mysqlpp::Connection& conn) { + string insert_query = "insert into "+registrations_table+ + " (subscriber_id) values ("+ + long2str(subscriber_id)+");"; + + try { + mysqlpp::Query query = conn.query(); + query << insert_query; + + mysqlpp::SimpleResult res = query.execute(); + if (!res) { + WARN("creating registration in DB with query '%s' failed: '%s'\n", + insert_query.c_str(), res.info()); + } + } catch (const mysqlpp::Exception& er) { + // Catch-all for any MySQL++ exceptions + ERROR("MySQL++ error: %s\n", er.what()); + return; + } +} + +void DBRegAgent::updateDBRegistration(long subscriber_id, int last_code, + const string& last_reason, + bool update_status, int status, + bool update_ts, unsigned int expiry) { + + string update_query = "update "+registrations_table+" set " + "last_code="+ int2str(last_code) +", " + "last_reason=\""+last_reason+"\""; + + if (update_status) { + update_query += ", registration_status="+int2str(status); + } + + if (update_ts) { + update_query += ", last_registration=NOW(), " + "expiry=TIMESTAMPADD(SECOND,"+int2str(expiry)+", NOW())"; + } + + update_query += " where " COLNAME_SUBSCRIBER_ID "="+long2str(subscriber_id) + ";"; + try { + DBG("updating registration in DB with query '%s'\n", + update_query.c_str()); + + mysqlpp::Query query = DBRegAgent::DBConnection.query(); + query << update_query; + + mysqlpp::SimpleResult res = query.execute(); + if (!res || !res.rows()) { + WARN("updating registration in DB with query '%s' failed: '%s'\n", + update_query.c_str(), res.info()); + } + } catch (const mysqlpp::Exception& er) { + // Catch-all for any MySQL++ exceptions + ERROR("MySQL++ error: %s\n", er.what()); + return; + } + +} + +void DBRegAgent::onSipReplyEvent(AmSipReplyEvent* ev) { + if (!ev) return; + DBG("received SIP reply event for '%s'\n", ev->reply.local_tag.c_str()); + + registrations_mut.lock(); + map::iterator it=registration_ltags.find(ev->reply.local_tag); + if (it!=registration_ltags.end()) { + long subscriber_id = it->second; + map::iterator r_it=registrations.find(subscriber_id); + if (r_it != registrations.end()) { + AmSIPRegistration* registration = r_it->second; + if (!registration) { + ERROR("Internal error: registration object missing\n"); + return; + } + + // AmSIPRegistration::RegistrationState state_before = r_it->second->getState(); + registration->getDlg()->updateStatus(ev->reply); + + AmSIPRegistration::RegistrationState current_state = registration->getState(); + + //update registrations set + bool update_status = false; + int status = 0; + bool update_ts = false; + unsigned int expiry = 0; + + if (ev->reply.code >= 300) { + if (current_state == AmSIPRegistration::RegisterPending) { + DBG("received negative reply, but still in pending state (auth).\n"); + } else { + // registration failed - mark in DB + DBG("registration failed - mark in DB\n"); + update_status = true; + status = REG_STATUS_FAILED; + // todo: schedule for retry? + } + } else if (ev->reply.code >= 200) { + // positive reply + if (!registration->getUnregistering()) { + setRegistrationTimer(subscriber_id, registration->getExpiresTS(), time(0)); + + update_status = true; + status = REG_STATUS_ACTIVE; + + update_ts = true; + expiry = registration->getExpiresLeft(); + } else { + update_status = true; + status = REG_STATUS_REMOVED; + } + } + + DBG("update DB with reply %u %s\n", ev->reply.code, ev->reply.reason.c_str()); + updateDBRegistration(subscriber_id, ev->reply.code, ev->reply.reason, + update_status, status, update_ts, expiry); + + } else { + ERROR("internal: inconsistent registration list\n"); + } + } else { + DBG("ignoring reply for unknown registration\n"); + } + registrations_mut.unlock(); +} + +void DBRegAgent::run() { + running = true; + + DBG("DBRegAgent thread: waiting 2 sec for server startup ...\n"); + sleep(2); + + if (enable_ratelimiting) { + DBG("starting processor thread\n"); + registration_processor.start(); + } + + DBG("running DBRegAgent thread...\n"); + while (running) { + processEvents(); + + usleep(1000); // 1ms + } + + DBG("DBRegAgent done, removing all registrations from Event Dispatcher...\n"); + registrations_mut.lock(); + for (map::iterator it=registration_ltags.begin(); + it != registration_ltags.end(); it++) { + AmEventDispatcher::instance()->delEventQueue(it->first); + } + registrations_mut.unlock(); + + DBG("removing "MOD_NAME" registrations from Event Dispatcher...\n"); + AmEventDispatcher::instance()->delEventQueue(MOD_NAME); + + DBG("DBRegAgent thread stopped.\n"); +} + +void DBRegAgent::on_stop() { + DBG("DBRegAgent on_stop()...\n"); + running = false; +} + +void DBRegAgent::setRegistrationTimer(long subscriber_id, + time_t expiry, time_t reg_start_ts) { + DBG("setting re-Register timer for subscription %ld, expiry %ld, reg_start_t %ld\n", + subscriber_id, expiry, reg_start_ts); + + RegTimer* timer = NULL; + map::iterator it=registration_timers.find(subscriber_id); + if (it==registration_timers.end()) { + DBG("timer object for subscription %ld not found\n", subscriber_id); + timer = new RegTimer(); + timer->data1 = subscriber_id; + timer->cb = _timer_cb; + DBG("created timer object [%p] for subscription %ld\n", timer, subscriber_id); + } else { + timer = it->second; + DBG("removing timer...\n"); + registration_timer.remove_timer(timer); + } + + registration_timers.insert(std::make_pair(subscriber_id, timer)); + + if (minimum_reregister_interval>0.0) { + time_t t_expiry_max = reg_start_ts; + time_t t_expiry_min = reg_start_ts; + if (expiry > reg_start_ts) + t_expiry_max+=(expiry - reg_start_ts) * reregister_interval; + if (expiry > reg_start_ts) + t_expiry_min+=(expiry - reg_start_ts) * minimum_reregister_interval; + timer->expires = t_expiry_max; + DBG("calculated re-registration at TS %ld .. %ld" + "(reg_start_ts=%ld, reg_expiry=%ld, reregister_interval=%f, " + "minimum_reregister_interval=%f)\n", + t_expiry_min, t_expiry_max, reg_start_ts, expiry, + reregister_interval, minimum_reregister_interval); + + registration_timer.insert_timer_leastloaded(timer, t_expiry_min, t_expiry_max); + + } else { + time_t t_expiry = reg_start_ts; + if (expiry > reg_start_ts) + t_expiry+=(expiry - reg_start_ts) * reregister_interval; + DBG("calculated re-registration at TS %ld " + "(reg_start_ts=%ld, reg_expiry=%ld, reregister_interval=%f)\n", + t_expiry, reg_start_ts, expiry, reregister_interval); + + timer->expires = t_expiry; + registration_timer.insert_timer(timer); + } +} + +void DBRegAgent::clearRegistrationTimer(long subscriber_id) { + DBG("removing timer for subscription %ld", subscriber_id); + + map::iterator it=registration_timers.find(subscriber_id); + if (it==registration_timers.end()) { + DBG("timer object for subscription %ld not found\n", subscriber_id); + return; + } + registration_timer.remove_timer(it->second); + + DBG("deleting timer object [%p]\n", it->second); + delete it->second; + + registration_timers.erase(it); +} + +void DBRegAgent::removeRegistrationTimer(long subscriber_id) { + DBG("removing timer object for subscription %ld", subscriber_id); + + map::iterator it=registration_timers.find(subscriber_id); + if (it==registration_timers.end()) { + DBG("timer object for subscription %ld not found\n", subscriber_id); + return; + } + + DBG("deleting timer object [%p]\n", it->second); + delete it->second; + + registration_timers.erase(it); +} + +void DBRegAgent::timer_cb(RegTimer* timer, long subscriber_id, void* data2) { + DBG("re-registration timer expired: subscriber %ld, timer=[%p]\n", subscriber_id, timer); + + scheduleRegistration(subscriber_id); + + registrations_mut.lock(); + removeRegistrationTimer(subscriber_id); + registrations_mut.unlock(); +} + + +void DBRegAgent::DIcreateRegistration(int subscriber_id, const string& user, + const string& pass, const string& realm, + AmArg& ret) { + DBG("DI method: createRegistration(%i, %s, %s, %s)\n", + subscriber_id, user.c_str(), + pass.c_str(), realm.c_str()); + + createRegistration(subscriber_id, user, pass, realm); + scheduleRegistration(subscriber_id); + ret.push(200); + ret.push("OK"); +} + +void DBRegAgent::DIupdateRegistration(int subscriber_id, const string& user, + const string& pass, const string& realm, + AmArg& ret) { + DBG("DI method: updateRegistration(%i, %s, %s, %s)\n", + subscriber_id, user.c_str(), + pass.c_str(), realm.c_str()); + updateRegistration(subscriber_id, user, pass, realm); + ret.push(200); + ret.push("OK"); +} + +void DBRegAgent::DIremoveRegistration(int subscriber_id, AmArg& ret) { + DBG("DI method: removeRegistration(%i)\n", + subscriber_id); + scheduleDeregistration(subscriber_id); + ret.push(200); + ret.push("OK"); +} + +// ///////// DI API /////////////////// + +void DBRegAgent::invoke(const string& method, + const AmArg& args, AmArg& ret) +{ + if (method == "createRegistration"){ + args.assertArrayFmt("isss"); // subscriber_id, user, pass, realm + DIcreateRegistration(args.get(0).asInt(), args.get(1).asCStr(), + args.get(2).asCStr(),args.get(3).asCStr(), + ret); + } else if (method == "updateRegistration"){ + args.assertArrayFmt("isss"); // subscriber_id, user, pass, realm + DIupdateRegistration(args.get(0).asInt(), args.get(1).asCStr(), + args.get(2).asCStr(),args.get(3).asCStr(), + ret); + } else if (method == "removeRegistration"){ + args.assertArrayFmt("i"); // subscriber_id + DIremoveRegistration(args.get(0).asInt(), ret); + } else if(method == "_list"){ + ret.push(AmArg("createRegistration")); + ret.push(AmArg("updateRegistration")); + ret.push(AmArg("removeRegistration")); + } else + throw AmDynInvoke::NotImplemented(method); +} + +// /////////////// processor thread ///////////////// + +DBRegAgentProcessorThread::DBRegAgentProcessorThread() + : AmEventQueue(this), stopped(false) { +} + +DBRegAgentProcessorThread::~DBRegAgentProcessorThread() { +} + +void DBRegAgentProcessorThread::on_stop() { +} + +void DBRegAgentProcessorThread::rateLimitWait() { + DBG("applying rate limit %u initial requests per %us\n", + DBRegAgent::ratelimit_rate, DBRegAgent::ratelimit_per); + + DBG("allowance before ratelimit: %f\n", allowance); + + struct timeval current; + struct timeval time_passed; + gettimeofday(¤t, 0); + timersub(¤t, &last_check, &time_passed); + last_check = current; + double seconds_passed = (double)time_passed.tv_sec + + (double)time_passed.tv_usec / 1000000.0; + allowance += seconds_passed * + (double) DBRegAgent::ratelimit_rate / (double)DBRegAgent::ratelimit_per; + + if (allowance > (double)DBRegAgent::ratelimit_rate) + allowance = (double)DBRegAgent::ratelimit_rate; // enough time passed, but limit to max + if (allowance < 1.0) { + useconds_t sleep_time = 1000000.0 * (1.0 - allowance) * + ((double)DBRegAgent::ratelimit_per/(double)DBRegAgent::ratelimit_rate); + DBG("not enough allowance (%f), sleeping %d useconds\n", allowance, sleep_time); + usleep(sleep_time); + allowance=0.0; + } else { + allowance -= 1.0; + } + + DBG("allowance left: %f\n", allowance); +} + +void DBRegAgentProcessorThread::run() { + DBG("DBRegAgentProcessorThread thread started\n"); + + // register us as SIP event receiver for MOD_NAME_processor + AmEventDispatcher::instance()->addEventQueue(MOD_NAME "_processor",this); + + // initialize ratelimit + gettimeofday(&last_check, NULL); + allowance = DBRegAgent::ratelimit_rate; + + reg_agent = DBRegAgent::instance(); + while (!stopped) { + waitForEvent(); + while (eventPending()) { + rateLimitWait(); + processSingleEvent(); + } + } + DBG("DBRegAgentProcessorThread thread stopped\n"); +} + +void DBRegAgentProcessorThread::process(AmEvent* ev) { + + if (ev->event_id == E_SYSTEM) { + AmSystemEvent* sys_ev = dynamic_cast(ev); + if(sys_ev){ + DBG("Session received system Event\n"); + if (sys_ev->sys_event == AmSystemEvent::ServerShutdown) { + DBG("stopping processor thread\n"); + stopped = true; + } + return; + } + } + + + if (ev->event_id == RegistrationActionEventID) { + RegistrationActionEvent* reg_action_ev = + dynamic_cast(ev); + if (reg_action_ev) { + reg_agent->onRegistrationActionEvent(reg_action_ev); + return; + } + } + + ERROR("unknown event received!\n"); +} +#if 0 +void test_cb(RegTimer* tr, long data1, void* data2) { + DBG("cb called: [%p], data %ld / [%p]\n", tr, data1, data2); +} + +void DBRegAgent::run_tests() { + + registration_timer.start(); + + struct timeval now; + gettimeofday(&now, 0); + + RegTimer rt; + rt.expires = now.tv_sec + 10; + rt.cb=test_cb; + registration_timer.insert_timer(&rt); + + RegTimer rt2; + rt2.expires = now.tv_sec + 5; + rt2.cb=test_cb; + registration_timer.insert_timer(&rt2); + + RegTimer rt3; + rt3.expires = now.tv_sec + 15; + rt3.cb=test_cb; + registration_timer.insert_timer(&rt3); + + RegTimer rt4; + rt4.expires = now.tv_sec - 1; + rt4.cb=test_cb; + registration_timer.insert_timer(&rt4); + + RegTimer rt5; + rt5.expires = now.tv_sec + 100000; + rt5.cb=test_cb; + registration_timer.insert_timer(&rt5); + + RegTimer rt6; + rt6.expires = now.tv_sec + 100; + rt6.cb=test_cb; + registration_timer.insert_timer_leastloaded(&rt6, now.tv_sec+5, now.tv_sec+50); + + + sleep(30); + gettimeofday(&now, 0); + + RegTimer rt7; + rt6.expires = now.tv_sec + 980; + rt6.cb=test_cb; + registration_timer.insert_timer_leastloaded(&rt6, now.tv_sec+9980, now.tv_sec+9990); + + vector rts; + + for (int i=0;i<1000;i++) { + RegTimer* t = new RegTimer(); + rts.push_back(t); + t->expires = now.tv_sec + i; + t->cb=test_cb; + registration_timer.insert_timer_leastloaded(t, now.tv_sec, now.tv_sec+1000); + } + + sleep(200); +} +#endif diff --git a/apps/db_reg_agent/DBRegAgent.h b/apps/db_reg_agent/DBRegAgent.h new file mode 100644 index 00000000..cde951bc --- /dev/null +++ b/apps/db_reg_agent/DBRegAgent.h @@ -0,0 +1,231 @@ +/* + * Copyright (C) 2011 Stefan Sayer + * + * This file is part of SEMS, a free SIP media server. + * + * SEMS is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * For a license to use the sems software under conditions + * other than those described here, or to purchase support for this + * software, please contact iptel.org by e-mail at the following addresses: + * info@iptel.org + * + * SEMS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +#ifndef _DB_REGAgent_h_ +#define _DB_REGAgent_h_ +#include + +#include + + +#include +using std::map; +#include +using std::queue; + +#include "AmApi.h" +#include "AmSipRegistration.h" + +#include "RegistrationTimer.h" + +#define REG_STATUS_INACTIVE 0 +#define REG_STATUS_PENDING 1 +#define REG_STATUS_ACTIVE 2 +#define REG_STATUS_FAILED 3 +#define REG_STATUS_REMOVED 4 + +#define REG_STATUS_INACTIVE_S "0" +#define REG_STATUS_PENDING_S "1" +#define REG_STATUS_ACTIVE_S "2" +#define REG_STATUS_FAILED_S "3" +#define REG_STATUS_REMOVED_S "4" + +#define COLNAME_SUBSCRIBER_ID "subscriber_id" +#define COLNAME_USER "user" +#define COLNAME_PASS "pass" +#define COLNAME_REALM "realm" + +#define COLNAME_STATUS "registration_status" +#define COLNAME_EXPIRY "expiry" +#define COLNAME_REGISTRATION_TS "last_registration" +#define COLNAME_LAST_CODE "last_code" +#define COLNAME_LAST_REASON "last_reason" + +#define RegistrationActionEventID 117 + +struct RegistrationActionEvent : public AmEvent { + + enum RegAction { Register, Deregister }; + +RegistrationActionEvent(RegAction action, int subscriber_id) + : AmEvent(RegistrationActionEventID), + action(action), subscriber_id(subscriber_id) { } + + RegAction action; + int subscriber_id; +}; + +class DBRegAgent; + +// separate thread for REGISTER sending, which can block for rate limiting +class DBRegAgentProcessorThread +: public AmThread, + public AmEventQueue, + public AmEventHandler +{ + + DBRegAgent* reg_agent; + bool stopped; + + void rateLimitWait(); + + double allowance; + struct timeval last_check; + + protected: + void process(AmEvent* ev); + + public: + DBRegAgentProcessorThread(); + ~DBRegAgentProcessorThread(); + + void run(); + void on_stop(); + +}; + +class DBRegAgent +: public AmDynInvokeFactory, + public AmDynInvoke, + public AmThread, + public AmEventQueue, + public AmEventHandler +{ + + static string joined_query; + static string registrations_table; + + static double reregister_interval; + static double minimum_reregister_interval; + + static bool enable_ratelimiting; + static unsigned int ratelimit_rate; + static unsigned int ratelimit_per; + + map registrations; + map registration_ltags; + map registration_timers; + AmMutex registrations_mut; + + // connection used in main DBRegAgent thread + static mysqlpp::Connection DBConnection; + + // connection used in other threads + static mysqlpp::Connection StatusDBConnection; + + int onLoad(); + + RegistrationTimer registration_timer; + DBRegAgentProcessorThread registration_processor; + + bool loadRegistrations(); + + void createDBRegistration(long subscriber_id, mysqlpp::Connection& conn); + void updateDBRegistration(long subscriber_id, int last_code, + const string& last_reason, + bool update_status = false, int status = 0, + bool update_ts=false, unsigned int expiry = 0); + + /** create registration in our list */ + void createRegistration(long subscriber_id, + const string& user, + const string& pass, + const string& realm); + /** update registration in our list */ + void updateRegistration(long subscriber_id, + const string& user, + const string& pass, + const string& realm); + + /** remove registration */ + void removeRegistration(long subscriber_id); + + /** schedule this subscriber to REGISTER imminently */ + void scheduleRegistration(long subscriber_id); + + /** schedule this subscriber to de-REGISTER imminently*/ + void scheduleDeregistration(long subscriber_id); + + /** create a timer for that registration + @param subscriber_id - ID of subscription + @param expiry - SIP registration expiry time + @param reg_start_ts - start TS of the SIP registration + */ + void setRegistrationTimer(long subscriber_id, time_t expiry, time_t reg_start_ts); + + /** clear re-registration timer and remove timer object */ + void clearRegistrationTimer(long subscriber_id); + + /** remove timer object */ + void removeRegistrationTimer(long subscriber_id); + + // void run_tests(); + + // amThread + void run(); + void on_stop(); + + // AmEventHandler + void process(AmEvent* ev); + + void onSipReplyEvent(AmSipReplyEvent* ev); + + void onRegistrationActionEvent(RegistrationActionEvent* reg_action_ev); + + + unsigned int expires; + + bool running; + + AmDynInvoke* uac_auth_i; + + void DIcreateRegistration(int subscriber_id, const string& user, + const string& pass, const string& realm, + AmArg& ret); + void DIupdateRegistration(int subscriber_id, const string& user, + const string& pass, const string& realm, + AmArg& ret); + void DIremoveRegistration(int subscriber_id, AmArg& ret); + + + public: + DBRegAgent(const string& _app_name); + ~DBRegAgent(); + + DECLARE_MODULE_INSTANCE(DBRegAgent); + + // DI + // DI factory + AmDynInvoke* getInstance() { return instance(); } + // DI API + void invoke(const string& method, + const AmArg& args, AmArg& ret); + /** re-registration timer callback */ + void timer_cb(RegTimer* timer, long subscriber_id, void* data2); + + friend class DBRegAgentProcessorThread; +}; + +#endif diff --git a/apps/db_reg_agent/Makefile b/apps/db_reg_agent/Makefile new file mode 100644 index 00000000..fc938786 --- /dev/null +++ b/apps/db_reg_agent/Makefile @@ -0,0 +1,7 @@ +plug_in_name = db_reg_agent + +module_ldflags = -lmysqlpp +module_cflags = -DMOD_NAME=\"$(plug_in_name)\" -I/usr/include/mysql++ -I/usr/include/mysql + +COREPATH ?=../../core +include $(COREPATH)/plug-in/Makefile.app_module diff --git a/apps/db_reg_agent/RegistrationTimer.cpp b/apps/db_reg_agent/RegistrationTimer.cpp new file mode 100644 index 00000000..e8f9b9e1 --- /dev/null +++ b/apps/db_reg_agent/RegistrationTimer.cpp @@ -0,0 +1,275 @@ +/* + * Copyright (C) 2011 Stefan Sayer + * + * This file is part of SEMS, a free SIP media server. + * + * SEMS is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * For a license to use the sems software under conditions + * other than those described here, or to purchase support for this + * software, please contact iptel.org by e-mail at the following addresses: + * info@iptel.org + * + * SEMS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +#include "RegistrationTimer.h" +#include + +RegistrationTimer::RegistrationTimer() + : current_bucket(0) +{ + struct timeval now; + gettimeofday(&now, 0); + current_bucket_start = now.tv_sec; +} + +// unsafe! +int RegistrationTimer::get_bucket_index(time_t tv) { + time_t buckets_start_time = current_bucket_start; + + if (tv < buckets_start_time) + return -1; + + // offset + int bucket_index = (tv - buckets_start_time); + bucket_index /= TIMER_BUCKET_LENGTH; + + if (bucket_index > TIMER_BUCKETS) // too far in the future + return -2; + + bucket_index += current_bucket; + bucket_index %= TIMER_BUCKETS; // circular array + + return bucket_index; +} + +void RegistrationTimer::place_timer(RegTimer* timer, int bucket_index) { + std::list::iterator it = buckets[bucket_index].timers.begin(); + while (it != buckets[bucket_index].timers.end() && + (timer->expires > (*it)->expires)) + it++; + + buckets[bucket_index].timers.insert(it, timer); + size_t b_size = buckets[bucket_index].timers.size(); + + DBG("inserted timer [%p] in bucket %i (now sized %zd)\n", + timer, bucket_index, b_size); +} + +void RegistrationTimer::fire_timer(RegTimer* timer) { + if (timer && timer->cb) { + DBG("firing timer [%p]\n", timer); + timer->cb(timer, timer->data1, timer->data2); + } +} + +bool RegistrationTimer::insert_timer(RegTimer* timer) { + if (!timer) + return false; + + buckets_mut.lock(); + int bucket_index = get_bucket_index(timer->expires); + + if (bucket_index == -1) { + // already expired, fire timer + buckets_mut.unlock(); + DBG("inserting already expired timer [%p], firing\n", timer); + fire_timer(timer); + return false; + } + + if (bucket_index == -2) { + ERROR("trying to place timer too far in the future\n"); + buckets_mut.unlock(); + return false; + } + + place_timer(timer, bucket_index); + + buckets_mut.unlock(); + + return true; +} + +bool RegistrationTimer::remove_timer(RegTimer* timer) { + if (!timer) + return false; + + bool res = false; + + buckets_mut.lock(); + int bucket_index = get_bucket_index(timer->expires); + + if (bucket_index < 0) { + buckets_mut.unlock(); + return false; + } + + std::list& timerlist = buckets[bucket_index].timers; + + for (std::list::iterator it = timerlist.begin(); + it != timerlist.end(); it++) { + if (*it == timer) { + timerlist.erase(it); + res = true; + break; + } + } + + buckets_mut.unlock(); + + if (res) { + DBG("successfully removed timer [%p]\n", timer); + } else { + DBG("timer [%p] not found for removing\n", timer); + } + return res; +} + +void RegistrationTimer::run_timers() { + std::list timers_tbf; + + struct timeval now; + gettimeofday(&now, 0); + + buckets_mut.lock(); + + // bucket over? + if (now.tv_sec > current_bucket_start + TIMER_BUCKET_LENGTH) { + timers_tbf.insert(timers_tbf.begin(), + buckets[current_bucket].timers.begin(), + buckets[current_bucket].timers.end()); + buckets[current_bucket].timers.clear(); + current_bucket++; + current_bucket %= TIMER_BUCKETS; + current_bucket_start += TIMER_BUCKET_LENGTH; + // DBG("turned bucket to %i\n", current_bucket); + } + + // move timers from current_bucket + RegTimerBucket& bucket = buckets[current_bucket]; + std::list::iterator it = bucket.timers.begin(); + while (it != bucket.timers.end() && + now.tv_sec > (*it)->expires) { + std::list::iterator c_it = it; + it++; + timers_tbf.push_back(*c_it); + bucket.timers.erase(c_it); + } + + buckets_mut.unlock(); + + if (!timers_tbf.empty()) { + DBG("firing %zd timers\n", timers_tbf.size()); + for (std::list::iterator it=timers_tbf.begin(); + it != timers_tbf.end(); it++) { + fire_timer(*it); + } + } +} + +void RegistrationTimer::run() +{ + struct timeval now,next_tick,diff,tick; + + tick.tv_sec = 0; + tick.tv_usec = TIMER_RESOLUTION; + + gettimeofday(&now, NULL); + timeradd(&tick,&now,&next_tick); + + while(true){ + + gettimeofday(&now,NULL); + + if(timercmp(&now,&next_tick,<)){ + + struct timespec sdiff,rem; + timersub(&next_tick, &now,&diff); + + sdiff.tv_sec = diff.tv_sec; + sdiff.tv_nsec = diff.tv_usec * 1000; + + if(sdiff.tv_nsec > 2000000) // 2 ms + nanosleep(&sdiff,&rem); + } + //else { + //printf("missed one tick\n"); + //} + + run_timers(); + timeradd(&tick,&next_tick,&next_tick); + } +} + +void RegistrationTimer::on_stop() { +} + +bool RegistrationTimer::insert_timer_leastloaded(RegTimer* timer, + time_t from_time, + time_t to_time) { + + buckets_mut.lock(); + + int from_index = get_bucket_index(from_time); + int to_index = get_bucket_index(to_time); + + if (from_index < 0 && to_index < 0) + return false; + + int res_index = from_index; + if (from_index < 0) { + // use to_index (should not occur) + res_index = to_index; + } else { + // find least loaded bucket + size_t least_load = buckets[from_index].timers.size(); + + int i = from_index; + while (i != to_index) { + if (buckets[i].timers.size() <= least_load) { + least_load = buckets[i].timers.size(); + res_index = i; + } + + i++; + i %= TIMER_BUCKETS; + } + DBG("found bucket %i with least load %zd (between %i and %i)\n", + res_index, least_load, from_index, to_index); + } + + // update expires to some random value inside the selected bucket + int diff = (unsigned)res_index - current_bucket; + + if ((unsigned)res_index < current_bucket) { + diff+=TIMER_BUCKETS; + } + + timer->expires = current_bucket_start + + diff * TIMER_BUCKET_LENGTH + // bucket start + rand() % TIMER_BUCKET_LENGTH; + DBG("setting expires to %ld (between %ld and %ld)\n", + timer->expires, from_time, to_time); + + place_timer(timer, res_index); + + buckets_mut.unlock(); + + return false; +} + + + + diff --git a/apps/db_reg_agent/RegistrationTimer.h b/apps/db_reg_agent/RegistrationTimer.h new file mode 100644 index 00000000..744f863a --- /dev/null +++ b/apps/db_reg_agent/RegistrationTimer.h @@ -0,0 +1,113 @@ +/* + * Copyright (C) 2011 Stefan Sayer + * + * This file is part of SEMS, a free SIP media server. + * + * SEMS is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * For a license to use the sems software under conditions + * other than those described here, or to purchase support for this + * software, please contact iptel.org by e-mail at the following addresses: + * info@iptel.org + * + * SEMS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +#ifndef _RegistrationTimer_h_ +#define _RegistrationTimer_h_ + +#include +#include + +#include + +#include "log.h" +#include "AmThread.h" + +#define TIMER_BUCKET_LENGTH 10 // 10 sec +#define TIMER_BUCKETS 1000 // 1000 buckets + +// 100 ms == 100000 us +#define TIMER_RESOLUTION 100000 + +class RegTimer; +typedef void (*timer_cb)(RegTimer*, long /*data1*/,void* /*data2*/); + +class RegTimerBucket; + +class RegTimer { + public: + time_t expires; + + timer_cb cb; + long data1; + void* data2; + + RegTimer() + : expires(0), cb(0), data1(0), data2(0) { } +}; + +class RegTimerBucket { + public: + std::list timers; + + RegTimerBucket() { } +}; + +/** + Additionally to normal timer operation (setting and removing timer, + fire the timer when it is expired), this RegistrationTimer timers + class needs to support insert_timer_leastloaded() which should insert + the timer in some least loaded interval between from_time and to_time + in order to flatten out re-register spikes (due to restart etc). + + Timer granularity is seconds. + + Timers are saved in buckets of TIMER_BUCKET_LENGTH seconds. the buckets + array is a circular one, the current bucket starts from the time + current_bucket_start (in seconds as in time(2)). + + The timer object is owned by the caller, and MUST be valid until it is + fired or removed. + */ + +class RegistrationTimer +: public AmThread +{ + time_t current_bucket_start; + // every bucket contains TIMER_BUCKET_LENGTH seconds of timers + RegTimerBucket buckets[TIMER_BUCKETS]; + unsigned int current_bucket; + AmMutex buckets_mut; + + int get_bucket_index(time_t tv); + void place_timer(RegTimer* timer, int bucket_index); + void fire_timer(RegTimer* timer); + void run_timers(); + + protected: + void run(); + void on_stop(); + + public: + bool insert_timer(RegTimer* timer); + bool remove_timer(RegTimer* timer); + + bool insert_timer_leastloaded(RegTimer* timer, + time_t from_time, + time_t to_time); + + RegistrationTimer(); +}; + +#endif diff --git a/apps/db_reg_agent/etc/db_reg_agent.conf b/apps/db_reg_agent/etc/db_reg_agent.conf new file mode 100644 index 00000000..1f7a25c0 --- /dev/null +++ b/apps/db_reg_agent/etc/db_reg_agent.conf @@ -0,0 +1,50 @@ +# Database connection +mysql_user=root +mysql_passwd=mypass123 + +# mysql_server, default: localhost +# mysql_server=localhost + +#mysql_db, default: sems +# mysql_db=sems + +# table for registration status +# default: registrations +# registrations_table="registrations" + +# query joining subscriber info with registration table +# (without trailing ';' such that where clause can appended) +joined_query="select subscribers.subscriber_id as subscriber_id, subscribers.user as user, subscribers.pass as pass, subscribers.realm as realm, registrations.registration_status as registration_status, registrations.expiry as expiry, registrations.last_registration as last_registration from subscribers left join registrations on subscribers.subscriber_id=registrations.subscriber_id" + +# expires: desired expires, i.e. expires value that is requested +# default: 7200 +# expires=300 + +# reregister_interval: fraction of actual expires after which register is refreshed +# default: reregister_interval=0.5 +#reregister_interval=0.5 + +# minimum_reregister_interval: if set, re-register is scheduled in least loaded time +# in minimum_reregister_interval .. reregister_interval i order to smooth load spikes +# must be smaller than reregister_interval +# default: off +# +# example: +# reregister_interval=0.5 +# minimum_reregister_interval=0.4 +# on a registration expiring in 3600s, the least loaded spot between 1440s and 1800s +# is chosen +# +#minimum_reregister_interval=0.4 + +# enable_ratelimiting=yes : Enable ratelimiting? +# default: no +# if enabled, the amount of initial REGISTER requests is limited (not counting re-trans- +# missions and requests re-sent for authentication) +#enable_ratelimiting=yes + +# ratelimit_rate=300 : rate of initial REGISTER requests to send as maximum +#ratelimit_rate=2 + +# ratelimit_per=1 : per time unit (in seconds, e.g. 300 REGISTER in 1 second) +#ratelimit_per=1 \ No newline at end of file diff --git a/doc/Readme.db_reg_agent.txt b/doc/Readme.db_reg_agent.txt new file mode 100644 index 00000000..2212ad1e --- /dev/null +++ b/doc/Readme.db_reg_agent.txt @@ -0,0 +1,69 @@ + + +db_reg_agent module Readme (c) 2011 Stefan Sayer + +Purpose +------- + +The db_reg_agent module allows SEMS to read SIP accounts from a database +and register the accounts to SIP a registrar. In that it serves a similar +purpose as the reg_agent/registrar_client modules, with the differences +that it reads accounts from mysql DB instead of the file system, and that +it is built to support many (up to several 10k) subscription. Additionally, +accounts may be added, changed and removed while SEMS is running; the +db_reg_agent then can be triggered via DI interface (XMLRPC/json-rpc) to +pick up the new registration. + +Features +- configurable subscription query +- configurable desired expires +- flatten out re-register spikes by intelligently planning registration refresh +- ratelimiting (x new REGISTER requests per y seconds) +- seamless restart of SEMS server possible; registration status is restored from DB. + +DI control functions +-------------------- + + createRegistration(int subscriber_id, string user, string pass, string realm) + updateRegistration(int subscriber_id, string user, string pass, string realm) + removeRegistration(int subscriber_id) + +After removing a registration by issuing removeRegistration, the subcriber entry will +be present with the status REMOVED. + +Registration status (registration_status) +----------------------------------------- + REG_STATUS_INACTIVE 0 + REG_STATUS_PENDING 1 + REG_STATUS_ACTIVE 2 + REG_STATUS_FAILED 3 + REG_STATUS_REMOVED 4 + +Database +-------- +There may be two tables, subscriptions and registration status. The query which +selects subscription and registration status as join may be configured as well +as the name of the registration status table (registrations_table). + +Clocks of SEMS host and DB host must be synchronized in order for restart without +massive re-registration to work. + +Example tables structure: + +CREATE TABLE IF NOT EXISTS `registrations` ( + `subscriber_id` int(11) NOT NULL, + `registration_status` tinyint(1) NOT NULL DEFAULT '0', + `last_registration` datetime, + `expiry` datetime, + `last_code` smallint(2), + `last_reason` varchar(256), + PRIMARY KEY (`subscriber_id`) +) ENGINE=MyISAM DEFAULT CHARSET=latin1; + +CREATE TABLE IF NOT EXISTS `subscribers` ( + `subscriber_id` int(10) NOT NULL AUTO_INCREMENT, + `user` varchar(256) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, + `pass` varchar(256) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, + `realm` varchar(256) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, + PRIMARY KEY (`subscriber_id`) +) ENGINE=MyISAM DEFAULT CHARSET=latin1 AUTO_INCREMENT=2 ;