SubscriptionParameter.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/RDADeviceServer/SubscriptionParameter.h>

#include <fesa-core/RDADeviceServer/SubscriptionFilter.h>
#include <fesa-core/Server/Property.h>
#include <fesa-core/DataStore/AbstractDevice.h>
#include <fesa-core/Diagnostic/Diagnostics.h>

#include <cmw-log/Logger.h>


namespace
{

const std::string diagTopic = fesa::DiagnosticUtils::rdaTrackingStr;

} // namespace


namespace fesa
{

SubscriptionParameter::SubscriptionParameter(const std::string& paramName, AbstractDevice& device,
                                             Property& property, bool useFastUpdate,
                                             NotificationThread* pNotificationThread) :
    name_(paramName),
    device_(device),
    property_(property),
    useFastUpdate_(useFastUpdate),
    pNotificationThread_(pNotificationThread)
{
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    std::ostringstream traceStrStream;
    traceStrStream << "Creating subscription parameter with name '" << name_ << "'";
    diagMsg.msg = traceStrStream.str();
    diagnostics->log(diagTopic, diagMsg);
}

SubscriptionParameter::~SubscriptionParameter()
{
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    std::ostringstream traceStrStream;
    traceStrStream << "Deleting subscription parameter with name '" << name_ << "'";
    diagMsg.msg = traceStrStream.str();
    diagnostics->log(diagTopic, diagMsg);
    SubscriptionFilterItr itr = filters_.begin();
    SubscriptionFilterItr itrEnd = filters_.end();
    filters_.erase(itr, itrEnd);
}

void SubscriptionParameter::addSubscriber(const std::string& cycleSelector, const rdaData& newFilter, rdaValueChangeListener* subscriber)
{
    // Adding a new subscriber, which correspond to a monitorOn request, consist of two phases:
    // 1) Register the new client
    // 2) Provide the client with a firstUpdate.

    //1) Register the new Subscriber
    //find matching filter
    SubscriptionFilterItr itr;
    for (itr = filters_.begin(); itr != filters_.end(); itr++) // iterate through all the registered SubscriptionFilter
    {
        if ((*itr)->matchFilter(newFilter)) // request the filter to proceed to a comparison with the newFilter
        {
            //register
            (*itr)->addSubscriber(cycleSelector, subscriber);
            break;
        }
    }
    if (itr == filters_.end())
    {
        // No registered SubscriptionFilter is matching the new one.
        // Create the new filter
        boost::shared_ptr<AbstractSubscriptionFilter> sf;
        sf.reset((AbstractSubscriptionFilter*)new SubscriptionFilter(device_, property_, newFilter, useFastUpdate_, pNotificationThread_));

        // add subscriber
        sf->addSubscriber(cycleSelector, subscriber);
        // insert the new filter in the collection
        filters_.push_back(sf);
        //Make itr pointing to the last element freshly added
        itr = filters_.end();
        --itr;
    }

    //2) Provide the firstUpdate: if the property is a Setting and no cycleSelector is specified and device/property is multiplexed
    // firstUpdate send back one by one the current value of all cycles.
    // itr points to the right subscription Filter
    if (cycleSelector.empty() && property_.getType()==SETTING && device_.isMultiplexed() && property_.isMultiplexed())
    {//we have to return a value for EACH cycle !
        (*itr)->allCyclesFirstUpdate(newFilter, subscriber);
    }
    else //standard first update
    {
        (*itr)->cycleFirstUpdate(cycleSelector, newFilter, subscriber);
    }
}

void SubscriptionParameter::removeSubscriber(const std::string& cycleSelector, rdaValueChangeListener* subscriber, bool& isEmpty)
{
    SubscriptionFilterItr itr;
    for (itr = filters_.begin(); itr != filters_.end(); ++itr) // iterate through all the registered SubscriptionFilter
    {
        bool isFilterEmpty = false; // tells if the SubscriptionFilter is empty
        bool toBeDeleted = true;    // tells if the SubscriptionFilter can be deleted in case it is in use by the NotificationThread
        if ((*itr)->removeSubscriber(cycleSelector, subscriber, isFilterEmpty, toBeDeleted))
        {
            if (isFilterEmpty)
            {
                filters_.erase(itr);
            }
            break;
        }
    }
    // set isEmpty
    isEmpty = filters_.empty();
}

void SubscriptionParameter::notify(MultiplexingContext& muxContext, RequestType& reqType,
                                   std::vector<std::string>& validCS)
{
    SubscriptionFilterItr itr = filters_.begin();
    SubscriptionFilterItr itrEnd = filters_.end();
    for (; itr != itrEnd; ++itr)
    {
        (*itr)->notify(muxContext, reqType, validCS);
    }
}

void SubscriptionParameter::reportInternalError(const std::string& errorCategory, int32_t errorCode, const std::string&  message, bool changed)
{
    SubscriptionFilterItr itr;
    for (itr = filters_.begin(); itr != filters_.end(); ++itr) // iterate through all the registered SubscriptionFilter
    {
        (*itr)->reportInternalError(errorCategory, errorCode, message, changed);
    }
}

void SubscriptionParameter::printDebugState(std::stringstream& str)
{
    str << "\t\tParam: " << name_ << "\n";
    SubscriptionFilterItr itr = filters_.begin();
    SubscriptionFilterItr itrEnd = filters_.end();
    for (; itr != itrEnd; ++itr)
    {
        (*itr)->printDebugState(str);
    }
}

} // fesa