MsgQueueFactory.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/Core/MsgQueueFactory.h>

#include <fesa-core/Core/AbstractEquipment.h>
#include <fesa-core/Core/LocalMsgQueue.h>
#include <fesa-core/Core/IpcMsgQueue.h>
#include <fesa-core/Core/AbstractMsgQueue.h>
#include <fesa-core/Utilities/Lock.h>
#include <fesa-core/Utilities/Mutex.h>
#include <fesa-core/Exception/FesaException.h>


namespace
{

static fesa::Mutex queueFactoryMutex_;

} // namespace

namespace fesa
{

std::map<std::string, AbstractMsgQueue*> MsgQueueFactory::blockingIPCQueueCol_;
std::map<std::string, AbstractMsgQueue*> MsgQueueFactory::nonBlockingIPCQueueCol_;
std::map<std::string, AbstractMsgQueue*> MsgQueueFactory::localQueueCol_;

AbstractMsgQueue* MsgQueueFactory::getOrCreateMsgQueue(const std::string& name, uint32_t queuelength, bool blocking)
{
    std::map<std::string, AbstractMsgQueue*>* queueCol = 0;
    if (AbstractEquipment::getInstance()->getProcessType() == unsplit)
    {
        queueCol = &localQueueCol_;
    }
    else
    {
        if (blocking)
        {
            queueCol =  &blockingIPCQueueCol_;
        }
        else
        {
            queueCol =  &nonBlockingIPCQueueCol_;
        }
    }

    Lock lock(queueFactoryMutex_);
    std::map<std::string, AbstractMsgQueue*>::const_iterator iter = queueCol->find(name);
    AbstractMsgQueue* msgQueuePtr;
    //Queue has not be found, so it is created and inserted into the map
    if (iter == queueCol->end())
    {
        if (AbstractEquipment::getInstance()->getProcessType() == unsplit)
        {
            msgQueuePtr = (AbstractMsgQueue*) new LocalMsgQueue("/" + name, queuelength);
        }
        else
        {
            // IPC message queue expects "/" as first character of the name
            msgQueuePtr = (AbstractMsgQueue*) new IpcMsgQueue("/" + name, queuelength, blocking);
        }
        std::pair<std::map<std::string, AbstractMsgQueue*>::iterator, bool> res = queueCol->insert(std::pair<
                                                                                                   std::string, AbstractMsgQueue*>(name, msgQueuePtr));
    }
    else
    {
        msgQueuePtr = (*iter).second;

        if (msgQueuePtr->getQueueLength() != queuelength)
        {
            throw FesaException(__FILE__, __LINE__, FesaErrorQueuenameDuplication.c_str(), name.c_str(),
                                "queue length");
        }
    }
    return msgQueuePtr;
}

MsgQueueFactory::~MsgQueueFactory()
{
    for (std::map<std::string, AbstractMsgQueue*>::iterator iter = blockingIPCQueueCol_.begin(); iter != blockingIPCQueueCol_.end(); iter++)
    {
        delete (*iter).second;
    }
    blockingIPCQueueCol_.clear();

    for (std::map<std::string, AbstractMsgQueue*>::iterator iter = nonBlockingIPCQueueCol_.begin(); iter != nonBlockingIPCQueueCol_.end(); iter++)
    {
        delete (*iter).second;
    }
    nonBlockingIPCQueueCol_.clear();

    for (std::map<std::string, AbstractMsgQueue*>::iterator iter = localQueueCol_.begin(); iter != localQueueCol_.end(); iter++)
    {
        delete (*iter).second;
    }
    localQueueCol_.clear();
}

} // fesa