X-Git-Url: https://git.stg.codes/stg.git/blobdiff_plain/597f6f31e3801612273886481381df509d8bcd12..1cd2b12bd4e4d86f6cd099240795f3ebeb3852b3:/projects/stargazer/plugins/other/smux/smux.cpp diff --git a/projects/stargazer/plugins/other/smux/smux.cpp b/projects/stargazer/plugins/other/smux/smux.cpp index 5367e249..07f9d200 100644 --- a/projects/stargazer/plugins/other/smux/smux.cpp +++ b/projects/stargazer/plugins/other/smux/smux.cpp @@ -15,17 +15,12 @@ #include #include "stg/common.h" -#include "stg/plugin_creator.h" #include "smux.h" #include "utils.h" -PLUGIN_CREATOR smc; - -PLUGIN * GetPlugin() +namespace { -return smc.GetPlugin(); -} bool SPrefixLess(const Sensors::value_type & a, const Sensors::value_type & b) @@ -33,38 +28,43 @@ bool SPrefixLess(const Sensors::value_type & a, return a.first.PrefixLess(b.first); } +} + +extern "C" STG::Plugin* GetPlugin() +{ + static SMUX plugin; + return &plugin; +} + SMUX_SETTINGS::SMUX_SETTINGS() - : errorStr(), - ip(0), - port(0), - password() + : ip(0), + port(0) {} -int SMUX_SETTINGS::ParseSettings(const MODULE_SETTINGS & s) +int SMUX_SETTINGS::ParseSettings(const STG::ModuleSettings & s) { -PARAM_VALUE pv; -std::vector::const_iterator pvi; +STG::ParamValue pv; int p; pv.param = "Port"; -pvi = std::find(s.moduleParams.begin(), s.moduleParams.end(), pv); -if (pvi == s.moduleParams.end()) +auto pvi = std::find(s.moduleParams.begin(), s.moduleParams.end(), pv); +if (pvi == s.moduleParams.end() || pvi->value.empty()) { errorStr = "Parameter \'Port\' not found."; printfd(__FILE__, "Parameter 'Port' not found\n"); return -1; } -if (ParseIntInRange(pvi->value[0], 2, 65535, &p)) +if (ParseIntInRange(pvi->value[0], 2, 65535, &p) != 0) { errorStr = "Cannot parse parameter \'Port\': " + errorStr; printfd(__FILE__, "Cannot parse parameter 'Port'\n"); return -1; } -port = p; +port = static_cast(p); pv.param = "Password"; pvi = std::find(s.moduleParams.begin(), s.moduleParams.end(), pv); -if (pvi == s.moduleParams.end()) +if (pvi == s.moduleParams.end() || pvi->value.empty()) { errorStr = "Parameter \'Password\' not found."; printfd(__FILE__, "Parameter 'Password' not found\n"); @@ -77,7 +77,7 @@ else pv.param = "Server"; pvi = std::find(s.moduleParams.begin(), s.moduleParams.end(), pv); -if (pvi == s.moduleParams.end()) +if (pvi == s.moduleParams.end() || pvi->value.empty()) { errorStr = "Parameter \'Server\' not found."; printfd(__FILE__, "Parameter 'Server' not found\n"); @@ -89,32 +89,19 @@ return 0; } SMUX::SMUX() - : PLUGIN(), - users(NULL), - tariffs(NULL), - admins(NULL), - services(NULL), - corporations(NULL), - traffcounter(NULL), - errorStr(), - smuxSettings(), - settings(), - thread(), - mutex(), - running(false), + : users(nullptr), + tariffs(nullptr), + admins(nullptr), + services(nullptr), + corporations(nullptr), + traffcounter(nullptr), stopped(true), + needReconnect(false), + lastReconnectTry(0), + reconnectTimeout(1), sock(-1), - smuxHandlers(), - pdusHandlers(), - sensors(), - tables(), - notifiers(), - addUserNotifier(*this), - delUserNotifier(*this), - addDelTariffNotifier(*this) + logger(STG::PluginLogger::get("smux")) { -pthread_mutex_init(&mutex, NULL); - smuxHandlers[SMUX_PDUs_PR_close] = &SMUX::CloseHandler; smuxHandlers[SMUX_PDUs_PR_registerResponse] = &SMUX::RegisterResponseHandler; smuxHandlers[SMUX_PDUs_PR_pdus] = &SMUX::PDUsRequestHandler; @@ -127,18 +114,11 @@ pdusHandlers[PDUs_PR_set_request] = &SMUX::SetRequestHandler; SMUX::~SMUX() { - { - Sensors::iterator it; - for (it = sensors.begin(); it != sensors.end(); ++it) - delete it->second; - } - { - Tables::iterator it; - for (it = tables.begin(); it != tables.end(); ++it) - delete it->second; - } -printfd(__FILE__, "SMUX::~SMUX()\n"); -pthread_mutex_destroy(&mutex); + for (auto& kv : sensors) + delete kv.second; + for (auto& kv : tables) + delete kv.second; + printfd(__FILE__, "SMUX::~SMUX()\n"); } int SMUX::ParseSettings() @@ -148,15 +128,15 @@ return smuxSettings.ParseSettings(settings); int SMUX::Start() { -assert(users != NULL && "users must not be NULL"); -assert(tariffs != NULL && "tariffs must not be NULL"); -assert(admins != NULL && "admins must not be NULL"); -assert(services != NULL && "services must not be NULL"); -assert(corporations != NULL && "corporations must not be NULL"); -assert(traffcounter != NULL && "traffcounter must not be NULL"); +assert(users != nullptr && "users must not be NULL"); +assert(tariffs != nullptr && "tariffs must not be NULL"); +assert(admins != nullptr && "admins must not be NULL"); +assert(services != nullptr && "services must not be NULL"); +assert(corporations != nullptr && "corporations must not be NULL"); +assert(traffcounter != nullptr && "traffcounter must not be NULL"); if (PrepareNet()) - return -1; + needReconnect = true; // Users sensors[OID(".1.3.6.1.4.1.38313.1.1.1")] = new TotalUsersSensor(*users); @@ -170,6 +150,7 @@ sensors[OID(".1.3.6.1.4.1.38313.1.1.8")] = new PassiveUsersSensor(*users); sensors[OID(".1.3.6.1.4.1.38313.1.1.9")] = new CreditUsersSensor(*users); sensors[OID(".1.3.6.1.4.1.38313.1.1.10")] = new FreeMbUsersSensor(*users); sensors[OID(".1.3.6.1.4.1.38313.1.1.11")] = new TariffChangeUsersSensor(*users); +sensors[OID(".1.3.6.1.4.1.38313.1.1.12")] = new ActiveUsersSensor(*users); // Tariffs sensors[OID(".1.3.6.1.4.1.38313.1.2.1")] = new TotalTariffsSensor(*tariffs); // Admins @@ -188,7 +169,7 @@ UpdateTables(); SetNotifiers(); #ifdef SMUX_DEBUG -Sensors::const_iterator it(sensors.begin()); +auto it = sensors.begin(); while (it != sensors.end()) { printfd(__FILE__, "%s = %s\n", @@ -198,15 +179,8 @@ while (it != sensors.end()) } #endif -if (!running) - { - if (pthread_create(&thread, NULL, Runner, this)) - { - errorStr = "Cannot create thread."; - printfd(__FILE__, "Cannot create thread\n"); - return -1; - } - } +if (!m_thread.joinable()) + m_thread = std::jthread([this](auto token){ Run(std::move(token)); }); return 0; } @@ -214,7 +188,7 @@ return 0; int SMUX::Stop() { printfd(__FILE__, "SMUX::Stop() - Before\n"); -running = false; +m_thread.request_stop(); if (!stopped) { @@ -222,91 +196,75 @@ if (!stopped) for (int i = 0; i < 25 && !stopped; i++) { struct timespec ts = {0, 200000000}; - nanosleep(&ts, NULL); - } - - //after 5 seconds waiting thread still running. now killing it - if (!stopped) - { - printfd(__FILE__, "SMUX::Stop() - failed to stop thread, killing it\n"); - if (pthread_kill(thread, SIGINT)) - { - errorStr = "Cannot kill thread."; - printfd(__FILE__, "SMUX::Stop() - Cannot kill thread\n"); - return -1; - } - printfd(__FILE__, "SMUX::Stop() - killed Run\n"); + nanosleep(&ts, nullptr); } } -pthread_join(thread, NULL); +if (!stopped) + m_thread.detach(); +else + m_thread.join(); ResetNotifiers(); - { - Tables::iterator it; - for (it = tables.begin(); it != tables.end(); ++it) - delete it->second; - } - { - Sensors::iterator it; - for (it = sensors.begin(); it != sensors.end(); ++it) - delete it->second; - } +for (auto& kv : sensors) + delete kv.second; +for (auto& kv : tables) + delete kv.second; tables.erase(tables.begin(), tables.end()); sensors.erase(sensors.begin(), sensors.end()); close(sock); +if (!stopped) + { + return -1; + } + printfd(__FILE__, "SMUX::Stop() - After\n"); return 0; } -int SMUX::Reload() +int SMUX::Reload(const STG::ModuleSettings & /*ms*/) { -if (Stop()) +if (Stop() != 0) return -1; -if (Start()) +if (Start() != 0) return -1; +if (!needReconnect) + { + printfd(__FILE__, "SMUX reconnected succesfully.\n"); + logger("Reconnected successfully."); + } return 0; } -void * SMUX::Runner(void * d) -{ -sigset_t signalSet; -sigfillset(&signalSet); -pthread_sigmask(SIG_BLOCK, &signalSet, NULL); - -SMUX * smux = static_cast(d); - -smux->Run(); - -return NULL; -} - -void SMUX::Run() +void SMUX::Run(std::stop_token token) { stopped = true; if (!SendOpenPDU(sock)) - return; + needReconnect = true; if (!SendRReqPDU(sock)) - return; -running = true; + needReconnect = true; stopped = false; -while(running) +while (!token.stop_requested()) { - if (WaitPackets(sock)) + if (WaitPackets(sock) && !needReconnect) { - SMUX_PDUs_t * pdus = RecvSMUXPDUs(sock); - if (pdus) + auto* pdus = RecvSMUXPDUs(sock); + if (pdus != nullptr) { DispatchPDUs(pdus); ASN_STRUCT_FREE(asn_DEF_SMUX_PDUs, pdus); } + else if (!token.stop_requested()) + Reconnect(); } - if (!running) + else if (!token.stop_requested() && needReconnect) + Reconnect(); + if (token.stop_requested()) break; } SendClosePDU(sock); @@ -320,6 +278,7 @@ sock = socket(AF_INET, SOCK_STREAM, 0); if (sock < 0) { errorStr = "Cannot create socket."; + logger("Cannot create a socket: %s", strerror(errno)); printfd(__FILE__, "Cannot create socket\n"); return true; } @@ -330,9 +289,10 @@ addr.sin_family = AF_INET; addr.sin_port = htons(smuxSettings.GetPort()); addr.sin_addr.s_addr = smuxSettings.GetIP(); -if (connect(sock, reinterpret_cast(&addr), sizeof(addr))) +if (connect(sock, reinterpret_cast(&addr), sizeof(addr)) != 0) { errorStr = "Cannot connect."; + logger("Cannot connect the socket: %s", strerror(errno)); printfd(__FILE__, "Cannot connect. Message: '%s'\n", strerror(errno)); return true; } @@ -340,17 +300,41 @@ if (connect(sock, reinterpret_cast(&addr), sizeof(addr))) return false; } +bool SMUX::Reconnect() +{ +if (needReconnect && difftime(time(nullptr), lastReconnectTry) < reconnectTimeout) + return true; + +time(&lastReconnectTry); +SendClosePDU(sock); +close(sock); +if (!PrepareNet()) + if (SendOpenPDU(sock)) + if (SendRReqPDU(sock)) + { + reconnectTimeout = 1; + needReconnect = false; + logger("Connected successfully"); + printfd(__FILE__, "Connected successfully\n"); + return false; + } + +if (needReconnect) + if (reconnectTimeout < 60) + reconnectTimeout *= 2; + +needReconnect = true; +return true; +} + bool SMUX::DispatchPDUs(const SMUX_PDUs_t * pdus) { -SMUXHandlers::iterator it; -it = smuxHandlers.find(pdus->present); +auto it = smuxHandlers.find(pdus->present); if (it != smuxHandlers.end()) - { return (this->*(it->second))(pdus); - } +#ifdef SMUX_DEBUG else { -#ifdef SMUX_DEBUG switch (pdus->present) { case SMUX_PDUs_PR_NOTHING: @@ -366,8 +350,8 @@ else printfd(__FILE__, "PDUs: undefined\n"); } asn_fprint(stderr, &asn_DEF_SMUX_PDUs, pdus); -#endif } +#endif return false; } @@ -375,7 +359,7 @@ bool SMUX::UpdateTables() { Sensors newSensors; bool done = true; -Tables::iterator it(tables.begin()); +auto it = tables.begin(); while (it != tables.end()) { try @@ -394,11 +378,11 @@ while (it != tables.end()) } if (!done) { - Sensors::iterator it(newSensors.begin()); - while (it != newSensors.end()) + auto sit = newSensors.begin(); + while (sit != newSensors.end()) { - delete it->second; - ++it; + delete sit->second; + ++sit; } return false; } @@ -406,12 +390,11 @@ if (!done) it = tables.begin(); while (it != tables.end()) { - std::pair res; - res = std::equal_range(sensors.begin(), - sensors.end(), - std::pair(OID(it->first), NULL), - SPrefixLess); - Sensors::iterator sit(res.first); + auto res = std::equal_range(sensors.begin(), + sensors.end(), + std::pair(OID(it->first), nullptr), + SPrefixLess); + auto sit = res.first; while (sit != res.second) { delete sit->second; @@ -426,20 +409,20 @@ sensors.insert(newSensors.begin(), newSensors.end()); return true; } -void SMUX::SetNotifier(USER_PTR userPtr) +void SMUX::SetNotifier(UserPtr userPtr) { -notifiers.push_back(CHG_AFTER_NOTIFIER(*this, userPtr)); -userPtr->GetProperty().tariffName.AddAfterNotifier(¬ifiers.back()); +notifiers.emplace_back(*this, userPtr); +userPtr->GetProperties().tariffName.AddAfterNotifier(¬ifiers.back()); } -void SMUX::UnsetNotifier(USER_PTR userPtr) +void SMUX::UnsetNotifier(UserPtr userPtr) { -std::list::iterator it = notifiers.begin(); +auto it = notifiers.begin(); while (it != notifiers.end()) { if (it->GetUserPtr() == userPtr) { - userPtr->GetProperty().tariffName.DelAfterNotifier(&(*it)); + userPtr->GetProperties().tariffName.DelAfterNotifier(&(*it)); notifiers.erase(it); break; } @@ -452,36 +435,39 @@ void SMUX::SetNotifiers() int h = users->OpenSearch(); assert(h && "USERS::OpenSearch is always correct"); -USER_PTR u; -while (!users->SearchNext(h, &u)) +UserPtr u; +while (users->SearchNext(h, &u) == 0) SetNotifier(u); users->CloseSearch(h); -users->AddNotifierUserAdd(&addUserNotifier); -users->AddNotifierUserDel(&delUserNotifier); - -tariffs->AddNotifierAdd(&addDelTariffNotifier); -tariffs->AddNotifierDel(&addDelTariffNotifier); +m_onAddUserConn = users->onAdd([this](auto user){ + SetNotifier(user); + UpdateTables(); +}); +m_onDelUserConn = users->onDel([this](auto user){ + UnsetNotifier(user); + UpdateTables(); +}); + +auto updateTables = [this](const STG::TariffData&){ UpdateTables(); }; +m_onAddTariffConn = tariffs->onAdd(updateTables); +m_onDelTariffConn = tariffs->onDel(updateTables); } void SMUX::ResetNotifiers() { -tariffs->DelNotifierDel(&addDelTariffNotifier); -tariffs->DelNotifierAdd(&addDelTariffNotifier); +m_onAddTariffConn.disconnect(); +m_onDelTariffConn.disconnect(); -users->DelNotifierUserDel(&delUserNotifier); -users->DelNotifierUserAdd(&addUserNotifier); +m_onAddUserConn.disconnect(); +m_onDelUserConn.disconnect(); -std::list::iterator it = notifiers.begin(); +auto it = notifiers.begin(); while (it != notifiers.end()) { - it->GetUserPtr()->GetProperty().tariffName.DelAfterNotifier(&(*it)); + it->GetUserPtr()->GetProperties().tariffName.DelAfterNotifier(&(*it)); ++it; } -} - -void CHG_AFTER_NOTIFIER::Notify(const std::string &, const std::string &) -{ -smux.UpdateTables(); +notifiers.clear(); }