example/diameter/launcher/OriginHost.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 // Standard
  10 #include <string>
  11
  12 // Project
  13 #include <anna/diameter.comm/Message.hpp>
  14 #include <anna/diameter/stack/Dictionary.hpp>
  15 #include <anna/diameter/codec/EngineManager.hpp>
  16 #include <anna/core/core.hpp>
  17 #include <anna/time/Date.hpp>
  18 #include <anna/xml/Compiler.hpp>
  19
  20 // Process
  21 #include <OriginHost.hpp>
  22 #include <MyDiameterEngine.hpp>
  23
  24
  25 namespace anna {
  26   namespace diameter {
  27     namespace stack {
  28       class Dictionary;
  29     }
  30   }
  31 }
  32
  33 OriginHost::OriginHost(const std::string &originHost, unsigned int applicationId, const anna::diameter::stack::Dictionary *baseProtocolDictionary) :
  34   a_originHost(originHost), a_applicationId(applicationId) {
  35
  36   std::string commEngineName = a_originHost + "_DiameterCommEngine";
  37   a_commEngine = new MyDiameterEngine(commEngineName.c_str(), baseProtocolDictionary);
  38   a_commEngine->setAutoBind(false);  // allow to create client-sessions without binding them, in order to set timeouts.
  39   a_codecEngine = anna::diameter::codec::EngineManager::instantiate().getCodecEngine(applicationId);
  40
  41   a_logFile = "";
  42   a_burstLogFile = "";
  43   a_splitLog = false;
  44   a_detailedLog = false;
  45   a_dumpLog = false;
  46   a_entity = NULL;
  47   a_diameterServer = NULL;
  48
  49   // Comm resources:
  50   a_allowedInactivityTime = (anna::Millisecond)90000;
  51   a_tcpConnectDelay = (anna::Millisecond)200;
  52   a_answersTimeout = (anna::Millisecond)10000;
  53   a_ceaTimeout = (anna::Millisecond)10000;
  54   a_watchdogPeriod = (anna::Millisecond)30000;
  55   a_requestRetransmissions = 0;
  56
  57   // Burst
  58   a_burstCycle = 1;
  59   a_burstRepeat = false;
  60   a_burstActive = false;
  61   a_burstLoadIndx = 0;
  62   a_burstDeliveryIt = a_burstMessages.begin();
  63   a_otaRequest = 0;
  64   a_burstPopCounter = 0;
  65 }
  66
  67
  68 void OriginHost::createEntity(const std::string &entityRepresentation, const anna::Millisecond &bindTimeout, const anna::Millisecond &applicationTimeout) throw(anna::RuntimeException) {
  69
  70   anna::socket_v servers = anna::functions::getSocketVectorFromString(entityRepresentation);
  71   std::string entityDescription = "Launcher diameter entity for "; entityDescription += a_originHost;
  72   a_entity = (MyDiameterEntity*)(a_commEngine->createEntity(servers, entityDescription));
  73   a_entity->setClassCodeTimeout(anna::diameter::comm::ClassCode::Bind, bindTimeout);
  74   a_entity->setClassCodeTimeout(anna::diameter::comm::ClassCode::ApplicationMessage, applicationTimeout);
  75 }
  76
  77 void OriginHost::startDiameterServer(const std::string &serverRepresentation, int sessions, const anna::Millisecond &inactivityTimeout) throw(anna::RuntimeException) {
  78
  79   //if(sessions <= 0) return; negative implies no limit for accepted connections
  80
  81   std::string address; int port;
  82   anna::functions::getAddressAndPortFromSocketLiteral(serverRepresentation, address, port);
  83   std::string serverDescription = "Launcher diameter local server for "; serverDescription += a_originHost;
  84   a_diameterServer = (MyLocalServer*)(a_commEngine->createLocalServer(address, port, sessions));
  85           // we could set sessions = 0, and after application run(), use setMaxConnections(real sessions)
  86           // over the local server in order to start it.
  87
  88   a_diameterServer->setDescription(serverDescription);
  89   a_diameterServer->setAllowedInactivityTime(inactivityTimeout);
  90 }
  91
  92 anna::diameter::comm::Message *OriginHost::createCommMessage() throw(anna::RuntimeException) {
  93   anna::diameter::comm::Message *result = a_commMessages.create();
  94   result->setRetries(a_requestRetransmissions);
  95   if (a_requestRetransmissions > 0) result->setOnExpiry(anna::diameter::comm::Message::OnExpiry::Retransmit);
  96   return result;
  97 }
  98
  99
 100 void OriginHost::releaseCommMessage(anna::diameter::comm::Message *msg) throw() {
 101   a_commMessages.release(msg);
 102 }
 103
 104
 105 void OriginHost::writeLogFile(const anna::DataBlock & db, const std::string &logExtension, const std::string &detail) const throw() {
 106   anna::diameter::codec::Message codecMsg;
 107   try { codecMsg.decode(db); } catch(anna::RuntimeException &ex) { ex.trace(); }
 108   writeLogFile(codecMsg, logExtension, detail);
 109 }
 110
 111 // Already decoded:
 112 void OriginHost::writeLogFile(const anna::diameter::codec::Message &decodedMessage, const std::string &logExtension, const std::string &detail) const throw() {
 113   // Open target file:
 114   std::string targetFile = a_logFile;
 115
 116   if(a_splitLog) {
 117     targetFile += ".";
 118     targetFile += logExtension;
 119   }
 120
 121   std::ofstream out(targetFile.c_str(), std::ifstream::out | std::ifstream::app);
 122   // Set text to dump:
 123   std::string title = "[";
 124   title += logExtension;
 125   title += "]";
 126   // Build complete log:
 127   std::string log = "\n";
 128   std::string xml = decodedMessage.asXMLString();
 129
 130
 131   if(a_detailedLog) {
 132     anna::time::Date now;
 133     now.setNow();
 134     title += " ";
 135     title += now.asString();
 136     log += anna::functions::highlight(title, anna::functions::TextHighlightMode::OverAndUnderline);
 137     log += xml;
 138     log += "\n";
 139     log += anna::functions::highlight("Used resource");
 140     log += detail;
 141     log += "\n";
 142   } else {
 143     log += title;
 144     log += "\n";
 145     log += xml;
 146     log += "\n";
 147   }
 148
 149   if(a_dumpLog) {
 150     // <unix ms timestamp>.<originHost>.<hop by hop>.<end to end>.<message code>.<request|answer>.<type of event>.xml
 151     std::string name = anna::functions::asString((anna::Millisecond)anna::functions::millisecond());
 152     name += ".";
 153     name += getMyDiameterEngine()->getOriginHost();
 154     name += ".";
 155     name += anna::functions::asString(decodedMessage.getHopByHop());
 156     name += ".";
 157     name += anna::functions::asString(decodedMessage.getEndToEnd());
 158     name += ".";
 159     name += anna::functions::asString(decodedMessage.getId().first);
 160     name += ".";
 161     name += ((decodedMessage.getId().second) ? "request.":"answer.");
 162     name += logExtension;
 163     name += ".xml";
 164     std::ofstream outMsg(name.c_str(), std::ifstream::out | std::ifstream::app);
 165     outMsg.write(xml.c_str(), xml.size());
 166     outMsg.close();
 167   }
 168
 169   // Write and close
 170   out.write(log.c_str(), log.size());
 171   out.close();
 172 }
 173
 174 void OriginHost::writeBurstLogFile(const std::string &buffer) throw() {
 175   std::ofstream out(a_burstLogFile.c_str(), std::ifstream::out | std::ifstream::app);
 176   out.write(buffer.c_str(), buffer.size());
 177   out.close();    // close() will be called when the object is destructed (i.e., when it goes out of scope).
 178   // you'd call close() only if you indeed for some reason wanted to close the filestream
 179   // earlier than it goes out of scope.
 180 }
 181
 182 int OriginHost::clearBurst() throw() {
 183   int size = a_burstMessages.size();
 184
 185   if(size) {
 186     std::map<int, anna::diameter::comm::Message*>::const_iterator it;
 187     std::map<int, anna::diameter::comm::Message*>::const_iterator it_min(a_burstMessages.begin());
 188     std::map<int, anna::diameter::comm::Message*>::const_iterator it_max(a_burstMessages.end());
 189
 190     for(it = it_min; it != it_max; it++) releaseCommMessage((*it).second);
 191
 192     a_burstMessages.clear();
 193   } else {
 194     std::string msg = "Burst list already empty. Nothing done";
 195     std::cout << msg << std::endl;
 196     LOGWARNING(anna::Logger::warning(msg, ANNA_FILE_LOCATION));
 197   }
 198
 199   a_burstActive = false;
 200   a_burstLoadIndx = 0;
 201   a_burstDeliveryIt = a_burstMessages.begin();
 202   return size;
 203 }
 204
 205 int OriginHost::loadBurstMessage(const anna::DataBlock & db) throw(anna::RuntimeException) {
 206   anna::diameter::comm::Message *msg = createCommMessage();
 207   msg->setBody(db);
 208   a_burstMessages[a_burstLoadIndx++] = msg;
 209   return (a_burstLoadIndx - 1);
 210 }
 211
 212 int OriginHost::stopBurst() throw() {
 213   if(!a_burstActive) {
 214     std::string msg = "Burst launch is already stopped. Nothing done";
 215     std::cout << msg << std::endl;
 216     LOGWARNING(anna::Logger::warning(msg, ANNA_FILE_LOCATION));
 217     return -1;
 218   }
 219
 220   a_burstActive = false;
 221   // Remaining on cycle:
 222   return (a_burstMessages.size() - (*a_burstDeliveryIt).first);
 223 }
 224
 225 int OriginHost::popBurst(int releaseAmount) throw() {
 226   if(!a_burstActive) {
 227     std::string msg = "Burst launch is stopped. Nothing done";
 228     std::cout << msg << std::endl;
 229     LOGWARNING(anna::Logger::warning(msg, ANNA_FILE_LOCATION));
 230     return -1;
 231   }
 232
 233   if(releaseAmount < 1) {
 234     std::string msg = "No valid release amount is specified. Ignoring burst pop";
 235     std::cout << msg << std::endl;
 236     LOGWARNING(anna::Logger::warning(msg, ANNA_FILE_LOCATION));
 237     return -2;
 238   }
 239
 240   int currentOTArequests = a_entity->getOTARequests();
 241   a_burstPopCounter = (releaseAmount > currentOTArequests) ? currentOTArequests : releaseAmount;
 242   return a_burstPopCounter;
 243 }
 244
 245 int OriginHost::pushBurst(int loadAmount) throw() {
 246   if(a_burstMessages.size() == 0) {
 247     std::string msg = "Burst data not found (empty list). Ignoring burst launch";
 248     std::cout << msg << std::endl;
 249     LOGWARNING(anna::Logger::warning(msg, ANNA_FILE_LOCATION));
 250     return -1;
 251   }
 252
 253   if(loadAmount < 1) {
 254     std::string msg = "No valid load amount is specified. Ignoring burst push";
 255     std::cout << msg << std::endl;
 256     LOGWARNING(anna::Logger::warning(msg, ANNA_FILE_LOCATION));
 257     return -2;
 258   }
 259
 260   a_burstActive = true;
 261   int count;
 262
 263   for(count = 0; count < loadAmount; count++)
 264     if(!sendBurstMessage()) break;
 265
 266   return count;
 267 }
 268
 269 int OriginHost::sendBurst(int loadAmount) throw() {
 270   if(a_burstMessages.size() == 0) {
 271     std::string msg = "Burst data not found (empty list). Ignoring burst launch";
 272     std::cout << msg << std::endl;
 273     LOGWARNING(anna::Logger::warning(msg, ANNA_FILE_LOCATION));
 274     return -1;
 275   }
 276
 277   if(loadAmount < 1) {
 278     std::string msg = "No valid load amount is specified. Ignoring burst send";
 279     std::cout << msg << std::endl;
 280     LOGWARNING(anna::Logger::warning(msg, ANNA_FILE_LOCATION));
 281     return -2;
 282   }
 283
 284   int count;
 285
 286   for(count = 0; count < loadAmount; count++)
 287     if(!sendBurstMessage(true /* anyway */)) break;
 288
 289   return count;
 290 }
 291
 292 int OriginHost::startBurst(int initialLoad) throw() {
 293   if(initialLoad < 1) {
 294     std::string msg = "No initial load is specified. Ignoring burst start";
 295     std::cout << msg << std::endl;
 296     LOGWARNING(anna::Logger::warning(msg, ANNA_FILE_LOCATION));
 297     return -2;
 298   }
 299
 300   a_burstActive = true;
 301   a_burstCycle = 1;
 302   a_burstDeliveryIt = a_burstMessages.begin();
 303   return (pushBurst(initialLoad));
 304 }
 305
 306 bool OriginHost::sendBurstMessage(bool anyway) throw() {
 307   if(!anyway && !burstActive()) return false;
 308
 309   if(a_burstPopCounter > 0) {
 310     if(burstLogEnabled()) writeBurstLogFile("x");
 311
 312     a_burstPopCounter--;
 313     return false;
 314   }
 315
 316   if(a_burstDeliveryIt == a_burstMessages.end()) {
 317     a_burstDeliveryIt = a_burstMessages.begin();
 318
 319     if(!anyway) {
 320       if(a_burstRepeat) {
 321         a_burstCycle++;
 322
 323         if(burstLogEnabled()) writeBurstLogFile(anna::functions::asString("\nCompleted burst cycle. Starting again (repeat mode) on cycle %d.\n", a_burstCycle));
 324       } else {
 325         if(burstLogEnabled()) writeBurstLogFile("\nCompleted burst cycle. Burst finished (repeat mode disabled).\n");
 326
 327         stopBurst();
 328         return false;
 329       }
 330     }
 331   }
 332
 333   anna::diameter::comm::Message *msg = (*a_burstDeliveryIt).second;
 334   int order = (*a_burstDeliveryIt).first + 1;
 335   a_burstDeliveryIt++;
 336   bool dot = true;
 337   // sending
 338   bool result = a_entity->send(msg, anna::CommandLine::instantiate().exists("balance"));
 339
 340   if(burstLogEnabled()) {
 341     if(a_burstMessages.size() >= 100)
 342       dot = (order  % (a_burstMessages.size() / 100));
 343
 344     if(dot) {
 345       writeBurstLogFile(".");
 346     } else {
 347       writeBurstLogFile(anna::functions::asString(" %d", order));
 348       int otaReqs  = a_entity->getOTARequests();
 349
 350       if(result && (otaReqs != a_otaRequest)) {
 351         // false if was a sending after an answer received (no OTA change in this case)
 352         // true after push and pop operations
 353         a_otaRequest = otaReqs;
 354         writeBurstLogFile(anna::functions::asString("[OTA %d]", a_otaRequest));
 355       }
 356     }
 357   }
 358
 359   // Detailed log:
 360   if(logEnabled()) {
 361     anna::diameter::comm::Server *usedServer = a_entity->getLastUsedResource();
 362     anna::diameter::comm::ClientSession *usedClientSession = usedServer ? usedServer->getLastUsedResource() : NULL;
 363     std::string detail = usedClientSession ? usedClientSession->asString() : "<null client session>"; // esto no deberia ocurrir
 364     writeLogFile(msg->getBody(), (result ? "sent2e" : "send2eError"), detail); // el del nodo de trabajo
 365   }
 366
 367   return result;
 368 }
 369
 370 std::string OriginHost::lookBurst(int order) const throw() {
 371
 372   if (order == -1) order = a_burstDeliveryIt->first;
 373
 374   std::string result = "No message found for order provided (";
 375   result += anna::functions::asString(order);
 376   result += ")";
 377   std::map<int, anna::diameter::comm::Message*>::const_iterator it = a_burstMessages.find(order - 1);
 378
 379   if(it != a_burstMessages.end()) {
 380     anna::diameter::codec::Message codecMsg;
 381     try { codecMsg.decode((*it).second->getBody()); result = codecMsg.asXMLString(); } catch(anna::RuntimeException &ex) { ex.trace(); }
 382   }
 383
 384   return result;
 385 }
 386
 387 std::string OriginHost::gotoBurst(int order) throw() {
 388   std::string result = "Position not found for order provided (";
 389   std::map<int, anna::diameter::comm::Message*>::iterator it = a_burstMessages.find(order - 1);
 390
 391   if(it != a_burstMessages.end()) {
 392     a_burstDeliveryIt = it;
 393     result = "Position updated for order provided (";
 394   }
 395
 396   result += anna::functions::asString(order);
 397   result += ")";
 398   return result;
 399 }
 400
 401 anna::xml::Node* OriginHost::asXML(anna::xml::Node* parent) const
 402 throw() {
 403   anna::xml::Node* result = parent->createChild("OriginHost");
 404
 405   result->createAttribute("originHost", a_originHost);
 406   result->createAttribute("ApplicationId", a_applicationId);
 407   result->createAttribute("originRealm", a_commEngine->getOriginRealm());
 408   result->createAttribute("LogFile", a_logFile);
 409   result->createAttribute("SplitLog", a_splitLog ? "yes" : "no");
 410   result->createAttribute("DetailedLog", a_detailedLog ? "yes" : "no");
 411   result->createAttribute("DumpLog", a_dumpLog ? "yes" : "no");
 412   result->createAttribute("BurstLogFile", a_burstLogFile);
 413   result->createAttribute("RequestRetransmissions", a_requestRetransmissions);
 414
 415   anna::xml::Node* commEngine = result->createChild("CommEngine");
 416   a_commEngine->asXML(commEngine);
 417
 418   return result;
 419 }
 420
 421 std::string OriginHost::asXMLString() const throw() {
 422   anna::xml::Node root("root");
 423   return anna::xml::Compiler().apply(asXML(&root));
 424 }