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