MediatorSystem.hpp

#pragma once
#include <samchon/API.hpp>

#include <samchon/templates/slave/SlaveSystem.hpp>
#include <samchon/protocol/IListener.hpp>

#include <samchon/templates/parallel/base/ParallelSystemArrayBase.hpp>
#include <samchon/templates/distributed/base/DistributedSystemArrayBase.hpp>
#include <samchon/templates/distributed/base/DistributedProcessBase.hpp>

#include <mutex>
#include <samchon/HashMap.hpp>

namespace samchon
{
namespace templates
{
namespace parallel
{
    class MediatorSystem
        : public virtual slave::SlaveSystem,
        public virtual protocol::IListener
    {
    private:
        typedef slave::SlaveSystem super;

        external::base::ExternalSystemArrayBase *system_array_;
        HashMap<size_t, std::shared_ptr<slave::InvokeHistory>> progress_list_;

        std::mutex mtx_;

    public:
        /* ---------------------------------------------------------
            CONSTRUCTORS
        --------------------------------------------------------- */
        MediatorSystem(external::base::ExternalSystemArrayBase* systemArray)
            : super()
        {
            this->system_array_ = systemArray;
        };

        virtual ~MediatorSystem() = default;

        virtual void start() = 0;

        /* ---------------------------------------------------------
            ACCESSORS
        --------------------------------------------------------- */
        template <class SystemArray>
        auto getSystemArray() const -> SystemArray*
        {
            return (SystemArray*)system_array_;
        };

        /* ---------------------------------------------------------
            INVOKE MESSAGE CHAIN
        --------------------------------------------------------- */
        void _Complete_history(size_t uid)
        {
            std::unique_lock<std::mutex> uk(mtx_);

            //--------
            // NEED TO REDEFINE START AND END TIME
            //--------
            // NO SUCH HISTORY; THE PROCESS HAD DONE ONLY IN THIS MEDIATOR LEVEL.
            if (progress_list_.has(uid) == false)
                return;

            // COMPLETE THE HISTORY
            std::shared_ptr<slave::InvokeHistory> history = progress_list_.get(uid);
            history->complete();

            // ERASE THE HISTORY ON PROGRESS LIST
            progress_list_.erase(uid);

            // REPORT THE HISTORY TO MASTER
            std::thread(&MediatorSystem::sendData, this, history->toInvoke()).detach();
        };

    private:
        virtual void _Reply_data(std::shared_ptr<protocol::Invoke> invoke) override final
        {
            if (invoke->has("_History_uid") == true)
            {
                // REGISTER THIS PROCESS ON HISTORY LIST
                std::shared_ptr<slave::InvokeHistory> history(new slave::InvokeHistory(invoke));
                progress_list_.insert({ history->getUID(), history });

                if (invoke->has("_Piece_first") == true)
                {
                    // PARALLEL PROCESS
                    size_t first = invoke->get("_Piece_first")->getValue<size_t>();
                    size_t last = invoke->get("_Piece_last")->getValue<size_t>();

                    invoke->erase(invoke->end() - 2, invoke->end());
                    ((base::ParallelSystemArrayBase*)system_array_)->sendPieceData(invoke, first, last);
                }
                else if (invoke->has("_Process_name") == true)
                {
                    // DISTRIBUTED PROCESS
                    auto ds_system_array = (distributed::base::DistributedSystemArrayBase*)system_array_;

                    // FIND THE MATCHED ROLE
                    const std::string &process_name = invoke->get("_Process_name")->getValue<std::string>();
                    if (ds_system_array->hasProcess(process_name) == false)
                        return;

                    // SEND DATA VIA THE ROLE
                    auto process = ds_system_array->getProcess(process_name);
                    ((distributed::base::DistributedProcessBase*)(process.get()))->sendData(invoke, 1.0);
                }
            }
            else
                replyData(invoke);
        };

    public:
        virtual void replyData(std::shared_ptr<protocol::Invoke> invoke) override
        {
            ((protocol::IProtocol*)system_array_)->sendData(invoke);
        };
    };
};
};
};

#include <samchon/templates/parallel/MediatorClient.hpp>
#include <samchon/templates/parallel/MediatorWebClient.hpp>
#include <samchon/templates/parallel/MediatorServer.hpp>
#include <samchon/templates/parallel/MediatorWebServer.hpp>