WindMIll Data Publisher using observer pattern

Question

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

1. 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]

1. 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.

2. 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.

3. Are the Event type and other enums for handling events and data used properly. : Use shared pointers and take care of typos

4. 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.

Answer 1

Here are some things that may help you improve your program.

Compile with warnings turned on

When I attempted to compile the project, it told me that this line was suspect:

if ((*clientitr)->handle = &observer)
{
    clientitr = clients.erase(clientitr);
}

Indeed, looking at the context, the condition within the if should be == instead of =. Your compiler can help you spot subtle bugs, so you should get into the habit of turning the warning levels all the way up.

Ensure every control path returns a proper value

The WindMill_Publisher::getCityData() routine returns a reference to a Windata structure under some set of conditions but then doesn't return anything at all otherwise. This is an error. The function must always return a valid reference. If it is not always possible, then the code should be changed to handle the error, as with returning a pointer and then having the caller check for nullptr.

Separate interface from implementation

The interface is the part in the .h file and the implementation is in the .cpp file. Users of this code should be able to read and understand everything they need from the implementation file. That means that, for example, the Subject.h file should be separated into two pieces: the interface (only) and the implementation (only). Having all of the code in the .h file defeats much of the purpose of having header files at all.

Fix spelling errors

The code has ClientDetials instead of ClientDetails and Banglore instead of Bangalore. These kinds of typos don't bother the compiler at all, but will bother human readers of the code and make it a little more difficult to understand and maintain.

Prefer throw rather than printing to cout

The current code includes this line within WindMill_Publisher::readCityData():

if(!res)
{
    std::cout << " unable to load xml file " << static_cast<int>(res.status)<<"\n";
    return false;
}

There are a few problems with this. First, the return value is never checked within the constructor, so the result is that the data may or may not have been correctly initialized. The second problem is that it's probably better and easier to simply throw an error from within the function. This helps assure that the created WindMill_Publisher object is valid if no exceptions were thrown.

Don't repeat yourself

The only difference between Client_1 and Client_2 is the data contained, so they should be different instances but the same class. This can easily be done by passing the required data in via a constructor. Even if your compiler doesn't support initializer lists, you can create an external function to allow adding of cities to the object.

Be careful with concurrency

As written, the code passes a reference to an Observer class to the registerClient class and then a pointer to that reference is stored within WindMill_Publisher. That's a serious problem because there is nothing to prevent the passed object from getting deleted after registration. The result is that the callback will attempt to reference a deleted object's member pointer which is not likely to be what you want. Instead, either create a copy and store that or use a std::shared_ptr.

Don't overcomplicate the design

There really isn't much need for the Observer class. All that's really required for the callback is a function pointer. This could be done just as well by passing in a freestanding function. It seems to me that this would make for a somewhat more flexible interface.

Consider a different approach

I find the code for the WindMill_Publisher to be much more complex than it needs to be. Fundamentally, there are a few different things happening. There is first, a data structure that holds current city data and this data structure is asynchronously updated from an external XML file. Next, there is the publisher of this code which sends updated data to each registered client. I would probably want to separate these a little more cleanly. Further, rather than waking up periodically to check the timers, this is a good candidate for a priority queue.

Here's how that might work. Calculate the deadlines for all notifications and store them in a priority queue. Then set the alarm for the item at the front of the queue. Example: if there are three items with deadlines that are 2, 5 and 7 seconds away, have the application sleep for 2 seconds. When it wakes, it handles the item at the top of the queue and recalculates. The next item is due 3 seconds later, so it sleeps for 3 seconds, etc. The queue then only needs to be adjusted when either a node is registered or unregistered and when the application wakes up. To avoid sleeping and then waking immediately, process the queue until there are no more items for the current second.

Omit return 0

When a C or C++ program reaches the end of main the compiler will automatically generate code to return 0, so there is no need to put return 0; explicitly at the end of main.

Note: when I make this suggestion, it's almost invariably followed by one of two kinds of comments: "I didn't know that." or "That's bad advice!" My rationale is that it's safe and useful to rely on compiler behavior explicitly supported by the standard. For C, since C99; see ISO/IEC 9899:1999 section 5.1.2.2.3:

[...] a return from the initial call to the main function is equivalent to calling the exit function with the value returned by the main function as its argument; reaching the } that terminates the main function returns a value of 0.

For C++, since the first standard in 1998; see ISO/IEC 14882:1998 section 3.6.1:

If control reaches the end of main without encountering a return statement, the effect is that of executing return 0;

All versions of both standards since then (C99 and C++98) have maintained the same idea. We rely on automatically generated member functions in C++, and few people write explicit return; statements at the end of a void function. Reasons against omitting seem to boil down to "it looks weird". If, like me, you're curious about the rationale for the change to the C standard read this question. Also note that in the early 1990s this was considered "sloppy practice" because it was undefined behavior (although widely supported) at the time.

So I advocate omitting it; others disagree (often vehemently!) In any case, if you encounter code that omits it, you'll know that it's explicitly supported by the standard and you'll know what it means.