gRPCModule.h

Go to the documentation of this file.

// This file is part of DynExp.
 
#pragma once
 
#include "stdafx.h"
#include "DynExpCore.h"
#include "../HardwareAdapters/HardwareAdaptergRPC.h"
#include "../MetaInstruments/gRPCInstrument.h"
 
namespace DynExpModule
{
    template <typename... gRPCServices>
    class gRPCModule;
 
    template <typename... gRPCServices>
    class gRPCModuleData : public DynExp::ModuleDataBase
    {
    public:
        gRPCModuleData() { Init(); }
        virtual ~gRPCModuleData() = default;
 
    private:
        void ResetImpl(dispatch_tag<ModuleDataBase>) override final { Init(); }
 
        virtual void ResetImpl(dispatch_tag<gRPCModuleData>) {};
 
        void Init() {}
    };
 
    template <typename... gRPCServices>
    class gRPCModuleParams : public DynExp::ModuleParamsBase
    {
    public:
        gRPCModuleParams(DynExp::ItemIDType ID, const DynExp::DynExpCore& Core)
            : ModuleParamsBase(ID, Core), NetworkParams(*this) {}
 
        virtual ~gRPCModuleParams() = default;
 
        virtual const char* GetParamClassTag() const noexcept override { return "gRPCModuleParams"; }
 
        DynExp::NetworkParamsExtension NetworkParams;   
 
    private:
        void ConfigureParamsImpl(dispatch_tag<ModuleParamsBase>) override final { ConfigureParamsImpl(dispatch_tag<gRPCModuleParams>()); }
 
        virtual void ConfigureParamsImpl(dispatch_tag<gRPCModuleParams>) {}
 
        virtual const DynExp::NetworkParamsExtension* GetNetworkAddressParamsChild() const noexcept override { return &NetworkParams; }
    };
 
    template <typename... gRPCServices>
    class gRPCModuleConfigurator : public DynExp::ModuleConfiguratorBase
    {
    public:
        using ObjectType = gRPCModule<gRPCServices...>;
        using ParamsType = gRPCModuleParams<gRPCServices...>;
 
        gRPCModuleConfigurator() = default;
        virtual ~gRPCModuleConfigurator() = default;
    };
 
    template <typename... gRPCServices>
    class gRPCModule : public DynExp::ModuleBase
    {
        template <typename gRPCService>
        using ServicePtrType = std::unique_ptr<typename gRPCService::AsyncService>;
 
    public:
        using ParamsType = gRPCModuleParams<gRPCServices...>;                               
        using ConfigType = gRPCModuleConfigurator<gRPCServices...>;                         
        using ModuleDataType = gRPCModuleData<gRPCServices...>;                             
 
        constexpr static auto Name() noexcept { return "gRPC Module"; }                     
        constexpr static auto Category() noexcept { return "Network Modules (Servers)"; }   
 
        gRPCModule(const std::thread::id OwnerThreadID, DynExp::ParamsBasePtrType&& Params)
            : ModuleBase(OwnerThreadID, std::move(Params)),
            ServicePtrs(MakeServicePtrTuple()), ServerRunning(false) {}
 
        virtual ~gRPCModule() = default;
 
        virtual std::string GetName() const override { return Name(); }                     
        virtual std::string GetCategory() const override { return Category(); }             
 
        bool TreatModuleExceptionsAsWarnings() const override { return ServerRunning; }
 
        std::chrono::milliseconds GetMainLoopDelay() const override final { return std::chrono::milliseconds(1); }
 
        grpc::ServerCompletionQueue* GetServerQueue() const noexcept { return ServerQueue.get(); }
 
        template <size_t Index>
        auto& GetService() const noexcept { return *std::get<Index>(ServicePtrs); }
 
        template <typename T>
        auto& GetService() const noexcept { return *std::get<ServicePtrType<T>>(ServicePtrs); }
 
    protected:
        class CallDataBase
        {
            enum class StateType {
                Init,       
                Process,    
                Exit        
            };
 
            using StateMachineStateType = Util::StateMachineState<StateType(CallDataBase::*)(DynExp::ModuleInstance&)>;
 
        protected:
            CallDataBase(const gRPCModule* const OwningModule) noexcept
                : OwningModule(OwningModule), StateMachine(InitState, ProcessState, ExitState) {}
 
        public:
            virtual ~CallDataBase() {}
 
            auto GetOwningModule() const noexcept { return OwningModule; }
 
            void Proceed(DynExp::ModuleInstance& Instance) { StateMachine.Invoke(*this, Instance); }
 
        protected:
            auto* GetServerContext() noexcept { return &ServerContext; }
 
        private:
 
            virtual void InitChild(DynExp::ModuleInstance& Instance) = 0;
 
            virtual void ProcessChild(DynExp::ModuleInstance& Instance) = 0;
 
            StateType InitStateFunc(DynExp::ModuleInstance& Instance)
            {
                InitChild(Instance);
 
                return StateType::Process;
            }
 
            StateType ProcessStateFunc(DynExp::ModuleInstance& Instance)
            {
                ProcessChild(Instance);
 
                return StateType::Exit;
            }
 
            StateType ExitStateFunc(DynExp::ModuleInstance& Instance)
            {
                delete this;
 
                return StateType::Exit;
            }
 
            static constexpr auto InitState = Util::StateMachineState(StateType::Init, &CallDataBase::InitStateFunc);
 
            static constexpr auto ProcessState = Util::StateMachineState(StateType::Process, &CallDataBase::ProcessStateFunc);
 
            static constexpr auto ExitState = Util::StateMachineState(StateType::Exit, &CallDataBase::ExitStateFunc, "", true);
 
            const gRPCModule* const OwningModule;                       
            Util::StateMachine<StateMachineStateType> StateMachine;     
            grpc::ServerContext ServerContext;                          
        };
 
        template <typename DerivedType, typename gRPCService, typename RequestMessageType, typename ResponseMessageType,
            typename std::enable_if_t<Util::is_contained_in_v<gRPCService, gRPCServices...>, int> = 0>
        class TypedCallDataBase : public CallDataBase
        {
        public:
            static void MakeCall(const gRPCModule* const OwningModule, DynExp::ModuleInstance& Instance) { (new DerivedType(OwningModule))->Proceed(Instance); }
 
        private:
            friend DerivedType;
 
            using ResponseWriterType = grpc::ServerAsyncResponseWriter<ResponseMessageType>;
 
            using RequestFuncType = std::function<void(typename gRPCService::AsyncService*, grpc::ServerContext*,
                RequestMessageType*, ResponseWriterType*, grpc::CompletionQueue*, grpc::ServerCompletionQueue*, void*)>;
 
            TypedCallDataBase(const gRPCModule* const OwningModule, const RequestFuncType RequestFunc) noexcept
                : CallDataBase(OwningModule), RequestFunc(RequestFunc), ResponseWriter(this->GetServerContext()) {}
            
            virtual ~TypedCallDataBase() = default;
 
            void InitChild(DynExp::ModuleInstance& Instance) override final
            {
                // The address of *this* instance serves as the tag to distinguish multiple remote procedure calls.
                RequestFunc(&this->GetOwningModule()->template GetService<gRPCService>(), this->GetServerContext(),
                    &RequestMessage, &ResponseWriter, this->GetOwningModule()->GetServerQueue(), this->GetOwningModule()->GetServerQueue(), this);
            }
 
            void ProcessChild(DynExp::ModuleInstance& Instance) override final
            {
                MakeCall(this->GetOwningModule(), Instance);
                
                ProcessChildImpl(Instance);
 
                ResponseWriter.Finish(ResponseMessage, grpc::Status::OK, this);
            }
 
 
            virtual void ProcessChildImpl(DynExp::ModuleInstance& Instance) = 0;
 
            const RequestFuncType RequestFunc;      
            RequestMessageType RequestMessage;      
            ResponseMessageType ResponseMessage;    
            ResponseWriterType ResponseWriter;      
        };
 
    private:
        auto MakeServicePtrTuple() { return std::make_tuple(std::make_unique<typename ServicePtrType<gRPCServices>::element_type>()...); }
 
        Util::DynExpErrorCodes::DynExpErrorCodes ModuleMainLoop(DynExp::ModuleInstance& Instance) override final
        {
            void* Tag;
            bool IsOK;
 
            auto Result = ServerQueue->AsyncNext(&Tag, &IsOK, std::chrono::system_clock::now() + std::chrono::milliseconds(80));
 
            if (Result == grpc::CompletionQueue::NextStatus::GOT_EVENT && Tag && IsOK)
                static_cast<CallDataBase*>(Tag)->Proceed(Instance);
 
            return Util::DynExpErrorCodes::NoError;
        }
 
        void ResetImpl(dispatch_tag<ModuleBase>) override final
        {
            ServerQueue.reset();
            Server.reset();
 
            ServicePtrs = MakeServicePtrTuple();
            ServerRunning = false;
 
            ResetImpl(dispatch_tag<gRPCModule>());
        }
 
        virtual void ResetImpl(DynExp::Object::dispatch_tag<gRPCModule>) {}
 
 
        virtual void CreateInitialCallDataObjectsImpl(DynExp::Object::dispatch_tag<gRPCModule>, DynExp::ModuleInstance& Instance) const = 0;
 
        virtual void OnInitChild(DynExp::ModuleInstance* Instance) const {}
 
        virtual void OnExitChild(DynExp::ModuleInstance* Instance) const {}
 
 
        void OnInit(DynExp::ModuleInstance* Instance) const override final
        {
            std::string Address;
            std::string ObjName;
 
            {
                auto ModuleParams = DynExp::dynamic_Params_cast<gRPCModule>(Instance->ParamsGetter());
                Address = ModuleParams->NetworkParams.MakeAddress();
                ObjName = ModuleParams->ObjectName;
            } // ModuleParams unlocked here.
 
            grpc::ServerBuilder ServerBuilder;
            ServerBuilder.AddListeningPort(Address, grpc::InsecureServerCredentials());
            std::apply([&ServerBuilder](auto&... ServicePtr) { (ServerBuilder.RegisterService(ServicePtr.get()), ...); }, ServicePtrs);
            ServerQueue = ServerBuilder.AddCompletionQueue();
            Server = ServerBuilder.BuildAndStart();
 
            CreateInitialCallDataObjectsImpl(DynExp::Object::dispatch_tag<gRPCModule>(), *Instance);
            OnInitChild(Instance);
 
            ServerRunning = true;
            Util::EventLog().Log("gRPC server \"" + ObjName + "\" (" + GetCategoryAndName() + ") listening on " + Address + ".");
        }
 
        void OnExit(DynExp::ModuleInstance* Instance) const override final
        {
            OnExitChild(Instance);
            Shutdown();
 
            Util::EventLog().Log("gRPC server \"" + GetObjectName() + "\" (" + GetCategoryAndName() + ") shut down.");
        }
 
        void OnErrorChild(DynExp::ModuleInstance& Instance) const override final
        {
            Shutdown();
        }
 
        void Shutdown() const
        {
            if (Server)
                Server->Shutdown();
            if (ServerQueue)
                ServerQueue->Shutdown();
 
            DrainServerQueue();
        }
 
        void DrainServerQueue() const
        {
            if (!ServerQueue)
                return;
 
            bool Result = true;
            while (Result)
            {
                void* Tag = nullptr;
                bool IsOK;
                Result = ServerQueue->Next(&Tag, &IsOK);
 
                if (Result && Tag)
                    delete static_cast<CallDataBase*>(Tag);
            }
        }
 
        mutable std::unique_ptr<grpc::ServerCompletionQueue> ServerQueue;
 
        mutable std::unique_ptr<grpc::Server> Server;
 
        std::tuple<ServicePtrType<gRPCServices>...> ServicePtrs;
 
        mutable std::atomic<bool> ServerRunning;
    };
}