SubscriptionTreeManager.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

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

#include <fesa-core/RDADeviceServer/Notification.h>
#include <fesa-core/RDADeviceServer/SubscriptionParameter.h>
#include <fesa-core/Server/Property.h>
#include <fesa-core/DataStore/AbstractDevice.h>
#include <fesa-core/Core/MessageTypes.h>
#include <fesa-core/Core/NotificationIDManager.h>
#include <fesa-core/Core/NotificationID.h>
#include <fesa-core/Utilities/Lock.h>

#include <cmw-rda/ValueChangeListener.h>
#include <cmw-log/Logger.h>

#include <boost/lexical_cast.hpp>

#include <iostream>

namespace
{

CMW::Log::Logger& logger = CMW::Log::LoggerFactory::getLogger("FESA.FWK.fesa-core.RDADeviceServer.SubscriptionTreeManager");

static fesa::Mutex singleMutex;

}

namespace fesa
{

SubscriptionTreeManager::SubscriptionTreeManager(bool useFastUpdate) :
    initializationSucessfull_(false), lastErrorCode_(0), useFastUpdate_(useFastUpdate)
{

}

SubscriptionTreeManager::~SubscriptionTreeManager()
{
    //Shotdown all Notifications
    std::map<uint32_t, boost::shared_ptr<Notification> >::iterator iterN;
    for (iterN = notifications_.begin(); iterN != notifications_.end(); iterN++)
    {
        notifications_.erase(iterN);
    }

    // Remove all the subscriptions
    std::map<std::string, boost::shared_ptr<SubscriptionParameter> >::iterator iterS;
    for (iterS = subscriptions_.begin(); iterS != subscriptions_.end(); iterS++)
    {
        subscriptions_.erase(iterS);
    }
}

//Adding a new subscriber consist to 5 steps:
// -1) Check if the notification are properly initialized.
// -2) Create the parameterName computed form deviceName and propertyname
// -3) Insert the new SubscriptionParameter in the subscriptionParameter collection
// -4) Register the new SubscriptionCycle in the SubscriptionCycle collection
// -5) Iterate over the Notification collection and if the new SubscriptionParameter is controlled by a Notification insert it

void SubscriptionTreeManager::addSubscriber(const std::string& className, AbstractDevice& device, Property& property, const std::string& cycleSelector,
                                            const rdaData& filter, rdaValueChangeListener* subscriber)
{
    // add the new subscriber to the tree: Inserting new element in the subscriptionTree requires to manage the concurrent access.
    // TODO for the time being the tree is fully locked during the whole insertion operation with the consequence of preventing the notification
    // execution. Investigate if this could be refine in order to reduce as much as possible the time between the lock acquisition and the release.
    // Add new subscriber ....lock the SubscriberParameter Collection  (unlock is done automatically)

    // TODO: order the subscribers using an additional level: className
    // ************ critical section: automatic unlock*************
    Lock l(treeMutex_);

    if (!initializationSucessfull_)
    {
        initNotifications();
    }

    std::string paramName = device.getName() + std::string("#") + property.getName();
    boost::shared_ptr<SubscriptionParameter> sp = addSubscriptionParam(paramName, device, property, cycleSelector, filter, subscriber);
    subscriptionCycle_.addCycle(cycleSelector);

    // addSubscriptionParam returns a SubscriptionParameter* only if a new SubscriptionParameter is created.
    // Otherwise, returns NULL
    NotificationItr itr = notifications_.begin();
    NotificationItr itrEnd = notifications_.end();
    for(; itr != itrEnd; ++itr)
    {
        if((*itr).second->contains(device.getName(), property.getName()))
        {
            if (sp.get() != NULL) // it's a new subscritionParameter
                (*itr).second->addSubscriber(sp);
            (*itr).second->addCycle(cycleSelector);
        }
    }

    if (subscriber->isOnChange() && !property.isOnChange())
    {
        std::string message("Warning: Subscription OnChange is not supported by this property");
        subscriber->ioFailed(rdaIOError("FESA_RDADeviceServer", 0, message.c_str()));
    }
    // Print the SubscriptionTree while adding new subscriber
    if (logger.isLoggable(CMW::Log::Level::LL_DEBUG))
    {
        std::string s("add parameter: ");
        s.append(paramName);
        s.append(" listener: ");
        char buf[16];
        snprintf(buf, 16, "%p", subscriber);
        s.append(buf);
        printState(s);
    }
    //******** end of critical section **********
}

void SubscriptionTreeManager::removeSubscriber(AbstractDevice& device, Property& property, const std::string& cycleSelector,
                                               rdaValueChangeListener* subscriber)
{
    // remove a subscriber from the tree: Inserting new element in the subscriptionTree requires to manage the concurrent access.
    // TODO for the time being the tree is fully locked during the whole insertion operation with the consequence of preventing the notification
    // execution. Investigate if this could be refine in order to reduce as much as possible the time between the lock acquisition and the release.
    // remove subscriber ....lock the SubscriberParameter Collection  (unlock is done automatically)
    Lock l(treeMutex_);
    std::string paramName = device.getName() + std::string("#") + property.getName();
    SubscriptionItr subParamItr = subscriptions_.find(paramName);
    if (subParamItr != subscriptions_.end())
    {
        // output parameter saying if the subscriptionParameter isEmpty (means no listeners anymore)
        // it has to be deleted from Notification
        bool isEmpty = false;
        (*subParamItr).second->removeSubscriber(cycleSelector, subscriber, isEmpty);
        // remove from NotificationID
        NotificationItr notifItr = notifications_.begin();
        NotificationItr notifItrEnd = notifications_.end();
        {
            for(; notifItr != notifItrEnd; ++notifItr)
            {
                (*notifItr).second->removeSubscriber((*subParamItr).second, cycleSelector, isEmpty);
            }
        }
        subscriptionCycle_.removeCycle(cycleSelector);
        if (isEmpty)
        {
            boost::shared_ptr<SubscriptionParameter> sp = (*subParamItr).second;
            subscriptions_.erase(subParamItr);
        }
    }
    else
    {
        //TODO log error "found no such subscriptionParameter"
    }
    // Print the SubscriptionTree while removing subscriber
    if (logger.isLoggable(CMW::Log::Level::LL_DEBUG))
    {
        std::string s("remove parameter: ");
        s.append(paramName);
        s.append(" listener: ");
        char buf[16];
        snprintf(buf, 16, "%p", subscriber);
        s.append(buf);
        printState(s);
    }
}

void SubscriptionTreeManager::notify(uint32_t IDKey, MultiplexingContext& muxContext,
                                     RequestType& reqType)
{
    // lock the SubscriberParameter Collection  (unlock is done automatically)
    Lock l(treeMutex_);
    // try to initialize the tree
    if (!initializationSucessfull_)
    {
        initNotifications();
    }
    // Locate the Notification Instance
    NotificationItr itr = notifications_.find(IDKey);
    if(itr != notifications_.end())
    {

        (*itr).second->notify(muxContext, reqType);
        if (useFastUpdate_) {
            const rdaData emptyData;
            rdaValueChangeListener::flush(emptyData);
        }
    }
    else
    {
        std::ostringstream msg;
        msg << "notify error, IDKey not found " << boost::lexical_cast<std::string>(IDKey);
        LOG_ERROR_IF(logger, msg.str());
    }
}

void SubscriptionTreeManager::notify(ManualNotificationMessage* msg, RequestType& reqType)
{
    std::vector<std::string> devices;
    std::vector<std::string> properties;

    std::vector<std::string> tmp;


    // Check if any cycleSelector is matching the current context
    std::vector<std::string> validCS;
    // fill the validCS by calling matchContext
    subscriptionCycle_.matchContext(*(msg->muxContext_), validCS);


    if (validCS.size() != 0) // At least one registered cycleSelector is matching the current context
    {
        try {
            // critical section
            Lock l(treeMutex_);
            for (std::map<std::string,std::vector<std::string> >::iterator pADItr = msg->propertiesAndDevicesCol_.begin(); pADItr != msg->propertiesAndDevicesCol_.end(); ++pADItr)
            {
                for (std::vector<std::string>::iterator devIter = pADItr->second.begin(); devIter != pADItr->second.end(); ++devIter)
                {
                    std::string paramName = *devIter + std::string("#") + pADItr->first;
                    SubscriptionItr subParamItr = subscriptions_.find(paramName);
                    if (subParamItr != subscriptions_.end())
                    {
                        (*subParamItr).second->notify(*(msg->muxContext_), reqType, validCS);
                    }
                }
            }
        }
        catch(FesaException& ex)
        {

        }
        if (useFastUpdate_) {
            const rdaData emptyData;
            rdaValueChangeListener::flush(emptyData);
        }
    }
}

void SubscriptionTreeManager::reportInternalError(const std::string& errorCategory, int32_t errorCode, const std::string&  message)
{
    // critical section
    Lock l(treeMutex_);
    bool changed = (errorCode == lastErrorCode_) ? false : true;
    SubscriptionItr subItr = subscriptions_.begin();
    SubscriptionItr subItrEnd = subscriptions_.end();
    for(; subItr != subItrEnd; ++subItr)
    {
        (*subItr).second->reportInternalError(errorCategory, errorCode, message, changed);
    }
    lastErrorCode_ = errorCode;
}

boost::shared_ptr<SubscriptionParameter> SubscriptionTreeManager::addSubscriptionParam(const std::string& paramName, AbstractDevice& device,
                                                                                       Property& property, const std::string& cycleSelector,
                                                                                       const rdaData& filter, rdaValueChangeListener* subscriber)
{
    boost::shared_ptr<SubscriptionParameter> sp((SubscriptionParameter*)NULL);
    SubscriptionItr itr = subscriptions_.find(paramName);
    if(itr != subscriptions_.end())
    {
        (*itr).second->addSubscriber(cycleSelector, filter, subscriber);
        // Returns NULL because it's not a new SubscriptionParameter
    }
    else
    {
        NotificationThread* pNotificationThread = &(notificationThreadFactory_.getThread(property.getClassName(), property.getNotificationThreadKey()));

        // Create a new SubscriptionParameter: is not necessary to lock the SubscriberParameter collection for the time being
        sp.reset(new SubscriptionParameter(paramName, device, property, useFastUpdate_, pNotificationThread));
        // Add subscriber
        sp->addSubscriber(cycleSelector, filter, subscriber);
        // Register : lock the SubscriptionParameter collection (unlock is done automatically)
        subscriptions_[paramName] = sp;
    }
    //Never executed
    return sp;
}

void SubscriptionTreeManager::initNotifications()
{
    NotificationIDManager* idGenerator = NotificationIDManager::getInstance();
    const std::map<uint32_t, NotificationID*>& idsCol = idGenerator->getNotificationIDCol();
    // TODO : check addGlobalDevice
    //        addGlobalDevicesToNotificationIDCollection(idsCol);

    std::map<uint32_t, NotificationID*>::const_iterator notIter;
    for (notIter = idsCol.begin(); notIter != idsCol.end(); notIter++)
    {
        //generate a Notification for each ID
        boost::shared_ptr<Notification> notif(new Notification(*((*notIter).second)));
        notifications_.insert(std::pair<uint32_t, boost::shared_ptr<Notification> >((*notIter).second->getIDKey(), notif));
    }
    initializationSucessfull_ = true;

    // TODO: for the time being this feature is disabled because we need to register the SubcriptionParameter and the cycleSelector
    // To be checked later while validating the initialization phase.
    //put all the waiting subscribers in the collection
    //        std::vector<SubscriptionParameter*>::iterator iter;
    //        std::map<uint32_t, Notification*>::iterator iter2;
    //        for (iter = waitingSubscriberCol_.begin(); iter != waitingSubscriberCol_.end(); iter++)
    //        {
    //            //add the new subscriber into the Notifications
    //            const std::string& propName = (*iter)->getProperty().getName();
    //            const std::string& deviceName = (*iter)->getDevice().getName();
    //
    //            for (iter2 = notifications_.begin(); iter2 != notifications_.end(); iter2++)
    //            {
    //                if ((*iter2).second->contains(propName, deviceName))
    //                    (*iter2).second->addSubscriber(*(*iter));
    //            }
    //        }
    //
    //        //empty the list
    //        waitingSubscriberCol_.clear();
}

void SubscriptionTreeManager::printState(std::string& header)
{
    std::stringstream str;
    str << header << "\nSubscriptionTreeManager Configuration:\n\tNotif Configuration:\n";
    NotificationItr itr = notifications_.begin();
    NotificationItr itrEnd = notifications_.end();
    for(; itr != itrEnd; ++itr)
    {
        (*itr).second->printDebugState(str);
    }

    str << "\tCycleSubscription Configuration:\n";
    subscriptionCycle_.printDebugState(str,2);

    str << "\tSubscriptionParameter Configuration:\n";
    SubscriptionItr subItr = subscriptions_.begin();
    SubscriptionItr subItrEnd = subscriptions_.end();
    for(; subItr != subItrEnd; ++subItr)
    {
        (*subItr).second->printDebugState(str);
    }
    LOG_DEBUG_IF(logger, str.str());
}
} // fesa