Problem statement : There is a windmill which collects and stores wind data for various cities. The data is maintained in an xml file by the aggregator part of windmill. Now I am trying to implement the publisher part of the mill. The responsibility of publisher is to notify the registered clients for their respective subscribed cities at registered interval of time for ex :
a. Client 1 registers for city DELHI and NOIDA at a refresh rate of 5 sec.
b. client 2 registers for city BANGLORE and MUMBAI at refresh rate of 3 sec.
The minimum refresh rate can be 1 Sec. The publisher also reads/receives the data from the xml file at rate of 100 ms. The publisher should notify clients about wind speed and wind direction.
Lets say the format of xml in which the data is stored is :
<CityList>
<City name="Delhi" WindSpeed="50" WindDirection="2"/>
<City name="Noida" WindSpeed="43" WindDirection="1"/>
<City name="Banglore" WindSpeed="54" WindDirection="3"/>
<City name="Mumbai" WindSpeed="21" WindDirection="2"/>
</CityList>
Where name is the name of city ,WindSpeed is the speed of wind and windDirection can 0,1,2,3 denoting NE,NW,SE,SW direction of wind.
My Solution : This seems to me to be an observer pattern type problem, Hence I took the help and overview of observer pattern from stackexchange answer.
The observer class :
EventType Denotes the city for which the WindData [ Event ] is being sent.
#ifndef _OBSERVER_
#define _OBSERVER_
struct Windata;
enum class EventType
{
DELHI,
NOIDA,
BANGLORE,
MUMBAI
};
class Observer
{
public:
virtual void onEvent(EventType const &,Windata const &) = 0;
};
#endif
Also for the periodic timer the implementation is taken from this link of stackoverflow.
#include <functional>
#include <chrono>
#include <future>
#include <atomic>
class CallBackTimer
{
public:
CallBackTimer()
:_execute(false)
{}
~CallBackTimer() {
if (_execute.load(std::memory_order_acquire)) {
stop();
};
}
void stop()
{
_execute.store(false, std::memory_order_release);
if (_thd.joinable())
_thd.join();
}
template <class callable, class... arguments>
void start(int interval, callable&& f, arguments&&... args)
{
if (_execute.load(std::memory_order_acquire)) {
stop();
};
_execute.store(true, std::memory_order_release);
std::function<typename std::result_of<callable(arguments...)>::type()> task(std::bind(std::forward<callable>(f), std::forward<arguments>(args)...));
_thd = std::thread([this, interval, task]()
{
while (_execute.load(std::memory_order_acquire)) {
std::this_thread::sleep_for(std::chrono::milliseconds(interval));
task();
}
});
}
bool is_running() const {
return (_execute.load(std::memory_order_acquire) &&
_thd.joinable());
}
private:
std::atomic<bool> _execute;
std::thread _thd;
};
The Wind Mill [ subject ] :
#ifndef _SUBJECT_
#define _SUBJECT_
#include "observer.h"
#include "Event.h"
#include "Timer.h"
#include "pugixml.hpp"
#include <vector>
#include <iostream>
#include <memory>
#include <map>
#include <string>
#include <mutex>
enum class WindDirection
{
NE,
NW,
SE,
SW
};
struct Windata
{
int speed;
WindDirection dir;
};
class WindMill_Publisher
{
struct ClientDetials
{
std::vector<EventType> cities;
Observer * handle{ nullptr };
long refresh_interval{ 0 };
long saved_refresh_interval{ 0 };
};
std::vector <std::unique_ptr<ClientDetials>> clients;
std::map<EventType, Windata> cityData;
CallBackTimer refresh_timer;
int client_refesh_rate = 1000;//1 sec
std::map<std::string, EventType> citycodemap;
CallBackTimer update_timer;
int update_refresh_rate = 100;//100ms
std::mutex cityMutex;
public:
WindMill_Publisher()
{
setCityCodes();
readCityData();
update_timer.start(update_refresh_rate, std::bind(&WindMill_Publisher::readCityData, this));
}
void registerClient(Observer& observer, std::vector<EventType> cities,long refresh_interval_sec)
{
for (auto const & myclient : clients) {
if (myclient->handle == &observer)
throw std::runtime_error("client already registered");
}
std::unique_ptr<ClientDetials> newclient(new ClientDetials());
newclient->handle = &observer;
newclient->cities = cities;
newclient->refresh_interval = refresh_interval_sec;
newclient->saved_refresh_interval = refresh_interval_sec;
clients.push_back(std::move(newclient));
if (clients.size() == 1)
{
refresh_timer.start(client_refesh_rate, std::bind(&WindMill_Publisher::notifyClients, this));
}
}
void unregisterClient(Observer& observer)
{
auto clientitr = clients.begin();
while (clientitr != clients.end())
{
if ((*clientitr)->handle = &observer)
{
clientitr = clients.erase(clientitr);
}
else
{
++clientitr;
}
}
if (clients.size() == 0)
{
refresh_timer.stop();
}
}
private:
Windata& getCityData(EventType const & city)
{
std::lock_guard<std::mutex> lock(cityMutex);
auto find = cityData.find(city);
if (find != cityData.end())
return find->second;
}
void notifyClients()
{
for (auto const & myclient : clients) {
if (myclient->refresh_interval <= 0)
{
for (auto myclientcities : myclient->cities)
{
if (myclient->handle)
myclient->handle->onEvent(myclientcities, getCityData(myclientcities));
}
myclient->refresh_interval = myclient->saved_refresh_interval;
}
myclient->refresh_interval -= (client_refesh_rate/ 1000);
}
}
bool readCityData()
{
pugi::xml_document doc;
std::string path = "...\\Citylist.xml";
pugi::xml_parse_result res = doc.load_file(path.c_str());
if(!res)
{
std::cout << " unable to load xml file " << static_cast<int>(res.status)<<"\n";
return false;
}
pugi::xml_node cities = doc.child("CityList");
for (pugi::xml_node city = cities.child("City"); city; city = city.next_sibling("City"))
{
std::string name = city.attribute("name").value();
int speed = city.attribute("WindSpeed").as_int();
int dir = city.attribute("WindDirection").as_int();
auto findcity = citycodemap.find(name);
if (findcity != citycodemap.end())
{
Windata data;
data.speed = speed;
data.dir = static_cast<WindDirection>(dir);
updateCityData(findcity->second, data);
}
}
return true;
}
void updateCityData(EventType city, Windata data)
{
std::lock_guard<std::mutex> lock(cityMutex);
cityData[city] = data;
}
void setCityCodes()
{
citycodemap["Delhi"] = EventType::DELHI;
citycodemap["Noida"] = EventType::NOIDA;
citycodemap["Banglore"] = EventType::BANGLORE;
citycodemap["Mumbai"] = EventType::MUMBAI;
}
};
#endif
Explanation about windmill : It contains
- A vector of struct ClientDetials which in turn hold the vector of EventType [cities] for which client wants notification, a handle to client and refresh interval. There are two refresh interval which i will explain in a moment.
2.A map of cityData which hold latest updated data for all the cities
3.It has two periodic timers refresh_timer and update_timer with period of 1 sec and 100 ms respectively.
The refresh_timer has period of 1 sec, and at every cycle it checks weather the refresh_interval of any client has become 0. if it becomes zero it notifies the client and update the refresh_interval with saved_refresh_interval.
Another approach in my mind was to have an separate periodic timer for each client but that seemed like much overhead. I am not sure but i chose this way.
I also tried to use std::type_index
as key but was unable to implement it correctly.
The update_timer is for reading the updated data from xml file. For reading the data from xml pugixml is used.
The client or test :
#include <iostream>
#include <vector>
#include "Observer.h"
#include "Subject.h"
class Client_1 :public Observer
{
public:
std::vector<EventType> cities;
Client_1(){
cities.push_back(EventType::DELHI);
cities.push_back(EventType::NOIDA);
}
void onEvent(EventType const &city,Windata const &data)
{
std::cout << "Client_1 called for city : " << static_cast<int>(city) << " with wind speed : " << data.speed << " and direction : " << static_cast<int>(data.dir) << std::endl;
}
};
class Client_2 :public Observer
{
public:
std::vector<EventType> cities;
Client_2(){
cities.push_back(EventType::BANGLORE);
cities.push_back(EventType::MUMBAI);
}
void onEvent(EventType const &city, Windata const &data)
{
std::cout << "Client_2 called for city : " << static_cast<int>(city) << " with wind speed : " << data.speed << " and direction : " << static_cast<int>(data.dir) << std::endl;
}
};
int main()
{
Client_1 c1;
Client_2 c2;
WindMill_Publisher mill;
try
{
mill.registerClient(c1, c1.cities, 5);
mill.registerClient(c2, c2.cities, 3);
}
catch (std::runtime_error &e)
{
std::cout << e.what()<<std::endl;
}
while (1)
{
}
return 0;
}
The cities vector in client is public just for the ease of testing. Also VS2013 was not supporting the list initialization ,so i had to initialize the vector in the constructor.
I have certain doubts like : [In bold are the suggestion as per @Edward]
Is it too much overhead for publisher class. The timer management and to notify client at refresh_rate should be delegated too another class. : Yes , may be use priority queue.
Should I use separate timer for every client or is there a better way to do it without timers at all. : The timer sleep and wake up time can managed using priority queue rather than every 1 sec.
Are the Event type and other enums for handling events and data used properly. : Use shared pointers and take care of typos
Is it better approach to have a function pointer from client for callback [as in the stackexchange link] rather than having client overload pure virtual function. : Yes, function pointers approach seems to be a better design
Thanks.