source/diameter.comm/Entity.cpp

   1 // ANNA - Anna is Not Nothingness Anymore                                                         //
   2 //                                                                                                //
   3 // (c) Copyright 2005-2015 Eduardo Ramos Testillano & Francisco Ruiz Rayo                         //
   4 //                                                                                                //
   5 // See project site at http://redmine.teslayout.com/projects/anna-suite                           //
   6 // See accompanying file LICENSE or copy at http://www.teslayout.com/projects/public/anna.LICENSE //
   7
   8
   9 #include <anna/diameter.comm/Server.hpp>
  10 #include <anna/diameter.comm/Entity.hpp>
  11 #include <anna/diameter.comm/Engine.hpp>
  12 #include <anna/diameter.comm/OamModule.hpp>
  13 #include <anna/diameter/codec/functions.hpp>
  14 #include <anna/diameter/helpers/base/defines.hpp>
  15 #include <anna/core/functions.hpp>
  16 #include <anna/time/functions.hpp>
  17 #include <anna/diameter/helpers/base/functions.hpp>
  18 #include <anna/diameter/helpers/dcca/functions.hpp>
  19
  20 #include <anna/core/tracing/Logger.hpp>
  21 #include <anna/core/functions.hpp>
  22 #include <anna/core/tracing/TraceMethod.hpp>
  23
  24
  25 using namespace anna;
  26 using namespace anna::diameter;
  27 using namespace anna::diameter::comm;
  28
  29
  30
  31 void Entity::initialize() throw() {
  32   a_engine = NULL;
  33   a_servers.clear(); // importante (el recycler creo que no lo tocaba)
  34   a_available = false;
  35   a_deprecated = false;
  36   a_socketListLiteral = "";
  37   a_primarySocketLiteral = "";
  38   a_secondarySocketLiteral = "";
  39   a_description = "";
  40   a_category = 0;
  41   a_lastUsedResource = NULL;
  42 }
  43
  44
  45 void Entity::assertReady() throw(anna::RuntimeException) {
  46   if(a_servers.size() != a_maxServers) {
  47     std::string msg(asString());
  48     msg += " | Non-configured entity: you must add the remaining servers before any operation (bind, send, etc.)";
  49     throw anna::RuntimeException(msg, ANNA_FILE_LOCATION);
  50   }
  51 }
  52
  53
  54 void Entity::addServer(const socket_t & serverId)
  55 throw(anna::RuntimeException) {
  56   if(a_servers.size() == a_maxServers) {
  57     LOGDEBUG
  58     (
  59       std::string msg = "Maximum number of servers reached for this entity (";
  60       msg += anna::functions::asString(a_maxServers);
  61       msg += "). Please use clear()/close() primitives";
  62       anna::Logger::debug(msg, ANNA_FILE_LOCATION);
  63     );
  64     return;
  65   }
  66
  67   if(!a_engine)
  68     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
  69
  70   Server *s = a_engine->createServer(this, serverId);
  71   a_servers.push_back(s);
  72   a_deliveryIterator = begin();
  73 }
  74
  75
  76 int Entity::readSocketId(const Message* message, int maxClientSessions) const throw() {
  77   try {
  78     // Service-Context-Id:
  79     anna::diameter::helpers::dcca::ChargingContext::_v chargingContext;
  80     std::string scid = anna::diameter::helpers::dcca::functions::getServiceContextId(message->getBody(), chargingContext);
  81
  82     switch(chargingContext) {
  83     case anna::diameter::helpers::dcca::ChargingContext::Data:
  84     case anna::diameter::helpers::dcca::ChargingContext::Voice:
  85     case anna::diameter::helpers::dcca::ChargingContext::Content: {
  86       // Session-Id: '<DiameterIdentity>;<high 32 bits>;<low 32 bits>[;<optional value>="">]'
  87       std::string sid = anna::diameter::helpers::base::functions::getSessionId(message->getBody());
  88       std::string diameterIdentity, optional;
  89       anna::U32 high, low;
  90       anna::diameter::helpers::base::functions::decodeSessionId(sid, diameterIdentity, high, low /* context-teid */, optional);
  91       return (low % maxClientSessions);
  92     }
  93     case anna::diameter::helpers::dcca::ChargingContext::SMS:
  94     case anna::diameter::helpers::dcca::ChargingContext::MMS:
  95     case anna::diameter::helpers::dcca::ChargingContext::Unknown:
  96     default:
  97        return -1; // IEC model and Unknown traffic types
  98     }
  99   } catch(anna::RuntimeException &ex) {
 100     LOGDEBUG(
 101       std::string msg = ex.getText();
 102       msg += " | Round-robin between sessions will be used to send";
 103       anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 104     );
 105   }
 106
 107   return -1;
 108 }
 109
 110
 111 bool Entity::send(const Message* message, bool balance) throw(anna::RuntimeException) {
 112   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "send", ANNA_FILE_LOCATION));
 113   assertReady();
 114   // Carried socket id (forwarding/proxy features):
 115   std::string carriedSocketId = message->getRequestClientSessionKey();
 116
 117   if(carriedSocketId != "") {
 118     // Send (it was a request, we forward the answer):
 119     try {
 120       ClientSession * fixedClientSession = a_engine->findClientSession(carriedSocketId); // exception if not found
 121       fixedClientSession->send(message);
 122       return true;
 123     } catch(anna::RuntimeException &ex) {
 124       std::string msg = "Cannot deliver answer through a fixed client session (";
 125       msg += carriedSocketId;
 126       msg += "). Perhaps it existed but not now. Ignore";
 127       anna::Logger::error(msg, ANNA_FILE_LOCATION);
 128       ex.trace();
 129     }
 130   }
 131
 132   ////////////////////////////////////////////////////////////////////////////////////////
 133   // BALANCE vs STANDARD
 134   // Balance algorythm remember last delivery resource used, balancing from start to end,
 135   // understanding start as next resource to last used, and end as last used. Standard
 136   // algorithm always starts at primary (first defined) server.
 137   ////////////////////////////////////////////////////////////////////////////////////////
 138
 139   // Balance
 140   if(balance) {
 141     for(int k = 0; k < getMaxServers(); k++) {   // try round-robin only over one cycle,
 142       // no matter where you are: don't repeat same server
 143       if(a_deliveryIterator == end()) a_deliveryIterator = begin();
 144
 145       a_lastUsedResource = (*a_deliveryIterator);
 146       a_deliveryIterator++;
 147       // At 'readSocketId' documentation:
 148       //      If server is configured as single session (max client sessions equal to 1), entity will ignore
 149       //      this method because it won't affect the session selection.
 150       int serverSessions = a_lastUsedResource->getMaxClientSessions();
 151       int socketId = (serverSessions > 1) ? readSocketId(message, serverSessions) : -1; // optimization
 152
 153       if(a_lastUsedResource->send(message, socketId))  // exception only possible at findClientSession within server send procedure
 154         return true;
 155     }
 156   } else {
 157     // Standard (no balance) //   start at begining, try secondary, and so on until end.
 158     std::vector<Server*>::iterator it = begin();
 159
 160     while(it != end()) {
 161       a_lastUsedResource = (*it);
 162       it++;
 163       // At 'readSocketId' documentation:
 164       //      If server is configured as single session (max client sessions equal to 1), entity will ignore
 165       //      this method because it won't affect the session selection.
 166       int serverSessions = a_lastUsedResource->getMaxClientSessions();
 167       int socketId = (serverSessions > 1) ? readSocketId(message, serverSessions) : -1; // optimization
 168
 169       if(a_lastUsedResource->send(message, socketId))  // exception only possible at findClientSession within server send procedure
 170         return true;
 171     }
 172   }
 173
 174   // END BALANCE AND TRY ALGORITHM or STANDARD ///////////////////////////////////////////
 175   ////////////////////////////////////////////////////////////////////////////////////////
 176   // Here, sent has failed:
 177   // OAM
 178   OamModule &oamModule = OamModule::instantiate();
 179
 180   if(a_maxServers != 2) {
 181     oamModule.activateAlarm(OamModule::Alarm::UnableToDeliverDiameterMessageToEntityDefinedAs__s__, a_socketListLiteral.c_str());
 182     oamModule.count(OamModule::Counter::UnableToDeliverOverEntity);
 183   } else {
 184     OamModule &oamModule = OamModule::instantiate();
 185     oamModule.activateAlarm(OamModule::Alarm::UnableToDeliverDiameterMessageToEntityDefinedAsPrimary__s__AndSecondary__s__,
 186                             a_primarySocketLiteral.c_str(),
 187                             a_secondarySocketLiteral.c_str());
 188     oamModule.count(OamModule::Counter::UnableToDeliverOverEntity);
 189   }
 190
 191   return false;
 192 }
 193
 194 bool Entity::broadcast(const Message* message) throw(anna::RuntimeException) {
 195   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "broadcast", ANNA_FILE_LOCATION));
 196   assertReady();
 197   bool allok = true;
 198   bool ok;
 199
 200   for(std::vector<Server*>::iterator it = begin(); it != end(); it++) {
 201     try {
 202       ok = (*it)->broadcast(message);
 203
 204       if(!ok) allok = false;
 205     } catch(anna::RuntimeException &ex) {
 206       ex.trace();
 207       allok = false;
 208     }
 209   }
 210
 211   return allok;
 212 }
 213
 214 bool Entity::bind() throw(anna::RuntimeException) {
 215   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "bind", ANNA_FILE_LOCATION));
 216   assertReady();
 217   bool result = true; // all OK return
 218
 219   for(std::vector<Server*>::iterator it = begin(); it != end(); it++) {
 220     try {
 221       (*it)->bind();
 222     } catch(anna::RuntimeException &ex) {
 223       ex.trace();
 224       result = false;
 225     }
 226   }
 227
 228   return result;
 229 }
 230
 231 void Entity::raiseAutoRecovery(bool autoRecovery) throw(anna::RuntimeException) {
 232   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "raiseAutoRecovery", ANNA_FILE_LOCATION));
 233   assertReady();
 234
 235   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 236     (*it)->raiseAutoRecovery(autoRecovery);
 237 }
 238
 239 void Entity::setClassCodeTimeout(const ClassCode::_v v, const anna::Millisecond & millisecond) throw() {
 240   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "setClassCodeTimeout", ANNA_FILE_LOCATION));
 241   assertReady();
 242
 243   for(std::vector<Server*>::iterator it = begin(); it != end(); it++) {
 244     try {
 245       (*it)->setClassCodeTimeout(v, millisecond);
 246     } catch(anna::RuntimeException &ex) {
 247       ex.trace();
 248     }
 249   }
 250 }
 251
 252
 253 // Private close/destroy method
 254 void Entity::close(bool destroy) throw(anna::RuntimeException) {
 255   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "close", ANNA_FILE_LOCATION));
 256
 257   if(!a_engine)
 258     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
 259
 260   assertReady();
 261
 262   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 263     a_engine->closeServer(*it, destroy);
 264 }
 265
 266
 267 socket_v Entity::getAddressPortList() const throw() {
 268   socket_v result;
 269
 270   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++) {
 271     socket_t address((*it)->getAddress(), (*it)->getPort());
 272     result.push_back(address);
 273   }
 274
 275   return result;
 276 }
 277
 278 int Entity::getOTARequests() const throw() {
 279   int result = 0;
 280
 281   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 282     result += (*it)->getOTARequests();
 283
 284   return result;
 285 }
 286
 287 void Entity::childIdle() const throw() {
 288   // Check father engine idleness:
 289   if(idle()) a_engine->eraseDeprecatedIdleEntities();
 290 }
 291
 292
 293 void Entity::hide() throw() {
 294   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 295     (*it)->hide();
 296 }
 297
 298 void Entity::show() throw() {
 299   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 300     (*it)->show();
 301 }
 302
 303 bool Entity::hidden() const throw() {
 304   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 305     if((*it)->shown()) return false;
 306
 307   return true;
 308 }
 309 bool Entity::shown() const throw() {
 310   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 311     if((*it)->hidden()) return false;
 312
 313   return true;
 314 }
 315
 316 void Entity::eventPeerShutdown(const ClientSession* clientSession) throw() {
 317   LOGWARNING(
 318     std::string msg(clientSession->asString());
 319     msg += " | eventPeerShutdown";
 320     anna::Logger::warning(msg, ANNA_FILE_LOCATION);
 321   );
 322 }
 323
 324 void Entity::eventRequestRetransmission(const ClientSession* clientSession, Message *request) throw() {
 325   LOGWARNING(
 326     std::string msg(clientSession->asString());
 327
 328     HopByHop hopByHop = codec::functions::getHopByHop(request->getBody()); // context identification
 329     int retries = request->getRetries();
 330
 331     msg += anna::functions::asString(" | eventRequestRetransmission: request with application HopByHop: %u; remaining %d retries", hopByHop, retries);
 332     anna::Logger::warning(msg, ANNA_FILE_LOCATION);
 333   );
 334 }
 335
 336 std::string Entity::asString() const throw() {
 337   std::string result("diameter::comm::Entity { ");
 338   std::string realm = a_engine->getRealm();
 339
 340   if(realm != "") {
 341     result += "Parent Engine (realm): ";
 342     result += realm;
 343   }
 344
 345   result += " | Category: ";
 346   result += anna::functions::asString(a_category);
 347
 348   if(a_description != "") {
 349     result += " | Description: '";
 350     result += a_description;
 351   }
 352
 353   result += " | Available: ";
 354   result += a_available ? "yes" : "no";
 355   result += " | Deprecated: ";
 356   result += a_deprecated ? "yes" : "no";
 357   result += " | Max servers supported: ";
 358   result += anna::functions::asString(a_maxServers);
 359   result += " | Currently configured servers: ";
 360   result += anna::functions::asString(a_servers.size());
 361   result += anna::functions::asString(" | OTA requests: %d%s", getOTARequests(), idle() ? " (idle)" : "");
 362   result += " | Last Incoming Activity Time: ";
 363   result += a_lastIncomingActivityTime.asString();
 364   result += " | Last Outgoing Activity Time: ";
 365   result += a_lastOutgoingActivityTime.asString();
 366   result += " | Hidden: ";
 367   result += (hidden() ? "yes" : "no");
 368   result += "\n";
 369
 370   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++) {
 371     result += "\n";
 372     result += (*it)->asString();
 373   }
 374
 375   return result;
 376 }
 377
 378 anna::xml::Node* Entity::asXML(anna::xml::Node* parent) const throw() {
 379   anna::xml::Node* result = parent->createChild("diameter.Entity");
 380   std::string realm = a_engine->getRealm();
 381
 382   if(realm != "") result->createAttribute("ParentEngineRealm", realm);
 383
 384   result->createAttribute("Category", anna::functions::asString(a_category));
 385
 386   if(a_description != "") result->createAttribute("Description", a_description);
 387
 388   result->createAttribute("Available", a_available ? "yes" : "no");
 389   result->createAttribute("Deprecated", a_deprecated ? "yes" : "no");
 390   result->createAttribute("MaxServersSupported", anna::functions::asString(a_maxServers));
 391   result->createAttribute("CurrentlyConfiguredServers", anna::functions::asString(a_servers.size()));
 392   result->createAttribute("OTArequests", anna::functions::asString("%d%s", getOTARequests(), idle() ? " (idle)" : ""));
 393   result->createAttribute("LastIncomingActivityTime", a_lastIncomingActivityTime.asString());
 394   result->createAttribute("LastOutgoingActivityTime", a_lastOutgoingActivityTime.asString());
 395   result->createAttribute("Hidden", hidden() ? "yes" : "no");
 396   anna::xml::Node* servers = result->createChild("Entity.Servers");
 397
 398   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 399     (*it)->asXML(servers);
 400
 401   return result;
 402 }
 403
 404 void Entity::availabilityLost() throw() {
 405   a_available = false;
 406   LOGDEBUG(
 407     std::string msg = "diameter::comm::Entity { Description: ";
 408     msg += getDescription();
 409     msg += " } has lost its availability";
 410     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 411   );
 412   OamModule &oamModule = OamModule::instantiate();
 413
 414   if(a_maxServers != 2) {
 415     oamModule.activateAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAs__s__, a_socketListLiteral.c_str());
 416     oamModule.count(OamModule::Counter::LostAvailabilityOverEntity);
 417   } else {
 418     oamModule.activateAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAsPrimary__s__AndSecondary__s__,
 419                             a_primarySocketLiteral.c_str(),
 420                             a_secondarySocketLiteral.c_str());
 421     oamModule.count(OamModule::Counter::LostAvailabilityOverEntity);
 422   }
 423
 424   a_engine->availabilityLost(this);
 425   a_engine->refreshAvailabilityForEntities();
 426 }
 427
 428
 429 void Entity::availabilityRecovered() throw() {
 430   a_available = true;
 431   LOGDEBUG(
 432     std::string msg = "diameter::comm::Entity { Description: ";
 433     msg += getDescription();
 434     msg += " } has recovered its availability";
 435     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 436   );
 437   OamModule &oamModule = OamModule::instantiate();
 438
 439   if(a_maxServers != 2) {
 440     oamModule.cancelAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAs__s__, a_socketListLiteral.c_str());
 441     oamModule.count(OamModule::Counter::RecoveredAvailabilityOverEntity);
 442   } else {
 443     oamModule.cancelAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAsPrimary__s__AndSecondary__s__,
 444                           a_primarySocketLiteral.c_str(),
 445                           a_secondarySocketLiteral.c_str());
 446     oamModule.count(OamModule::Counter::RecoveredAvailabilityOverEntity);
 447   }
 448
 449   a_engine->availabilityRecovered(this);
 450   a_engine->refreshAvailabilityForEntities();
 451 }
 452
 453
 454 bool Entity::refreshAvailability() throw() {
 455   // Here available
 456   if(a_available) {  // check not-bound state for all servers:
 457     bool isolate = true;
 458
 459     for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 460       if((*it)->isAvailable()) { isolate = false; break; }
 461
 462     if(isolate) {
 463       availabilityLost();
 464       return true;
 465     }
 466
 467     return false;
 468   }
 469
 470   // Here not available
 471   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 472     if((*it)->isAvailable()) {
 473       availabilityRecovered();
 474       return true;
 475     }
 476
 477   return false;
 478 }
 479
 480 //------------------------------------------------------------------------------
 481 //----------------------------------------- Entity::updateIncomingActivityTime()
 482 //------------------------------------------------------------------------------
 483 void Entity::updateIncomingActivityTime() throw() {
 484   a_lastIncomingActivityTime = anna::functions::millisecond();
 485   LOGDEBUG
 486   (
 487     std::string msg = "Updated INCOMING activity on entity (milliseconds unix): ";
 488     msg += anna::functions::asString(a_lastIncomingActivityTime.getValue());
 489     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 490   );
 491 }
 492
 493
 494 //------------------------------------------------------------------------------
 495 //----------------------------------------- Entity::updateOutgoingActivityTime()
 496 //------------------------------------------------------------------------------
 497 void Entity::updateOutgoingActivityTime(void) throw() {
 498   a_lastOutgoingActivityTime = anna::functions::millisecond();
 499   LOGDEBUG
 500   (
 501     std::string msg = "Updated OUTGOING activity on entity (milliseconds unix): ";
 502     msg += anna::functions::asString(a_lastOutgoingActivityTime.getValue());
 503     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 504   );
 505 }
 506