FluxioCompositeExecutor.h

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <experimental/memory>
#include <optional>
#include <shared_mutex>
#include <string>
 
#include "RobotAPI/libraries/aron/core/data/variant/container/Dict.h"
#include "RobotAPI/libraries/skills/core/FluxioResult.h"
#include "RobotAPI/libraries/skills/core/error/FluxioException.h"
 
#include "../FluxioControlNode.h"
#include "../FluxioParameter.h"
#include "../FluxioParameterNode.h"
#include "../FluxioSkill.h"
#include "../FluxioSkillStatusUpdate.h"
#include "../FluxioSubSkillNode.h"
#include "../SkillExecutionID.h"
#include "../SkillStatusUpdate.h"
#include "FluxioExecutor.h"
 
namespace armarx::skills
{
    class FluxioCompositeExecutor : public FluxioExecutor
    {
    public:
        FluxioCompositeExecutor(
            const std::string& id,
            const skills::FluxioSkill& skill,
            const std::function<skills::Result<std::experimental::observer_ptr<FluxioExecutor>,
                                               skills::error::FluxioException>(
                const std::string& skillId,
                const std::string& profileId,
                const std::string& executorName,
                armarx::aron::data::DictPtr parameters)>&& executeFluxioSkillFunc,
            const std::function<std::experimental::observer_ptr<FluxioExecutor>(
                const std::vector<std::string>& parameterIds)>&& addMergerExecutorToDCFunc,
            const std::function<std::experimental::observer_ptr<FluxioExecutor>(
                const std::string& id,
                const skills::FluxioSkill& skill,
                bool isRetry)>&& addLoopExecutorToDCFunc);
 
        void run(std::string executorName,
                 armarx::aron::data::DictPtr parameters,
                 std::experimental::observer_ptr<const FluxioProfile> profilePtr) override;
        void abort() override;
        std::optional<std::vector<skills::FluxioSkillStatusUpdate>> getStatusUpdate() override;
        bool validateSkill(skills::FluxioEdge& ret) const;
 
    private:
        void startSubRoutine(
            const std::experimental::observer_ptr<const skills::FluxioNode>& startNode,
            const std::experimental::observer_ptr<const skills::FluxioParameter>& startParameter,
            std::atomic_bool& running,
            const std::string& executorName,
            const std::experimental::observer_ptr<const FluxioProfile> profilePtr);
 
        void handleParameterRoutine(
            std::experimental::observer_ptr<const skills::FluxioParameterNode> parameterNode,
            std::atomic_bool& running,
            const std::string& executorName);
 
        void handleSubSkillRoutine(
            std::experimental::observer_ptr<const skills::FluxioSubSkillNode> subSkillNode,
            std::atomic_bool& running,
            const std::string& executorName,
            const std::experimental::observer_ptr<const FluxioProfile> profilePtr);
 
        void handleControlRoutine(
            std::experimental::observer_ptr<const skills::FluxioControlNode> controlNode,
            const std::experimental::observer_ptr<const skills::FluxioParameter>& startParameter,
            std::atomic_bool& running,
            const std::string& executorName,
            const std::experimental::observer_ptr<const FluxioProfile> profilePtr);
 
        void abortSubExecutions();
 
        void setStatus(skills::SkillStatus status, const std::string& nodeId = "noId") override;
 
        void pollSubStatuses();
 
        const std::function<void(const std::string& executionId)> abortFluxioSkill;
 
        const std::function<skills::Result<std::experimental::observer_ptr<FluxioExecutor>,
                                           skills::error::FluxioException>(
            const std::string& skillId,
            const std::string& profileId,
            const std::string& executorName,
            armarx::aron::data::DictPtr parameters)>
            executeFluxioSkill;
 
        const std::function<std::experimental::observer_ptr<FluxioExecutor>(
            const std::vector<std::string>& parameterIds)>
            addMergerExecutorToDC;
 
        const std::function<std::experimental::observer_ptr<
            FluxioExecutor>(const std::string& id, const skills::FluxioSkill& skill, bool isRetry)>
            addLoopExecutorToDC;
 
        std::optional<skills::SkillExecutionID> executionId = std::nullopt;
        const skills::FluxioSkill& skill;
        std::map<std::string, std::experimental::observer_ptr<FluxioExecutor>>
            subExecutionsMap; // key is node id
        std::shared_mutex subExecutionsMapMutex;
 
        const armarx::Duration pollingFrequency = armarx::Duration::MilliSeconds(250);
    };
} // namespace armarx::skills