mirror of https://github.com/sipwise/sems.git
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.sayer/1.4-spce2.6
Stefan Sayer
15 years ago
committed by
Stefan Sayer
parent
f5adf80c80
commit
7ae3ca4d4d
@ -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 <unistd.h>
|
||||
#include <stdlib.h>
|
||||
|
||||
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<AmSessionEventHandler*>(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<long, AmSIPRegistration*>::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<long, AmSIPRegistration*>::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<RegistrationActionEvent*>(ev);
|
||||
if (reg_action_ev) {
|
||||
onRegistrationActionEvent(reg_action_ev);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
AmSipReplyEvent* sip_rep = dynamic_cast<AmSipReplyEvent*>(ev);
|
||||
if (sip_rep) {
|
||||
onSipReplyEvent(sip_rep);
|
||||
return;
|
||||
}
|
||||
|
||||
if (ev->event_id == E_SYSTEM) {
|
||||
AmSystemEvent* sys_ev = dynamic_cast<AmSystemEvent*>(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<long, AmSIPRegistration*>::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<long, AmSIPRegistration*>::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<string, long>::iterator it=registration_ltags.find(ev->reply.local_tag);
|
||||
if (it!=registration_ltags.end()) {
|
||||
long subscriber_id = it->second;
|
||||
map<long, AmSIPRegistration*>::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<string, long>::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<long, RegTimer*>::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<long, RegTimer*>::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<long, RegTimer*>::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<AmSystemEvent*>(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<RegistrationActionEvent*>(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<RegTimer*> 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
|
||||
@ -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 <sys/time.h>
|
||||
|
||||
#include <mysql++/mysql++.h>
|
||||
|
||||
|
||||
#include <map>
|
||||
using std::map;
|
||||
#include <queue>
|
||||
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<long, AmSIPRegistration*> registrations;
|
||||
map<string, long> registration_ltags;
|
||||
map<long, RegTimer*> 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
|
||||
@ -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
|
||||
@ -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 <stdlib.h>
|
||||
|
||||
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<RegTimer*>::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<RegTimer*>& timerlist = buckets[bucket_index].timers;
|
||||
|
||||
for (std::list<RegTimer*>::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<RegTimer*> 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<RegTimer*>::iterator it = bucket.timers.begin();
|
||||
while (it != bucket.timers.end() &&
|
||||
now.tv_sec > (*it)->expires) {
|
||||
std::list<RegTimer*>::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<RegTimer*>::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;
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
@ -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 <list>
|
||||
#include <vector>
|
||||
|
||||
#include <sys/time.h>
|
||||
|
||||
#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<RegTimer*> 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
|
||||
@ -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
|
||||
@ -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 ;
|
||||
Loading…
Reference in new issue