actor.hpp

Go to the documentation of this file.

#pragma once
 
#include <memory>
#include <coroio/corochain.hpp>
#include "messages.hpp"

 
namespace NNet {
namespace NActors {
 
class TActorSystem;

using TLocalActorId = uint32_t;

using TNodeId = uint16_t;

using TCookie = uint16_t;

using TMessageId = uint32_t;

class TActorId {
public:
    TActorId() = default;

    operator bool() const {
        return !((NodeId_ == 0) & (ActorId_ == 0) & (Cookie_ == 0));
    }

    TNodeId NodeId() const {
        return NodeId_;
    }

    TLocalActorId ActorId() const {
        return ActorId_;
    }

    TCookie Cookie() const {
        return Cookie_;
    }

    std::string ToString() const {
        return "ActorId:"
                + std::to_string(NodeId_) + ":"
                + std::to_string(ActorId_) + ":"
                + std::to_string(Cookie_);
    }

    TActorId(TNodeId nodeId, TLocalActorId actorId, TCookie cookie)
        : NodeId_(nodeId)
        , ActorId_(actorId)
        , Cookie_(cookie)
    { }
 
private:
    TLocalActorId ActorId_ = 0;  
    TNodeId NodeId_ = 0;         
    TCookie Cookie_ = 0;         
};

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>
    void Send(TActorId to, T&& message);

    template<typename T>
    void Forward(TActorId to, T&& message);

    TEvent Schedule(TTime when, TActorId sender, TActorId recipient, TMessageId messageId, TBlob blob);

    template<typename T>
    TEvent Schedule(TTime when, TActorId sender, TActorId recipient, T&& message);

    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