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     //default:
  96     //   return -1; // IEC model and Unknown traffic types
  97     }
  98   } catch(anna::RuntimeException &ex) {
  99     LOGDEBUG(
 100       std::string msg = ex.getText();
 101       msg += " | Round-robin between sessions will be used to send";
 102       anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 103     );
 104   }
 105
 106   return -1;
 107 }
 108
 109
 110 bool Entity::send(const Message* message, bool balance) throw(anna::RuntimeException) {
 111   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "send", ANNA_FILE_LOCATION));
 112   assertReady();
 113   // Carried socket id (forwarding/proxy features):
 114   std::string carriedSocketId = message->getRequestClientSessionKey();
 115
 116   if(carriedSocketId != "") {
 117     // Send (it was a request, we forward the answer):
 118     try {
 119       ClientSession * fixedClientSession = a_engine->findClientSession(carriedSocketId); // exception if not found
 120       fixedClientSession->send(message);
 121       return true;
 122     } catch(anna::RuntimeException &ex) {
 123       std::string msg = "Cannot deliver answer through a fixed client session (";
 124       msg += carriedSocketId;
 125       msg += "). Perhaps it existed but not now. Ignore";
 126       anna::Logger::error(msg, ANNA_FILE_LOCATION);
 127       ex.trace();
 128     }
 129   }
 130
 131   ////////////////////////////////////////////////////////////////////////////////////////
 132   // BALANCE vs STANDARD
 133   // Balance algorythm remember last delivery resource used, balancing from start to end,
 134   // understanding start as next resource to last used, and end as last used. Standard
 135   // algorithm always starts at primary (first defined) server.
 136   ////////////////////////////////////////////////////////////////////////////////////////
 137
 138   // Balance
 139   if(balance) {
 140     for(int k = 0; k < getMaxServers(); k++) {   // try round-robin only over one cycle,
 141       // no matter where you are: don't repeat same server
 142       if(a_deliveryIterator == end()) a_deliveryIterator = begin();
 143
 144       a_lastUsedResource = (*a_deliveryIterator);
 145       a_deliveryIterator++;
 146       // At 'readSocketId' documentation:
 147       //      If server is configured as single session (max client sessions equal to 1), entity will ignore
 148       //      this method because it won't affect the session selection.
 149       int serverSessions = a_lastUsedResource->getMaxClientSessions();
 150       int socketId = (serverSessions > 1) ? readSocketId(message, serverSessions) : -1; // optimization
 151
 152       if(a_lastUsedResource->send(message, socketId))  // exception only possible at findClientSession within server send procedure
 153         return true;
 154     }
 155   } else {
 156     // Standard (no balance) //   start at begining, try secondary, and so on until end.
 157     std::vector<Server*>::iterator it = begin();
 158
 159     while(it != end()) {
 160       a_lastUsedResource = (*it);
 161       it++;
 162       // At 'readSocketId' documentation:
 163       //      If server is configured as single session (max client sessions equal to 1), entity will ignore
 164       //      this method because it won't affect the session selection.
 165       int serverSessions = a_lastUsedResource->getMaxClientSessions();
 166       int socketId = (serverSessions > 1) ? readSocketId(message, serverSessions) : -1; // optimization
 167
 168       if(a_lastUsedResource->send(message, socketId))  // exception only possible at findClientSession within server send procedure
 169         return true;
 170     }
 171   }
 172
 173   // END BALANCE AND TRY ALGORITHM or STANDARD ///////////////////////////////////////////
 174   ////////////////////////////////////////////////////////////////////////////////////////
 175   // Here, sent has failed:
 176   // OAM
 177   OamModule &oamModule = OamModule::instantiate();
 178
 179   if(a_maxServers != 2) {
 180     oamModule.activateAlarm(OamModule::Alarm::UnableToDeliverDiameterMessageToEntityDefinedAs__s__, a_socketListLiteral.c_str());
 181     oamModule.count(OamModule::Counter::UnableToDeliverOverEntity);
 182   } else {
 183     OamModule &oamModule = OamModule::instantiate();
 184     oamModule.activateAlarm(OamModule::Alarm::UnableToDeliverDiameterMessageToEntityDefinedAsPrimary__s__AndSecondary__s__,
 185                             a_primarySocketLiteral.c_str(),
 186                             a_secondarySocketLiteral.c_str());
 187     oamModule.count(OamModule::Counter::UnableToDeliverOverEntity);
 188   }
 189
 190   return false;
 191 }
 192
 193 bool Entity::broadcast(const Message* message) throw(anna::RuntimeException) {
 194   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "broadcast", ANNA_FILE_LOCATION));
 195   assertReady();
 196   bool allok = true;
 197   bool ok;
 198
 199   for(std::vector<Server*>::iterator it = begin(); it != end(); it++) {
 200     try {
 201       ok = (*it)->broadcast(message);
 202
 203       if(!ok) allok = false;
 204     } catch(anna::RuntimeException &ex) {
 205       ex.trace();
 206       allok = false;
 207     }
 208   }
 209
 210   return allok;
 211 }
 212
 213 bool Entity::bind() throw(anna::RuntimeException) {
 214   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "bind", ANNA_FILE_LOCATION));
 215   assertReady();
 216   bool result = true; // all OK return
 217
 218   for(std::vector<Server*>::iterator it = begin(); it != end(); it++) {
 219     try {
 220       (*it)->bind();
 221     } catch(anna::RuntimeException &ex) {
 222       ex.trace();
 223       result = false;
 224     }
 225   }
 226
 227   return result;
 228 }
 229
 230 void Entity::raiseAutoRecovery(bool autoRecovery) throw(anna::RuntimeException) {
 231   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "raiseAutoRecovery", ANNA_FILE_LOCATION));
 232   assertReady();
 233
 234   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 235     (*it)->raiseAutoRecovery(autoRecovery);
 236 }
 237
 238 void Entity::setClassCodeTimeout(const ClassCode::_v v, const anna::Millisecond & millisecond) throw() {
 239   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "setClassCodeTimeout", ANNA_FILE_LOCATION));
 240   assertReady();
 241
 242   for(std::vector<Server*>::iterator it = begin(); it != end(); it++) {
 243     try {
 244       (*it)->setClassCodeTimeout(v, millisecond);
 245     } catch(anna::RuntimeException &ex) {
 246       ex.trace();
 247     }
 248   }
 249 }
 250
 251
 252 // Private close/destroy method
 253 void Entity::close(bool destroy) throw(anna::RuntimeException) {
 254   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Entity", "close", ANNA_FILE_LOCATION));
 255
 256   if(!a_engine)
 257     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
 258
 259   assertReady();
 260
 261   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 262     a_engine->closeServer(*it, destroy);
 263 }
 264
 265
 266 socket_v Entity::getAddressPortList() const throw() {
 267   socket_v result;
 268
 269   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++) {
 270     socket_t address((*it)->getAddress(), (*it)->getPort());
 271     result.push_back(address);
 272   }
 273
 274   return result;
 275 }
 276
 277 int Entity::getOTARequests() const throw() {
 278   int result = 0;
 279
 280   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 281     result += (*it)->getOTARequests();
 282
 283   return result;
 284 }
 285
 286 void Entity::childIdle() const throw() {
 287   // Check father engine idleness:
 288   if(idle()) a_engine->eraseDeprecatedIdleEntities();
 289 }
 290
 291
 292 void Entity::hide() throw() {
 293   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 294     (*it)->hide();
 295 }
 296
 297 void Entity::show() throw() {
 298   for(std::vector<Server*>::iterator it = begin(); it != end(); it++)
 299     (*it)->show();
 300 }
 301
 302 bool Entity::hidden() const throw() {
 303   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 304     if((*it)->shown()) return false;
 305
 306   return true;
 307 }
 308 bool Entity::shown() const throw() {
 309   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 310     if((*it)->hidden()) return false;
 311
 312   return true;
 313 }
 314
 315 void Entity::eventPeerShutdown(const ClientSession* clientSession) throw() {
 316   LOGWARNING(
 317     std::string msg(clientSession->asString());
 318     msg += " | eventPeerShutdown";
 319     anna::Logger::warning(msg, ANNA_FILE_LOCATION);
 320   );
 321 }
 322
 323 std::string Entity::asString() const throw() {
 324   std::string result("diameter::comm::Entity { ");
 325   std::string realm = a_engine->getRealm();
 326
 327   if(realm != "") {
 328     result += "Parent Engine (realm): ";
 329     result += realm;
 330   }
 331
 332   result += " | Category: ";
 333   result += anna::functions::asString(a_category);
 334
 335   if(a_description != "") {
 336     result += " | Description: '";
 337     result += a_description;
 338   }
 339
 340   result += " | Available: ";
 341   result += a_available ? "yes" : "no";
 342   result += " | Deprecated: ";
 343   result += a_deprecated ? "yes" : "no";
 344   result += " | Max servers supported: ";
 345   result += anna::functions::asString(a_maxServers);
 346   result += " | Currently configured servers: ";
 347   result += anna::functions::asString(a_servers.size());
 348   result += anna::functions::asString(" | OTA requests: %d%s", getOTARequests(), idle() ? " (idle)" : "");
 349   result += " | Last Incoming Activity Time: ";
 350   result += a_lastIncomingActivityTime.asString();
 351   result += " | Last Outgoing Activity Time: ";
 352   result += a_lastOutgoingActivityTime.asString();
 353   result += " | Hidden: ";
 354   result += (hidden() ? "yes" : "no");
 355   result += "\n";
 356
 357   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++) {
 358     result += "\n";
 359     result += (*it)->asString();
 360   }
 361
 362   return result;
 363 }
 364
 365 anna::xml::Node* Entity::asXML(anna::xml::Node* parent) const throw() {
 366   anna::xml::Node* result = parent->createChild("diameter.Entity");
 367   std::string realm = a_engine->getRealm();
 368
 369   if(realm != "") result->createAttribute("ParentEngineRealm", realm);
 370
 371   result->createAttribute("Category", anna::functions::asString(a_category));
 372
 373   if(a_description != "") result->createAttribute("Description", a_description);
 374
 375   result->createAttribute("Available", a_available ? "yes" : "no");
 376   result->createAttribute("Deprecated", a_deprecated ? "yes" : "no");
 377   result->createAttribute("MaxServersSupported", anna::functions::asString(a_maxServers));
 378   result->createAttribute("CurrentlyConfiguredServers", anna::functions::asString(a_servers.size()));
 379   result->createAttribute("OTArequests", anna::functions::asString("%d%s", getOTARequests(), idle() ? " (idle)" : ""));
 380   result->createAttribute("LastIncomingActivityTime", a_lastIncomingActivityTime.asString());
 381   result->createAttribute("LastOutgoingActivityTime", a_lastOutgoingActivityTime.asString());
 382   result->createAttribute("Hidden", hidden() ? "yes" : "no");
 383   anna::xml::Node* servers = result->createChild("Entity.Servers");
 384
 385   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 386     (*it)->asXML(servers);
 387
 388   return result;
 389 }
 390
 391 void Entity::availabilityLost() throw() {
 392   a_available = false;
 393   LOGDEBUG(
 394     std::string msg = "diameter::comm::Entity { Description: ";
 395     msg += getDescription();
 396     msg += " } has lost its availability";
 397     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 398   );
 399   OamModule &oamModule = OamModule::instantiate();
 400
 401   if(a_maxServers != 2) {
 402     oamModule.activateAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAs__s__, a_socketListLiteral.c_str());
 403     oamModule.count(OamModule::Counter::LostAvailabilityOverEntity);
 404   } else {
 405     oamModule.activateAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAsPrimary__s__AndSecondary__s__,
 406                             a_primarySocketLiteral.c_str(),
 407                             a_secondarySocketLiteral.c_str());
 408     oamModule.count(OamModule::Counter::LostAvailabilityOverEntity);
 409   }
 410
 411   a_engine->availabilityLost(this);
 412   a_engine->refreshAvailabilityForEntities();
 413 }
 414
 415
 416 void Entity::availabilityRecovered() throw() {
 417   a_available = true;
 418   LOGDEBUG(
 419     std::string msg = "diameter::comm::Entity { Description: ";
 420     msg += getDescription();
 421     msg += " } has recovered its availability";
 422     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 423   );
 424   OamModule &oamModule = OamModule::instantiate();
 425
 426   if(a_maxServers != 2) {
 427     oamModule.cancelAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAs__s__, a_socketListLiteral.c_str());
 428     oamModule.count(OamModule::Counter::RecoveredAvailabilityOverEntity);
 429   } else {
 430     oamModule.cancelAlarm(OamModule::Alarm::c_LostAvailabilityOverEntityDefinedAsPrimary__s__AndSecondary__s__,
 431                           a_primarySocketLiteral.c_str(),
 432                           a_secondarySocketLiteral.c_str());
 433     oamModule.count(OamModule::Counter::RecoveredAvailabilityOverEntity);
 434   }
 435
 436   a_engine->availabilityRecovered(this);
 437   a_engine->refreshAvailabilityForEntities();
 438 }
 439
 440
 441 bool Entity::refreshAvailability() throw() {
 442   // Here available
 443   if(a_available) {  // check not-bound state for all servers:
 444     bool isolate = true;
 445
 446     for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 447       if((*it)->isAvailable()) { isolate = false; break; }
 448
 449     if(isolate) {
 450       availabilityLost();
 451       return true;
 452     }
 453
 454     return false;
 455   }
 456
 457   // Here not available
 458   for(std::vector<Server*>::const_iterator it = begin(); it != end(); it++)
 459     if((*it)->isAvailable()) {
 460       availabilityRecovered();
 461       return true;
 462     }
 463
 464   return false;
 465 }
 466
 467 //------------------------------------------------------------------------------
 468 //----------------------------------------- Entity::updateIncomingActivityTime()
 469 //------------------------------------------------------------------------------
 470 void Entity::updateIncomingActivityTime() throw() {
 471   a_lastIncomingActivityTime = anna::functions::millisecond();
 472   LOGDEBUG
 473   (
 474     std::string msg = "Updated INCOMING activity on entity (milliseconds unix): ";
 475     msg += anna::functions::asString(a_lastIncomingActivityTime.getValue());
 476     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 477   );
 478 }
 479
 480
 481 //------------------------------------------------------------------------------
 482 //----------------------------------------- Entity::updateOutgoingActivityTime()
 483 //------------------------------------------------------------------------------
 484 void Entity::updateOutgoingActivityTime(void) throw() {
 485   a_lastOutgoingActivityTime = anna::functions::millisecond();
 486   LOGDEBUG
 487   (
 488     std::string msg = "Updated OUTGOING activity on entity (milliseconds unix): ";
 489     msg += anna::functions::asString(a_lastOutgoingActivityTime.getValue());
 490     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 491   );
 492 }
 493