NotificationThread.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

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

#include <fesa-core/RDADeviceServer/SubscriptionFilter.h>
#include <fesa-core/Core/AbstractEquipment.h>
#include <fesa-core/Core/ThreadPriorityConfiguration.h>
#include <fesa-core/Diagnostic/Diagnostics.h>
#include <fesa-core/Exception/FesaException.h>
#include <fesa-core/Utilities/Lock.h>
#include <fesa-core/Utilities/XMLParser.h>

#include <cmw-log/Logger.h>


namespace
{

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

} // namespace


namespace fesa
{

NotificationThread::NotificationThread(const std::string& className, const std::string& notificationThreadKey) :
    inUseSubFilter_(NULL)
{
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    const std::string threadName("Notif_" + notificationThreadKey);
    {
        AbstractEquipment* eqp = AbstractEquipment::getInstance();
        std::string instanceFile = eqp->getDeviceDataFileName();
        XMLParser xmlParser(instanceFile, false);
        ThreadPriorityConfigurationFromFile conf(xmlParser, eqp->getProcessConfiguration());
        int32_t prio =  conf.getPrioNotificationthread(className, notificationThreadKey);
        setPriority(prio);
    }
    try
    {
        std::ostringstream traceStrStream;
        traceStrStream << "Starting notification thread with priority " << getPriority();
        diagMsg.msg = traceStrStream.str();
        diagnostics->log(diagTopic, diagMsg);
        start(false, threadName);
    }
    catch (FesaException& ex)
    {
        std::ostringstream traceStrStream;
        traceStrStream << "Failed to start notification thread: " << ex.getMessage();
        diagMsg.msg = traceStrStream.str();
        diagnostics->log(diagTopic, diagMsg);
    }
}

NotificationThread::~NotificationThread()
{
}

void NotificationThread::addNotificationEvent(MultiplexingContext& muxContext, RequestType& reqType,
                                              std::vector<std::string>& validCS, SubscriptionFilter* subFilter)
{
    // the Stl queue is NOT threadsave .. so we have to use mutexes
    boost::shared_ptr<NotificationThread::NotificationQueueElement> pEl(new NotificationThread::NotificationQueueElement(muxContext, reqType, validCS, subFilter));
    {
        Lock lock(mutexQueue_);
        queue_.push_back(pEl);
    }
    startConsumption_.signal();
}

bool NotificationThread::removeNotificationEvent(SubscriptionFilter* subFilter)
{
    Lock lock(mutexQueue_); //get a lock on the queue
    std::deque<boost::shared_ptr<NotificationQueueElement> >::iterator iter = queue_.begin();

    //erase all entrys, with matching subscriber
    while (iter != queue_.end())
    {
        if ((*iter)->subFilter_ == subFilter)// compare the reference of both
        {
            boost::shared_ptr<NotificationQueueElement> pEl = (*iter);
            iter = queue_.erase(iter);//return value is a iterator-ref to the next element
        }
        else
        {
            iter++;
        }
    }
    bool isInUse = (inUseSubFilter_ == subFilter) ? true : false;
    return isInUse;
}

void NotificationThread::run()
{
    while (isRunning_)
    {
        boost::shared_ptr<NotificationQueueElement> pEl;
        {
            Lock lock(mutexQueue_);
            inUseSubFilter_ = NULL;
            while (queue_.empty())
            {
                startConsumption_.wait(mutexQueue_);
            }

            //dequeue element
            pEl = queue_.front();
            queue_.pop_front(); //delete this element from the queue
            // keep the current in use Subscription Filter in case a remove Subscriber is waiting.
            inUseSubFilter_ = pEl->subFilter_;
        }
        pEl->subFilter_->doUpdate(pEl->muxContext_, pEl->reqType_, pEl->validCS_);
    }
    hasFinished_ = true;
}

NotificationThread::NotificationQueueElement::NotificationQueueElement(MultiplexingContext& muxContext, RequestType& reqType,
                                                                       std::vector<std::string>& validCS, SubscriptionFilter* subFilter) :
    subFilter_(subFilter),
    muxContext_(muxContext),
    reqType_(reqType),
    validCS_(validCS)
{
}


NotificationThread::NotificationQueueElement::~NotificationQueueElement()
{
}

} // fesa