X-Git-Url: https://git.stg.codes/stg.git/blobdiff_plain/07802098b4ebcddaa55994f7a87fe0b0e8514d04..80aca6d78d4ff5d514216fe96c44e2ee75c4e933:/projects/stargazer/plugins/other/smux/smux.cpp?ds=sidebyside diff --git a/projects/stargazer/plugins/other/smux/smux.cpp b/projects/stargazer/plugins/other/smux/smux.cpp index bd83f1ab..f79085df 100644 --- a/projects/stargazer/plugins/other/smux/smux.cpp +++ b/projects/stargazer/plugins/other/smux/smux.cpp @@ -6,213 +6,34 @@ #include #include #include +#include #include #include - -#include "asn1/OpenPDU.h" -#include "asn1/ClosePDU.h" -#include "asn1/RReqPDU.h" -#include "asn1/GetRequest-PDU.h" -#include "asn1/GetResponse-PDU.h" -#include "asn1/VarBindList.h" -#include "asn1/VarBind.h" -#include "asn1/OBJECT_IDENTIFIER.h" -#include "asn1/ber_decoder.h" -#include "asn1/der_encoder.h" +#include +#include +#include #include "stg/common.h" #include "smux.h" +#include "utils.h" -bool WaitPackets(int sd); - -bool String2OI(const std::string & str, OBJECT_IDENTIFIER_t * oi) -{ -size_t left = 0, pos = 0, arcPos = 0; -int arcs[1024]; -pos = str.find_first_of('.', left); -if (pos == 0) - { - left = 1; - pos = str.find_first_of('.', left); - } -while (pos != std::string::npos) - { - int arc = 0; - if (str2x(str.substr(left, left - pos), arc)) - { - return false; - } - arcs[arcPos++] = arc; - left = pos + 1; - pos = str.find_first_of('.', left); - } -if (left < str.length()) - { - int arc = 0; - if (str2x(str.substr(left, left - pos), arc)) - { - return false; - } - arcs[arcPos++] = arc; - } -printfd(__FILE__, "String2OI() - arcPos: %d\n", arcPos); -OBJECT_IDENTIFIER_set_arcs(oi, arcs, sizeof(arcs[0]), arcPos); -return true; -} - -class SMUX_CREATOR -{ -private: - SMUX * smux; - -public: - SMUX_CREATOR() : smux(new SMUX()) {} - ~SMUX_CREATOR() { delete smux; } - - SMUX * GetPlugin() { return smux; } -}; - -SMUX_CREATOR sac; - -PLUGIN * GetPlugin() -{ -return sac.GetPlugin(); -} - -int SendOpenPDU(int fd) -{ -const char * description = "Stg SMUX Plugin"; -//int oid[] = {1, 3, 6, 1, 4, 1, 38313, 1, 5, 2, 1, 1}; -asn_enc_rval_t error; -OpenPDU_t msg; - -memset(&msg, 0, sizeof(msg)); - -msg.present = OpenPDU_PR_simple; -asn_long2INTEGER(&msg.choice.simple.version, SimpleOpen__version_version_1); -if (!String2OI(".1.3.6.1.4.1.38313", &msg.choice.simple.identity)) - { - printfd(__FILE__, - "SendOpenPDU() - failed to convert string to OBJECT_IDENTIFIER\n"); - return -1; - } -OCTET_STRING_fromString(&msg.choice.simple.description, description); -OCTET_STRING_fromString(&msg.choice.simple.password, ""); - -char buffer[1024]; -error = der_encode_to_buffer(&asn_DEF_OpenPDU, &msg, buffer, sizeof(buffer)); - -if (error.encoded == -1) - { - printfd(__FILE__, "Could not encode OpenPDU (at %s)\n", - error.failed_type ? error.failed_type->name : "unknown"); - return -1; - } -else - { - write(fd, buffer, error.encoded); - printfd(__FILE__, "OpenPDU encoded successfully to %d bytes\n", - error.encoded); - } -return 0; -} - -int SendClosePDU(int fd) +namespace { -ClosePDU_t msg; - -memset(&msg, 0, sizeof(msg)); - -asn_long2INTEGER(&msg, ClosePDU_goingDown); - -char buffer[1024]; -asn_enc_rval_t error; -error = der_encode_to_buffer(&asn_DEF_ClosePDU, &msg, buffer, sizeof(buffer)); -if (error.encoded == -1) - { - printfd(__FILE__, "Could not encode ClosePDU (at %s)\n", - error.failed_type ? error.failed_type->name : "unknown"); - return -1; - } -else - { - write(fd, buffer, error.encoded); - printfd(__FILE__, "ClosePDU encoded successfully\n"); - } -return 0; -} - -int SendRReqPDU(int fd) +bool SPrefixLess(const Sensors::value_type & a, + const Sensors::value_type & b) { -int oid[] = {1, 3, 6, 1, 4, 1, 38313, 1}; -asn_enc_rval_t error; -RReqPDU_t msg; - -memset(&msg, 0, sizeof(msg)); - -msg.priority = 0; -asn_long2INTEGER(&msg.operation, RReqPDU__operation_readOnly); -OBJECT_IDENTIFIER_set_arcs(&msg.subtree, - oid, - sizeof(oid[0]), - 8); - -char buffer[1024]; -error = der_encode_to_buffer(&asn_DEF_RReqPDU, &msg, buffer, sizeof(buffer)); - -if (error.encoded == -1) - { - printfd(__FILE__, "Could not encode RReqPDU (at %s)\n", - error.failed_type ? error.failed_type->name : "unknown"); - return -1; - } -else - { - write(fd, buffer, error.encoded); - printfd(__FILE__, "RReqPDU encoded successfully to %d bytes\n", - error.encoded); - } -return 0; +return a.first.PrefixLess(b.first); } -SMUX_PDUs_t * RecvSMUXPDUs(int fd) -{ -char buffer[1024]; -SMUX_PDUs_t * pdus = NULL; - -memset(buffer, 0, sizeof(buffer)); - -size_t length = read(fd, buffer, sizeof(buffer)); -if (length < 1) - return NULL; -asn_dec_rval_t error; -error = ber_decode(0, &asn_DEF_SMUX_PDUs, (void **)&pdus, buffer, length); -if(error.code != RC_OK) - { - printfd(__FILE__, "Failed to decode PDUs at byte %ld\n", - (long)error.consumed); - return NULL; - } -return pdus; } -int ParseIntInRange(const std::string & str, - int min, - int max, - int * val) +extern "C" STG::Plugin* GetPlugin() { -if (str2x(str.c_str(), *val)) - { - return -1; - } -if (*val < min || *val > max) - { - return -1; - } -return 0; + static SMUX plugin; + return &plugin; } SMUX_SETTINGS::SMUX_SETTINGS() @@ -220,31 +41,30 @@ SMUX_SETTINGS::SMUX_SETTINGS() 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"); @@ -257,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"); @@ -269,15 +89,22 @@ return 0; } SMUX::SMUX() - : PLUGIN(), - users(NULL), - tariffs(NULL), - running(false), + : users(nullptr), + tariffs(nullptr), + admins(nullptr), + services(nullptr), + corporations(nullptr), + traffcounter(nullptr), stopped(true), - sock(-1) + needReconnect(false), + lastReconnectTry(0), + reconnectTimeout(1), + sock(-1), + 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; @@ -290,11 +117,11 @@ pdusHandlers[PDUs_PR_set_request] = &SMUX::SetRequestHandler; SMUX::~SMUX() { -Sensors::iterator it; -for (it = sensors.begin(); it != sensors.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() @@ -304,33 +131,59 @@ return smuxSettings.ParseSettings(settings); int SMUX::Start() { +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[".1.3.6.1.4.1.38313.1.1.1"] = new TotalUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.2"] = new ConnectedUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.3"] = new AuthorizedUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.4"] = new AlwaysOnlineUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.5"] = new NoCashUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.7"] = new DisabledDetailStatsUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.8"] = new DisabledUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.9"] = new PassiveUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.10"] = new CreditUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.11"] = new FreeMbUsersSensor(*users); -sensors[".1.3.6.1.4.1.38313.1.1.12"] = new TariffChangeUsersSensor(*users); +sensors[OID(".1.3.6.1.4.1.38313.1.1.1")] = new TotalUsersSensor(*users); +sensors[OID(".1.3.6.1.4.1.38313.1.1.2")] = new ConnectedUsersSensor(*users); +sensors[OID(".1.3.6.1.4.1.38313.1.1.3")] = new AuthorizedUsersSensor(*users); +sensors[OID(".1.3.6.1.4.1.38313.1.1.4")] = new AlwaysOnlineUsersSensor(*users); +sensors[OID(".1.3.6.1.4.1.38313.1.1.5")] = new NoCashUsersSensor(*users); +sensors[OID(".1.3.6.1.4.1.38313.1.1.6")] = new DisabledDetailStatsUsersSensor(*users); +sensors[OID(".1.3.6.1.4.1.38313.1.1.7")] = new DisabledUsersSensor(*users); +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[".1.3.6.1.4.1.38313.1.2.1"] = new TotalTariffsSensor(*tariffs); - -if (!running) +sensors[OID(".1.3.6.1.4.1.38313.1.2.1")] = new TotalTariffsSensor(*tariffs); +// Admins +sensors[OID(".1.3.6.1.4.1.38313.1.3.1")] = new TotalAdminsSensor(*admins); +// Services +sensors[OID(".1.3.6.1.4.1.38313.1.4.1")] = new TotalServicesSensor(*services); +// Corporations +sensors[OID(".1.3.6.1.4.1.38313.1.5.1")] = new TotalCorporationsSensor(*corporations); +// Traffcounter +sensors[OID(".1.3.6.1.4.1.38313.1.6.1")] = new TotalRulesSensor(*traffcounter); + +// Table data +tables[".1.3.6.1.4.1.38313.1.2.2"] = new TariffUsersTable(".1.3.6.1.4.1.38313.1.2.2", *tariffs, *users); + +UpdateTables(); +SetNotifiers(); + +#ifdef SMUX_DEBUG +auto it = sensors.begin(); +while (it != sensors.end()) { - if (pthread_create(&thread, NULL, Runner, this)) - { - errorStr = "Cannot create thread."; - printfd(__FILE__, "Cannot create thread\n"); - return -1; - } + printfd(__FILE__, "%s = %s\n", + it->first.ToString().c_str(), + it->second->ToString().c_str()); + ++it; } +#endif + +if (!m_thread.joinable()) + m_thread = std::jthread([this](auto token){ Run(std::move(token)); }); return 0; } @@ -338,7 +191,7 @@ return 0; int SMUX::Stop() { printfd(__FILE__, "SMUX::Stop() - Before\n"); -running = false; +m_thread.request_stop(); if (!stopped) { @@ -346,55 +199,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(); + +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; } -void * SMUX::Runner(void * d) +int SMUX::Reload(const STG::ModuleSettings & /*ms*/) { -SMUX * smux = static_cast(d); - -smux->Run(); - -return NULL; +if (Stop() != 0) + return -1; +if (Start() != 0) + return -1; +if (!needReconnect) + { + printfd(__FILE__, "SMUX reconnected succesfully.\n"); + logger("Reconnected successfully."); + } +return 0; } -void SMUX::Run() +void SMUX::Run(std::stop_token token) { -SendOpenPDU(sock); -SendRReqPDU(sock); -running = true; +stopped = true; +if (!SendOpenPDU(sock)) + needReconnect = true; +if (!SendRReqPDU(sock)) + 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); @@ -408,6 +281,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; } @@ -418,9 +292,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; } @@ -428,42 +303,39 @@ if (connect(sock, reinterpret_cast(&addr), sizeof(addr))) return false; } -bool WaitPackets(int sd) +bool SMUX::Reconnect() { -fd_set rfds; -FD_ZERO(&rfds); -FD_SET(sd, &rfds); - -struct timeval tv; -tv.tv_sec = 0; -tv.tv_usec = 500000; +if (needReconnect && difftime(time(nullptr), lastReconnectTry) < reconnectTimeout) + return true; -int res = select(sd + 1, &rfds, NULL, NULL, &tv); -if (res == -1) // Error - { - if (errno != EINTR) - { - printfd(__FILE__, "Error on select: '%s'\n", strerror(errno)); - } - return false; - } +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 (res == 0) // Timeout - { - 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 { switch (pdus->present) @@ -482,5 +354,116 @@ else } asn_fprint(stderr, &asn_DEF_SMUX_PDUs, pdus); } +#endif return false; } + +bool SMUX::UpdateTables() +{ +Sensors newSensors; +bool done = true; +auto it = tables.begin(); +while (it != tables.end()) + { + try + { + it->second->UpdateSensors(newSensors); + } + catch (const std::runtime_error & ex) + { + printfd(__FILE__, + "SMUX::UpdateTables - failed to update table '%s': '%s'\n", + it->first.c_str(), ex.what()); + done = false; + break; + } + ++it; + } +if (!done) + { + auto sit = newSensors.begin(); + while (sit != newSensors.end()) + { + delete sit->second; + ++sit; + } + return false; + } + +it = tables.begin(); +while (it != tables.end()) + { + 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; + ++sit; + } + sensors.erase(res.first, res.second); + ++it; + } + +sensors.insert(newSensors.begin(), newSensors.end()); + +return true; +} + +void SMUX::SetNotifier(UserPtr userPtr) +{ +notifiers.emplace_back(*this, userPtr); +userPtr->GetProperties().tariffName.AddAfterNotifier(¬ifiers.back()); +} + +void SMUX::UnsetNotifier(UserPtr userPtr) +{ +auto it = notifiers.begin(); +while (it != notifiers.end()) + { + if (it->GetUserPtr() == userPtr) + { + userPtr->GetProperties().tariffName.DelAfterNotifier(&(*it)); + notifiers.erase(it); + break; + } + ++it; + } +} + +void SMUX::SetNotifiers() +{ +int h = users->OpenSearch(); +assert(h && "USERS::OpenSearch is always correct"); + +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); +} + +void SMUX::ResetNotifiers() +{ +tariffs->DelNotifierDel(&addDelTariffNotifier); +tariffs->DelNotifierAdd(&addDelTariffNotifier); + +users->DelNotifierUserDel(&delUserNotifier); +users->DelNotifierUserAdd(&addUserNotifier); + +auto it = notifiers.begin(); +while (it != notifiers.end()) + { + it->GetUserPtr()->GetProperties().tariffName.DelAfterNotifier(&(*it)); + ++it; + } +notifiers.clear(); +}