source/diameter.comm/Server.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/ClientSession.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/statistics/Engine.hpp>
  17 #include <anna/time/functions.hpp>
  18
  19 #include <anna/core/tracing/Logger.hpp>
  20 #include <anna/core/functions.hpp>
  21 #include <anna/core/tracing/TraceMethod.hpp>
  22
  23 // STL
  24 #include <vector>
  25
  26 using namespace anna;
  27 using namespace anna::diameter;
  28 using namespace anna::diameter::comm;
  29
  30
  31 void Server::initialize() throw() {
  32   a_parent = NULL;
  33   a_engine = NULL;
  34   a_clientSessions.clear(); // importante (el recycler creo que no lo tocaba)
  35   a_available = false;
  36   a_maxClientSessions = 1; // mono client connection
  37   a_lastIncomingActivityTime = (anna::Millisecond)0;
  38   a_lastOutgoingActivityTime = (anna::Millisecond)0;
  39   a_lastUsedResource = NULL;
  40 }
  41
  42 void Server::initializeStatisticResources() throw() {
  43   std::string accName = "remote server '";
  44   accName += anna::functions::socketLiteralAsString(a_socket.first, a_socket.second);
  45   accName += "' from origin-realm '";
  46   accName += a_engine ? a_engine->getOriginRealmName() : "unknown"; // it should be known (createServer)
  47   accName += "' and origin-host '";
  48   accName += a_engine ? a_engine->getOriginHostName() : "unknown"; // it should be known (createServer)
  49   accName += "'";
  50   a_messageStatistics.initialize(accName);
  51 }
  52
  53 void Server::resetStatistics() throw() {
  54   a_messageStatistics.getAccumulator()->reset();
  55 }
  56
  57 void Server::updateProcessingTimeStatisticConcept(const double &value, const anna::diameter::CommandId &cid) throw() {
  58   a_messageStatistics.process(MessageStatistics::ConceptType::SentRequestProcessingTime, cid, value);
  59   LOGDEBUG(anna::Logger::debug(a_messageStatistics.getAccumulator()->asString(), ANNA_FILE_LOCATION));
  60 }
  61
  62 void Server::updateReceivedMessageSizeStatisticConcept(const double &value, const anna::diameter::CommandId &cid) throw() {
  63   a_messageStatistics.process(MessageStatistics::ConceptType::ReceivedMessageSize, cid, value);
  64   LOGDEBUG(anna::Logger::debug(a_messageStatistics.getAccumulator()->asString(), ANNA_FILE_LOCATION));
  65 }
  66
  67
  68 void Server::assertReady() throw(anna::RuntimeException) {
  69   if(a_clientSessions.size() != a_maxClientSessions) {
  70     std::string msg(asString());
  71     msg += " | Non-configured server: you must add the remaining client-sessions before any operation (bind, send, etc.)";
  72     throw anna::RuntimeException(msg, ANNA_FILE_LOCATION);
  73   }
  74 }
  75
  76
  77 void Server::addClientSession(int socketId)
  78 throw(anna::RuntimeException) {
  79   if(a_clientSessions.size() == a_maxClientSessions) {
  80     LOGDEBUG
  81     (
  82       std::string msg = "Maximum number of client-sessions reached for this server (";
  83       msg += anna::functions::asString(a_maxClientSessions);
  84       msg += ").";
  85       anna::Logger::debug(msg, ANNA_FILE_LOCATION);
  86     );
  87     return;
  88   }
  89
  90   if(!a_engine)
  91     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
  92
  93   ClientSession *s = a_engine->createClientSession(this, socketId);
  94   a_clientSessions.push_back(s);
  95 }
  96
  97 int Server::getOTARequests() const throw() {
  98   int result = 0;
  99
 100   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 101     result += (*it)->getOTARequests();
 102
 103   return result;
 104 }
 105
 106
 107 bool Server::send(const Message* message, int socketId) throw(anna::RuntimeException) {
 108   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "send", ANNA_FILE_LOCATION));
 109   assertReady();
 110   bool fixedSocket = (socketId != -1);
 111   int clientSessions = getNumberOfClientSessions();
 112
 113   for(int k = 0; k < clientSessions; k++) {    // try round-robin only over one cycle,
 114     // no matter where you are: don't repeat same socket
 115     if(fixedSocket)
 116       a_lastUsedResource = a_engine->findClientSession(a_socket.first /*ip*/, a_socket.second /*port*/, socketId); // exception if not found
 117     else {
 118       if(a_deliveryIterator == end()) a_deliveryIterator = begin();
 119
 120       a_lastUsedResource = (*a_deliveryIterator);
 121
 122       if(clientSessions > 1) a_deliveryIterator++;  // Round robin
 123     }
 124
 125     try {
 126       // Send:
 127       const Response* response = a_lastUsedResource->send(message);
 128       return true; // no matter if response is NULL (answers, i.e.) or not.
 129     } catch(anna::RuntimeException &ex) {
 130       LOGDEBUG(
 131         std::string msg = ex.getText();
 132         msg += " | Send failed on socket id ";
 133         msg += anna::functions::asString(a_lastUsedResource->getSocketId());
 134         anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 135       );
 136     }
 137
 138     // Specific socket sending doesn't try
 139     if(fixedSocket) return false;
 140   }
 141
 142   return false;
 143 }
 144
 145
 146 bool Server::broadcast(const Message* message) throw(anna::RuntimeException) {
 147   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "broadcast", ANNA_FILE_LOCATION));
 148   assertReady();
 149   const Response* response;
 150   bool allok = true;
 151
 152   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++) {
 153     try {
 154       response = (*it)->send(message);
 155     } catch(anna::RuntimeException &ex) {
 156       ex.trace();
 157       allok = false;
 158     }
 159   }
 160
 161   return allok;
 162 }
 163
 164
 165 bool Server::bind() throw(anna::RuntimeException) {
 166   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "bind", ANNA_FILE_LOCATION));
 167   assertReady();
 168   a_deliveryIterator = begin();
 169   bool result = true; // all OK return
 170
 171   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++) {
 172     try {
 173       (*it)->bind();
 174     } catch(anna::RuntimeException &ex) {
 175       ex.trace();
 176       result = false;
 177     }
 178   }
 179
 180   return result;
 181 }
 182
 183 void Server::raiseAutoRecovery(bool autoRecovery) throw(anna::RuntimeException) {
 184   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "raiseAutoRecovery", ANNA_FILE_LOCATION));
 185   assertReady();
 186   a_deliveryIterator = begin();
 187
 188   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 189     (*it)->setAutoRecovery(autoRecovery);
 190 }
 191
 192 void Server::setClassCodeTimeout(const ClassCode::_v v, const anna::Millisecond & millisecond) throw() {
 193   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "setClassCodeTimeout", ANNA_FILE_LOCATION));
 194   assertReady();
 195
 196   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++) {
 197     try {
 198       (*it)->setClassCodeTimeout(v, millisecond);
 199     } catch(anna::RuntimeException &ex) {
 200       ex.trace();
 201     }
 202   }
 203 }
 204
 205
 206 // Private close/destroy method
 207 void Server::close(bool destroy) throw(anna::RuntimeException) {
 208   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "close", ANNA_FILE_LOCATION));
 209
 210   if(!a_engine)
 211     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
 212
 213   assertReady();
 214
 215   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 216     a_engine->closeClientSession(*it, destroy);
 217 }
 218
 219
 220 void Server::childIdle() const throw() {
 221   // Check father entity idleness:
 222   if(idle()) a_parent->childIdle();
 223 }
 224
 225
 226 void Server::hide() throw() {
 227   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 228     (*it)->hide();
 229 }
 230
 231 void Server::show() throw() {
 232   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 233     (*it)->show();
 234 }
 235
 236 bool Server::hidden() const throw() {
 237   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 238     if((*it)->shown()) return false;
 239
 240   return true;
 241 }
 242 bool Server::shown() const throw() {
 243   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 244     if((*it)->hidden()) return false;
 245
 246   return true;
 247 }
 248
 249
 250
 251 std::string Server::socketAsString() const throw() {
 252   std::string result = getAddress();
 253   result += ":";
 254   result += anna::functions::asString(getPort());
 255   return result;
 256 }
 257
 258
 259 std::string Server::asString() const throw() {
 260   std::string result("diameter::comm::Server { ");
 261   result += " | Parent Entity: ";
 262   result += a_parent->getDescription();
 263   result += " | Server Address: ";
 264   result += a_socket.first;
 265   result += " | Server Port: ";
 266   result += anna::functions::asString(a_socket.second);
 267   result += " | Available: ";
 268   result += a_available ? "yes" : "no";
 269   result += " | Max client-sessions supported: ";
 270   result += anna::functions::asString(a_maxClientSessions);
 271   result += " | Currently configured client-sessions: ";
 272   result += anna::functions::asString(a_clientSessions.size());
 273   result += anna::functions::asString(" | OTA requests: %d%s", getOTARequests(), idle() ? " (idle)" : "");
 274   result += " | Last Incoming Activity Time: ";
 275   result += a_lastIncomingActivityTime.asString();
 276   result += " | Last Outgoing Activity Time: ";
 277   result += a_lastOutgoingActivityTime.asString();
 278   result += " | Hidden: ";
 279   result += (hidden() ? "yes" : "no");
 280   result += "\n";
 281   result += a_messageStatistics.getAccumulator()->asString();
 282
 283   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++) {
 284     result += "\n";
 285     result += (*it)->asString();
 286   }
 287
 288   return result;
 289 }
 290
 291 anna::xml::Node* Server::asXML(anna::xml::Node* parent) const throw() {
 292   anna::xml::Node* result = parent->createChild("diameter.Server");
 293   result->createAttribute("ParentEntity", a_parent->getDescription());
 294   result->createAttribute("ServerAddress", a_socket.first);
 295   result->createAttribute("ServerPort", anna::functions::asString(a_socket.second));
 296   result->createAttribute("Available", a_available ? "yes" : "no");
 297   result->createAttribute("MaxClientSessionsSupported", anna::functions::asString(a_maxClientSessions));
 298   result->createAttribute("CurrentlyConfiguredClientSessions", anna::functions::asString(a_clientSessions.size()));
 299   result->createAttribute("OTArequests", anna::functions::asString("%d%s", getOTARequests(), idle() ? " (idle)" : ""));
 300   result->createAttribute("LastIncomingActivityTime", a_lastIncomingActivityTime.asString());
 301   result->createAttribute("LastOutgoingActivityTime", a_lastOutgoingActivityTime.asString());
 302   result->createAttribute("Hidden", hidden() ? "yes" : "no");
 303   // Statistics
 304   anna::xml::Node* stats = result->createChild("Statistics");
 305   a_messageStatistics.getAccumulator()->asXML(stats);
 306   anna::xml::Node* clientSessions = result->createChild("Server.ClientSessions");
 307
 308   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 309     (*it)->asXML(clientSessions);
 310
 311   return result;
 312 }
 313
 314
 315 void Server::eventPeerShutdown(const ClientSession *clientSession) throw() {
 316   // Inform father entity:
 317   a_parent->eventPeerShutdown(clientSession);
 318 }
 319
 320 void Server::eventRequestRetransmission(const ClientSession* clientSession, Message *request) throw() {
 321   // Inform father entity:
 322   a_parent->eventRequestRetransmission(clientSession, request);
 323 }
 324
 325 void Server::eventResponse(const Response& response) throw(anna::RuntimeException) {
 326   // Inform father entity:
 327   a_parent->eventResponse(response);
 328 }
 329
 330 void Server::eventRequest(ClientSession *clientSession, const anna::DataBlock & request) throw(anna::RuntimeException) {
 331   // Inform father entity:
 332   a_parent->eventRequest(clientSession, request);
 333 }
 334
 335 void Server::eventUnknownResponse(ClientSession *clientSession, const anna::DataBlock & response) throw(anna::RuntimeException) {
 336   // Inform father entity:
 337   a_parent->eventUnknownResponse(clientSession, response);
 338 }
 339
 340 void Server::eventDPA(ClientSession *clientSession, const anna::DataBlock & response) throw(anna::RuntimeException) {
 341   // Inform father entity:
 342   a_parent->eventDPA(clientSession, response);
 343 }
 344
 345
 346 void Server::availabilityLost() throw() {
 347   a_available = false;
 348   std::string socket = anna::functions::socketLiteralAsString(a_socket.first, a_socket.second);
 349   LOGDEBUG(
 350     std::string msg = "diameter::comm::Server { Socket: ";
 351     msg += socket;
 352     msg += " } has lost its availability";
 353     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 354   );
 355   // OAM
 356   OamModule &oamModule = OamModule::instantiate();
 357   oamModule.activateAlarm(OamModule::Alarm::c_LostAvailabilityOverServerDefinedAs__s__, socket.c_str());
 358   oamModule.count(OamModule::Counter::LostAvailabilityOverServer);
 359   a_engine->availabilityLost(this);
 360   a_parent->refreshAvailability();
 361 }
 362
 363
 364 void Server::availabilityRecovered() throw() {
 365   a_available = true;
 366   std::string socket = anna::functions::socketLiteralAsString(a_socket.first, a_socket.second);
 367   LOGDEBUG(
 368     std::string msg = "diameter::comm::Server { Socket: ";
 369     msg += socket;
 370     msg += " } has recovered its availability";
 371     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 372   );
 373   // OAM
 374   OamModule &oamModule = OamModule::instantiate();
 375   oamModule.cancelAlarm(OamModule::Alarm::c_LostAvailabilityOverServerDefinedAs__s__, socket.c_str());
 376   oamModule.count(OamModule::Counter::RecoveredAvailabilityOverServer);
 377   a_engine->availabilityRecovered(this);
 378   a_parent->refreshAvailability();
 379 }
 380
 381
 382
 383 bool Server::refreshAvailability() throw() {
 384   // Here available
 385   if(a_available) {  // check not-bound state for all client-sessions:
 386     bool isolate = true;
 387
 388     for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 389
 390       //if ((*it)->getState() == ClientSession::State::Bound) { isolate = false; break; }
 391       if((*it)->getState() != ClientSession::State::Closed) { isolate = false; break; }
 392
 393     if(isolate) {
 394       availabilityLost();
 395       return true;
 396     }
 397
 398     return false;
 399   }
 400
 401   // Here not available
 402   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 403     if((*it)->getState() == ClientSession::State::Bound) {
 404       availabilityRecovered();
 405       return true;
 406     }
 407
 408   return false;
 409 }
 410
 411
 412 //------------------------------------------------------------------------------
 413 //---------------------------------------- Server::updateIncomingActivityTime()
 414 //------------------------------------------------------------------------------
 415 void Server::updateIncomingActivityTime() throw() {
 416   a_lastIncomingActivityTime = anna::functions::millisecond();
 417   LOGDEBUG
 418   (
 419     std::string msg = "Updated INCOMING activity on server (milliseconds unix): ";
 420     msg += anna::functions::asString(a_lastIncomingActivityTime.getValue());
 421     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 422   );
 423   a_parent->updateIncomingActivityTime();
 424 }
 425
 426
 427 //------------------------------------------------------------------------------
 428 //---------------------------------------- Server::updateOutgoingActivityTime()
 429 //------------------------------------------------------------------------------
 430 void Server::updateOutgoingActivityTime(void) throw() {
 431   a_lastOutgoingActivityTime = anna::functions::millisecond();
 432   LOGDEBUG
 433   (
 434     std::string msg = "Updated OUTGOING activity on server (milliseconds unix): "; msg += anna::functions::asString(a_lastOutgoingActivityTime.getValue());
 435     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 436   );
 437   a_parent->updateOutgoingActivityTime();
 438 }