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       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   bool allok = true;
 150
 151   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++) {
 152     try {
 153       (*it)->send(message);
 154     } catch(anna::RuntimeException &ex) {
 155       ex.trace();
 156       allok = false;
 157     }
 158   }
 159
 160   return allok;
 161 }
 162
 163
 164 bool Server::bind() throw(anna::RuntimeException) {
 165   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "bind", ANNA_FILE_LOCATION));
 166   assertReady();
 167   a_deliveryIterator = begin();
 168   bool result = true; // all OK return
 169
 170   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++) {
 171     try {
 172       (*it)->bind();
 173     } catch(anna::RuntimeException &ex) {
 174       ex.trace();
 175       result = false;
 176     }
 177   }
 178
 179   return result;
 180 }
 181
 182 void Server::raiseAutoRecovery(bool autoRecovery) throw(anna::RuntimeException) {
 183   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "raiseAutoRecovery", ANNA_FILE_LOCATION));
 184   assertReady();
 185   a_deliveryIterator = begin();
 186
 187   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 188     (*it)->setAutoRecovery(autoRecovery);
 189 }
 190
 191 void Server::setClassCodeTimeout(const ClassCode::_v v, const anna::Millisecond & millisecond) throw() {
 192   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "setClassCodeTimeout", ANNA_FILE_LOCATION));
 193   assertReady();
 194
 195   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++) {
 196     try {
 197       (*it)->setClassCodeTimeout(v, millisecond);
 198     } catch(anna::RuntimeException &ex) {
 199       ex.trace();
 200     }
 201   }
 202 }
 203
 204
 205 // Private close/destroy method
 206 void Server::close(bool destroy) throw(anna::RuntimeException) {
 207   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "close", ANNA_FILE_LOCATION));
 208
 209   if(!a_engine)
 210     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
 211
 212   assertReady();
 213
 214   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 215     a_engine->closeClientSession(*it, destroy);
 216 }
 217
 218
 219 void Server::childIdle() const throw() {
 220   // Check father entity idleness:
 221   if(idle()) a_parent->childIdle();
 222 }
 223
 224
 225 void Server::hide() throw() {
 226   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 227     (*it)->hide();
 228 }
 229
 230 void Server::show() throw() {
 231   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 232     (*it)->show();
 233 }
 234
 235 bool Server::hidden() const throw() {
 236   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 237     if((*it)->shown()) return false;
 238
 239   return true;
 240 }
 241 bool Server::shown() const throw() {
 242   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 243     if((*it)->hidden()) return false;
 244
 245   return true;
 246 }
 247
 248
 249
 250 std::string Server::socketAsString() const throw() {
 251   std::string result = getAddress();
 252   result += ":";
 253   result += anna::functions::asString(getPort());
 254   return result;
 255 }
 256
 257
 258 std::string Server::asString() const throw() {
 259   std::string result("diameter::comm::Server { ");
 260   result += " | Parent Entity: ";
 261   result += a_parent->getDescription();
 262   result += " | Server Address: ";
 263   result += a_socket.first;
 264   result += " | Server Port: ";
 265   result += anna::functions::asString(a_socket.second);
 266   result += " | Available: ";
 267   result += a_available ? "yes" : "no";
 268   result += " | Max client-sessions supported: ";
 269   result += anna::functions::asString(a_maxClientSessions);
 270   result += " | Currently configured client-sessions: ";
 271   result += anna::functions::asString(a_clientSessions.size());
 272   result += anna::functions::asString(" | OTA requests: %d%s", getOTARequests(), idle() ? " (idle)" : "");
 273   result += " | Last Incoming Activity Time: ";
 274   result += a_lastIncomingActivityTime.asString();
 275   result += " | Last Outgoing Activity Time: ";
 276   result += a_lastOutgoingActivityTime.asString();
 277   result += " | Hidden: ";
 278   result += (hidden() ? "yes" : "no");
 279   result += "\n";
 280   result += a_messageStatistics.getAccumulator()->asString();
 281
 282   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++) {
 283     result += "\n";
 284     result += (*it)->asString();
 285   }
 286
 287   return result;
 288 }
 289
 290 anna::xml::Node* Server::asXML(anna::xml::Node* parent) const throw() {
 291   anna::xml::Node* result = parent->createChild("diameter.Server");
 292   result->createAttribute("ParentEntity", a_parent->getDescription());
 293   result->createAttribute("ServerAddress", a_socket.first);
 294   result->createAttribute("ServerPort", anna::functions::asString(a_socket.second));
 295   result->createAttribute("Available", a_available ? "yes" : "no");
 296   result->createAttribute("MaxClientSessionsSupported", anna::functions::asString(a_maxClientSessions));
 297   result->createAttribute("CurrentlyConfiguredClientSessions", anna::functions::asString(a_clientSessions.size()));
 298   result->createAttribute("OTArequests", anna::functions::asString("%d%s", getOTARequests(), idle() ? " (idle)" : ""));
 299   result->createAttribute("LastIncomingActivityTime", a_lastIncomingActivityTime.asString());
 300   result->createAttribute("LastOutgoingActivityTime", a_lastOutgoingActivityTime.asString());
 301   result->createAttribute("Hidden", hidden() ? "yes" : "no");
 302   // Statistics
 303   anna::xml::Node* stats = result->createChild("Statistics");
 304   a_messageStatistics.getAccumulator()->asXML(stats);
 305   anna::xml::Node* clientSessions = result->createChild("Server.ClientSessions");
 306
 307   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 308     (*it)->asXML(clientSessions);
 309
 310   return result;
 311 }
 312
 313
 314 void Server::eventPeerShutdown(const ClientSession *clientSession) throw() {
 315   // Inform father entity:
 316   a_parent->eventPeerShutdown(clientSession);
 317 }
 318
 319 void Server::eventRequestRetransmission(const ClientSession* clientSession, Message *request) throw() {
 320   // Inform father entity:
 321   a_parent->eventRequestRetransmission(clientSession, request);
 322 }
 323
 324 void Server::eventResponse(const Response& response) throw(anna::RuntimeException) {
 325   // Inform father entity:
 326   a_parent->eventResponse(response);
 327 }
 328
 329 void Server::eventRequest(ClientSession *clientSession, const anna::DataBlock & request) throw(anna::RuntimeException) {
 330   // Inform father entity:
 331   a_parent->eventRequest(clientSession, request);
 332 }
 333
 334 void Server::eventUnknownResponse(ClientSession *clientSession, const anna::DataBlock & response) throw(anna::RuntimeException) {
 335   // Inform father entity:
 336   a_parent->eventUnknownResponse(clientSession, response);
 337 }
 338
 339 void Server::eventDPA(ClientSession *clientSession, const anna::DataBlock & response) throw(anna::RuntimeException) {
 340   // Inform father entity:
 341   a_parent->eventDPA(clientSession, response);
 342 }
 343
 344
 345 void Server::availabilityLost() throw() {
 346   a_available = false;
 347   std::string socket = anna::functions::socketLiteralAsString(a_socket.first, a_socket.second);
 348   LOGDEBUG(
 349     std::string msg = "diameter::comm::Server { Socket: ";
 350     msg += socket;
 351     msg += " } has lost its availability";
 352     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 353   );
 354   // OAM
 355   OamModule &oamModule = OamModule::instantiate();
 356   oamModule.activateAlarm(OamModule::Alarm::c_LostAvailabilityOverServerDefinedAs__s__, socket.c_str());
 357   oamModule.count(OamModule::Counter::LostAvailabilityOverServer);
 358   a_engine->availabilityLost(this);
 359   a_parent->refreshAvailability();
 360 }
 361
 362
 363 void Server::availabilityRecovered() throw() {
 364   a_available = true;
 365   std::string socket = anna::functions::socketLiteralAsString(a_socket.first, a_socket.second);
 366   LOGDEBUG(
 367     std::string msg = "diameter::comm::Server { Socket: ";
 368     msg += socket;
 369     msg += " } has recovered its availability";
 370     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 371   );
 372   // OAM
 373   OamModule &oamModule = OamModule::instantiate();
 374   oamModule.cancelAlarm(OamModule::Alarm::c_LostAvailabilityOverServerDefinedAs__s__, socket.c_str());
 375   oamModule.count(OamModule::Counter::RecoveredAvailabilityOverServer);
 376   a_engine->availabilityRecovered(this);
 377   a_parent->refreshAvailability();
 378 }
 379
 380
 381
 382 bool Server::refreshAvailability() throw() {
 383   // Here available
 384   if(a_available) {  // check not-bound state for all client-sessions:
 385     bool isolate = true;
 386
 387     for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 388
 389       //if ((*it)->getState() == ClientSession::State::Bound) { isolate = false; break; }
 390       if((*it)->getState() != ClientSession::State::Closed) { isolate = false; break; }
 391
 392     if(isolate) {
 393       availabilityLost();
 394       return true;
 395     }
 396
 397     return false;
 398   }
 399
 400   // Here not available
 401   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 402     if((*it)->getState() == ClientSession::State::Bound) {
 403       availabilityRecovered();
 404       return true;
 405     }
 406
 407   return false;
 408 }
 409
 410
 411 //------------------------------------------------------------------------------
 412 //---------------------------------------- Server::updateIncomingActivityTime()
 413 //------------------------------------------------------------------------------
 414 void Server::updateIncomingActivityTime() throw() {
 415   a_lastIncomingActivityTime = anna::functions::millisecond();
 416   LOGDEBUG
 417   (
 418     std::string msg = "Updated INCOMING activity on server (milliseconds unix): ";
 419     msg += anna::functions::asString(a_lastIncomingActivityTime.getValue());
 420     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 421   );
 422   a_parent->updateIncomingActivityTime();
 423 }
 424
 425
 426 //------------------------------------------------------------------------------
 427 //---------------------------------------- Server::updateOutgoingActivityTime()
 428 //------------------------------------------------------------------------------
 429 void Server::updateOutgoingActivityTime(void) throw() {
 430   a_lastOutgoingActivityTime = anna::functions::millisecond();
 431   LOGDEBUG
 432   (
 433     std::string msg = "Updated OUTGOING activity on server (milliseconds unix): "; msg += anna::functions::asString(a_lastOutgoingActivityTime.getValue());
 434     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 435   );
 436   a_parent->updateOutgoingActivityTime();
 437 }