source/diameter.comm/Server.cpp

   1 // ANNA - Anna is Not Nothingness Anymore
   2 //
   3 // (c) Copyright 2005-2014 Eduardo Ramos Testillano & Francisco Ruiz Rayo
   4 //
   5 // http://redmine.teslayout.com/projects/anna-suite
   6 //
   7 // Redistribution and use in source and binary forms, with or without
   8 // modification, are permitted provided that the following conditions
   9 // are met:
  10 //
  11 //     * Redistributions of source code must retain the above copyright
  12 // notice, this list of conditions and the following disclaimer.
  13 //     * Redistributions in binary form must reproduce the above
  14 // copyright notice, this list of conditions and the following disclaimer
  15 // in the documentation and/or other materials provided with the
  16 // distribution.
  17 //     *  Neither the name of the copyright holder nor the names of its
  18 // contributors may be used to endorse or promote products derived from
  19 // this software without specific prior written permission.
  20 //
  21 // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  22 // "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  23 // LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  24 // A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  25 // OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  26 // SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  27 // LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  28 // DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  29 // THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  30 // (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  31 // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  32 //
  33 // Authors: eduardo.ramos.testillano@gmail.com
  34 //          cisco.tierra@gmail.com
  35
  36
  37 #include <anna/diameter.comm/ClientSession.hpp>
  38 #include <anna/diameter.comm/Entity.hpp>
  39 #include <anna/diameter.comm/Engine.hpp>
  40 #include <anna/diameter.comm/OamModule.hpp>
  41 #include <anna/diameter/codec/functions.hpp>
  42 #include <anna/diameter/helpers/base/defines.hpp>
  43 #include <anna/core/functions.hpp>
  44 #include <anna/statistics/Engine.hpp>
  45 #include <anna/time/functions.hpp>
  46
  47 #include <anna/core/tracing/Logger.hpp>
  48 #include <anna/core/functions.hpp>
  49 #include <anna/core/tracing/TraceMethod.hpp>
  50
  51 // STL
  52 #include <vector>
  53
  54 using namespace anna;
  55 using namespace anna::diameter;
  56 using namespace anna::diameter::comm;
  57
  58
  59 void Server::initialize() throw() {
  60   a_parent = NULL;
  61   a_engine = NULL;
  62   a_clientSessions.clear(); // importante (el recycler creo que no lo tocaba)
  63   a_available = false;
  64   a_maxClientSessions = 1; // mono client connection
  65   a_lastIncomingActivityTime = (anna::Millisecond)0;
  66   a_lastOutgoingActivityTime = (anna::Millisecond)0;
  67   a_statisticsAccumulator.reset();
  68   a_lastUsedResource = NULL;
  69 }
  70
  71 void Server::initializeStatisticConcepts() throw() {
  72   // Statistics:
  73   anna::statistics::Engine& statsEngine = anna::statistics::Engine::instantiate();
  74   // Concepts descriptions:
  75   std::string serverAsString = anna::functions::socketLiteralAsString(a_socket.first, a_socket.second);
  76   std::string c1desc = "Diameter processing time (for requests) at servers on "; c1desc += serverAsString;
  77   std::string c2desc = "Diameter message sizes received from servers on "; c2desc += serverAsString;
  78   // Registering
  79   a_processing_time__StatisticConceptId = statsEngine.addConcept(c1desc.c_str(), "ms", true/* integer values */);
  80   a_received_message_size__StatisticConceptId = statsEngine.addConcept(c2desc.c_str(), "bytes", true/* integer values */);
  81 }
  82
  83 void Server::resetStatistics() throw() {
  84   a_statisticsAccumulator.reset();
  85 }
  86
  87 void Server::updateProcessingTimeStatisticConcept(const double &value) throw() {
  88   a_statisticsAccumulator.process(a_processing_time__StatisticConceptId, value);
  89   LOGDEBUG(anna::Logger::debug(a_statisticsAccumulator.asString(), ANNA_FILE_LOCATION));
  90 }
  91
  92 void Server::updateReceivedMessageSizeStatisticConcept(const double &value) throw() {
  93   a_statisticsAccumulator.process(a_received_message_size__StatisticConceptId, value);
  94   //LOGDEBUG(anna::Logger::debug(a_statisticsAccumulator.asString(), ANNA_FILE_LOCATION));
  95 }
  96
  97
  98 void Server::assertReady() throw(anna::RuntimeException) {
  99   if(a_clientSessions.size() != a_maxClientSessions) {
 100     std::string msg(asString());
 101     msg += " | Non-configured server: you must add the remaining client-sessions before any operation (bind, send, etc.)";
 102     throw anna::RuntimeException(msg, ANNA_FILE_LOCATION);
 103   }
 104 }
 105
 106
 107 void Server::addClientSession(int socketId)
 108 throw(anna::RuntimeException) {
 109   if(a_clientSessions.size() == a_maxClientSessions) {
 110     LOGDEBUG
 111     (
 112       std::string msg = "Maximum number of client-sessions reached for this server (";
 113       msg += anna::functions::asString(a_maxClientSessions);
 114       msg += ").";
 115       anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 116     );
 117     return;
 118   }
 119
 120   if(!a_engine)
 121     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
 122
 123   ClientSession *s = a_engine->createClientSession(this, socketId);
 124   a_clientSessions.push_back(s);
 125 }
 126
 127 int Server::getOTARequests() const throw() {
 128   int result = 0;
 129
 130   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 131     result += (*it)->getOTARequests();
 132
 133   return result;
 134 }
 135
 136
 137 bool Server::send(const Message* message, int socketId) throw(anna::RuntimeException) {
 138   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "send", ANNA_FILE_LOCATION));
 139   assertReady();
 140   bool fixedSocket = (socketId != -1);
 141   int clientSessions = getNumberOfClientSessions();
 142
 143   for(int k = 0; k < clientSessions; k++) {    // try round-robin only over one cycle,
 144     // no matter where you are: don't repeat same socket
 145     if(fixedSocket)
 146       a_lastUsedResource = a_engine->findClientSession(a_socket.first /*ip*/, a_socket.second /*port*/, socketId); // exception if not found
 147     else {
 148       if(a_deliveryIterator == end()) a_deliveryIterator = begin();
 149
 150       a_lastUsedResource = (*a_deliveryIterator);
 151
 152       if(clientSessions > 1) a_deliveryIterator++;  // Round robin
 153     }
 154
 155     try {
 156       // Send:
 157       const Response* response = a_lastUsedResource->send(message);
 158       return true; // no matter if response is NULL (answers, i.e.) or not.
 159     } catch(anna::RuntimeException &ex) {
 160       LOGDEBUG(
 161         std::string msg = ex.getText();
 162         msg += " | Send failed on socket id ";
 163         msg += anna::functions::asString(a_lastUsedResource->getSocketId());
 164         anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 165       );
 166     }
 167
 168     // Specific socket sending doesn't try
 169     if(fixedSocket) return false;
 170   }
 171
 172   return false;
 173 }
 174
 175
 176 bool Server::broadcast(const Message* message) throw(anna::RuntimeException) {
 177   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "broadcast", ANNA_FILE_LOCATION));
 178   assertReady();
 179   const Response* response;
 180   bool allok = true;
 181
 182   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++) {
 183     try {
 184       response = (*it)->send(message);
 185     } catch(anna::RuntimeException &ex) {
 186       ex.trace();
 187       allok = false;
 188     }
 189   }
 190
 191   return allok;
 192 }
 193
 194
 195 bool Server::bind() throw(anna::RuntimeException) {
 196   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "bind", ANNA_FILE_LOCATION));
 197   assertReady();
 198   a_deliveryIterator = begin();
 199   bool result = true; // all OK return
 200
 201   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++) {
 202     try {
 203       (*it)->bind();
 204     } catch(anna::RuntimeException &ex) {
 205       ex.trace();
 206       result = false;
 207     }
 208   }
 209
 210   return result;
 211 }
 212
 213 void Server::raiseAutoRecovery(bool autoRecovery) throw(anna::RuntimeException) {
 214   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "raiseAutoRecovery", ANNA_FILE_LOCATION));
 215   assertReady();
 216   a_deliveryIterator = begin();
 217
 218   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 219     (*it)->setAutoRecovery(autoRecovery);
 220 }
 221
 222 void Server::setClassCodeTimeout(const ClassCode::_v v, const anna::Millisecond & millisecond) throw() {
 223   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "setClassCodeTimeout", ANNA_FILE_LOCATION));
 224   assertReady();
 225
 226   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++) {
 227     try {
 228       (*it)->setClassCodeTimeout(v, millisecond);
 229     } catch(anna::RuntimeException &ex) {
 230       ex.trace();
 231     }
 232   }
 233 }
 234
 235
 236 // Private close/destroy method
 237 void Server::close(bool destroy) throw(anna::RuntimeException) {
 238   LOGMETHOD(anna::TraceMethod tttm("diameter::comm::Server", "close", ANNA_FILE_LOCATION));
 239
 240   if(!a_engine)
 241     throw anna::RuntimeException("Invalid engine reference (NULL)", ANNA_FILE_LOCATION);
 242
 243   assertReady();
 244
 245   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 246     a_engine->closeClientSession(*it, destroy);
 247 }
 248
 249
 250 void Server::childIdle() const throw() {
 251   // Check father entity idleness:
 252   if(idle()) a_parent->childIdle();
 253 }
 254
 255
 256 void Server::hide() throw() {
 257   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 258     (*it)->hide();
 259 }
 260
 261 void Server::show() throw() {
 262   for(std::vector<ClientSession*>::iterator it = begin(); it != end(); it++)
 263     (*it)->show();
 264 }
 265
 266 bool Server::hidden() const throw() {
 267   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 268     if((*it)->shown()) return false;
 269
 270   return true;
 271 }
 272 bool Server::shown() const throw() {
 273   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 274     if((*it)->hidden()) return false;
 275
 276   return true;
 277 }
 278
 279
 280
 281 std::string Server::socketAsString() const throw() {
 282   std::string result = getAddress();
 283   result += ":";
 284   result += anna::functions::asString(getPort());
 285   return result;
 286 }
 287
 288
 289 std::string Server::asString() const throw() {
 290   std::string result("diameter::comm::Server { ");
 291   result += " | Parent Entity: ";
 292   result += a_parent->getDescription();
 293   result += " | Server Address: ";
 294   result += a_socket.first;
 295   result += " | Server Port: ";
 296   result += anna::functions::asString(a_socket.second);
 297   result += " | Available: ";
 298   result += a_available ? "yes" : "no";
 299   result += " | Max client-sessions supported: ";
 300   result += anna::functions::asString(a_maxClientSessions);
 301   result += " | Currently configured client-sessions: ";
 302   result += anna::functions::asString(a_clientSessions.size());
 303   result += anna::functions::asString(" | OTA requests: %d%s", getOTARequests(), idle() ? " (idle)" : "");
 304   result += " | Last Incoming Activity Time: ";
 305   result += a_lastIncomingActivityTime.asString();
 306   result += " | Last Outgoing Activity Time: ";
 307   result += a_lastOutgoingActivityTime.asString();
 308   result += " | Hidden: ";
 309   result += (hidden() ? "yes" : "no");
 310   result += "\n";
 311   result += a_statisticsAccumulator.asString();
 312
 313   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++) {
 314     result += "\n";
 315     result += (*it)->asString();
 316   }
 317
 318   return result;
 319 }
 320
 321 anna::xml::Node* Server::asXML(anna::xml::Node* parent) const throw() {
 322   anna::xml::Node* result = parent->createChild("diameter.Server");
 323   result->createAttribute("ParentEntity", a_parent->getDescription());
 324   result->createAttribute("ServerAddress", a_socket.first);
 325   result->createAttribute("ServerPort", anna::functions::asString(a_socket.second));
 326   result->createAttribute("Available", a_available ? "yes" : "no");
 327   result->createAttribute("MaxClientSessionsSupported", anna::functions::asString(a_maxClientSessions));
 328   result->createAttribute("CurrentlyConfiguredClientSessions", anna::functions::asString(a_clientSessions.size()));
 329   result->createAttribute("OTArequests", anna::functions::asString("%d%s", getOTARequests(), idle() ? " (idle)" : ""));
 330   result->createAttribute("LastIncomingActivityTime", a_lastIncomingActivityTime.asString());
 331   result->createAttribute("LastOutgoingActivityTime", a_lastOutgoingActivityTime.asString());
 332   result->createAttribute("Hidden", hidden() ? "yes" : "no");
 333   // Statistics
 334   anna::xml::Node* stats = result->createChild("Statistics");
 335   a_statisticsAccumulator.asXML(stats);
 336   anna::xml::Node* clientSessions = result->createChild("Server.ClientSessions");
 337
 338   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 339     (*it)->asXML(clientSessions);
 340
 341   return result;
 342 }
 343
 344
 345 void Server::eventPeerShutdown(const ClientSession *clientSession) throw() {
 346   // Inform father server:
 347   a_parent->eventPeerShutdown(clientSession);
 348 }
 349
 350 void Server::eventResponse(const Response& response) throw(anna::RuntimeException) {
 351   // Inform father server:
 352   a_parent->eventResponse(response);
 353 }
 354
 355 void Server::eventRequest(ClientSession *clientSession, const anna::DataBlock & request) throw(anna::RuntimeException) {
 356   // Inform father server:
 357   a_parent->eventRequest(clientSession, request);
 358 }
 359
 360 void Server::eventUnknownResponse(ClientSession *clientSession, const anna::DataBlock & response) throw(anna::RuntimeException) {
 361   // Inform father server:
 362   a_parent->eventUnknownResponse(clientSession, response);
 363 }
 364
 365 void Server::availabilityLost() throw() {
 366   a_available = false;
 367   std::string socket = anna::functions::socketLiteralAsString(a_socket.first, a_socket.second);
 368   LOGDEBUG(
 369     std::string msg = "diameter::comm::Server { Socket: ";
 370     msg += socket;
 371     msg += " } has lost its availability";
 372     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 373   );
 374   // OAM
 375   OamModule &oamModule = OamModule::instantiate();
 376   oamModule.activateAlarm(OamModule::Alarm::c_LostAvailabilityOverServerDefinedAs__s__, socket.c_str());
 377   oamModule.count(OamModule::Counter::LostAvailabilityOverServer);
 378   a_engine->availabilityLost(this);
 379   a_parent->refreshAvailability();
 380 }
 381
 382
 383 void Server::availabilityRecovered() throw() {
 384   a_available = true;
 385   std::string socket = anna::functions::socketLiteralAsString(a_socket.first, a_socket.second);
 386   LOGDEBUG(
 387     std::string msg = "diameter::comm::Server { Socket: ";
 388     msg += socket;
 389     msg += " } has recovered its availability";
 390     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 391   );
 392   // OAM
 393   OamModule &oamModule = OamModule::instantiate();
 394   oamModule.cancelAlarm(OamModule::Alarm::c_LostAvailabilityOverServerDefinedAs__s__, socket.c_str());
 395   oamModule.count(OamModule::Counter::RecoveredAvailabilityOverServer);
 396   a_engine->availabilityRecovered(this);
 397   a_parent->refreshAvailability();
 398 }
 399
 400
 401
 402 bool Server::refreshAvailability() throw() {
 403   // Here available
 404   if(a_available) {  // check not-bound state for all client-sessions:
 405     bool isolate = true;
 406
 407     for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 408
 409       //if ((*it)->getState() == ClientSession::State::Bound) { isolate = false; break; }
 410       if((*it)->getState() != ClientSession::State::Closed) { isolate = false; break; }
 411
 412     if(isolate) {
 413       availabilityLost();
 414       return true;
 415     }
 416
 417     return false;
 418   }
 419
 420   // Here not available
 421   for(std::vector<ClientSession*>::const_iterator it = begin(); it != end(); it++)
 422     if((*it)->getState() == ClientSession::State::Bound) {
 423       availabilityRecovered();
 424       return true;
 425     }
 426
 427   return false;
 428 }
 429
 430
 431 //------------------------------------------------------------------------------
 432 //---------------------------------------- Server::updateIncomingActivityTime()
 433 //------------------------------------------------------------------------------
 434 void Server::updateIncomingActivityTime() throw() {
 435   a_lastIncomingActivityTime = anna::functions::millisecond();
 436   LOGDEBUG
 437   (
 438     std::string msg = "Updated INCOMING activity on server (milliseconds unix): ";
 439     msg += anna::functions::asString(a_lastIncomingActivityTime.getValue());
 440     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 441   );
 442   a_parent->updateIncomingActivityTime();
 443 }
 444
 445
 446 //------------------------------------------------------------------------------
 447 //---------------------------------------- Server::updateOutgoingActivityTime()
 448 //------------------------------------------------------------------------------
 449 void Server::updateOutgoingActivityTime(void) throw() {
 450   a_lastOutgoingActivityTime = anna::functions::millisecond();
 451   LOGDEBUG
 452   (
 453     std::string msg = "Updated OUTGOING activity on server (milliseconds unix): "; msg += anna::functions::asString(a_lastOutgoingActivityTime.getValue());
 454     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 455   );
 456   a_parent->updateOutgoingActivityTime();
 457 }
 458