FluxioMergerExecutor.cpp

Go to the documentation of this file.

#include "FluxioMergerExecutor.h"
 
#include <mutex>
#include <optional>
#include <string>
#include <thread>
#include <vector>
 
#include <ArmarXCore/core/logging/Logging.h>
 
#include "../SkillStatusUpdate.h"
 
namespace armarx::skills
{
    FluxioMergerExecutor::FluxioMergerExecutor(const std::string& id,
                                               const std::vector<std::string>& parameterIds) :
        FluxioExecutor(id, false)
    {
        for (const auto& id : parameterIds)
        {
            tokenHasArrivedMap[id] = false;
        }
        ARMARX_WARNING << "Fluxio Merger is waiting for the following tokens to arrive: "
                       << parameterIds;
    }
 
    void
    FluxioMergerExecutor::run(const std::string executorName,
                              armarx::aron::data::DictPtr /*parameters*/,
                              std::experimental::observer_ptr<const FluxioProfile> /*profilePtr*/)
    {
        // set status running
        setStatus(skills::SkillStatus::Running);
        this->executorName = executorName;
 
        // wait for all tokens to arrive
        while (true)
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(250));
 
            // check if all tokens have arrived
            bool allTokensArrived = true;
            for (const auto& [id, hasArrived] : tokenHasArrivedMap)
            {
                if (!hasArrived)
                {
                    allTokensArrived = false;
                    break;
                }
            }
 
            if (status->status == skills::SkillStatus::Aborted ||
                status->status == skills::SkillStatus::Failed ||
                status->status == skills::SkillStatus::Succeeded)
            {
                return;
            }
 
            if (allTokensArrived)
            {
                setStatus(skills::SkillStatus::Succeeded);
                return;
            }
        }
    }
 
    void
    FluxioMergerExecutor::abort()
    {
        setStatus(skills::SkillStatus::Aborted);
    }
 
    void
    FluxioMergerExecutor::checkInToken(const std::string& parameterId)
    {
        std::unique_lock l(tokenHasArrivedMapMutex);
        const auto it = tokenHasArrivedMap.find(parameterId);
 
        if (it == tokenHasArrivedMap.end())
        {
            ARMARX_WARNING << "Unexpected parameterId: " << parameterId;
        }
        else
        {
            ARMARX_WARNING << "Fluxio Merger received token for parameterId: " << parameterId;
            tokenHasArrivedMap[parameterId] = true;
        }
 
        l.unlock();
    }
 
    std::optional<std::vector<skills::FluxioSkillStatusUpdate>>
    FluxioMergerExecutor::getStatusUpdate()
    {
        // unused method
        return std::nullopt;
    }
} // namespace armarx::skills