1 // ANNA - Anna is Not Nothingness Anymore //
3 // (c) Copyright 2005-2015 Eduardo Ramos Testillano & Francisco Ruiz Rayo //
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 //
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>
21 #include <OriginHost.hpp>
22 #include <MyDiameterEngine.hpp>
33 OriginHost::OriginHost(const std::string &originHost, unsigned int applicationId, const anna::diameter::stack::Dictionary *baseProtocolDictionary) :
34 a_originHost(originHost), a_applicationId(applicationId) {
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);
44 a_detailedLog = false;
47 a_diameterServer = NULL;
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;
59 a_burstRepeat = false;
60 a_burstActive = false;
62 a_burstDeliveryIt = a_burstMessages.begin();
64 a_burstPopCounter = 0;
68 void OriginHost::createEntity(const std::string &entityRepresentation, const anna::Millisecond &bindTimeout, const anna::Millisecond &applicationTimeout) throw(anna::RuntimeException) {
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);
77 void OriginHost::startDiameterServer(const std::string &serverRepresentation, int sessions, const anna::Millisecond &inactivityTimeout) throw(anna::RuntimeException) {
79 //if(sessions <= 0) return; negative implies no limit for accepted connections
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.
88 a_diameterServer->setDescription(serverDescription);
89 a_diameterServer->setAllowedInactivityTime(inactivityTimeout);
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);
100 void OriginHost::releaseCommMessage(anna::diameter::comm::Message *msg) throw() {
101 a_commMessages.release(msg);
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);
112 void OriginHost::writeLogFile(const anna::diameter::codec::Message &decodedMessage, const std::string &logExtension, const std::string &detail) const throw() {
114 std::string targetFile = a_logFile;
118 targetFile += logExtension;
121 std::ofstream out(targetFile.c_str(), std::ifstream::out | std::ifstream::app);
123 std::string title = "[";
124 title += logExtension;
126 // Build complete log:
127 std::string log = "\n";
128 std::string xml = decodedMessage.asXMLString();
132 anna::time::Date now;
135 title += now.asString();
136 log += anna::functions::highlight(title, anna::functions::TextHighlightMode::OverAndUnderline);
139 log += anna::functions::highlight("Used resource");
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());
153 name += getMyDiameterEngine()->getOriginHost();
155 name += anna::functions::asString(decodedMessage.getHopByHop());
157 name += anna::functions::asString(decodedMessage.getEndToEnd());
159 name += anna::functions::asString(decodedMessage.getId().first);
161 name += ((decodedMessage.getId().second) ? "request.":"answer.");
162 name += logExtension;
164 std::ofstream outMsg(name.c_str(), std::ifstream::out | std::ifstream::app);
165 outMsg.write(xml.c_str(), xml.size());
170 out.write(log.c_str(), log.size());
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.
182 int OriginHost::clearBurst() throw() {
183 int size = a_burstMessages.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());
190 for(it = it_min; it != it_max; it++) releaseCommMessage((*it).second);
192 a_burstMessages.clear();
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));
199 a_burstActive = false;
201 a_burstDeliveryIt = a_burstMessages.begin();
205 int OriginHost::loadBurstMessage(const anna::DataBlock & db) throw(anna::RuntimeException) {
206 anna::diameter::comm::Message *msg = createCommMessage();
208 a_burstMessages[a_burstLoadIndx++] = msg;
209 return (a_burstLoadIndx - 1);
212 int OriginHost::stopBurst() throw() {
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));
220 a_burstActive = false;
221 // Remaining on cycle:
222 return (a_burstMessages.size() - (*a_burstDeliveryIt).first);
225 int OriginHost::popBurst(int releaseAmount) throw() {
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));
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));
240 int currentOTArequests = a_entity->getOTARequests();
241 a_burstPopCounter = (releaseAmount > currentOTArequests) ? currentOTArequests : releaseAmount;
242 return a_burstPopCounter;
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));
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));
260 a_burstActive = true;
263 for(count = 0; count < loadAmount; count++)
264 if(!sendBurstMessage()) break;
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));
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));
286 for(count = 0; count < loadAmount; count++)
287 if(!sendBurstMessage(true /* anyway */)) break;
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));
300 a_burstActive = true;
302 a_burstDeliveryIt = a_burstMessages.begin();
303 return (pushBurst(initialLoad));
306 bool OriginHost::sendBurstMessage(bool anyway) throw() {
307 if(!anyway && !burstActive()) return false;
309 if(a_burstPopCounter > 0) {
310 if(burstLogEnabled()) writeBurstLogFile("x");
316 if(a_burstDeliveryIt == a_burstMessages.end()) {
317 a_burstDeliveryIt = a_burstMessages.begin();
323 if(burstLogEnabled()) writeBurstLogFile(anna::functions::asString("\nCompleted burst cycle. Starting again (repeat mode) on cycle %d.\n", a_burstCycle));
325 if(burstLogEnabled()) writeBurstLogFile("\nCompleted burst cycle. Burst finished (repeat mode disabled).\n");
333 anna::diameter::comm::Message *msg = (*a_burstDeliveryIt).second;
334 int order = (*a_burstDeliveryIt).first + 1;
338 bool result = a_entity->send(msg, anna::CommandLine::instantiate().exists("balance"));
340 if(burstLogEnabled()) {
341 if(a_burstMessages.size() >= 100)
342 dot = (order % (a_burstMessages.size() / 100));
345 writeBurstLogFile(".");
347 writeBurstLogFile(anna::functions::asString(" %d", order));
348 int otaReqs = a_entity->getOTARequests();
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));
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
370 std::string OriginHost::lookBurst(int order) const throw() {
372 if (order == -1) order = a_burstDeliveryIt->first;
374 std::string result = "No message found for order provided (";
375 result += anna::functions::asString(order);
377 std::map<int, anna::diameter::comm::Message*>::const_iterator it = a_burstMessages.find(order - 1);
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(); }
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);
391 if(it != a_burstMessages.end()) {
392 a_burstDeliveryIt = it;
393 result = "Position updated for order provided (";
396 result += anna::functions::asString(order);
401 anna::xml::Node* OriginHost::asXML(anna::xml::Node* parent) const
403 anna::xml::Node* result = parent->createChild("OriginHost");
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);
415 anna::xml::Node* commEngine = result->createChild("CommEngine");
416 a_commEngine->asXML(commEngine);
421 std::string OriginHost::asXMLString() const throw() {
422 anna::xml::Node root("root");
423 return anna::xml::Compiler().apply(asXML(&root));