NotificationConsumer.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/Server/NotificationConsumer.h>

#include <fesa-core/Server/AbstractServerEquipment.h>
#include <fesa-core/Core/AbstractEquipment.h>
#include <fesa-core/Core/AbstractMsgQueue.h>
#include <fesa-core/Core/MsgQueueFactory.h>
#include <fesa-core/Core/MessageTypes.h>
#include <fesa-core/Core/ThreadPriorityConfiguration.h>
#include <fesa-core/DataStore/EquipmentData.h>
#include <fesa-core/Diagnostic/Diagnostics.h>
#include <fesa-core/Persistency/PersistencyManager.h>
#include <fesa-core/RDADeviceServer/SubscriptionTreeManager.h>
#include <fesa-core/Synchronization/MultiplexingContext.h>
#include <fesa-core/Utilities/ProcessConfiguration.h>
#include <fesa-core/Utilities/XMLParser.h>

#include <cmw-log/Logger.h>


namespace
{

CMW::Log::Logger& logger = CMW::Log::LoggerFactory::getLogger("FESA.FWK.fesa-core.Server.NotificationConsumer");
const std::string diagTopic = fesa::DiagnosticUtils::startUpTrackingStr;

} // namespace


namespace fesa
{

//init static members
NotificationConsumer* NotificationConsumer::theInstance_ = NULL;

NotificationConsumer::NotificationConsumer(const std::string& queueName,  SubscriptionTreeManager& subscriptionTreeManager) :  subscriptionTreeManager_(subscriptionTreeManager)
{
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    std::ostringstream traceStrStream;
    traceStrStream << "Notification consumer is connecting to message queue '" << queueName << "'";
    diagMsg.msg = traceStrStream.str();
    diagnostics->log(diagTopic, diagMsg);
    int32_t size = AbstractEquipment::getInstance()->getProcessConfiguration()->getIntValue(PropertyTag::MSG_NUM_MAX);
    msgQueuePtr_ = MsgQueueFactory::getOrCreateMsgQueue(queueName, size, true);
    setDetachState( PTHREAD_CREATE_DETACHED);
    {
        AbstractEquipment* eqp = AbstractEquipment::getInstance();
        std::string instanceFile = eqp->getDeviceDataFileName();
        XMLParser xmlParser(instanceFile, false);
        ThreadPriorityConfigurationFromFile conf(xmlParser, eqp->getProcessConfiguration());
        int32_t prio =  conf.getPrioNotificationConsumer();
        setPriority(prio);
    }
}

NotificationConsumer::~NotificationConsumer()
{
}

NotificationConsumer* NotificationConsumer::getInstance( SubscriptionTreeManager& subscriptionTreeManager)
{
    if (theInstance_ == NULL)
    {
        AbstractEquipment* eqp = AbstractEquipment::getInstance();
        theInstance_ = new NotificationConsumer(eqp->getEquipmentName(),  subscriptionTreeManager);
    }
    return theInstance_;
}

void NotificationConsumer::releaseInstance()
{
    if (theInstance_ != NULL)
    {
        delete theInstance_;
        theInstance_ = NULL;
    }
}

void NotificationConsumer::run()
{
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    AbstractServerEquipment* eqp = AbstractServerEquipment::getInstance();
    RequestType reqType = GET;
    diagMsg.msg = "Notification consumer is purging the queue before consumption";
    diagnostics->log(diagTopic, diagMsg);
    msgQueuePtr_->purge();
    AbstractMessage* abstractMsg = NULL;
    while (isRunning_)
    {
        try
        {
            // reset the pointer just in case an exception is thrown by the messageQueue::consumeMsg method
            abstractMsg = NULL;
            //this thread will get stuck here, until a message arrives
            abstractMsg = msgQueuePtr_->consumeMsg();

            switch (abstractMsg->msgType_)
            {
                case AutomaticNotificationMsg:
                {
                    AutomaticNotificationMessage* concreteMsg =
                        static_cast<AutomaticNotificationMessage*> (abstractMsg);

                    std::vector<AbstractDeviceClass*> deviceCol = eqp->getDeviceClassCol();

                    EquipmentData* equipmentData = EquipmentData::getInstance();
                    if (*(equipmentData->notificationFailure_) == true)
                    {
                        *(equipmentData->notificationFailure_) = false;
                    }

                    try
                    {

                        subscriptionTreeManager_.notify(concreteMsg->notificationID_, *concreteMsg->muxContext_, reqType);
                        delete concreteMsg->muxContext_;
                    }
                    catch (...)
                    {
                        delete concreteMsg->muxContext_;
                        throw;
                    }
                    break;
                }
                case ManualNotificationMsg:
                {
                    ManualNotificationMessage* concreteMsg = static_cast<ManualNotificationMessage*> (abstractMsg);

                    EquipmentData* equipmentData = EquipmentData::getInstance();
                    if (*(equipmentData->notificationFailure_) == true)
                    {
                        *(equipmentData->notificationFailure_) = false;
                    }

                    try
                    {
                        subscriptionTreeManager_.notify(concreteMsg, reqType);
                        delete concreteMsg->muxContext_;
                    }
                    catch (...)
                    {
                        delete concreteMsg->muxContext_;
                        throw;
                    }

                    break;
                }
                case CommandMsg:
                {
                    CommandMessage *concreteMsg = static_cast<CommandMessage*> (abstractMsg);

                    if (concreteMsg->command_ == "PersistencyTrigger")
                    {
                        PersistencyManager::getInstance()->trigger(concreteMsg->parameter1_);
                    }
                    else
                    {
                        //unknown command
                        throw FesaException(__FILE__, __LINE__, FesaErrorUnknownMessageType.c_str(),
                                            msgQueuePtr_->getQueueName().c_str());
                    }
                    break;
                }
                default:
                    throw FesaException(__FILE__, __LINE__, FesaErrorUnknownMessageType.c_str(),
                                        msgQueuePtr_->getQueueName().c_str());
            }

            delete abstractMsg;

            //MONITORING: Notification-JAM detection TODO : figure out out if needed

        }
        catch (FesaException ex)
        {
            if (abstractMsg != NULL)
                delete abstractMsg;
            std::ostringstream errorStrStream;
            errorStrStream << "Notification consuming failure: " << ex.getMessage();
            LOG_ERROR_IF(logger, errorStrStream.str());
        }

    }
    isRunning_ = false;
}

} // fesa