actor.hpp

Go to the documentation of this file.

#pragma once
 
#include <memory>
#include <coroio/corochain.hpp>
#include "messages.hpp"
#include "actorid.hpp"
 
namespace NNet {
namespace NActors {
 
class TActorSystem;
 
class TEnvelope
{
public:
    TActorId Sender;        
    TActorId Recipient;     
    TMessageId MessageId;   
    TBlob Blob;            
};
 
using TEvent = std::pair<unsigned, TTime>;
 
class TActorContext
{
public:
    using TPtr = std::unique_ptr<TActorContext>;
 
    TActorId Sender() const {
        return Sender_;
    }
 
    TActorId Self() const {
        return Self_;
    }
 
    void Send(TActorId to, TMessageId messageId, TBlob blob);
 
    void Forward(TActorId to, TMessageId messageId, TBlob blob);
 
    template<typename T, typename... Args>
    void Send(TActorId to, Args&&... args);
 
    template<typename T, typename... Args>
    void Forward(TActorId to, Args&&... args);
 
    TEvent Schedule(TTime when, TActorId sender, TActorId recipient, TMessageId messageId, TBlob blob);
 
    template<typename T, typename... Args>
    TEvent Schedule(TTime when, TActorId sender, TActorId recipient, Args&&... args);
 
    void Cancel(TEvent event);
 
    TFuture<void> Sleep(TTime until);
 
    template<typename Rep, typename Period>
    TFuture<void> Sleep(std::chrono::duration<Rep,Period> duration);
 
    template<typename T, typename TQuestion>
    TFuture<T> Ask(TActorId recipient, TQuestion&& question);
 
    struct TAsync {
        TAsync(TActorSystem* actorSystem, TLocalActorId actorId)
            : ActorSystem_(actorSystem)
            , ActorId_(actorId)
        { }
 
        void Commit(TFuture<void>&& future);
 
    private:
        TLocalActorId ActorId_;
        TActorSystem* ActorSystem_;
    };
 
    TAsync StartAsync();
 
    static void* operator new(size_t size, TActorSystem* actorSystem);
    static void operator delete(void* ptr);
 
private:
    TActorContext(TActorId sender, TActorId self, TActorSystem* actorSystem)
        : Sender_(sender)
        , Self_(self)
        , ActorSystem(actorSystem)
    { }
 
    TActorId Sender_;              
    TActorId Self_;                
    TActorSystem* ActorSystem = nullptr;  
 
    friend class TActorSystem;
    friend class TMockActorContext;
};
 
class TMockActorContext : public TActorContext {
public:
    TMockActorContext(TActorId sender, TActorId self, TActorSystem* actorSystem)
        : TActorContext(sender, self, actorSystem)
    { }
};
 
class IActor {
public:
    using TPtr = std::unique_ptr<IActor>;
    friend class TActorSystem;
 
    IActor() = default;
    virtual ~IActor() = default;
 
    virtual void Receive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) = 0;
};
 
class ICoroActor : public IActor {
public:
    void Receive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) override;
 
    virtual TFuture<void> CoReceive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) = 0;
};
 
class IBehavior {
public:
    virtual ~IBehavior() = default;
 
    virtual void Receive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) = 0;
};
 
template<typename TBaseBehavior, typename... TMessages>
class TBehavior : public IBehavior
{
public:
    void Receive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) override {
        bool handled = (TryHandleMessage<TMessages>(
            messageId,
            blob,
            ctx
        ) || ...);
 
        if (!handled) {
            static_cast<TBaseBehavior*>(this)->HandleUnknownMessage(messageId, std::move(blob), std::move(ctx));
        }
    }
 
private:
    template<typename TMessage>
    bool TryHandleMessage(TMessageId messageId, TBlob& blob, TActorContext::TPtr& ctx) {
        if (TMessage::MessageId == messageId) {
            if (blob.Type == TBlob::PointerType::Near) {
                auto&& mes = DeserializeNear<TMessage>(blob);
                HandleMessage<TMessage>(
                    std::move(mes),
                    std::move(blob),
                    std::move(ctx)
                );
            } else {
                auto&& mes = DeserializeFar<TMessage>(blob);
                HandleMessage<TMessage>(
                    std::move(mes),
                    std::move(blob),
                    std::move(ctx)
                );
            }
            return true;
        }
        return false;
    }
 
    template<typename TMessage>
    void HandleMessage(TMessage&& message, TBlob blob, TActorContext::TPtr ctx)
    {
        using ReturnType = decltype(static_cast<TBaseBehavior*>(this)->Receive(
            std::declval<TMessage>(),
            std::declval<TBlob>(),
            std::declval<TActorContext::TPtr>()
        ));
 
        if constexpr (std::is_same_v<ReturnType, void>) {
            static_cast<TBaseBehavior*>(this)->Receive(
                std::move(message),
                std::move(blob),
                std::move(ctx)
            );
        } else {
            auto async = ctx->StartAsync();
            auto future = static_cast<TBaseBehavior*>(this)->Receive(
                std::move(message),
                std::move(blob),
                std::move(ctx)
            );
            if (!future.done()) {
                async.Commit(std::move(future));
            }
        }
    }
};
 
class IBehaviorActor : public IActor {
public:
    void Become(IBehavior* behavior) {
        CurrentBehavior_ = behavior;
    }
 
    void Receive(TMessageId messageId, TBlob blob, TActorContext::TPtr ctx) override {
        CurrentBehavior_->Receive(messageId, std::move(blob), std::move(ctx));
    }
 
private:
    IBehavior* CurrentBehavior_;
};
 
} // namespace NActors
} // namespace NNet