FairMQ/fairmq/StateQueue.h

/********************************************************************************
 *    Copyright (C) 2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
 *                                                                              *
 *              This software is distributed under the terms of the             *
 *              GNU Lesser General Public Licence (LGPL) version 3,             *
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/

#ifndef FAIRMQSTATEQUEUE_H_
#define FAIRMQSTATEQUEUE_H_

#include <fairmq/States.h>

#include <queue>
#include <mutex>
#include <chrono>
#include <utility> // pair
#include <condition_variable>

namespace fair::mq
{

class StateQueue
{
  public:
    StateQueue() = default;
    ~StateQueue() = default;

    fair::mq::State WaitForNext()
    {
        std::unique_lock<std::mutex> lock(fMtx);
        while (fStates.empty()) {
            fCV.wait_for(lock, std::chrono::milliseconds(50));
        }

        fair::mq::State state = fStates.front();

        if (state == fair::mq::State::Error) {
            throw DeviceErrorState("Controlled device transitioned to error state.");
        }

        fStates.pop();
        return state;
    }

    template<typename Rep, typename Period>
    std::pair<bool, fair::mq::State> WaitForNext(std::chrono::duration<Rep, Period> const& duration)
    {
        std::unique_lock<std::mutex> lock(fMtx);
        fCV.wait_for(lock, duration);

        if (fStates.empty()) {
            return { false, fair::mq::State::Ok };
        }

        fair::mq::State state = fStates.front();

        if (state == fair::mq::State::Error) {
            throw DeviceErrorState("Controlled device transitioned to error state.");
        }

        fStates.pop();
        return { true, state };
    }

    void WaitForState(fair::mq::State state) { while (WaitForNext() != state) {} }

    void Push(fair::mq::State state)
    {
        {
            std::lock_guard<std::mutex> lock(fMtx);
            fStates.push(state);
        }
        fCV.notify_all();
    }

    void Clear()
    {
        std::lock_guard<std::mutex> lock(fMtx);
        fStates = std::queue<fair::mq::State>();
    }

  private:
    std::queue<fair::mq::State> fStates;
    std::mutex fMtx;
    std::condition_variable fCV;
};

} // namespace fair::mq

#endif /* FAIRMQSTATEQUEUE_H_ */