source: trunk/Cosy/base/msgqueue.cc@ 20026

#include "msgqueue.h"

#include <iostream>

using namespace std;

// --------------------------------------------------------------------------
//
// This creates the Message queue thread,
//
MsgQueue::MsgQueue() : fState(kRun), fThread(std::bind(&MsgQueue::Thread, this))
{
}

// --------------------------------------------------------------------------
//
// The destructor terminates the thread.
//
MsgQueue::~MsgQueue()
{
    CancelThread();
}

void MsgQueue::CancelThread()
{
    const std::lock_guard<std::mutex> lock(fMutex);
    if (fState!=kIdle)
        return;

    fState = kAbort;
    fCond.notify_one();
    fThread.join();
}

// --------------------------------------------------------------------------
//
// This is the function which must be overloaded.
// Here you process the messages. mp is a pointer which you can
// specify when posting the message.
//
// If a new messages is posted while the old one is not yet
// finished the fBreak flag is set. Please test this flag with
// Break() and try to finish (or stop) the current action as soon
// as possible. This makes sure, that before a new action is started
// the old action can be finished correctly by the user.
//
int MsgQueue::Proc(int msg, void *mp)
{
    return 0;
}

int MsgQueue::Break() const
{
    return fSize>0 || fState!=kRun;
}

// --------------------------------------------------------------------------
//
// This is the thread which handles the processing.
// As soon as a message is posted the fBreak flag is set (see PostMsg)
// And as soon as the current action is finished the new action is executed
// in this thread. This makes sure, that the calling program is not stalled.
//
void MsgQueue::Thread()
{
    std::unique_lock<std::mutex> lock(fMutex);

    while (1)
    {
        while (fList.empty() && fState==kRun)
            fCond.wait(lock);

        if (fState==kAbort)
            break;

        if (fState==kStop && fList.empty())
            break;

        const auto &val = fList.front();
        fSize--;

        lock.unlock();
        Proc(val.first, const_cast<char*>(val.second.data()));
        lock.lock();

        fList.pop_front();
    }

    fList.clear();
    fSize = 0;

    fState = kIdle;
}


// --------------------------------------------------------------------------
//
// Use this function to post a message.
// mp can be a pointer to a data structure. size should be the size of it.
// size bytes of this structure are copied and a pointer to the copy
// is forwarded to the Proc function.
//
void *MsgQueue::PostMsg(int msg, void *mp, int size)
{
    const std::lock_guard<std::mutex> lock(fMutex);
    if (fState==kIdle)
        return false;

    fSize++;
    fList.emplace_back(msg, vector<char>(reinterpret_cast<char*>(mp),reinterpret_cast<char*>(mp)+size));

    fCond.notify_one();

    return 0;

}