SubscriptionFilter.h

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#ifndef SUBSCRIPTION_FILTER_H_
#define SUBSCRIPTION_FILTER_H_

#include <fesa-core/Core/FesaDefs.h>

#include <cmw-rda/Data.h>

#include <string>
#include <vector>
#include <map>
#include <sstream>
#include <boost/shared_ptr.hpp>

class rdaValueChangeListener;

namespace fesa
{

class AbstractDevice;
class Property;
class MultiplexingContext;
class NotificationThread;

class AbstractSubscriptionFilter
{
  public:
    AbstractSubscriptionFilter(AbstractDevice& device, Property& property, const rdaData& newFilter, bool useFastUpdate);

    virtual ~AbstractSubscriptionFilter();

    virtual void addSubscriber(const std::string& cycleSelector, rdaValueChangeListener* subscriber);

    virtual bool removeSubscriber(const std::string& cycleSelector, rdaValueChangeListener* subscriber, bool& isEmpty, bool& toBeDeleted);

    bool matchFilter(const rdaData& filter);

    virtual void notify(MultiplexingContext& muxContext, RequestType& reqType, std::vector<std::string>& validCS)  = 0;

    void cycleFirstUpdate(const std::string& cycleSelector, const rdaData& newFilter, rdaValueChangeListener* l);

    void allCyclesFirstUpdate(const rdaData& newFilter, rdaValueChangeListener* l);

    void reportInternalError(const std::string& category, int32_t errorCode, const std::string& message, bool changed);

    void cancelInternalError(const std::string& message, bool changed);

    void virtual printDebugState(std::stringstream& str);

  protected:

    AbstractDevice& device_;

    Property& property_;

    rdaData filter_;

    bool useFastUpdate_;

    struct FesaListener
    {
        // Constructor to initialize members
        FesaListener(rdaValueChangeListener* l) : subscriber_(l), updateStamp_(0), reportErrorOnce_(false) {};

        ~FesaListener() {};
        //the rda listener itself
        rdaValueChangeListener* subscriber_;

        //the time of the last update
        int64_t updateStamp_;

        //tells if the client never received any error in case we are still
        //reporting the same error
        bool reportErrorOnce_;
    };

    typedef std::vector<boost::shared_ptr<FesaListener> > ListenerVector;
    typedef ListenerVector::iterator LstnrItr;

    typedef std::map<std::string, boost::shared_ptr<ListenerVector> > ListenerMap;
    typedef ListenerMap::iterator LstnrMapItr;
    ListenerMap listeners_;

};


class SubscriptionFilter: public AbstractSubscriptionFilter
{
  public:
    SubscriptionFilter(AbstractDevice& device, Property& property, const rdaData& newFilter, bool useFastUpdate, NotificationThread* pNotificationThread);

    ~SubscriptionFilter();

    void addSubscriber(const std::string& cycleSelector, rdaValueChangeListener* subscriber);

    bool removeSubscriber(const std::string& cycleSelector, rdaValueChangeListener* subscriber, bool& isEmpty, bool& toBeDeleted);

    void notify(MultiplexingContext& muxContext, RequestType& reqType, std::vector<std::string>& validCS);

    void doUpdate(MultiplexingContext& muxContext, RequestType& reqType, std::vector<std::string>& validCS);

  private:

    NotificationThread* pNotificationThread_;

    bool cancelled_;

    uint32_t pendingNotification_;
};

} // fesa

#endif // SUBSCRIPTION_FILTER_H_