FesaDeviceServer.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

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

#include <fesa-core/RDADeviceServer/SubscriptionTreeManager.h>
#include <fesa-core/Core/AbstractEquipment.h>
#include <fesa-core/Core/FesaDefs.h>
#include <fesa-core/Core/ThreadPriorityConfiguration.h>
#include <fesa-core/DataStore/AbstractDevice.h>
#include <fesa-core/DataStore/AbstractDeviceFactory.h>
#include <fesa-core/DataStore/EquipmentData.h>
#include <fesa-core/DataStore/GlobalDevice.h>
#include <fesa-core/Diagnostic/Diagnostics.h>
#include <fesa-core/Exception/FesaException.h>
#include <fesa-core/Server/AbstractServerDeviceClass.h>
#include <fesa-core/Server/AbstractServerEquipment.h>
#include <fesa-core/Server/Property.h>
#include <fesa-core/Utilities/Lock.h>
#include <fesa-core/Utilities/ProcessConfiguration.h>
#include <fesa-core/Utilities/XMLParser.h>

#include <cmw-rda/Config.h>
#include <cmw-rda/Data.h>
#include <cmw-rda/Exception.h>
#include <cmw-log/Logger.h>


namespace
{

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

} // namespace


namespace fesa
{

const std::string ERROR_CATEGORY = "FESA";

FesaDeviceServer::FesaDeviceServer(const std::string& name, const ProcessConfiguration* processConfiguration, SubscriptionTreeManager* pSubscriptionTreeManager) :
    rdaDeviceServerBase(name.c_str()),
    pTreeManager_(pSubscriptionTreeManager)
{
    firstUpdateAcquisitionEnabled_ = processConfiguration->getBoolValue(PropertyTag::FIRST_UPDATE_ACQUISITION);
    firstUpdateSettingEnabled_ = processConfiguration->getBoolValue(PropertyTag::FIRST_UPDATE_SETTING);
    try
    {
        rdaConfig::useConfigFile(processConfiguration->getCMWConfigFile().c_str());
    }
    catch (rdaInternalError& ex)
    {
        std::ostringstream msg;
        msg << "rdaConfig::useConfigFile failed: " << ex.getMessage();
        LOG_ERROR_IF(logger, msg.str());
    }
}

FesaDeviceServer::~FesaDeviceServer()
{
    shutDown();//stops the RDA Server ( this makes the runServer() method return)
    mutex_.lock();
}

void FesaDeviceServer::runDeviceServer()
{
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    std::ostringstream traceStrStream;
    traceStrStream << "Starting FESA device server with name '" << getName() << "' - Server side startup should be finished here";
    diagMsg.msg = traceStrStream.str();
    diagnostics->log(diagTopic, diagMsg);
    try
    {
        //starts the RDA Server ... this method will only return if the shutDown(); method is called
        Lock lock(mutex_);
        runServer();
    }
    catch (const rdaInternalError& ex)
    {
        std::ostringstream errorStrStream;
        errorStrStream << "RDA server failure. RDA internal error:" << ex.getFile() << " : " << ex.getLine() << " : " << ex.getMessage();
        LOG_ERROR_IF(logger, errorStrStream.str());
        throw;
    }
    // all calls below throw FesaExceptions instead of returning NULL
    // but only RDAException can be propagate to the caller
    catch (const rdaException& ex)
    {
        std::ostringstream errorStrStream;
        errorStrStream << "RDA server failure. RDA internal error: " << ex.getType() << "  " << ex.getMessage();
        LOG_ERROR_IF(logger, errorStrStream.str());
        throw;
    }
    catch (...)
    {
        LOG_ERROR_IF(logger, "RDA Server failure. Unknown exception in 'runRDADeviceServer'");
        throw;
    }
}

rdaData* FesaDeviceServer::get(const rdaIOPoint& iop, const rdaData& ctx)
{
    const std::string threadName = "GetSetLayer";
    Thread::registerThreadIdName(pthread_self(), threadName);
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    std::auto_ptr<rdaData> value;
    // All calls below throw FesaExceptions instead of returning NULL
    // but only RDAException can be propagated to the caller
    try
    {
        if (isRTProcessDown())
        {
            throw FesaException(__FILE__, __LINE__, FesaErrorRTProcessDown.c_str());
        }
        std::ostringstream traceStrStream;
        traceStrStream << "Device: " << iop.getDeviceName() << ". Property: " << iop.getPropertyName()
                       << ". Client: " << getClientInfo()->getUserName() << " from " << getClientInfo()->getHostName();
        diagMsg.msg = traceStrStream.str();
        diagnostics->log(diagTopic, diagMsg);
        //get pointers to all needed parameters
        Property* property;
        AbstractDevice* device;
        AbstractServerDeviceClass* srvDeviceClass;
        std::string cycleSelector("");
        decodeIOPoint(iop, srvDeviceClass, property, device, cycleSelector);
        value.reset(new rdaData());
        property->get(*device, cycleSelector, ctx, *value, GET);//perform the get
    }
    catch (FesaException& ex)
    {
        throw rdaIOError(ERROR_CATEGORY.c_str(), ex.getErrorCodeAsLong(), ex.getMessage().c_str());
    }
    catch (rdaException& ex)
    {
        throw rdaIOError("RDA", 0, ex.getMessage());
    }
    catch (...)// in case of any other exceptions, just throw them to the next level (the subscriber)
    {
        throw rdaIOError(ERROR_CATEGORY.c_str(), 0, "failure performing 'Get': Unknown Error");
    }
    rdaData* pRDAData = value.get();
    value.release();
    return pRDAData;
}

void FesaDeviceServer::set(const rdaIOPoint& iop, const rdaData& ctx, const rdaData& value)
{
    const std::string threadName = "GetSetLayer";
    Thread::registerThreadIdName(pthread_self(), threadName);
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    // All calls below throw FesaExceptions instead of returning NULL
    // but only RDAException can be propagated to the caller
    try
    {
        if (isRTProcessDown())
        {
            throw FesaException(__FILE__, __LINE__, FesaErrorRTProcessDown.c_str());
        }
        std::ostringstream traceStrStream;
        traceStrStream << "Device: " << iop.getDeviceName() << ". Property: " << iop.getPropertyName()
                       << ". Client: " << getClientInfo()->getUserName() << " from " << getClientInfo()->getHostName();
        diagMsg.msg = traceStrStream.str();
        diagnostics->log(diagTopic, diagMsg);
        // Get pointers to all needed parameters
        Property* property;
        AbstractDevice* device;
        AbstractServerDeviceClass* srvDeviceClass;
        std::string cycleSelector("");
        decodeIOPoint(iop, srvDeviceClass, property, device, cycleSelector);
        property->set(*device, cycleSelector, ctx, value); // Perform set
    }
    catch (rdaException& ex)
    {
        throw rdaIOError("RDA", 0, ex.getMessage());
    }
    catch (FesaException& ex)
    {
        throw rdaIOError(ERROR_CATEGORY.c_str(), ex.getErrorCodeAsLong(), ex.getMessage().c_str());
    }
    catch (...)// in case of any other exceptions, just throw them to the next level (the subscriber)
    {
        throw rdaIOError(ERROR_CATEGORY.c_str(), 0, "failure performing 'Set': Unknown Error");
    }
}

void FesaDeviceServer::monitorOn(const rdaIOPoint& iop, const rdaData& ctx, rdaValueChangeListener* listener)
{
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    Property* property;
    AbstractDevice* device;
    std::string cycleSelector("");
    AbstractServerDeviceClass* srvDeviceClass;
    // all calls below throw FesaExceptions instead of returning NULL
    // but only RDAException can be propagated to the caller
    try
    {
        if (isRTProcessDown())
        {
            throw FesaException(__FILE__, __LINE__, FesaErrorRTProcessDown.c_str());
        }
        std::ostringstream traceStrStream;
        traceStrStream << "Device: " << iop.getDeviceName() << ". Property: " << iop.getPropertyName()
                       << ". Client: " << getClientInfo()->getUserName() << " from " << getClientInfo()->getHostName();
        diagMsg.msg = traceStrStream.str();
        diagnostics->log(diagTopic, diagMsg);
        decodeIOPoint(iop, srvDeviceClass, property, device, cycleSelector);
        property->validateParameters(*device, cycleSelector, (ctx.isEmpty() ? false : true), SUBSCRIBE);
        // Checking if the subscription is disabled due to notification failure
        if (*(EquipmentData::getInstance()->notificationFailure_) == true)
        {
            throw FesaException(__FILE__, __LINE__, FesaErrorPropertySubscriptionDisabled.c_str());
        }
        pTreeManager_->addSubscriber(srvDeviceClass->getName(), *device, *property, cycleSelector, ctx, listener);  //This builds the subscription tree, and adds the enties in the notification objects
    }
    catch (rdaException& ex)
    {
        throw rdaIOError("RDA", 0, ex.getMessage());
    }
    catch (FesaException& ex)
    {
        throw rdaIOError(ERROR_CATEGORY.c_str(), ex.getErrorCodeAsLong(), ex.getMessage().c_str());
    }
    catch (...)// in case of any other exceptions, just throw them to the next level (the subscriber)
    {
        throw rdaIOError(ERROR_CATEGORY.c_str(), 0, "failure adding new Subscriber: Unknown Error");
    }
}

void FesaDeviceServer::monitorOff(const rdaIOPoint& iop, rdaValueChangeListener* listener)
{
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::server;
    AbstractServerDeviceClass* srvDeviceClass;
    Property* property;
    AbstractDevice* device;
    std::string cycleSelector;

    std::ostringstream traceStrStream;
    traceStrStream << "Device: " << iop.getDeviceName() << ". Property: " << iop.getPropertyName()
                   << ". Client: " << getClientInfo()->getUserName() << " from " << getClientInfo()->getHostName();
    diagMsg.msg = traceStrStream.str();
    diagnostics->log(diagTopic, diagMsg);
    try //try+catch is needed here in order to unlock the tree on an exception
    {
        //get pointers to all needed parameters
        decodeIOPoint(iop, srvDeviceClass, property, device, cycleSelector);
        pTreeManager_->removeSubscriber(*device, *property, cycleSelector, listener);//this will remove the subscriber from the subscription tree AND from the Notifications

    }
    catch (rdaException& ex)
    {
        throw rdaIOError("RDA", 0, ex.getMessage());
    }
    catch (FesaException& ex)
    {
        throw rdaIOError(ERROR_CATEGORY.c_str(), ex.getErrorCodeAsLong(), ex.getMessage().c_str());
    }
    catch (...)// in case of any other exceptions, just throw them to the next level (the subscriber)
    {
        throw rdaIOError(ERROR_CATEGORY.c_str(), 0, "failure removing Subscriber: Unknown Error");
    }
}

void FesaDeviceServer::decodeIOPoint(const rdaIOPoint& iop, AbstractServerDeviceClass*& srvDeviceClass, Property*& prop, AbstractDevice*& device,
                                     std::string& cycleSelector)
{
    const std::string propertyName(iop.getPropertyName());
    const std::string deviceName(iop.getDeviceName());

    //TODO check if IOPoint return a null pointer
    // this is necessary, to be able to handle empty strings
    const char* cs = const_cast<char*> (iop.getCycleSelector());
    if (!cs)
        cs = "";
    cycleSelector = cs;

    //get pointers to all needed parameters
    AbstractServerEquipment* equipment = AbstractServerEquipment::getInstance();
    srvDeviceClass = equipment->getDeviceClassFromDeviceName(deviceName);
    prop = srvDeviceClass->getProperty(propertyName);
    if (prop == NULL)
    {
        throw FesaException(__FILE__, __LINE__, FesaErrorPropertyNotFound.c_str(), propertyName.c_str());
    }

    if (prop->isGlobal())
    {
        device = srvDeviceClass->getDeviceFactory()->getGlobalDevice();
        if (device == NULL)
        {
            device = srvDeviceClass->getDeviceFactory()->getDevice(deviceName);
            if (device != NULL)
                // Global property cannot be accessed using a deviceInstance
                throw FesaException(__FILE__, __LINE__, FesaErrorGlobalPropertyDeviceInstance.c_str(), propertyName.c_str());
            // unknown device
            throw FesaException(__FILE__, __LINE__, FesaErrorUnknownDevice.c_str(), deviceName.c_str());
        }
    }
    else
    {
        device = srvDeviceClass->getDeviceFactory()->getDevice(deviceName);
        if (device == NULL) {
            device = srvDeviceClass->getDeviceFactory()->getGlobalDevice();
            if (device != NULL)
                // Device property cannot be accessed using teh global device
                throw FesaException(__FILE__, __LINE__, FesaErrorDevicePropertyGlobalDevice.c_str(), propertyName.c_str());
            // unknown device
            throw FesaException(__FILE__, __LINE__, FesaErrorUnknownDevice.c_str(), deviceName.c_str());
        }
    }
}


bool FesaDeviceServer::isRTProcessDown()
{
    EquipmentData* eqData = EquipmentData::getInstance();
    if (*(eqData->hasRt_) && !(*(eqData->rtUp_)))
    {
        return true;
    }
    return false;
}

void FesaDeviceServer::start()
{
    {
        AbstractEquipment* eqp = AbstractEquipment::getInstance();
        std::string instanceFile = eqp->getDeviceDataFileName();
        XMLParser xmlParser(instanceFile, false);
        ThreadPriorityConfigurationFromFile conf(xmlParser, eqp->getProcessConfiguration());
        int32_t prio =  conf.getPrioCMW();
        setPriority(prio);
    }
    Thread::start(false, "DeviceServer");
}

void FesaDeviceServer::run()
{
    try
    {
        runDeviceServer();
    }
    catch(const rdaInternalError& ex)
    {
        std::ostringstream errorStrStream;
        errorStrStream << "Fatal error in RDA runtime:" << ex.getFile() << " : " << ex.getLine() << " : " << ex.getMessage() << std::endl;
        LOG_ERROR_IF(logger, errorStrStream.str());
        throw FesaException(__FILE__, __LINE__, FesaErrorIRunningDeviceServer.c_str(), ex.getMessage());
    }
    catch(const rdaInternalException& ex)
    {
        std::ostringstream errorStrStream;
        errorStrStream << "Recoverable error in the RDA runtime:" << ex.getMessage() << std::endl;
        LOG_ERROR_IF(logger, errorStrStream.str());
        throw FesaException(__FILE__, __LINE__, FesaErrorIRunningDeviceServer.c_str(), ex.getMessage());
    }
}

void FesaDeviceServer::initCMW()
{
    const int argc = 0;
    char** argv = NULL;
    try
    {
        rdaDeviceServerBase::init(argc, argv); // argc and argv are not used in RDA except in some restart procedure, which is obsolete according to CMW team
        std::ostringstream infoStringStream;
        infoStringStream << "CMW initialized from " << AbstractEquipment::getInstance()->getProcessConfiguration()->getCMWConfigFile();
        LOG_INFO_IF(logger, infoStringStream.str());
    }
    catch (const rdaInternalError& ex)
    {
        throw FesaException(__FILE__, __LINE__, FesaErrorInitializingRdaDeviceServerBase.c_str(), ex.getMessage());
    }
}

} // fesa