src/syncer.cpp

   1 #include <sys/select.h>
   2
   3 #include <curl/curl.h>
   4 #include <curl/easy.h>
   5
   6 #include <cassert>
   7 #include <algorithm>
   8 #include <exception>
   9
  10 #include <boost/bind.hpp>
  11
  12 #include "snmp_pp/snmp_pp.h"
  13
  14 #include "syncer.h"
  15 #include "settings.h"
  16 #include "switch.h"
  17 #include "subscriber.h"
  18 #include "datatypes.h"
  19 #include "logger.h"
  20
  21 using SSMD::Syncer;
  22 using SSMD::Timer;
  23 using SSMD::Switch;
  24 using SSMD::Subscriber;
  25 using SSMD::Lines;
  26
  27 Timer::Timer(boost::function<void ()> callback, time_t interval)
  28     : _interval(interval),
  29       _lastFire(0),
  30       _callback(callback)
  31 {
  32 }
  33
  34 Timer::~Timer()
  35 {
  36 }
  37
  38 int Timer::getTimeout() const
  39 {
  40     double delta = difftime(time(NULL), _lastFire);
  41     return _interval - delta;
  42 }
  43
  44 void Timer::fire()
  45 {
  46     _callback();
  47     time(&_lastFire);
  48 }
  49
  50 Syncer::Syncer(SettingsParser & sp,
  51                Snmp & snmp)
  52     : _settingsParser(sp),
  53       _snmp(snmp)
  54 {
  55     // 1 db syncer
  56     _timers.push_back(Timer(boost::bind(&Syncer::syncInfo, this),
  57                             _settingsParser.settings().infoSyncInterval()));
  58 }
  59
  60 Syncer::~Syncer()
  61 {
  62 }
  63
  64 void Syncer::run(const bool & running, bool & reload)
  65 {
  66     logger << "Syncer::run()" << std::endl;
  67     while (running) {
  68         if (wait()) {
  69             logger << "Syncer::run() - wait stopped by signal" << std::endl;
  70             if (!running)
  71                 break;
  72             if (reload) {
  73                 logger << "Syncer::run() - reload" << std::endl;
  74                 try {
  75                     _settingsParser.reloadConfig();
  76                 }
  77                 catch (std::exception & ex) {
  78                     logger << "Syncer::run() - exception: " << ex.what() << std::endl;
  79                 }
  80                 reload = false;
  81             }
  82         }
  83     }
  84 }
  85
  86 bool Syncer::wait()
  87 {
  88     Timer & timer(getNextTimer());
  89
  90     if (timer.getTimeout() > 0) {
  91         fd_set rfds;
  92
  93         FD_ZERO(&rfds);
  94
  95         struct timeval tv;
  96         tv.tv_sec = timer.getTimeout();
  97         tv.tv_usec = 0;
  98
  99         int retval = select(1, &rfds, NULL, NULL, &tv);
 100
 101         if (retval == -1) {
 102             return true;
 103         }
 104     }
 105
 106     timer.fire();
 107
 108     return false;
 109 }
 110
 111 Timer & Syncer::getNextTimer()
 112 {
 113     assert(_timers.size() && "Timer list must not be empty!");
 114     int timeout = _timers.begin()->getTimeout();
 115     std::list<Timer>::iterator it(_timers.begin());
 116     std::list<Timer>::iterator pos(_timers.begin());
 117     ++it;
 118     while (it != _timers.end()) {
 119         if (it->getTimeout() < timeout) {
 120             pos = it;
 121             timeout = pos->getTimeout();
 122         }
 123         ++it;
 124     }
 125     return *pos;
 126 }
 127
 128 void Syncer::syncInfo()
 129 {
 130     std::map<std::string, Switch> switches;
 131     if (!getSwitchesState(switches)) {
 132         logger << "Syncer::syncInfo() - failed to get new switch states" << std::endl;
 133         return;
 134     }
 135     std::list<TimedSwitch>::iterator it(_switches.begin());
 136     while (it != _switches.end()) {
 137         _timers.erase(it->second);
 138         _switches.erase(it++);
 139     }
 140     std::map<std::string, Switch>::const_iterator sit;
 141     for (sit = switches.begin(); sit != switches.end(); ++sit) {
 142         // Insert switch with no timer
 143         _switches.push_back(std::make_pair(sit->second, _timers.end()));
 144         // Insert timer for this switch
 145         TimerIterator tit = _timers.insert(
 146                 _timers.end(),
 147                 Timer(boost::bind(&Switch::sync, &_switches.back().first),
 148                       _settingsParser.settings().switchSyncInterval()));
 149         // Set timer iterator for this switch
 150         _switches.back().second = tit;
 151     }
 152     logger << "Syncer::syncInfo() - data synchronization successfull, switches: " << _switches.size() << std::endl;
 153 }
 154
 155 size_t curlWriteFunction(void * ptr, size_t size, size_t nmemb, void * userdata)
 156 {
 157     char * data = static_cast<char *>(ptr);
 158     std::string * dest = static_cast<std::string *>(userdata);
 159     dest->append(data, size * nmemb);
 160     return size * nmemb;
 161 }
 162
 163 bool Syncer::getDBData(std::string & data) const
 164 {
 165     CURL * handle = curl_easy_init();
 166     if (handle) {
 167         char errorBuffer[CURL_ERROR_SIZE];
 168         curl_easy_setopt(handle, CURLOPT_SSL_VERIFYPEER, 0); // Accept self-signed certs
 169         curl_easy_setopt(handle, CURLOPT_LOW_SPEED_LIMIT, 1); // Less than 1 bps
 170         curl_easy_setopt(handle, CURLOPT_LOW_SPEED_TIME, 60); // During 60 secs
 171         curl_easy_setopt(handle, CURLOPT_URL, _settingsParser.settings().dataURL().c_str()); // Our URL
 172         curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, curlWriteFunction); // Our write callback
 173         curl_easy_setopt(handle, CURLOPT_WRITEDATA, &data); // Our callback data
 174         curl_easy_setopt(handle, CURLOPT_ERRORBUFFER, errorBuffer); // Buffer for an error messages
 175         CURLcode res = curl_easy_perform(handle);
 176         if (res) {
 177             logger << "Syncer::getDBData() - DB communication error: '" << errorBuffer << "'" << std::endl;
 178             curl_easy_cleanup(handle);
 179             return false;
 180         }
 181         curl_easy_cleanup(handle);
 182         return true;
 183     }
 184     logger << "Syncer::getDBData() - failed to init CURL library" << std::endl;
 185     return false;
 186 }
 187
 188 bool Syncer::getSwitchesState(std::map<std::string, Switch> & switches)
 189 {
 190     if (_settingsParser.settings().dataURL().empty()) {
 191         logger << "Switch::getSwitchesState() - data URL is empty" << std::endl;
 192         return false;
 193     }
 194     std::string data;
 195     if (!getDBData(data)) {
 196         logger << "Syncer::getSwitchesState() - failed to fetch data from the URL: '" << _settingsParser.settings().dataURL() << "'" << std::endl;
 197         return false;
 198     }
 199     Lines lines;
 200     if (!parseData(data, lines)) {
 201         logger << "Syncer::getSwitchesState() - failed to parse data:\n" << data << std::endl;
 202         return false;
 203     }
 204     Lines::const_iterator it;
 205     for (it = lines.begin(); it != lines.end(); ++it) {
 206         std::pair<std::map<std::string, Switch>::iterator, bool> res(
 207             switches.insert(
 208                 std::make_pair(
 209                     it->switchIP,
 210                     Switch(
 211                         _settingsParser.settings(),
 212                         _snmp,
 213                         it->switchIP,
 214                         it->readCommunity,
 215                         it->writeCommunity,
 216                         it->uplinkPort
 217                     )
 218                 )
 219             )
 220         );
 221         res.first->second.addSubscriber(
 222             Subscriber(
 223                 it->mac,
 224                 it->userPort,
 225                 it->upShape,
 226                 it->downShape,
 227                 it->upBurst,
 228                 it->downBurst
 229             )
 230         );
 231     }
 232     return true;
 233 }