diff --git a/debian/patches/series b/debian/patches/series index 7bb6093a..fe03d4ed 100644 --- a/debian/patches/series +++ b/debian/patches/series @@ -39,3 +39,4 @@ sipwise/upgrade_to_bullseye.patch sipwise/port_to_py3k.patch sipwise/db_reg_agent_auth_header.patch sipwise/skip_outbound_auth_if_credentials_empty.patch +sipwise/add_support_for_peering_outbound_registration.patch diff --git a/debian/patches/sipwise/add_support_for_peering_outbound_registration.patch b/debian/patches/sipwise/add_support_for_peering_outbound_registration.patch new file mode 100644 index 00000000..8f6208de --- /dev/null +++ b/debian/patches/sipwise/add_support_for_peering_outbound_registration.patch @@ -0,0 +1,1788 @@ +--- a/apps/db_reg_agent/DBRegAgent.cpp ++++ b/apps/db_reg_agent/DBRegAgent.cpp +@@ -36,7 +36,8 @@ DEFINE_MODULE_INSTANCE(DBRegAgent, MOD_N + mysqlpp::Connection DBRegAgent::MainDBConnection(mysqlpp::use_exceptions); + mysqlpp::Connection DBRegAgent::ProcessorDBConnection(mysqlpp::use_exceptions); + +-string DBRegAgent::joined_query; ++string DBRegAgent::joined_query_subscribers; ++string DBRegAgent::joined_query_peerings; + string DBRegAgent::registrations_table = "registrations"; + + double DBRegAgent::reregister_interval = 0.5; +@@ -59,8 +60,8 @@ bool DBRegAgent::save_auth_replies = fal + + unsigned int DBRegAgent::error_retry_interval = 300; + +-static void _timer_cb(RegTimer* timer, long subscriber_id, int data2) { +- DBRegAgent::instance()->timer_cb(timer, subscriber_id, data2); ++static void _timer_cb(RegTimer* timer, long object_id, int data2, const string& type) { ++ DBRegAgent::instance()->timer_cb(timer, object_id, data2, type); + } + + DBRegAgent::DBRegAgent(const string& _app_name) +@@ -213,12 +214,13 @@ int DBRegAgent::onLoad() + AmEventDispatcher::instance()->addEventQueue(MOD_NAME,this); + + if (!AmPlugIn::registerDIInterface(MOD_NAME, this)) { +- ERROR("registering "MOD_NAME" DI interface\n"); ++ ERROR("registering %s DI interface\n", MOD_NAME); + return -1; + } + +- joined_query = cfg.getParameter("joined_query"); +- if (joined_query.empty()) { ++ joined_query_subscribers = cfg.getParameter("joined_query_subscribers"); ++ joined_query_peerings = cfg.getParameter("joined_query_peerings"); ++ if (joined_query_subscribers.empty() || joined_query_peerings.empty()) { + // todo: name! + ERROR("joined_query must be set\n"); + return -1; +@@ -230,7 +232,11 @@ int DBRegAgent::onLoad() + DBG("using registrations table '%s'\n", registrations_table.c_str()); + + if (!loadRegistrations()) { +- ERROR("loading registrations from DB\n"); ++ ERROR("REGISTER: loading registrations for subscribers from DB\n"); ++ return -1; ++ } ++ if (!loadRegistrationsPeerings()) { ++ ERROR("REGISTER: loading registrations for peerings from DB\n"); + return -1; + } + +@@ -270,117 +276,242 @@ bool DBRegAgent::loadRegistrations() { + try { + time_t now_time = time(NULL); + +- mysqlpp::Query query = DBRegAgent::MainDBConnection.query(); ++ mysqlpp::Query query_sb = DBRegAgent::MainDBConnection.query(); ++ string query_string_subscribers, table; ++ query_string_subscribers = joined_query_subscribers; + +- string query_string, table; ++ DBG("REGISTER: querying all registrations for subscribers with : '%s'\n", query_string_subscribers.c_str()); + +- query_string = joined_query; ++ query_sb << query_string_subscribers; ++ mysqlpp::UseQueryResult res_sb = query_sb.use(); + +- DBG("querying all registrations with : '%s'\n", +- query_string.c_str()); ++ while (mysqlpp::Row row = res_sb.fetch_row()) { ++ int status = 0; ++ string type = TYPE_SUBSCRIBER; ++ long object_id = row[COLNAME_SUBSCRIBER_ID]; + +- query << query_string; +- mysqlpp::UseQueryResult res = query.use(); +- +- // mysqlpp::Row::size_type row_count = res.num_rows(); +- // DBG("got %zd subscriptions\n", row_count); ++ if (object_id == NULL || object_id == 0) { ++ WARN("REGISTER: object_id is NULL or 0 for this subscriber, skipping..\n"); ++ continue; ++ } + +- while (mysqlpp::Row row = res.fetch_row()) { +- int status = 0; +- long subscriber_id = row[COLNAME_SUBSCRIBER_ID]; ++ DBG("REGISTER: Triggering for subscriber with object_id=<%d>\n", object_id); + + string contact_uri; + if (db_read_contact && row[COLNAME_CONTACT] != mysqlpp::null) { +- contact_uri = (string) row[COLNAME_CONTACT]; ++ contact_uri = (string) row[COLNAME_CONTACT]; + } + + if (row[COLNAME_STATUS] != mysqlpp::null) +- status = row[COLNAME_STATUS]; ++ status = row[COLNAME_STATUS]; + else { +- DBG("registration status entry for id %ld does not exist, creating...\n", +- subscriber_id); +- createDBRegistration(subscriber_id, ProcessorDBConnection); ++ DBG("registration status entry for id %ld does not exist, creating...\n", ++ object_id); ++ createDBRegistration(object_id, type, ProcessorDBConnection); + } + + DBG("got subscriber '%s@%s' status %i\n", +- string(row[COLNAME_USER]).c_str(), string(row[COLNAME_REALM]).c_str(), +- status); ++ 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_AUTH_USER], +- (string)row[COLNAME_USER], +- (string)row[COLNAME_PASS], +- (string)row[COLNAME_REALM], +- contact_uri +- ); +- scheduleRegistration(subscriber_id); +- }; break; +- +- case REG_STATUS_ACTIVE: +- { +- createRegistration(subscriber_id, +- (string)row[COLNAME_AUTH_USER], +- (string)row[COLNAME_USER], +- (string)row[COLNAME_PASS], +- (string)row[COLNAME_REALM], +- contact_uri +- ); +- +- 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, now_time); +- } +- +- }; 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; +- +- case REG_STATUS_TO_BE_REMOVED: +- { +- DBG("Scheduling Deregister of registration %ld %s@%s", subscriber_id, +- ((string)row[COLNAME_USER]).c_str(), ((string)row[COLNAME_REALM]).c_str()); +- createRegistration(subscriber_id, +- (string)row[COLNAME_AUTH_USER], +- (string)row[COLNAME_USER], +- (string)row[COLNAME_PASS], +- (string)row[COLNAME_REALM], +- contact_uri +- ); +- scheduleDeregistration(subscriber_id); +- }; ++ case REG_STATUS_INACTIVE: ++ case REG_STATUS_PENDING: // try again ++ case REG_STATUS_FAILED: // try again ++ { ++ createRegistration(object_id, ++ (string)row[COLNAME_AUTH_USER], ++ (string)row[COLNAME_USER], ++ (string)row[COLNAME_PASS], ++ (string)row[COLNAME_REALM], ++ contact_uri, ++ type); ++ scheduleRegistration(object_id, type); ++ }; break; ++ ++ case REG_STATUS_ACTIVE: ++ { ++ createRegistration(object_id, ++ (string)row[COLNAME_AUTH_USER], ++ (string)row[COLNAME_USER], ++ (string)row[COLNAME_PASS], ++ (string)row[COLNAME_REALM], ++ contact_uri, ++ type); ++ ++ 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", object_id); ++ scheduleRegistration(object_id, type); ++ } else { ++ setRegistrationTimer(object_id, dt_expiry, dt_registration_ts, now_time, type); ++ } ++ }; break; ++ ++ case REG_STATUS_REMOVED: ++ { ++ DBG("ignoring removed registration %ld %s@%s of type: %s\n", object_id, ++ ((string)row[COLNAME_USER]).c_str(), ((string)row[COLNAME_REALM]).c_str(), type.c_str()); ++ } break; ++ ++ case REG_STATUS_TO_BE_REMOVED: ++ { ++ DBG("Scheduling Deregister of registration %ld %s@%s of type: %s\n", object_id, ++ ((string)row[COLNAME_USER]).c_str(), ((string)row[COLNAME_REALM]).c_str(), type.c_str()); ++ createRegistration(object_id, ++ (string)row[COLNAME_AUTH_USER], ++ (string)row[COLNAME_USER], ++ (string)row[COLNAME_PASS], ++ (string)row[COLNAME_REALM], ++ contact_uri, ++ type); ++ scheduleDeregistration(object_id, type); ++ }; + } + } + ++ } catch (const mysqlpp::Exception& er) { ++ // Catch-all for any MySQL++ exceptions ++ ERROR("MySQL++ error: %s\n", er.what()); ++ return false; ++ } ++ ++ return true; ++} ++ ++bool DBRegAgent::loadRegistrationsPeerings() { ++ try { ++ time_t now_time = time(NULL); ++ ++ mysqlpp::Query query_pr = DBRegAgent::MainDBConnection.query(); ++ string query_string_peerings, table; ++ query_string_peerings = joined_query_peerings; ++ ++ DBG("REGISTER: querying all registrations for peerings with : '%s'\n", query_string_peerings.c_str()); ++ ++ query_pr << query_string_peerings; ++ mysqlpp::UseQueryResult res_pr = query_pr.use(); ++ ++ while (mysqlpp::Row row = res_pr.fetch_row()) { ++ int status = 0; ++ string type = TYPE_PEERING; ++ long object_id = row[COLNAME_PEER_ID]; ++ ++ if (object_id == NULL || object_id == 0) { ++ WARN("REGISTER: object_id is NULL or 0 for this peering, skipping..\n"); ++ continue; ++ } ++ ++ DBG("REGISTER: Triggering for peering with object_id=<%d>\n", object_id); ++ ++ string contact_uri; ++ if (db_read_contact && row[COLNAME_CONTACT] != mysqlpp::null) { ++ contact_uri = (string) row[COLNAME_CONTACT]; ++ } ++ ++ if (row[COLNAME_STATUS] != mysqlpp::null) ++ status = row[COLNAME_STATUS]; ++ else { ++ DBG("registration status entry for id %ld does not exist, creating...\n", ++ object_id); ++ createDBRegistration(object_id, type, ProcessorDBConnection); ++ } ++ ++ 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(object_id, ++ (string)row[COLNAME_AUTH_USER], ++ (string)row[COLNAME_USER], ++ (string)row[COLNAME_PASS], ++ (string)row[COLNAME_REALM], ++ contact_uri, ++ type); ++ scheduleRegistration(object_id, type); ++ }; break; ++ ++ case REG_STATUS_ACTIVE: ++ { ++ createRegistration(object_id, ++ (string)row[COLNAME_AUTH_USER], ++ (string)row[COLNAME_USER], ++ (string)row[COLNAME_PASS], ++ (string)row[COLNAME_REALM], ++ contact_uri, ++ type); ++ ++ 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", object_id); ++ scheduleRegistration(object_id, type); ++ } else { ++ setRegistrationTimer(object_id, dt_expiry, dt_registration_ts, now_time, type); ++ } ++ }; break; ++ ++ case REG_STATUS_REMOVED: ++ { ++ DBG("ignoring removed registration %ld %s@%s of type: %s\n", object_id, ++ ((string)row[COLNAME_USER]).c_str(), ((string)row[COLNAME_REALM]).c_str(), type.c_str()); ++ } break; ++ ++ case REG_STATUS_TO_BE_REMOVED: ++ { ++ DBG("Scheduling Deregister of registration %ld %s@%s of type: %s\n", object_id, ++ ((string)row[COLNAME_USER]).c_str(), ((string)row[COLNAME_REALM]).c_str(), type.c_str()); ++ createRegistration(object_id, ++ (string)row[COLNAME_AUTH_USER], ++ (string)row[COLNAME_USER], ++ (string)row[COLNAME_PASS], ++ (string)row[COLNAME_REALM], ++ contact_uri, ++ type); ++ scheduleDeregistration(object_id, type); ++ }; ++ } ++ } + + } catch (const mysqlpp::Exception& er) { + // Catch-all for any MySQL++ exceptions +@@ -392,14 +523,16 @@ bool DBRegAgent::loadRegistrations() { + } + + /** create registration in our list */ +-void DBRegAgent::createRegistration(long subscriber_id, ++void DBRegAgent::createRegistration(long object_id, + const string& auth_user, + const string& user, + const string& pass, + const string& realm, +- const string& contact) { ++ const string& contact, ++ const string& type) { + +- string auth_user_temp = auth_user.empty() ? user : auth_user; ++ string auth_user_temp = (auth_user.empty() || auth_user == "" || auth_user == "NULL") ? user : auth_user; ++ DBG("REGISTER: authentication user picked out: <%s> \n", auth_user_temp.c_str()); + + string _user = user; + if (username_with_domain && user.find('@')!=string::npos) { +@@ -420,30 +553,51 @@ void DBRegAgent::createRegistration(long + contact_uri // contact + ); + +-DBG(" >>> realm '%s' - user '%s' - auth_user '%s' - pass: '%s' outbound_proxy '%s' contact_uri '%s'\n", +- realm.c_str(), _user.c_str(), auth_user.c_str(), pass.c_str(), outbound_proxy.c_str(), contact_uri.c_str()); ++ DBG(" >>> realm '%s', user '%s', auth_user '%s', pass '%s', outbound_proxy '%s', contact_uri '%s', type '%s'\n", ++ realm.c_str(), user.c_str(), auth_user.c_str(), pass.c_str(), ++ outbound_proxy.c_str(), contact_uri.c_str(), type.c_str()); + + 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); +- clearRegistrationTimer(subscriber_id); +- registrations_mut.lock(); ++ // remove already existing registration for a peering ++ if (type == TYPE_PEERING) { ++ if (registrations_peers.find(object_id) != registrations_peers.end()) { ++ registrations_mut.unlock(); ++ WARN("registration for a Peering with ID %ld already exists, removing\n", object_id); ++ removeRegistration(object_id, type); ++ clearRegistrationTimer(object_id, type); ++ registrations_mut.lock(); ++ } ++ // remove already existing for a usual subscriber ++ } else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) { ++ if (registrations.find(object_id) != registrations.end()) { ++ registrations_mut.unlock(); ++ WARN("registration for a Subscriber with ID %ld already exists, removing\n", object_id); ++ removeRegistration(object_id, type); ++ clearRegistrationTimer(object_id, type); ++ registrations_mut.lock(); ++ } + } + + AmSIPRegistration* reg = new AmSIPRegistration(handle, reg_info, "" /*MOD_NAME*/); ++ ++ // a simple fix in case expires is for some reason 0. Not the best solution. ++ //if (expires == 0) expires = 60; + reg->setExpiresInterval(expires); + +- registrations[subscriber_id] = reg; +- registration_ltags[handle] = subscriber_id; ++ if (type == TYPE_PEERING) { ++ registrations_peers[object_id] = reg; ++ registration_ltags_peers[handle] = object_id; ++ } else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) { ++ registrations[object_id] = reg; ++ registration_ltags[handle] = object_id; ++ } + + if (NULL != uac_auth_i) { +- DBG("enabling UAC Auth for new registration.\n"); ++ DBG("REGISTER: Enabling UAC Auth for new registration of type: <%s>\n", type.c_str()); + + // get a sessionEventHandler from uac_auth +- AmArg di_args,ret; ++ AmArg di_args, ret; + AmArg a; + a.setBorrowedPointer(reg); + di_args.push(a); +@@ -451,20 +605,17 @@ DBG(" >>> realm '%s' - user '%s' - auth_ + + uac_auth_i->invoke("getHandler", di_args, ret); + if (!ret.size()) { +- ERROR("Can not add auth handler to new registration!\n"); ++ ERROR("Can not add auth handler to new registration!\n"); + } else { +- AmObject* p = ret.get(0).asObject(); +- if (p != NULL) { +- AmSessionEventHandler* h = dynamic_cast(p); +- if (h != NULL) +- reg->setSessionEventHandler(h); +- } ++ AmObject* 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 (const std::exception& e) { ++ ERROR("%s", e.what()); + } catch (...) { + ERROR("unknown exception occured\n"); + } +@@ -474,18 +625,20 @@ DBG(" >>> realm '%s' - user '%s' - auth_ + // 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()); ++ DBG("created new registration with ID <%ld>, ltag '%s' and type '%s'\n", ++ object_id, handle.c_str(), type.c_str()); + } + +-void DBRegAgent::updateRegistration(long subscriber_id, ++void DBRegAgent::updateRegistration(long object_id, + const string& auth_user, + const string& user, + const string& pass, + const string& realm, +- const string& contact) { ++ const string& contact, ++ const string& type) { + +- string auth_user_temp = auth_user.empty() ? user : auth_user; ++ string auth_user_temp = (auth_user.empty() || auth_user == "" || auth_user == "NULL") ? user : auth_user; ++ DBG("REGISTER: authentication user picked out: <%s> \n", auth_user_temp.c_str()); + + string _user = user; + if (username_with_domain && user.find('@')!=string::npos) { +@@ -493,14 +646,29 @@ void DBRegAgent::updateRegistration(long + } + + 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, auth_user_temp, _user, pass, realm, contact); +- scheduleRegistration(subscriber_id); +- return; ++ ++ map::iterator it; ++ ++ if (type == TYPE_PEERING) { ++ it=registrations_peers.find(object_id); ++ if (it == registrations_peers.end()) { ++ registrations_mut.unlock(); ++ WARN("updateRegistration - registration %ld %s@%s unknown, creating. Type: %s\n", ++ object_id, user.c_str(), realm.c_str(), type.c_str()); ++ createRegistration(object_id, auth_user_temp, user, pass, realm, contact, type); ++ scheduleRegistration(object_id, type); ++ return; ++ } ++ } else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) { ++ it=registrations.find(object_id); ++ if (it == registrations.end()) { ++ registrations_mut.unlock(); ++ WARN("updateRegistration - registration %ld %s@%s unknown, creating. Type: %s\n", ++ object_id, user.c_str(), realm.c_str(), type.c_str()); ++ createRegistration(object_id, auth_user_temp, user, pass, realm, contact, type); ++ scheduleRegistration(object_id, type); ++ return; ++ } + } + + bool need_reregister = it->second->getInfo().domain != realm +@@ -511,6 +679,7 @@ void DBRegAgent::updateRegistration(long + + string old_realm = it->second->getInfo().domain; + string old_user = it->second->getInfo().user; ++ string old_auth_user = it->second->getInfo().auth_user; + it->second->setRegistrationInfo(SIPRegistrationInfo(realm, _user, + _user, // name + auth_user_temp, // auth_user +@@ -519,64 +688,78 @@ void DBRegAgent::updateRegistration(long + contact)); // contact + registrations_mut.unlock(); + if (need_reregister) { +- DBG("user/realm for registration %ld changed (%s@%s -> %s@%s). " +- "Triggering immediate re-registration\n", +- subscriber_id, old_user.c_str(), old_realm.c_str(), _user.c_str(), realm.c_str()); +- scheduleRegistration(subscriber_id); ++ DBG("user/realm for registration %ld changed (%s@%s -> %s@%s). Auth user (%s -> %s)." ++ "Triggering immediate re-registration\n", ++ object_id, old_user.c_str(), old_realm.c_str(), ++ user.c_str(), realm.c_str(), old_auth_user.c_str(), auth_user_temp.c_str()); ++ scheduleRegistration(object_id, type); + } + } + + /** remove registration from our list */ +-void DBRegAgent::removeRegistration(long subscriber_id) { ++void DBRegAgent::removeRegistration(long object_id, const string& type) { + 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; ++ ++ map::iterator it; ++ if (type == TYPE_PEERING) { // remove reg for peerings ++ it = registrations_peers.find(object_id); ++ if (it != registrations_peers.end()) { ++ handle = it->second->getHandle(); ++ registration_ltags_peers.erase(handle); ++ delete it->second; ++ registrations_peers.erase(it); ++ res = true; ++ } ++ } else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) { // remove reg for subscribers ++ it = registrations.find(object_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); ++ DBG("removed registration with ID %ld, type: %s \n", object_id, type.c_str()); + } else { +- DBG("registration with ID %ld not found for removing\n", subscriber_id); ++ DBG("registration with ID %ld not found for removing, type: %s \n", object_id, type.c_str()); + } + } + + /** schedule this registration to REGISTER (immediately) */ +-void DBRegAgent::scheduleRegistration(long subscriber_id) { ++void DBRegAgent::scheduleRegistration(long object_id, const string& type) { + if (enable_ratelimiting) { + registration_processor. +- postEvent(new RegistrationActionEvent(RegistrationActionEvent::Register, +- subscriber_id)); ++ postEvent(new RegistrationActionEvent(RegistrationActionEvent::Register, ++ object_id, type)); + } else { + // use our own thread + postEvent(new RegistrationActionEvent(RegistrationActionEvent::Register, +- subscriber_id)); ++ object_id, type)); + } +- DBG("added to pending actions: REGISTER of %ld\n", subscriber_id); ++ DBG("Added to pending actions: REGISTER of %ld, type: %s\n", object_id, type.c_str()); + } + + /** schedule this registration to de-REGISTER (immediately) */ +-void DBRegAgent::scheduleDeregistration(long subscriber_id) { ++void DBRegAgent::scheduleDeregistration(long object_id, const string& type) { + if (enable_ratelimiting) { + registration_processor. + postEvent(new RegistrationActionEvent(RegistrationActionEvent::Deregister, +- subscriber_id)); ++ object_id, type)); + } else { + // use our own thread + postEvent(new RegistrationActionEvent(RegistrationActionEvent::Deregister, +- subscriber_id)); ++ object_id, type)); + } +- DBG("added to pending actions: DEREGISTER of %ld\n", subscriber_id); ++ DBG("added to pending actions: DEREGISTER of %ld, type: %s\n", object_id, type.c_str()); + } + + void DBRegAgent::process(AmEvent* ev) { +@@ -614,71 +797,108 @@ void DBRegAgent::process(AmEvent* ev) { + // uses ProcessorDBConnection + void DBRegAgent::onRegistrationActionEvent(RegistrationActionEvent* reg_action_ev) { + switch (reg_action_ev->action) { ++ + case RegistrationActionEvent::Register: + { +- DBG("REGISTER of registration %ld\n", reg_action_ev->subscriber_id); ++ DBG("REGISTER of registration %ld, type: %s\n", ++ reg_action_ev->object_id, reg_action_ev->type.c_str()); ++ + registrations_mut.lock(); +- map::iterator it= +- registrations.find(reg_action_ev->subscriber_id); +- if (it==registrations.end()) { +- DBG("ignoring scheduled REGISTER of unknown registration %ld\n", +- reg_action_ev->subscriber_id); +- } else { +- if (!it->second->doRegistration()) { +- updateDBRegistration(ProcessorDBConnection, +- reg_action_ev->subscriber_id, +- 480, ERR_REASON_UNABLE_TO_SEND_REQUEST, +- true, REG_STATUS_FAILED); +- if (error_retry_interval) { +- // schedule register-refresh after error_retry_interval +- setRegistrationTimer(reg_action_ev->subscriber_id, error_retry_interval, +- RegistrationActionEvent::Register); +- } +- } ++ map::iterator it; ++ bool marker = true; ++ ++ if (reg_action_ev->type == TYPE_PEERING) { ++ it = registrations_peers.find(reg_action_ev->object_id); ++ if (it==registrations_peers.end()) { ++ DBG("ignoring scheduled REGISTER of unknown registration %ld\n", ++ reg_action_ev->object_id); ++ marker = false; ++ } ++ } else if (reg_action_ev->type == TYPE_SUBSCRIBER || reg_action_ev->type == TYPE_UNDEFINED) { ++ it = registrations.find(reg_action_ev->object_id); ++ if (it==registrations.end()) { ++ DBG("ignoring scheduled REGISTER of unknown registration %ld\n", ++ reg_action_ev->object_id); ++ marker = false; ++ } + } ++ ++ if (marker) { ++ if (!it->second->doRegistration()) { ++ updateDBRegistration(ProcessorDBConnection, ++ reg_action_ev->object_id, reg_action_ev->type, ++ 480, ERR_REASON_UNABLE_TO_SEND_REQUEST, ++ true, REG_STATUS_FAILED); ++ if (error_retry_interval) { ++ // schedule register-refresh after error_retry_interval ++ setRegistrationTimer(reg_action_ev->object_id, error_retry_interval, ++ RegistrationActionEvent::Register, reg_action_ev->type); ++ } ++ } ++ } ++ + registrations_mut.unlock(); + } break; ++ + case RegistrationActionEvent::Deregister: + { +- DBG("De-REGISTER of registration %ld\n", reg_action_ev->subscriber_id); ++ DBG("De-REGISTER of registration %ld, type: %s\n", ++ reg_action_ev->object_id, reg_action_ev->type.c_str()); ++ + 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 %ld\n", +- reg_action_ev->subscriber_id); +- } else { +- if (!it->second->doUnregister()) { +- if (delete_removed_registrations && delete_failed_deregistrations) { +- DBG("sending de-Register failed - deleting registration %ld " +- "(delete_failed_deregistrations=yes)\n", reg_action_ev->subscriber_id); +- deleteDBRegistration(reg_action_ev->subscriber_id, ProcessorDBConnection); +- } else { +- DBG("failed sending de-register, updating DB with REG_STATUS_TO_BE_REMOVED " +- ERR_REASON_UNABLE_TO_SEND_REQUEST "for subscriber %ld\n", +- reg_action_ev->subscriber_id); +- updateDBRegistration(ProcessorDBConnection, +- reg_action_ev->subscriber_id, +- 480, ERR_REASON_UNABLE_TO_SEND_REQUEST, +- true, REG_STATUS_TO_BE_REMOVED); +- // don't re-try de-registrations if sending failed +- // if (error_retry_interval) { +- // // schedule register-refresh after error_retry_interval +- // setRegistrationTimer(reg_action_ev->subscriber_id, error_retry_interval, +- // RegistrationActionEvent::Deregister); +- // } +- } +- } ++ map::iterator it; ++ bool marker = true; ++ ++ if (reg_action_ev->type == TYPE_PEERING) { ++ it = registrations_peers.find(reg_action_ev->object_id); ++ if (it==registrations_peers.end()) { ++ DBG("ignoring scheduled De-REGISTER of unknown registration %ld\n", ++ reg_action_ev->object_id); ++ marker = false; ++ } ++ } else if (reg_action_ev->type == TYPE_SUBSCRIBER || reg_action_ev->type == TYPE_UNDEFINED) { ++ it = registrations.find(reg_action_ev->object_id); ++ if (it==registrations.end()) { ++ DBG("ignoring scheduled De-REGISTER of unknown registration %ld\n", ++ reg_action_ev->object_id); ++ marker = false; ++ } ++ } ++ ++ if (marker) { ++ if (!it->second->doUnregister()) { ++ if (delete_removed_registrations && delete_failed_deregistrations) { ++ DBG("sending de-Register failed - deleting registration %ld " ++ "(delete_failed_deregistrations=yes)\n", reg_action_ev->object_id); ++ deleteDBRegistration(reg_action_ev->object_id, reg_action_ev->type, ProcessorDBConnection); ++ } else { ++ DBG("failed sending de-register, updating DB with REG_STATUS_TO_BE_REMOVED " ++ ERR_REASON_UNABLE_TO_SEND_REQUEST "for subscriber %ld\n", ++ reg_action_ev->object_id); ++ updateDBRegistration(ProcessorDBConnection, ++ reg_action_ev->object_id, reg_action_ev->type, ++ 480, ERR_REASON_UNABLE_TO_SEND_REQUEST, ++ true, REG_STATUS_TO_BE_REMOVED); ++ } ++ } + } ++ + registrations_mut.unlock(); + } break; + } + } + +-void DBRegAgent::createDBRegistration(long subscriber_id, mysqlpp::Connection& conn) { ++void DBRegAgent::createDBRegistration(long object_id, const string& type, mysqlpp::Connection& conn) { ++ ++ string column_id = COLNAME_SUBSCRIBER_ID; ++ if (type == TYPE_PEERING) column_id = COLNAME_PEER_ID; ++ ++ // depending on if that is a registration for a subscriber or for a peering ++ // do a mysql insertion + string insert_query = "insert into "+registrations_table+ +- " (subscriber_id) values ("+ +- long2str(subscriber_id)+");"; ++ " (" + column_id.c_str() + ")" + "values ("+ long2str(object_id)+");"; ++ ++ DBG("MYSQL: trying to execute: <%s>\n", insert_query.c_str()); + + try { + mysqlpp::Query query = conn.query(); +@@ -686,8 +906,8 @@ void DBRegAgent::createDBRegistration(lo + + mysqlpp::SimpleResult res = query.execute(); + if (!res) { +- WARN("creating registration in DB with query '%s' failed: '%s'\n", +- insert_query.c_str(), res.info()); ++ WARN("creating registration in DB with query '%s' failed: '%s', type: %s\n", ++ insert_query.c_str(), res.info(), type.c_str()); + } + } catch (const mysqlpp::Exception& er) { + // Catch-all for any MySQL++ exceptions +@@ -696,9 +916,15 @@ void DBRegAgent::createDBRegistration(lo + } + } + +-void DBRegAgent::deleteDBRegistration(long subscriber_id, mysqlpp::Connection& conn) { ++void DBRegAgent::deleteDBRegistration(long object_id, const string& type, mysqlpp::Connection& conn) { ++ ++ string column_id = COLNAME_SUBSCRIBER_ID; ++ if (type == TYPE_PEERING) column_id = COLNAME_PEER_ID; ++ ++ // depending on if that is a de-registration for a subscriber or for a peering ++ // do a mysql deletion + string insert_query = "delete from "+registrations_table+ +- " where subscriber_id=" + long2str(subscriber_id)+";"; ++ " where " + column_id.c_str() + "=" + long2str(object_id) + ";"; + + try { + mysqlpp::Query query = conn.query(); +@@ -706,8 +932,8 @@ void DBRegAgent::deleteDBRegistration(lo + + mysqlpp::SimpleResult res = query.execute(); + if (!res) { +- WARN("removing registration in DB with query '%s' failed: '%s'\n", +- insert_query.c_str(), res.info()); ++ WARN("removing registration in DB with query '%s' failed: '%s', type: %s\n", ++ insert_query.c_str(), res.info(), type.c_str()); + } + } catch (const mysqlpp::Exception& er) { + // Catch-all for any MySQL++ exceptions +@@ -717,7 +943,7 @@ void DBRegAgent::deleteDBRegistration(lo + } + + void DBRegAgent::updateDBRegistration(mysqlpp::Connection& db_connection, +- long subscriber_id, int last_code, ++ long object_id, const string& type, int last_code, + const string& last_reason, + bool update_status, int status, + bool update_ts, unsigned int expiry, +@@ -743,7 +969,12 @@ void DBRegAgent::updateDBRegistration(my + query << ", contacts=" << mysqlpp::quote << contacts; + } + +- query << " where " COLNAME_SUBSCRIBER_ID "="+long2str(subscriber_id) + ";"; ++ // depending on if that is an update for a subscriber or for a peering ++ // do a mysql update ++ if (type == TYPE_SUBSCRIBER) ++ query << " where " COLNAME_SUBSCRIBER_ID "="+long2str(object_id) + ";"; ++ else ++ query << " where " COLNAME_PEER_ID "="+long2str(object_id) + ";"; + string query_str = query.str(); + DBG("updating registration in DB with query '%s'\n", query_str.c_str()); + +@@ -754,8 +985,8 @@ void DBRegAgent::updateDBRegistration(my + } else { + if (!res.rows()) { + // should not happen - DB entry is created on load or on createRegistration +- DBG("creating registration DB entry for subscriber %ld\n", subscriber_id); +- createDBRegistration(subscriber_id, db_connection); ++ DBG("creating registration DB entry for subscriber %ld, type: %s\n", object_id, type.c_str()); ++ createDBRegistration(object_id, type, db_connection); + query.reset(); + query << query_str; + +@@ -794,16 +1025,52 @@ void DBRegAgent::onSipReplyEvent(AmSipRe + #else + ev->reply.local_tag; + #endif +- +- map::iterator it=registration_ltags.find(local_tag); ++ ++ map::iterator it; ++ ++ bool marker = false; ++ string type; ++ ++ // not the best solution to match coming reply against needed object ++ // we need to find a way how to better differentiate, ++ // if that is related to a peering type or subscriber type ++ // now a basic attempt to first look into subscribers cache buckets then into ++ // peerings cache buckets (can this happen that ltag will overlap? must be unique) ++ ++ ++ // first try to find a registration object in a cache for subscribers (most common case) ++ it=registration_ltags.find(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()) { ++ marker=true; ++ type = TYPE_SUBSCRIBER; ++ ++ // secondly, if we didn't find anything before, try in a cache for peerings ++ } else { ++ it=registration_ltags_peers.find(local_tag); ++ if (it!=registration_ltags_peers.end()) { ++ marker=true; ++ type = TYPE_PEERING; ++ } ++ } ++ ++ if (marker) { ++ long object_id = it->second; ++ map::iterator r_it; ++ ++ marker = false; ++ if (type == TYPE_SUBSCRIBER) { // find registration for peering ++ r_it=registrations.find(object_id); ++ if (r_it != registrations.end()) marker = true; ++ } else { // find registration for subscriber ++ r_it=registrations_peers.find(object_id); ++ if (r_it != registrations_peers.end()) marker = true; ++ } ++ ++ if (marker) { + AmSIPRegistration* registration = r_it->second; + if (!registration) { +- ERROR("Internal error: registration object missing\n"); +- return; ++ ERROR("Internal error: registration object missing, type: %s\n", type.c_str()); ++ return; + } + unsigned int cseq_before = registration->getDlg()->cseq; + +@@ -813,7 +1080,7 @@ void DBRegAgent::onSipReplyEvent(AmSipRe + registration->getDlg()->updateStatus(ev->reply); + #endif + +- //update registrations set ++ //update registrations set + bool update_status = false; + int status = 0; + bool update_ts = false; +@@ -822,82 +1089,82 @@ void DBRegAgent::onSipReplyEvent(AmSipRe + bool auth_pending = false; + + if (ev->reply.code >= 300) { +- // REGISTER or de-REGISTER failed +- if ((ev->reply.code == 401 || ev->reply.code == 407) && +- // auth response codes +- // processing reply triggered sending request: resent by auth +- (cseq_before != registration->getDlg()->cseq)) { +- DBG("received negative reply, but still in pending state (auth).\n"); +- auth_pending = true; +- } else { +- if (!registration->getUnregistering()) { +- // REGISTER failed - mark in DB +- DBG("registration failed - mark in DB\n"); +- update_status = true; +- status = REG_STATUS_FAILED; +- if (error_retry_interval) { +- // schedule register-refresh after error_retry_interval +- setRegistrationTimer(subscriber_id, error_retry_interval, +- RegistrationActionEvent::Register); +- } +- } else { +- // de-REGISTER failed +- if (delete_removed_registrations && delete_failed_deregistrations) { +- DBG("de-Register failed - deleting registration %ld " +- "(delete_failed_deregistrations=yes)\n", subscriber_id); +- delete_status = true; +- } else { +- update_status = true; +- status = REG_STATUS_TO_BE_REMOVED; +- } +- } +- } ++ // REGISTER or de-REGISTER failed ++ if ((ev->reply.code == 401 || ev->reply.code == 407) && ++ // auth response codes ++ // processing reply triggered sending request: resent by auth ++ (cseq_before != registration->getDlg()->cseq)) { ++ DBG("received negative reply, but still in pending state (auth).\n"); ++ auth_pending = true; ++ } else { ++ if (!registration->getUnregistering()) { ++ // REGISTER failed - mark in DB ++ DBG("registration failed - mark in DB\n"); ++ update_status = true; ++ status = REG_STATUS_FAILED; ++ if (error_retry_interval) { ++ // schedule register-refresh after error_retry_interval ++ setRegistrationTimer(object_id, error_retry_interval, ++ RegistrationActionEvent::Register, type); ++ } ++ } else { ++ // de-REGISTER failed ++ if (delete_removed_registrations && delete_failed_deregistrations) { ++ DBG("de-Register failed - deleting registration %ld " ++ "(delete_failed_deregistrations=yes)\n", object_id); ++ delete_status = true; ++ } else { ++ update_status = true; ++ status = REG_STATUS_TO_BE_REMOVED; ++ } ++ } ++ } + } else if (ev->reply.code >= 200) { +- // positive reply +- if (!registration->getUnregistering()) { +- time_t now_time = time(0); +- setRegistrationTimer(subscriber_id, registration->getExpiresTS(), +- now_time, now_time); +- +- update_status = true; +- status = REG_STATUS_ACTIVE; +- +- update_ts = true; +- expiry = registration->getExpiresLeft(); +- } else { +- if (delete_removed_registrations) { +- delete_status = true; +- } else { +- update_status = true; +- status = REG_STATUS_REMOVED; +- } +- } ++ // positive reply ++ if (!registration->getUnregistering()) { ++ time_t now_time = time(0); ++ setRegistrationTimer(object_id, registration->getExpiresTS(), ++ now_time, now_time, type); ++ ++ update_status = true; ++ status = REG_STATUS_ACTIVE; ++ ++ update_ts = true; ++ expiry = registration->getExpiresLeft(); ++ } else { ++ if (delete_removed_registrations) { ++ delete_status = true; ++ } else { ++ update_status = true; ++ status = REG_STATUS_REMOVED; ++ } ++ } + } + + // skip provisional replies & auth + if (ev->reply.code >= 200 && !auth_pending) { +- // remove unregistered +- if (registration->getUnregistering()) { +- registrations_mut.unlock(); +- removeRegistration(subscriber_id); +- registrations_mut.lock(); +- } ++ // remove unregistered ++ if (registration->getUnregistering()) { ++ registrations_mut.unlock(); ++ removeRegistration(object_id, type); ++ registrations_mut.lock(); ++ } + } + + if (!delete_status) { +- if (auth_pending && !save_auth_replies) { +- DBG("not updating DB with auth reply %u %s\n", +- ev->reply.code, ev->reply.reason.c_str()); +- } else { +- DBG("update DB with reply %u %s\n", ev->reply.code, ev->reply.reason.c_str()); +- updateDBRegistration(MainDBConnection, +- subscriber_id, ev->reply.code, ev->reply.reason, +- update_status, status, update_ts, expiry, +- save_contacts, ev->reply.contact); +- } ++ if (auth_pending && !save_auth_replies) { ++ DBG("not updating DB with auth reply %u %s\n", ++ ev->reply.code, ev->reply.reason.c_str()); ++ } else { ++ DBG("update DB with reply %u %s\n", ev->reply.code, ev->reply.reason.c_str()); ++ updateDBRegistration(MainDBConnection, ++ object_id, type, ev->reply.code, ev->reply.reason, ++ update_status, status, update_ts, expiry, ++ save_contacts, ev->reply.contact); ++ } + } else { +- DBG("delete DB registration of subscriber %ld\n", subscriber_id); +- deleteDBRegistration(subscriber_id, MainDBConnection); ++ DBG("delete DB registration of subscriber %ld\n", object_id); ++ deleteDBRegistration(object_id, type, MainDBConnection); + } + + } else { +@@ -929,15 +1196,15 @@ void DBRegAgent::run() { + processEvents(); + } + +- DBG("DBRegAgent done, removing all registrations from Event Dispatcher...\n"); ++ DBG("DBRegAgent done, removing all registrations from Event Dispatcher for peerings...\n"); + registrations_mut.lock(); +- for (map::iterator it=registration_ltags.begin(); +- it != registration_ltags.end(); it++) { ++ for (map::iterator it=registration_ltags_peers.begin(); ++ it != registration_ltags_peers.end(); it++) { + AmEventDispatcher::instance()->delEventQueue(it->first); + } + registrations_mut.unlock(); + +- DBG("removing "MOD_NAME" registrations from Event Dispatcher...\n"); ++ DBG("removing %s registrations from Event Dispatcher...\n", MOD_NAME); + AmEventDispatcher::instance()->delEventQueue(MOD_NAME); + + mysqlpp::Connection::thread_end(); +@@ -951,19 +1218,31 @@ void DBRegAgent::on_stop() { + running = false; + } + +-void DBRegAgent::setRegistrationTimer(long subscriber_id, unsigned int timeout, +- RegistrationActionEvent::RegAction reg_action) { ++void DBRegAgent::setRegistrationTimer(long object_id, unsigned int timeout, ++ RegistrationActionEvent::RegAction reg_action, const string& type) { + DBG("setting Register timer for subscription %ld, timeout %u, reg_action %u\n", +- subscriber_id, timeout, reg_action); ++ object_id, timeout, reg_action); + + 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); ++ map::iterator it; ++ bool marker = false; ++ ++ if (type == TYPE_PEERING) { ++ it=registration_timers_peers.find(object_id); ++ if (it==registration_timers_peers.end()) marker = true; ++ ++ } else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) { ++ it=registration_timers.find(object_id); ++ if (it==registration_timers.end()) marker = true; ++ } ++ ++ if (marker) { ++ DBG("timer object for subscription %ld not found, type: %s\n", object_id, type.c_str()); + timer = new RegTimer(); +- timer->data1 = subscriber_id; ++ timer->data1 = object_id; ++ timer->data3 = type; // 'peering' or 'subscriber' + timer->cb = _timer_cb; +- DBG("created timer object [%p] for subscription %ld\n", timer, subscriber_id); ++ DBG("created timer object [%p] for subscription %ld, type: %s\n", timer, object_id, type.c_str()); + } else { + timer = it->second; + DBG("removing scheduled timer...\n"); +@@ -973,28 +1252,42 @@ void DBRegAgent::setRegistrationTimer(lo + timer->data2 = reg_action; + timer->expires = time(0) + timeout; + +- DBG("placing timer for %ld in T-%u\n", subscriber_id, timeout); ++ DBG("placing timer for %ld in T-%u, type: %s\n", object_id, timeout, type.c_str()); + registration_scheduler.insert_timer(timer); + +- registration_timers.insert(std::make_pair(subscriber_id, timer)); +- ++ if (type == TYPE_PEERING) ++ registration_timers_peers.insert(std::make_pair(object_id, timer)); ++ else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) ++ registration_timers.insert(std::make_pair(object_id, timer)); + } + +-void DBRegAgent::setRegistrationTimer(long subscriber_id, +- time_t expiry, time_t reg_start_ts, +- time_t now_time) { +- DBG("setting re-Register timer for subscription %ld, expiry %ld, reg_start_t %ld\n", +- subscriber_id, expiry, reg_start_ts); ++void DBRegAgent::setRegistrationTimer(long object_id, ++ time_t expiry, time_t reg_start_ts, ++ time_t now_time, const string& type) { ++ ++ DBG("setting re-Register timer for subscription %ld, expiry %ld, reg_start_t %ld, type: %s\n", ++ object_id, expiry, reg_start_ts, type.c_str()); + + 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); ++ map::iterator it; ++ bool marker = false; ++ ++ if (type == TYPE_PEERING) { ++ it=registration_timers_peers.find(object_id); ++ if (it==registration_timers_peers.end()) marker = true; ++ } else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) { ++ it=registration_timers.find(object_id); ++ if (it==registration_timers.end()) marker = true; ++ } ++ ++ if (marker) { ++ DBG("timer object for subscription %ld not found, type: %s\n", object_id, type.c_str()); + timer = new RegTimer(); +- timer->data1 = subscriber_id; ++ timer->data1 = object_id; ++ timer->data3 = type; // 'peering' or 'subscriber' + timer->cb = _timer_cb; +- DBG("created timer object [%p] for subscription %ld\n", timer, subscriber_id); +- registration_timers.insert(std::make_pair(subscriber_id, timer)); ++ DBG("created timer object [%p] for subscription %ld, type: %s\n", timer, object_id, type.c_str()); ++ registration_timers.insert(std::make_pair(object_id, timer)); + } else { + timer = it->second; + DBG("removing scheduled timer...\n"); +@@ -1058,82 +1351,114 @@ void DBRegAgent::setRegistrationTimer(lo + } + } + +-void DBRegAgent::clearRegistrationTimer(long subscriber_id) { +- DBG("removing timer for subscription %ld", subscriber_id); ++void DBRegAgent::clearRegistrationTimer(long object_id, const string& type) { ++ DBG("Removing timer for subscription %ld, type: %s", object_id, type.c_str()); ++ ++ map::iterator it; + +- map::iterator it=registration_timers.find(subscriber_id); +- if (it==registration_timers.end()) { +- DBG("timer object for subscription %ld not found\n", subscriber_id); ++ // clear registration timer for peerings ++ if (type == TYPE_PEERING) { ++ it=registration_timers_peers.find(object_id); ++ if (it==registration_timers_peers.end()) { ++ DBG("timer object for subscription %ld not found, type: %s\n", object_id, type.c_str()); ++ return; ++ } ++ // clear registration timer for subscribers ++ } else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) { ++ it=registration_timers.find(object_id); ++ if (it==registration_timers.end()) { ++ DBG("timer object for subscription %ld not found, type: %s\n", object_id, type.c_str()); + return; ++ } + } ++ + DBG("removing timer [%p] from scheduler\n", it->second); + registration_scheduler.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); ++ if (type == TYPE_PEERING) ++ registration_timers_peers.erase(it); ++ else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) ++ registration_timers.erase(it); ++} ++ ++void DBRegAgent::removeRegistrationTimer(long object_id, const string& type) { ++ DBG("removing timer object for subscription %ld, type: %s", object_id, type.c_str()); ++ ++ map::iterator it; ++ ++ // remove registration timer for peerings ++ if (type == TYPE_PEERING) { ++ it=registration_timers_peers.find(object_id); ++ if (it==registration_timers_peers.end()) { ++ DBG("timer object for subscription %ld not found, type: %s\n", object_id, type.c_str()); ++ return; ++ } + +- 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; ++ // remove registration timer for subscribers ++ } else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) { ++ it=registration_timers.find(object_id); ++ if (it==registration_timers.end()) { ++ DBG("timer object for subscription %ld not found, type: %s\n", object_id, type.c_str()); ++ return; ++ } + } + + DBG("deleting timer object [%p]\n", it->second); + delete it->second; + +- registration_timers.erase(it); ++ if (type == TYPE_PEERING) ++ registration_timers_peers.erase(it); ++ else if (type == TYPE_SUBSCRIBER || type == TYPE_UNDEFINED) ++ registration_timers.erase(it); + } + +-void DBRegAgent::timer_cb(RegTimer* timer, long subscriber_id, int reg_action) { +- DBG("re-registration timer expired: subscriber %ld, timer=[%p], action %d\n", +- subscriber_id, timer, reg_action); ++void DBRegAgent::timer_cb(RegTimer* timer, long object_id, int reg_action, const string& type) { ++ DBG("re-registration timer expired: subscriber %ld, timer=[%p], action %d, type %s\n", ++ object_id, timer, reg_action, type.c_str()); + + registrations_mut.lock(); +- removeRegistrationTimer(subscriber_id); ++ removeRegistrationTimer(object_id, type); + registrations_mut.unlock(); ++ + switch (reg_action) { + case RegistrationActionEvent::Register: +- scheduleRegistration(subscriber_id); break; ++ scheduleRegistration(object_id, type); break; + case RegistrationActionEvent::Deregister: +- scheduleDeregistration(subscriber_id); break; ++ scheduleDeregistration(object_id, type); break; + default: ERROR("internal: unknown reg_action %d for subscriber %ld timer event\n", +- reg_action, subscriber_id); ++ reg_action, object_id); + }; + } + ++void DBRegAgent::DIcreateRegistration(int object_id, const string& user, ++ const string& pass, const string& realm, ++ const string& contact, const string& auth_user, ++ const string& type, AmArg& ret) { + +-void DBRegAgent::DIcreateRegistration(int subscriber_id, const string& user, +- const string& pass, const string& realm, +- const string& contact, const string& auth_user, +- AmArg& ret) { +- +- string auth_user_temp = auth_user.empty() ? user : auth_user; ++ string auth_user_temp = (auth_user.empty() || auth_user == "" || auth_user == "NULL") ? user : auth_user; + + DBG("DI method: createRegistration(%i, %s, %s, %s, %s, %s)\n", +- subscriber_id, auth_user_temp.c_str(), user.c_str(), ++ object_id, auth_user_temp.c_str(), user.c_str(), + pass.c_str(), realm.c_str(), contact.c_str()); + +- createRegistration(subscriber_id, auth_user_temp, user, pass, realm, contact); +- scheduleRegistration(subscriber_id); ++ createRegistration(object_id, auth_user_temp, user, pass, realm, contact, type); ++ scheduleRegistration(object_id, type); + ret.push(200); + ret.push("OK"); + } + +-void DBRegAgent::DIupdateRegistration(int subscriber_id, const string& user, +- const string& pass, const string& realm, +- const string& contact, const string& auth_user, +- AmArg& ret) { ++void DBRegAgent::DIupdateRegistration(int object_id, const string& user, ++ const string& pass, const string& realm, ++ const string& contact, const string& auth_user, ++ const string& type, AmArg& ret) { + +- string auth_user_temp = auth_user.empty() ? user : auth_user; ++ string auth_user_temp = (auth_user.empty() || auth_user == "" || auth_user == "NULL") ? user : auth_user; + + DBG("DI method: updateRegistration(%i, %s, %s, %s, %s)\n", +- subscriber_id, auth_user_temp.c_str(), user.c_str(), ++ object_id, auth_user_temp.c_str(), user.c_str(), + pass.c_str(), realm.c_str()); + + string contact_uri = contact; +@@ -1141,28 +1466,28 @@ void DBRegAgent::DIupdateRegistration(in + contact_uri = "sip:"+ user + "@" + contact_hostport; + } + +- updateRegistration(subscriber_id, auth_user_temp, user, pass, realm, contact_uri); ++ updateRegistration(object_id, auth_user_temp, user, pass, realm, contact_uri, type); + + ret.push(200); + ret.push("OK"); + } + +-void DBRegAgent::DIremoveRegistration(int subscriber_id, AmArg& ret) { ++void DBRegAgent::DIremoveRegistration(int object_id, const string& type, AmArg& ret) { + DBG("DI method: removeRegistration(%i)\n", +- subscriber_id); +- scheduleDeregistration(subscriber_id); ++ object_id); ++ scheduleDeregistration(object_id, type); + + registrations_mut.lock(); +- clearRegistrationTimer(subscriber_id); ++ clearRegistrationTimer(object_id, type); + registrations_mut.unlock(); + + ret.push(200); + ret.push("OK"); + } + +-void DBRegAgent::DIrefreshRegistration(int subscriber_id, AmArg& ret) { +- DBG("DI method: refreshRegistration(%i)\n", subscriber_id); +- scheduleRegistration(subscriber_id); ++void DBRegAgent::DIrefreshRegistration(int object_id, const string& type, AmArg& ret) { ++ DBG("DI method: refreshRegistration(%i)\n", object_id); ++ scheduleRegistration(object_id, type); + + ret.push(200); + ret.push("OK"); +@@ -1173,49 +1498,110 @@ void DBRegAgent::DIrefreshRegistration(i + void DBRegAgent::invoke(const string& method, + const AmArg& args, AmArg& ret) + { ++ // create a brand new registration + if (method == "createRegistration"){ +- args.assertArrayFmt("issss"); // subscriber_id, user, pass, realm , auth_user ++ args.assertArrayFmt("issssss"); // object_id, user, pass, realm, contact, auth_user, type + string contact; + string auth_user; ++ string type; // 'peering' or 'subscriber' ++ ++ // for case when: object_id, user, pass, realm, contact + if (args.size() == 5) { + assertArgCStr(args.get(4)); + contact = args.get(4).asCStr(); +- } else if (args.size() > 5) { ++ ++ // for case when: object_id, user, pass, realm, contact, auth_user ++ } else if (args.size() == 6) { ++ assertArgCStr(args.get(4)); ++ assertArgCStr(args.get(5)); ++ contact = args.get(4).asCStr(); ++ auth_user = args.get(5).asCStr(); ++ ++ // for case when: object_id, user, pass, realm, contact, auth_user, type ++ } else if (args.size() == 7) { + assertArgCStr(args.get(4)); + assertArgCStr(args.get(5)); ++ assertArgCStr(args.get(6)); + contact = args.get(4).asCStr(); + auth_user = args.get(5).asCStr(); ++ type = args.get(6).asCStr(); + } ++ ++ // we only allow three possible types: 'peering', 'subscriber' and 'undefined' ++ if (type.empty() || (type != TYPE_PEERING && type != TYPE_SUBSCRIBER && type != TYPE_UNDEFINED)) { ++ DBG("REGISTER: Wrong type of the registration object defined: <%s>. Trying to fix.\n", type.c_str()); ++ type = TYPE_UNDEFINED; ++ } ++ ++ DBG("REGISTER: SEMS is about to Create a registration for: object_id=<%d>, type=<%s>, user=<%s>, realm=<%s> \n", ++ args.get(0).asInt(), type.c_str(), args.get(1).asCStr(), args.get(3).asCStr()); + DIcreateRegistration(args.get(0).asInt(), args.get(1).asCStr(), +- args.get(2).asCStr(), args.get(3).asCStr(), contact, auth_user, ret); ++ args.get(2).asCStr(), args.get(3).asCStr(), contact, auth_user, type, ret); ++ ++ // update an existing registration + } else if (method == "updateRegistration"){ +- args.assertArrayFmt("issss"); // subscriber_id, auth_user, user, pass, realm ++ args.assertArrayFmt("issssss"); // object_id, user, pass, realm, contact, auth_user, type + string contact; + string auth_user; ++ string type; // 'peering' or 'subscriber' ++ ++ // for case when: object_id, user, pass, realm, contact + if (args.size() == 5) { + assertArgCStr(args.get(4)); + contact = args.get(4).asCStr(); +- } else if (args.size() > 5) { ++ ++ // for case when: object_id, user, pass, realm, contact, auth_user ++ } else if (args.size() == 6) { ++ assertArgCStr(args.get(4)); ++ assertArgCStr(args.get(5)); ++ contact = args.get(4).asCStr(); ++ auth_user = args.get(5).asCStr(); ++ ++ // for case when: object_id, user, pass, realm, contact, auth_user, type ++ } else if (args.size() == 7) { + assertArgCStr(args.get(4)); + assertArgCStr(args.get(5)); ++ assertArgCStr(args.get(6)); + contact = args.get(4).asCStr(); + auth_user = args.get(5).asCStr(); ++ type = args.get(6).asCStr(); + } ++ ++ // we only allow three possible types: 'peering', 'subscriber' and 'undefined' ++ if (type.empty() || (type != TYPE_PEERING && type != TYPE_SUBSCRIBER && type != TYPE_UNDEFINED)) { ++ DBG("REGISTER: Wrong type of the registration object defined: <%s>. Trying to fix.\n", type.c_str()); ++ type = TYPE_UNDEFINED; ++ } ++ ++ DBG("REGISTER: SEMS is about to Update a registration for: object_id=<%d>, type=<%s>, user=<%s>, realm=<%s> \n", ++ args.get(0).asInt(), type.c_str(), args.get(1).asCStr(), args.get(3).asCStr()); + DIupdateRegistration(args.get(0).asInt(), args.get(1).asCStr(), +- args.get(2).asCStr(), args.get(3).asCStr(), contact, auth_user, ret); +- } else if (method == "removeRegistration"){ +- args.assertArrayFmt("i"); // subscriber_id +- DIremoveRegistration(args.get(0).asInt(), ret); +- } else if (method == "refreshRegistration"){ +- args.assertArrayFmt("i"); // subscriber_id +- DIrefreshRegistration(args.get(0).asInt(), ret); +- } else if(method == "_list"){ ++ args.get(2).asCStr(), args.get(3).asCStr(), contact, auth_user, type, ret); ++ ++ // remove an existing registration ++ } else if (method == "removeRegistration") { ++ args.assertArrayFmt("is"); // object_id, type ++ string type; // must be 'peering' or 'subscriber' ++ if (args.size() == 2) type = args.get(1).asCStr(); ++ if (type.empty()) type = TYPE_UNDEFINED; ++ DIremoveRegistration(args.get(0).asInt(), type, ret); ++ ++ // refresh an existing registration ++ } else if (method == "refreshRegistration") { ++ args.assertArrayFmt("is"); // object_id, type ++ string type; // must be 'peering' or 'subscriber' ++ if (args.size() == 2) type = args.get(1).asCStr(); ++ if (type.empty()) type = TYPE_UNDEFINED; ++ DIrefreshRegistration(args.get(0).asInt(), type, ret); ++ ++ } else if(method == "_list") { + ret.push(AmArg("createRegistration")); + ret.push(AmArg("updateRegistration")); + ret.push(AmArg("removeRegistration")); + ret.push(AmArg("refreshRegistration")); +- } else ++ } else { + throw AmDynInvoke::NotImplemented(method); ++ } + } + + // /////////////// processor thread ///////////////// +--- a/apps/db_reg_agent/DBRegAgent.h ++++ b/apps/db_reg_agent/DBRegAgent.h +@@ -55,6 +55,7 @@ using std::queue; + #define REG_STATUS_TO_BE_REMOVED_S "5" + + #define COLNAME_SUBSCRIBER_ID "subscriber_id" ++#define COLNAME_PEER_ID "peer_host_id" + #define COLNAME_AUTH_USER "auth_user" + #define COLNAME_USER "user" + #define COLNAME_PASS "pass" +@@ -66,6 +67,11 @@ using std::queue; + #define COLNAME_REGISTRATION_TS "last_registration" + #define COLNAME_LAST_CODE "last_code" + #define COLNAME_LAST_REASON "last_reason" ++#define COLNAME_ID_PK "id" ++ ++#define TYPE_PEERING "peering" ++#define TYPE_SUBSCRIBER "subscriber" ++#define TYPE_UNDEFINED "undefined" + + #define RegistrationActionEventID 117 + +@@ -75,12 +81,13 @@ struct RegistrationActionEvent : public + + enum RegAction { Register=0, Deregister }; + +-RegistrationActionEvent(RegAction action, long subscriber_id) ++RegistrationActionEvent(RegAction action, long object_id, const string& type) + : AmEvent(RegistrationActionEventID), +- action(action), subscriber_id(subscriber_id) { } ++ action(action), object_id(object_id), type(type) { } + + RegAction action; +- long subscriber_id; ++ long object_id; ++ const string type; + }; + + class DBRegAgent; +@@ -120,7 +127,8 @@ class DBRegAgent + public AmEventHandler + { + +- static string joined_query; ++ static string joined_query_subscribers; ++ static string joined_query_peerings; + static string registrations_table; + + static double reregister_interval; +@@ -147,9 +155,14 @@ class DBRegAgent + + static unsigned int error_retry_interval; + +- map registrations; ++ map registrations; // usual subscribers + map registration_ltags; + map registration_timers; ++ ++ map registrations_peers; // SIP peerings ++ map registration_ltags_peers; ++ map registration_timers_peers; ++ + AmMutex registrations_mut; + + // connection used in main DBRegAgent thread +@@ -170,59 +183,62 @@ class DBRegAgent + RegistrationTimer registration_scheduler; + DBRegAgentProcessorThread registration_processor; + +- bool loadRegistrations(); ++ bool loadRegistrations(); // for loading subscribers ++ bool loadRegistrationsPeerings(); // for loading peerings + +- void createDBRegistration(long subscriber_id, mysqlpp::Connection& conn); +- void deleteDBRegistration(long subscriber_id, mysqlpp::Connection& conn); ++ void createDBRegistration(long object_id, const string& type, mysqlpp::Connection& conn); ++ void deleteDBRegistration(long object_id, const string& type, mysqlpp::Connection& conn); + void updateDBRegistration(mysqlpp::Connection& db_connection, +- long subscriber_id, int last_code, ++ long object_id, const string& type, int last_code, + const string& last_reason, + bool update_status = false, int status = 0, + bool update_ts=false, unsigned int expiry = 0, + bool update_contacts=false, const string& contacts = ""); + + /** create registration in our list */ +- void createRegistration(long subscriber_id, ++ void createRegistration(long object_id, + const string& auth_user, + const string& user, + const string& pass, + const string& realm, +- const string& contact); ++ const string& contact, ++ const string& type); + /** update registration in our list */ + void updateRegistration(long subscriber_id, + const string& auth_user, + const string& user, + const string& pass, + const string& realm, +- const string& contact); ++ const string& contact, ++ const string& type); + + /** remove registration */ +- void removeRegistration(long subscriber_id); ++ void removeRegistration(long object_id, const string& type); + + /** schedule this subscriber to REGISTER imminently */ +- void scheduleRegistration(long subscriber_id); ++ void scheduleRegistration(long object_id, const string& type); + + /** schedule this subscriber to de-REGISTER imminently*/ +- void scheduleDeregistration(long subscriber_id); ++ void scheduleDeregistration(long object_id, const string& type); + + /** create a timer for the registration - fixed expiry + action */ +- void setRegistrationTimer(long subscriber_id, unsigned int timeout, +- RegistrationActionEvent::RegAction reg_action); ++ void setRegistrationTimer(long object_id, unsigned int timeout, ++ RegistrationActionEvent::RegAction reg_action, const string& type); + + /** create a registration refresh timer for that registration +- @param subscriber_id - ID of subscription ++ @param object_id - ID of subscription + @param expiry - SIP registration expiry time + @param reg_start_ts - start TS of the SIP registration + @param now_time - current time + */ +- void setRegistrationTimer(long subscriber_id, +- time_t expiry, time_t reg_start_ts, time_t now_time); ++ void setRegistrationTimer(long object_id, ++ time_t expiry, time_t reg_start_ts, time_t now_time, const string& type); + + /** clear re-registration timer and remove timer object */ +- void clearRegistrationTimer(long subscriber_id); ++ void clearRegistrationTimer(long object_id, const string& type); + + /** remove timer object */ +- void removeRegistrationTimer(long subscriber_id); ++ void removeRegistrationTimer(long object_id, const string& type); + + // void run_tests(); + +@@ -248,14 +264,14 @@ class DBRegAgent + + AmDynInvoke* uac_auth_i; + +- void DIcreateRegistration(int subscriber_id, const string& user, ++ void DIcreateRegistration(int object_id, const string& user, + const string& pass, const string& realm, +- const string& contact, const string& auth_user, AmArg& ret); +- void DIupdateRegistration(int subscriber_id, const string& user, ++ const string& contact, const string& auth_user, const string& type, AmArg& ret); ++ void DIupdateRegistration(int object_id, const string& user, + const string& pass, const string& realm, +- const string& contact, const string& auth_user, AmArg& ret); +- void DIremoveRegistration(int subscriber_id, AmArg& ret); +- void DIrefreshRegistration(int subscriber_id, AmArg& ret); ++ const string& contact, const string& auth_user, const string& type, AmArg& ret); ++ void DIremoveRegistration(int object_id, const string& type, AmArg& ret); ++ void DIrefreshRegistration(int object_id, const string& type, AmArg& ret); + + + public: +@@ -271,7 +287,7 @@ class DBRegAgent + void invoke(const string& method, + const AmArg& args, AmArg& ret); + /** re-registration timer callback */ +- void timer_cb(RegTimer* timer, long subscriber_id, int data2); ++ void timer_cb(RegTimer* timer, long object_id, int data2, const string& type); + + friend class DBRegAgentProcessorThread; + }; +--- a/apps/db_reg_agent/RegistrationTimer.cpp ++++ b/apps/db_reg_agent/RegistrationTimer.cpp +@@ -84,7 +84,7 @@ void RegistrationTimer::place_timer(RegT + void RegistrationTimer::fire_timer(RegTimer* timer) { + if (timer && timer->cb) { + DBG("firing timer [%p]\n", timer); +- timer->cb(timer, timer->data1, timer->data2); ++ timer->cb(timer, timer->data1, timer->data2, timer->data3); + } + } + +--- a/apps/db_reg_agent/RegistrationTimer.h ++++ b/apps/db_reg_agent/RegistrationTimer.h +@@ -34,6 +34,9 @@ + #include "log.h" + #include "AmThread.h" + ++#include ++using std::string; ++ + #define TIMER_BUCKET_LENGTH 10 // 10 sec + #define TIMER_BUCKETS 40000 // 40000 buckets (400000 sec, 111 hrs) + +@@ -41,7 +44,7 @@ + #define TIMER_RESOLUTION 100000 + + class RegTimer; +-typedef void (*timer_cb)(RegTimer*, long /*data1*/,int /*data2*/); ++typedef void (*timer_cb)(RegTimer*, long /*data1*/, int /*data2*/, const string& type); + + class RegTimerBucket; + +@@ -52,6 +55,7 @@ class RegTimer { + timer_cb cb; + long data1; + int data2; ++ string data3; + + RegTimer() + : expires(0), cb(0), data1(0), data2(0) { } +--- a/core/AmSipRegistration.cpp ++++ b/core/AmSipRegistration.cpp +@@ -72,9 +72,12 @@ void AmSIPRegistration::setRegistrationI + info = _info; + + cred.realm = info.domain; +- cred.user = info.user; ++ cred.user = info.auth_user.empty() ? info.user : info.auth_user; + cred.pwd = info.pwd; + ++ DBG("REGISTER: new credentials for registrations are: user=<%s>, pass=<%s>, realm=<%s>\n", ++ cred.user.c_str(), cred.pwd.c_str(), cred.realm.c_str()); ++ + req.user = info.user; + req.r_uri = "sip:"+info.domain; + req.from = info.name+" ";