OnSubscriptionEventSource.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/RealTime/OnSubscriptionEventSource.h>
#include <fesa-core/RealTime/OnSubscriptionRTEventPayload.h>
#include <fesa-core/Utilities/ParserElements/EventElement.h>
#include <fesa-core/Synchronization/NoneContext.h>
#include <fesa-core/Synchronization/TimingContext.h>
#include <fesa-core/Synchronization/SynchronizationLabObjectFactory.h>
#include <fesa-core/Core/AbstractEquipment.h>
#include <fesa-core/Utilities/ParserElements/ParserElementDefs.h>
#include <fesa-core/RealTime/RTEvent.h>
#include <cmw-log/Logger.h>


namespace
{

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

} // namespace


namespace fesa
{

OnSubscriptionEventSource::OnSubscriptionEventSource() :
    AbstractEventSource(ON_SUBSCRIPTION_EVENT_SOURCE_NAME, OnSubscriptionSource)
{
    itr1_ = onSubscriptionEventMapping_.begin();
    itr2_ = onSubscriptionEventMapping_.end();
    IsSubscriptionStarted_ = false;
    subscriptionThread_ = NULL;
}

OnSubscriptionEventSource::~OnSubscriptionEventSource()
{
    if (subscriptionThread_ != NULL)
        deleteSubscriptionThread();
}

void OnSubscriptionEventSource::connect(boost::shared_ptr<fesa::EventElement>& eventElement) // For the time being onChange is always false and subscription to ALL
{
    std::string eventName = eventElement->getConcreteName();
    std::string deviceName = eventElement->getSourceTypeSpecificData(ON_SUBSCRIPTION_TAG_DEVICE);
    std::string propertyName = eventElement->getSourceTypeSpecificData(ON_SUBSCRIPTION_TAG_PROPERTY);
    std::pair<std::string, std::string> tmp(deviceName, propertyName);
    onSubscriptionEventMapping_.insert(std::pair<std::pair<std::string, std::string>, std::string>(tmp, eventName));
    pendingSubscriptions_.insert(std::pair<std::string, std::string>(deviceName, propertyName));
}

RTEvent* OnSubscriptionEventSource::wait()
{
    if (!IsSubscriptionStarted_) // if it the first time
    {
        if (!startSubscriptions()) // If some of the subscriptions failed
        {
            // create a thread that every 5 second retries the connection of the remaining subscriptions
            createSubscriptionThread();
        }
    }

    if (itr1_ == itr2_)
    {
        eventInfo_.data.removeAll();
        proxy_.waitNotification(eventInfo_.deviceName, eventInfo_.propertyName, eventInfo_.cycleSelectorName,
                                eventInfo_.data);
        std::pair<OnSubscriptionEventMapping::iterator, OnSubscriptionEventMapping::iterator> itr =
            onSubscriptionEventMapping_.equal_range(std::pair<std::string, std::string>(
                                                        eventInfo_.deviceName, eventInfo_.propertyName));
        itr1_ = itr.first;
        itr2_ = itr.second;
    }

    if ((eventInfo_.data.contains("updateFlags"))) // Inmediate update should be ignored
    {
        std::string msg = " Inmediate update; event should be ignored";
        ++itr1_;
        return NULL;//TODO Returning NULL is not nice. Better would be to have the RTEvent as parameter
    }

    OnSubscriptionRTEventPayload* pRTPayload = new OnSubscriptionRTEventPayload(eventInfo_.deviceName,
                                                                                eventInfo_.propertyName, eventInfo_.cycleSelectorName, eventInfo_.data);

    // check if the subscription is coming from a multiplexed property or not

    MultiplexingContext* context;

    if (pRTPayload->rdaData_.contains("cycleStamp")) //subscription done to a timing related action
    {
        int64_t cycleStamp = pRTPayload->rdaData_.extractLong("cycleStamp");
        try
        {
            std::string cycleName = pRTPayload->rdaData_.extractString("cycleName");
            context = SynchronizationLabObjectFactory::getInstance()->createTimingContext(cycleStamp, cycleName);

        }
        catch (FesaException& ex)
        {
            delete pRTPayload;
            throw;//will be caught & logged in the AbstractEventSource
        }
    }
    else // subscription done to a non-timing related action
    {
        context = new NoneContext();
    }

    RTEvent* pRTEvt = new RTEvent(itr1_->second, context, this, pRTPayload);
    ++itr1_;
    return pRTEvt;
}

bool OnSubscriptionEventSource::startSubscriptions()
{
    IsSubscriptionStarted_ = true;
    OnSubscriptionEventMapping::iterator itr;
    for (itr = onSubscriptionEventMapping_.begin(); itr != onSubscriptionEventMapping_.end(); ++itr)
    {
        try
        {
            proxy_.subscribe(itr->first.first, itr->first.second);
            pendingSubscriptions_.erase(std::pair<std::string, std::string>(itr->first.first, itr->first.second));
        }
        catch (FesaException& ex)
        {
            LOG_INFO_IF(logger, ex.getMessage());
        }
    }
    if (pendingSubscriptions_.empty())
    {
        return true;
    }
    else
    {
        return false;
    }
}

void OnSubscriptionEventSource::createSubscriptionThread()
{
    subscriptionThread_ = new SubscriptionThread(this);
    subscriptionThread_->setPriority(getPriority());
    subscriptionThread_->start(false, "SubscriptionThread");
}

void OnSubscriptionEventSource::deleteSubscriptionThread()
{
    subscriptionThread_->stop();
    delete subscriptionThread_;
}

OnSubscriptionEventSource::SubscriptionThread::SubscriptionThread(OnSubscriptionEventSource* evtSrc)
{
    evtSrc_ = evtSrc;
}

OnSubscriptionEventSource::SubscriptionThread::~SubscriptionThread()
{

}

void OnSubscriptionEventSource::SubscriptionThread::run()
{
    while (!evtSrc_->pendingSubscriptions_.empty())
    {
        for (PendingSubscriptions::iterator itr = evtSrc_->pendingSubscriptions_.begin(); itr
                 != evtSrc_->pendingSubscriptions_.end(); ++itr)
        {
            try
            {
                evtSrc_->proxy_.subscribe(itr->first, itr->second);
                evtSrc_->pendingSubscriptions_.erase(std::pair<std::string, std::string>(itr->first, itr->second));
            }
            catch (FesaException& ex)
            {
                LOG_INFO_IF(logger, ex.getMessage());
            }
        }
        struct timespec req;
        req.tv_sec = 5;
        req.tv_nsec = 0;
        int32_t ret;
        do
        {
            ret = nanosleep(&req, &req);
        }
        while ((ret == -1) && (errno == EINTR));
    }
    evtSrc_->deleteSubscriptionThread();
}

} // fesa