source/diameter.comm/Session.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/core/functions.hpp>
  10 #include <anna/diameter/defines.hpp>
  11 #include <anna/diameter/functions.hpp>
  12 #include <anna/diameter/helpers/helpers.hpp>
  13 #include <anna/diameter/codec/functions.hpp>
  14 #include <anna/diameter/helpers/base/functions.hpp>
  15 #include <anna/time/functions.hpp>
  16
  17 // Local
  18 #include <anna/diameter.comm/Session.hpp>
  19 #include <anna/diameter.comm/Engine.hpp>
  20 #include <anna/diameter.comm/Entity.hpp>
  21 #include <anna/diameter.comm/Server.hpp>
  22 #include <anna/diameter.comm/Transport.hpp>
  23 #include <anna/diameter.comm/Response.hpp>
  24 #include <anna/diameter.comm/Message.hpp>
  25 #include <anna/diameter.comm/TimerManager.hpp>
  26 #include <anna/diameter.comm/Timer.hpp>
  27 #include <anna/diameter.comm/OriginHostManager.hpp>
  28 #include <anna/diameter.comm/OriginHost.hpp>
  29
  30 #include <anna/comm/Network.hpp>
  31 #include <anna/comm/ClientSocket.hpp>
  32 #include <anna/core/functions.hpp>
  33 #include <anna/core/DataBlock.hpp>
  34 #include <anna/core/tracing/Logger.hpp>
  35 #include <anna/core/tracing/TraceMethod.hpp>
  36 #include <anna/xml/Node.hpp>
  37 #include <anna/timex/Engine.hpp>
  38 #include <anna/app/functions.hpp>
  39
  40 // Standard
  41 #include <stdlib.h> // rand()
  42 #include <time.h>
  43
  44
  45
  46 using namespace std;
  47 using namespace anna::diameter;
  48 using namespace anna::diameter::comm;
  49
  50 //static
  51 const anna::Millisecond Session::DefaultTimeout(10000); // Application messages timeout
  52 const int Session::DefaultPort(3868);
  53
  54
  55 Session::Session(const char *className, const char *timerName) : anna::timex::Timer(timerName, (anna::Millisecond)0) /* not assigned */,
  56   a_className(className),
  57   a_timeController(NULL),
  58   a_notifyOrphansOnExpiration(true),
  59   a_actionTimer(NULL) {
  60   initialize();
  61 }
  62
  63 void Session::initialize() {
  64   a_state = State::Closed;
  65   a_socketId = 0;
  66   a_lastIncomingActivityTime = (anna::Millisecond)0;
  67   a_lastOutgoingActivityTime = (anna::Millisecond)0;
  68   a_onDisconnect = OnDisconnect::WaitPendings;
  69
  70   for(int i = ClassCode::Min; i < ClassCode::Max; i ++)
  71     a_timeouts [i] = DefaultTimeout;
  72 }
  73
  74 //Session::~Session() {;}
  75
  76
  77 void Session::initializeSequences() {
  78   // Sequences
  79   //
  80   //   Hop-by-Hop Identifier
  81   //      The Hop-by-Hop Identifier is an unsigned 32-bit integer field (in
  82   //      network byte order) and aids in matching requests and replies.
  83   //      The sender MUST ensure that the Hop-by-Hop identifier in a request
  84   //      is unique on a given connection at any given time, and MAY attempt
  85   //      to ensure that the number is unique across reboots.  The sender of
  86   //      an Answer message MUST ensure that the Hop-by-Hop Identifier field
  87   //      contains the same value that was found in the corresponding
  88   //      request.  The Hop-by-Hop identifier is normally a monotonically
  89   //      increasing number, whose start value was randomly generated.  An
  90   //      answer message that is received with an unknown Hop-by-Hop
  91   //      Identifier MUST be discarded.
  92   //
  93   //   End-to-End Identifier
  94   //      The End-to-End Identifier is an unsigned 32-bit integer field (in
  95   //      network byte order) and is used to detect duplicate messages.
  96   //      Upon reboot implementations MAY set the high order 12 bits to
  97   //      contain the low order 12 bits of current time, and the low order
  98   //      20 bits to a random value.  Senders of request messages MUST
  99   //      insert a unique identifier on each message.  The identifier MUST
 100   //      remain locally unique for a period of at least 4 minutes, even
 101   //      across reboots.  The originator of an Answer message MUST ensure
 102   //      that the End-to-End Identifier field contains the same value that
 103   //      was found in the corresponding request.  The End-to-End Identifier
 104   //      MUST NOT be modified by Diameter agents of any kind.  The
 105   //      combination of the Origin-Host (see Section 6.3) and this field is
 106   //      used to detect duplicates.  Duplicate requests SHOULD cause the
 107   //      same answer to be transmitted (modulo the hop-by-hop Identifier
 108   //      field and any routing AVPs that may be present), and MUST NOT
 109   //      affect any state that was set when the original request was
 110   //      processed.  Duplicate answer messages that are to be locally
 111   //      consumed (see Section 6.2) SHOULD be silently discarded.
 112   srand(::time(NULL) + anna::functions::hash(anna::functions::asString("%s:%d|%d", getAddress().c_str(), getPort(), a_socketId).c_str()));
 113   a_nextHopByHop = rand();
 114   a_nextEndToEnd = ((::time(NULL) & 0xFFF) << 20) + (rand() & 0xFFFFF);
 115 }
 116
 117 void Session::sendDPA(const Engine *commEngine, const anna::DataBlock &dprDataBlock)
 118 noexcept(false) {
 119   LOGMETHOD(anna::TraceMethod traceMethod("anna::diameter::comm::Session", "sendDPA", ANNA_FILE_LOCATION));
 120   anna::DataBlock dpa(true);
 121   commEngine->readDPA(dpa, dprDataBlock); // Asume that DPA is valid ...
 122
 123   if(dpa.isEmpty()) {
 124     LOGWARNING(anna::Logger::warning("This diameter agent defines an empty DPA message. Remote disconnection DPR will be ignored going to the Bound state", ANNA_FILE_LOCATION));
 125     setState(State::Bound);
 126     return;
 127   }
 128
 129   Message msgDpa;
 130   msgDpa.setBody(dpa);
 131   send(&msgDpa);
 132   LOGWARNING(anna::Logger::warning("DPA has been sent to the peer", ANNA_FILE_LOCATION));
 133   // Temporizador de proteccion por si el servidor no cierra:
 134   //    state            event              action           next state
 135   //    ---------------------------------------------------------------
 136   //    R-Open           R-Rcv-DPR          R-Snd-DPA        Closing
 137   //    Closing          Timeout            Error            Closed
 138   activateActionTimer(anna::diameter::comm::Timer::Type::SessionUnbind);
 139 }
 140
 141 void Session::setState(State::_v state) {
 142   LOGDEBUG(
 143
 144   if(state != a_state) {
 145   std::string msg("Session state change: ");
 146     msg += asText(a_state);
 147     msg += " -> ";
 148     msg += asText(state);
 149     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 150   }
 151   );
 152   a_state = state;
 153 }
 154
 155
 156 void Session::activateActionTimer(const anna::diameter::comm::Timer::Type::_v type) {
 157   LOGMETHOD(anna::TraceMethod traceMethod("anna::diameter::comm::Session", "activateActionTimer", ANNA_FILE_LOCATION));
 158   cancelTimer(); // Session timer
 159
 160   try {
 161     if(a_actionTimer && a_actionTimer->isActive()) cancelActionTimer();  // no ocurrira
 162
 163     a_actionTimer = TimerManager::instantiate().createTimer(this, type);
 164   } catch(anna::RuntimeException& ex) {
 165     std::string msg = "CAPTURED EXCEPTION during action timer activation (activateActionTimer): ";
 166     msg += ex.getText();
 167     anna::Logger::error(msg, ANNA_FILE_LOCATION);
 168   }
 169 }
 170
 171
 172 void Session::cancelActionTimer() {
 173   LOGMETHOD(anna::TraceMethod traceMethod("anna::diameter::comm::Session", "cancelActionTimer", ANNA_FILE_LOCATION));
 174
 175   if(a_actionTimer) {
 176     if(a_actionTimer->isActive()) {
 177       try {
 178         TimerManager::instantiate().cancelTimer(a_actionTimer);
 179       } catch(anna::RuntimeException& ex) {
 180         ex.trace();
 181       }
 182     } else { // por aqui no deberia pasar ...
 183       LOGDEBUG(anna::Logger::debug("Timer not activated!", ANNA_FILE_LOCATION));
 184     }
 185
 186     a_actionTimer = NULL;
 187   }
 188 }
 189
 190
 191 void Session::activateTimer() {
 192   LOGMETHOD(anna::TraceMethod traceMethod("anna::diameter::comm::Session", "activateTimer", ANNA_FILE_LOCATION));
 193   cancelActionTimer();
 194
 195   try {
 196     if(a_timeController == NULL)  // Application must created a timex engine
 197       a_timeController = anna::app::functions::component <anna::timex::Engine> (ANNA_FILE_LOCATION);
 198
 199     if(isActive()) cancelTimer();
 200
 201     a_timeController->activate(this);
 202   } catch(anna::RuntimeException& ex) {
 203     std::string msg = "CAPTURED EXCEPTION during session timer activation (activateTimer): ";
 204     msg += ex.getText();
 205     anna::Logger::error(msg, ANNA_FILE_LOCATION);
 206   }
 207
 208   timerStarted();
 209 }
 210
 211
 212 void Session::cancelTimer() {
 213   LOGMETHOD(anna::TraceMethod traceMethod("anna::diameter::comm::Session", "cancelTimer", ANNA_FILE_LOCATION));
 214
 215   if(isActive()) {
 216     try {
 217       if(a_timeController == NULL)  // Application must created a timex engine
 218         a_timeController = anna::app::functions::component <anna::timex::Engine> (ANNA_FILE_LOCATION);
 219
 220       a_timeController->cancel(this);
 221     } catch(anna::RuntimeException& ex) {
 222       std::string msg = "CAPTURED EXCEPTION during session timer cancellation (cancelTimer): ";
 223       msg += ex.getText();
 224       anna::Logger::error(msg, ANNA_FILE_LOCATION);
 225     }
 226
 227     timerStopped();
 228   } else {
 229     LOGDEBUG(anna::Logger::debug("Timer not activated!", ANNA_FILE_LOCATION));
 230   }
 231 }
 232
 233
 234 //-------------------------------------------------------------------------
 235 // Se invoca desde diameter::comm::Timer
 236 //-------------------------------------------------------------------------
 237 void Session::expireResponse(diameter::comm::Response* response)
 238 {
 239   LOGMETHOD(anna::TraceMethod traceMethod("anna::diameter::comm::Session", "expireResponse", ANNA_FILE_LOCATION));
 240   bool doUnbind = false;
 241   bool doRetransmission = false;
 242
 243   if(response->getClassCode() != ClassCode::Bind) {
 244     if(response->getRequest()->getOnExpiry() == Message::OnExpiry::Abandon) {
 245       a_onDisconnect = OnDisconnect::IgnorePendings; // Abandon is not graceful
 246       doUnbind = true;
 247     }
 248     else if(response->getRequest()->getOnExpiry() == Message::OnExpiry::Retransmit) {
 249       doRetransmission = true;
 250     }
 251   } else
 252     doUnbind = true; // (*)
 253
 254
 255   // Get origin host corresponding to the message:
 256   anna::diameter::comm::OriginHostManager &ohm = anna::diameter::comm::OriginHostManager::instantiate();
 257
 258   // Extract OriginHost from datablock (db):
 259   std::string originHostName = anna::diameter::helpers::base::functions::getOriginHost(response->getRequest()->getBody());
 260   LOGDEBUG(anna::Logger::debug(anna::functions::asString("ORIGIN HOST FOR THE MESSAGE WHICH WAS EXPIRED: %s", originHostName.c_str()), ANNA_FILE_LOCATION));
 261   anna::diameter::comm::OriginHost *originHost = ohm.getOriginHost(originHostName);
 262
 263   try {
 264     response->setMessage(NULL);
 265     eventResponse(*response, originHost);
 266   } catch(anna::RuntimeException& ex) {
 267     ex.trace();
 268   }
 269
 270   // DPR special case:
 271   diameter::CommandId cid = response->getRequest()->getCommandId();
 272
 273   if((cid == helpers::base::COMMANDID__Disconnect_Peer_Request) && (a_state == State::WaitingDPA)) {
 274     LOGDEBUG(anna::Logger::debug("Expired DPR sent to remote diameter point: local DPR procedure will be ignored going to the Bound state", ANNA_FILE_LOCATION));
 275     setState(State::Bound);
 276   }
 277
 278   if(doRetransmission) {
 279     diameter::comm::Message *request = const_cast<Message*>(response->getRequest());
 280     int retries = request->getRetries();
 281     if (retries > 0) {
 282       retries--;
 283       request->setRetries(retries);
 284       LOGDEBUG
 285       (
 286         std::string msg(asString());
 287         msg += anna::functions::asString(" | Retransmission initiated for request with HopByHop: %u; remaining %d retries", response->getHopByHop(), retries);
 288         anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 289       );
 290       diameter::codec::functions::setPotentiallyReTransmittedMessageBit(*request);
 291       eventRequestRetransmission(request);
 292       send(request);
 293     }
 294   }
 295
 296   response_erase(response);
 297
 298   if(doUnbind) unbind();
 299 }
 300
 301 void Session::finalize() {
 302   LOGMETHOD(anna::TraceMethod traceMethod("anna::diameter::comm::Session", "finalize", ANNA_FILE_LOCATION));
 303   setState(State::Closed);
 304   cancelTimer(); // Session timer
 305   cancelActionTimer(); // Action timer
 306   eventPeerShutdown();
 307 ///////////////////////////////////////////////////////////////////////
 308 // Notificar la finalizaci�n de las respuestas pendientes de recibir //
 309 ///////////////////////////////////////////////////////////////////////
 310 // RFC 3588 - 5.5.4.  Failover and Failback Procedures
 311 //
 312 //      In the event that a transport failure is detected with a peer, it is
 313 //      necessary for all pending request messages to be forwarded to an
 314 //      alternate agent, if possible.  This is commonly referred to as
 315 //      failover.
 316 //
 317 //      In order for a Diameter node to perform failover procedures, it is
 318 //      necessary for the node to maintain a pending message queue for a
 319 //      given peer.  When an answer message is received, the corresponding
 320 //      request is removed from the queue.  The Hop-by-Hop Identifier field
 321 //      is used to match the answer with the queued request.
 322 //
 323 //      When a transport failure is detected, if possible all messages in the
 324 //      queue are sent to an alternate agent with the T flag set.  On booting
 325 //      a Diameter client or agent, the T flag is also set on any records
 326 //      still remaining to be transmitted in non-volatile storage.  An
 327 //      example of a case where it is not possible to forward the message to
 328 //      an alternate server is when the message has a fixed destination, and
 329 //      the unavailable peer is the message's final destination (see
 330 //      Destination-Host AVP).  Such an error requires that the agent return
 331 //      an answer message with the 'E' bit set and the Result-Code AVP set to
 332 //      DIAMETER_UNABLE_TO_DELIVER.
 333 //
 334 //      It is important to note that multiple identical requests or answers
 335 //      MAY be received as a result of a failover.  The End-to-End Identifier
 336 //      field in the Diameter header along with the Origin-Host AVP MUST be
 337 //      used to identify duplicate messages.
 338   Response* response;
 339
 340   for(response_iterator ii = response_begin(), maxii = response_end(); ii != maxii; ii ++) {
 341     response = Session::response(ii);
 342     response->setResultCode(Response::ResultCode::DiameterUnavailable);
 343
 344     if(!a_notifyOrphansOnExpiration) {  // to avoid message bursts (to alternate servers for client-session context), we will manage at expireResponse
 345       response->cancelTimer();
 346
 347       try {
 348         response->setMessage(NULL);
 349         eventResponse(*response, nullptr); // upstream, we need to check second argument to know if comes from here.
 350                                            // If originHost is NULL, and we are client, we could access through engine: a_engine->getOriginHostName(), and then ohm.getOriginHost(name). BUT CLIENTS DO NOTHING WITH THIS ARGUMENT.
 351                                            // If server, we must return there: add protection for second argument.
 352       } catch(anna::RuntimeException& ex) {
 353         ex.trace();
 354       }
 355
 356       Response::release(response);
 357     }
 358   }
 359
 360   if(a_notifyOrphansOnExpiration) return;
 361
 362   a_responses.clear();
 363 }
 364
 365 void Session::response_add(Response* response)
 366 {
 367   a_responses.add(response);
 368   response->setSession(this);
 369
 370   try {
 371     response->activateTimer();
 372   } catch(anna::Exception& ex) {
 373     ex.trace();
 374   }
 375 }
 376
 377 void Session::response_erase(Response* response)
 378 {
 379   a_responses.erase(response);
 380   Response::release(response);
 381
 382   if(a_state == State::Disconnecting)  // only OnDisconnect::WaitPendings arrives here (the other disconnect suddently)
 383     if(getOTARequests() == 0) {
 384       // TODO: decode response->getRequest(), which is a comm message, so we get Origin-Host, and then, from OriginHostManager, we get the commEngine to provide:
 385       //sendDPA(commEngine, dprDataBlock);
 386     }
 387
 388   if(a_state == State::Closing)  // only OnDisconnect::WaitPendings arrives here (the other disconnect suddently)
 389     if(getOTARequests() == 0) unbind();
 390 }
 391
 392 Response* Session::response_find(const HopByHop hopByHop)
 393 noexcept(false) {
 394   diameter::comm::Response* result = a_responses.find(hopByHop);
 395 //   if (result == NULL) {
 396 //      string msg(asString());
 397 //      msg += anna::functions::asString(" | Response received for non registered context (HopByHop: %u)", hopByHop);
 398 //      throw anna::RuntimeException(msg, ANNA_FILE_LOCATION);
 399 //   }
 400   //Message * message = const_cast<Message*>(result->getRequest());
 401   return result;
 402 }
 403
 404 std::string Session::asString() const
 405 {
 406   string result(a_className);
 407   result += " { ";
 408   result += anna::timex::Timer::asString();
 409   result += " | Socket Id: ";
 410   result += anna::functions::asString(a_socketId);
 411   result += " | State: ";
 412   result += asText(a_state);
 413   result += " | OnDisconnect: ";
 414   result += asText(a_onDisconnect);
 415   result += " | Next Hop by hop: ";
 416   result += anna::functions::asString(a_nextHopByHop);
 417   result += " | Next End to end: ";
 418   result += anna::functions::asString(a_nextEndToEnd);
 419   result += anna::functions::asString(" | OTA requests: %d%s", getOTARequests(), idle() ? " (idle)" : "");
 420   result += " | Last Incoming Activity Time: ";
 421   result += a_lastIncomingActivityTime.asString();
 422   result += " | Last Outgoing Activity Time: ";
 423   result += a_lastOutgoingActivityTime.asString();
 424
 425   for(int i = ClassCode::Bind; i < ClassCode::Max; i ++) {
 426     result += " | Timeout for ClassCode '";
 427     result += ClassCode::asText((ClassCode::_v)i);
 428     result += "': ";
 429     result += a_timeouts[i].asString();
 430   }
 431
 432   return result;
 433 }
 434
 435 anna::xml::Node* Session::asXML(anna::xml::Node* parent) const
 436 {
 437   //parent = anna::timex::Timer::asXML(parent);
 438   anna::xml::Node* result = parent->createChild("diameter.comm.Session");
 439   result->createAttribute("SocketId", anna::functions::asString(a_socketId));
 440   result->createAttribute("State", asText(a_state));
 441   result->createAttribute("OnDisconnect", asText(a_onDisconnect));
 442   result->createAttribute("NextHopByHop", anna::functions::asString(a_nextHopByHop));
 443   result->createAttribute("NextEndToEnd", anna::functions::asString(a_nextEndToEnd));
 444   result->createAttribute("OTArequests", anna::functions::asString("%d%s", getOTARequests(), idle() ? " (idle)" : ""));
 445   result->createAttribute("LastIncomingActivityTime", a_lastIncomingActivityTime.asString());
 446   result->createAttribute("LastOutgoingActivityTime", a_lastOutgoingActivityTime.asString());
 447
 448   for(int i = ClassCode::Bind; i < ClassCode::Max; i ++) {
 449     std::string name = "TimeoutFor"; name += ClassCode::asText((ClassCode::_v)i);
 450     result->createAttribute(name.c_str(), a_timeouts[i].asString());
 451   }
 452
 453   // Messages
 454   anna::xml::Node* messages = result->createChild("diameter.comm.Messages");
 455   const Message* message;
 456
 457   for(const_response_iterator ii = response_begin(), maxii = response_end(); ii != maxii; ii ++) {
 458     if((message = Session::response(ii)->getRequest()) != NULL)
 459       message->asXML(messages);
 460   }
 461
 462   return result;
 463 }
 464
 465 const char* Session::asText(const State::_v state)
 466 {
 467   static const char* text [] = { "Closed", "WaitingBind", "Bound", "Failover", "Suspect", "WaitingDPA", "Disconnecting", "Closing" };
 468   return text [state];
 469 }
 470
 471 const char* Session::asText(const OnDisconnect::_v onDisconnect)
 472 {
 473   static const char* text [] = { "IgnorePendings", "WaitPendings" };
 474   return text [onDisconnect];
 475 }
 476
 477
 478
 479 HopByHop Session::SortById::value(const Response* response)
 480 {
 481   return response->getHopByHop();
 482 }
 483
 484
 485 //------------------------------------------------------------------------------
 486 //---------------------------------------- Session::updateIncomingActivityTime()
 487 //------------------------------------------------------------------------------
 488 void Session::updateIncomingActivityTime() {
 489   a_lastIncomingActivityTime = anna::functions::millisecond();
 490   LOGDEBUG
 491   (
 492     std::string msg = "Updated INCOMING activity on session (milliseconds unix): ";
 493     msg += anna::functions::asString(a_lastIncomingActivityTime.getValue());
 494     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 495   );
 496 }
 497
 498
 499 //------------------------------------------------------------------------------
 500 //---------------------------------------- Session::updateOutgoingActivityTime()
 501 //------------------------------------------------------------------------------
 502 void Session::updateOutgoingActivityTime(void) {
 503   a_lastOutgoingActivityTime = anna::functions::millisecond();
 504   LOGDEBUG
 505   (
 506     std::string msg = "Updated OUTGOING activity on session (milliseconds unix): ";
 507     msg += anna::functions::asString(a_lastOutgoingActivityTime.getValue());
 508     anna::Logger::debug(msg, ANNA_FILE_LOCATION);
 509   );
 510 }