source/diameter.comm/Session.cpp

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