MessageEndpoint.h

Go to the documentation of this file.

// Copyright Epic Games, Inc. All Rights Reserved.
 
#pragma once
 
#include "CoreTypes.h"
#include "Async/TaskGraphInterfaces.h"
#include "Containers/Array.h"
#include "Containers/ArrayBuilder.h"
#include "Containers/Queue.h"
#include "IMessageAttachment.h"
#include "IMessageBus.h"
#include "IMessageContext.h"
#include "IMessageHandler.h"
#include "IMessageReceiver.h"
#include "IMessageSender.h"
#include "IMessageBusListener.h"
#include "Misc/Guid.h"
#include "Templates/SharedPointer.h"
#include "UObject/NameTypes.h"
#include "Misc/ScopeLock.h"
 
 
DECLARE_DELEGATE_TwoParams(FOnMessageEndpointError, const IMessageContext&, const FString&);
 
 
struct FMessageBusNotification
{
    EMessageBusNotification NotificationType;
 
    FMessageAddress RegistrationAddress;
};
 
DECLARE_DELEGATE_OneParam(FOnBusNotification, const FMessageBusNotification&);
 
class FMessageEndpoint
    : public TSharedFromThis<FMessageEndpoint, ESPMode::ThreadSafe>
    , public IMessageReceiver
    , public IMessageSender
    , public IBusListener
{
public:
 
    typedef struct FMessageEndpointBuilder Builder;
 
    FMessageEndpoint(const FName& InName, const TSharedRef<IMessageBus, ESPMode::ThreadSafe>& InBus, const TArray<TSharedPtr<IMessageHandler, ESPMode::ThreadSafe>>& InHandlers, const FOnBusNotification InNotificationDelegate)
        : Address(FMessageAddress::NewAddress())
        , BusPtr(InBus)
        , Enabled(true)
        , Handlers(InHandlers)
        , NotificationDelegate(InNotificationDelegate)
        , Id(FGuid::NewGuid())
        , InboxEnabled(false)
        , Name(InName)
    {
        SetRecipientThread(FTaskGraphInterface::Get().GetCurrentThreadIfKnown());
    }
 
    ~FMessageEndpoint()
    {   
        auto Bus = BusPtr.Pin();
 
        if (Bus.IsValid())
        {
            Bus->Unregister(Address);
        }
    }
 
public:
 
    void Disable()
    {
        Enabled = false;
    }
 
    void Enable()
    {
        Enabled = true;
    }
 
    const FMessageAddress& GetAddress() const
    {
        return Address;
    }
 
    bool IsConnected() const
    {
        return BusPtr.IsValid();
    }
 
    bool IsEnabled() const
    {
        return Enabled;
    }
 
    void SetRecipientThread(const ENamedThreads::Type& NamedThread)
    {
        RecipientThread = ENamedThreads::GetThreadIndex(NamedThread);
    }
 
public:
 
    void Defer(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context, const FTimespan& Delay)
    {
        TSharedPtr<IMessageBus, ESPMode::ThreadSafe> Bus = GetBusIfEnabled();
 
        if (Bus.IsValid())
        {
            Bus->Forward(Context, TArrayBuilder<FMessageAddress>().Add(Address), Delay, AsShared());
        }
    }
 
    void Forward(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context, const TArray<FMessageAddress>& Recipients, const FTimespan& Delay)
    {
        TSharedPtr<IMessageBus, ESPMode::ThreadSafe> Bus = GetBusIfEnabled();
 
        if (Bus.IsValid())
        {
            Bus->Forward(Context, Recipients, Delay, AsShared());
        }
    }
 
    void Publish(void* Message, UScriptStruct* TypeInfo, EMessageScope Scope, const FTimespan& Delay, const FDateTime& Expiration)
    {
        Publish(Message, TypeInfo, Scope, TMap<FName, FString>(), Delay, Expiration);
    }
 
    void Publish(void* Message, UScriptStruct* TypeInfo, EMessageScope Scope, const TMap<FName, FString> Annotations, const FTimespan& Delay, const FDateTime& Expiration)
    {
        TSharedPtr<IMessageBus, ESPMode::ThreadSafe> Bus = GetBusIfEnabled();
 
        if (Bus.IsValid())
        {
            Bus->Publish(Message, TypeInfo, Scope, Annotations, Delay, Expiration, AsShared());
        }
    }
 
    UE_DEPRECATED(4.21, "FMessageEndpoint::Send with 6 params is deprecated. Please use FMessageEndpoint::Send that takes additionnal EMessageFlags instead!")
    void Send(void* Message, UScriptStruct* TypeInfo, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const TArray<FMessageAddress>& Recipients, const FTimespan& Delay, const FDateTime& Expiration)
    {
        Send(Message, TypeInfo, EMessageFlags::None, Attachment, Recipients, Delay, Expiration);
    }
 
    void Send(void* Message, UScriptStruct* TypeInfo, EMessageFlags Flags, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const TArray<FMessageAddress>& Recipients, const FTimespan& Delay, const FDateTime& Expiration)
    {
        TSharedPtr<IMessageBus, ESPMode::ThreadSafe> Bus = GetBusIfEnabled();
 
        if (Bus.IsValid())
        {
            Bus->Send(Message, TypeInfo, Flags, TMap<FName, FString>(), Attachment, Recipients, Delay, Expiration, AsShared());
        }
    }
 
    void Send(void* Message, UScriptStruct* TypeInfo, EMessageFlags Flags, const TMap<FName, FString> Annotations, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const TArray<FMessageAddress>& Recipients, const FTimespan& Delay, const FDateTime& Expiration)
    {
        TSharedPtr<IMessageBus, ESPMode::ThreadSafe> Bus = GetBusIfEnabled();
 
        if (Bus.IsValid())
        {
            Bus->Send(Message, TypeInfo, Flags, Annotations, Attachment, Recipients, Delay, Expiration, AsShared());
        }
    }
 
    UE_DEPRECATED(5.1, "Types names are now represented by path names. Please use a version of this function that takes an FTopLevelAssetPath as MessageType.")
    void Subscribe(const FName& MessageType, const FMessageScopeRange& ScopeRange)
    {
        Subscribe(UClass::TryConvertShortTypeNameToPathName<UStruct>(MessageType.ToString()), ScopeRange);
    }
 
    void Subscribe(const FTopLevelAssetPath& MessageType, const FMessageScopeRange& ScopeRange)
    {
        TSharedPtr<IMessageBus, ESPMode::ThreadSafe> Bus = GetBusIfEnabled();
 
        if (Bus.IsValid())
        {
            Bus->Subscribe(AsShared(), MessageType, ScopeRange);
        }
    }
 
    UE_DEPRECATED(5.1, "Types names are now represented by path names. Please use a version of this function that takes an FTopLevelAssetPath as MessageType.")
    void Unsubscribe(const FName& MessageType)
    {
        Unsubscribe(UClass::TryConvertShortTypeNameToPathName<UStruct>(MessageType.ToString()));
    }
 
    void Unsubscribe(const FTopLevelAssetPath& MessageType)
    {
        TSharedPtr<IMessageBus, ESPMode::ThreadSafe> Bus = GetBusIfEnabled();
 
        if (Bus.IsValid())
        {
            Bus->Unsubscribe(AsShared(), MessageType);
        }
    }
 
public:
 
    void DisableInbox()
    {
        InboxEnabled = false;
    }
 
    void EnableInbox()
    {
        InboxEnabled = true;
    }
 
    bool IsInboxEmpty() const
    {
        return Inbox.IsEmpty();
    }
 
    bool IsInboxEnabled() const
    {
        return InboxEnabled;
    }
 
    void ProcessInbox()
    {
        TSharedPtr<IMessageContext, ESPMode::ThreadSafe> Context;
 
        while (Inbox.Dequeue(Context))
        {
            ProcessMessage(Context.ToSharedRef());
        }
    }
 
    bool ReceiveFromInbox(TSharedPtr<IMessageContext, ESPMode::ThreadSafe>& OutContext)
    {
        return Inbox.Dequeue(OutContext);
    }
 
public:
 
    //~ IMessageReceiver interface
 
    virtual FName GetDebugName() const override
    {
        return Name;
    }
 
    virtual const FGuid& GetRecipientId() const override
    {
        return Id;
    }
 
    virtual ENamedThreads::Type GetRecipientThread() const override
    {
        return RecipientThread;
    }
 
    virtual bool IsLocal() const override
    {
        return true;
    }
 
    virtual void ReceiveMessage(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context) override
    {
        if (!Enabled)
        {
            return;
        }
 
        if (InboxEnabled)
        {
            Inbox.Enqueue(Context);
        }
        else
        {
            ProcessMessage(Context);
        }
    }
 
    //~ IBusListener interface
 
    virtual ENamedThreads::Type GetListenerThread() const override
    {
        return RecipientThread;
    }
 
    virtual void NotifyRegistration(const FMessageAddress& InAddress, EMessageBusNotification InNotification)
    {
        if (!Enabled)
        {
            return;
        }
 
        NotificationDelegate.ExecuteIfBound(FMessageBusNotification{ InNotification, InAddress });
    }
 
public:
 
    //~ IMessageSender interface
 
    virtual FMessageAddress GetSenderAddress() override
    {
        return Address;
    }
 
    virtual void NotifyMessageError(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context, const FString& Error) override
    {
        ErrorDelegate.ExecuteIfBound(Context.Get(), Error);
    }
 
public:
 
    template<typename T, typename... InArgTypes>
    static T* MakeMessage(InArgTypes&&... Args)
    {
        void* Buffer = FMemory::Malloc(sizeof(T));
 
        T* Message = new (Buffer) T(::Forward<InArgTypes>(Args)...);
 
        return Message;
    }
 
    void Forward(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context, const FMessageAddress& Recipient)
    {
        Forward(Context, TArrayBuilder<FMessageAddress>().Add(Recipient), FTimespan::Zero());
    }
 
    void Forward(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context, const FMessageAddress& Recipient, const FTimespan& Delay)
    {
        Forward(Context, TArrayBuilder<FMessageAddress>().Add(Recipient), Delay);
    }
 
    void Forward(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context, const TArray<FMessageAddress>& Recipients)
    {
        Forward(Context, Recipients, FTimespan::Zero());
    }
 
    template<typename MessageType>
    void Publish(MessageType* Message)
    {
        Publish(Message, MessageType::StaticStruct(), EMessageScope::Network, FTimespan::Zero(), FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Publish(MessageType* Message, EMessageScope Scope)
    {
        Publish(Message, MessageType::StaticStruct(), Scope, FTimespan::Zero(), FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Publish(MessageType* Message, const TMap<FName, FString> Annotations)
    {
        Publish(Message, MessageType::StaticStruct(), EMessageScope::Network, Annotations, FTimespan::Zero(), FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Publish(MessageType* Message, const TMap<FName, FString> Annotations, EMessageScope Scope)
    {
        Publish(Message, MessageType::StaticStruct(), Scope, Annotations, FTimespan::Zero(), FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Publish(MessageType* Message, const FTimespan& Delay)
    {
        Publish(Message, MessageType::StaticStruct(), EMessageScope::Network, Delay, FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Publish(MessageType* Message, EMessageScope Scope, const FTimespan& Delay)
    {
        Publish(Message, MessageType::StaticStruct(), Scope, Delay, FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Publish(MessageType* Message, EMessageScope Scope, const FTimespan& Delay, const FDateTime& Expiration)
    {
        Publish(Message, MessageType::StaticStruct(), Scope, Delay, Expiration);
    }
 
    template<typename MessageType>
    void Publish(MessageType* Message, const TMap<FName, FString> Annotations, EMessageScope Scope, const FTimespan& Delay, const FDateTime& Expiration)
    {
        Publish(Message, MessageType::StaticStruct(), Scope, Annotations, Delay, Expiration);
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const FMessageAddress& Recipient)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, nullptr, TArrayBuilder<FMessageAddress>().Add(Recipient), FTimespan::Zero(), FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const TMap<FName, FString> Annotations, const FMessageAddress& Recipient)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, Annotations, nullptr, TArrayBuilder<FMessageAddress>().Add(Recipient), FTimespan::Zero(), FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const FMessageAddress& Recipient, const FTimespan& Delay)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, nullptr, TArrayBuilder<FMessageAddress>().Add(Recipient), Delay, FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const FMessageAddress& Recipient, const FTimespan& Delay, const FDateTime& Expiration)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, nullptr, TArrayBuilder<FMessageAddress>().Add(Recipient), Delay, Expiration);
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const FMessageAddress& Recipient)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, Attachment, TArrayBuilder<FMessageAddress>().Add(Recipient), FTimespan::Zero(), FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const FMessageAddress& Recipient, const FDateTime& Expiration, const FTimespan& Delay)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, Attachment, TArrayBuilder<FMessageAddress>().Add(Recipient), Delay, Expiration);
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const TMap<FName, FString> Annotations, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const FMessageAddress& Recipient, const FDateTime& Expiration, const FTimespan& Delay)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, Annotations, Attachment, TArrayBuilder<FMessageAddress>().Add(Recipient), Delay, Expiration);
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const TArray<FMessageAddress>& Recipients)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, nullptr, Recipients, FTimespan::Zero(), FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const TArray<FMessageAddress>& Recipients, const FTimespan& Delay)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, nullptr, Recipients, Delay, FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const TArray<FMessageAddress>& Recipients, const FTimespan& Delay)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, Attachment, Recipients, Delay, FDateTime::MaxValue());
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const TArray<FMessageAddress>& Recipients, const FTimespan& Delay, const FDateTime& Expiration)
    {
        Send(Message, MessageType::StaticStruct(), EMessageFlags::None, Attachment, Recipients, Delay, Expiration);
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, EMessageFlags Flags, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const TArray<FMessageAddress>& Recipients, const FTimespan& Delay, const FDateTime& Expiration)
    {
        Send(Message, MessageType::StaticStruct(), Flags, Attachment, Recipients, Delay, Expiration);
    }
 
    template<typename MessageType>
    void Send(MessageType* Message, EMessageFlags Flags, const TMap<FName, FString> Annotations, const TSharedPtr<IMessageAttachment, ESPMode::ThreadSafe>& Attachment, const TArray<FMessageAddress>& Recipients, const FTimespan& Delay, const FDateTime& Expiration)
    {
        Send(Message, MessageType::StaticStruct(), Flags, Annotations, Attachment, Recipients, Delay, Expiration);
    }
 
    template<class MessageType>
    void Subscribe()
    {
        Subscribe(MessageType::StaticStruct()->GetStructPathName(), FMessageScopeRange::AtLeast(EMessageScope::Thread));
    }
 
    template<class MessageType>
    void Subscribe(const FMessageScopeRange& ScopeRange)
    {
        Subscribe(MessageType::StaticStruct()->GetStructPathName(), ScopeRange);
    }
 
    void Unsubscribe()
    {
        Unsubscribe(IMessageBus::PATHNAME_All);
    }
 
    template<class MessageType>
    void Unsubscribe()
    {
        Unsubscribe(MessageType::StaticStruct()->GetStructPathName());
    }
 
public:
 
    static void SafeRelease(TSharedPtr<FMessageEndpoint, ESPMode::ThreadSafe>& Endpoint)
    {
        if (Endpoint.IsValid())
        {
            Endpoint->ClearHandlers();
            Endpoint.Reset();
        }
    }
 
protected:
 
    void ClearHandlers()
    {
        FScopeLock Lock(&HandlersCS);
        Handlers.Empty();
    }
 
    inline TSharedPtr<IMessageBus, ESPMode::ThreadSafe> GetBusIfEnabled() const
    {
        if (Enabled)
        {
            return BusPtr.Pin();
        }
 
        return nullptr;
    }
 
    void ProcessMessage(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context)
    {
        if (!Context->IsValid())
        {
            return;
        }
 
        FScopeLock Lock(&HandlersCS);
 
        for (int32 HandlerIndex = 0; HandlerIndex < Handlers.Num(); ++HandlerIndex)
        {
            Handlers[HandlerIndex]->HandleMessage(Context);
        }
    }
 
private:
 
    const FMessageAddress Address;
 
    TWeakPtr<IMessageBus, ESPMode::ThreadSafe> BusPtr;
 
    bool Enabled;
 
    TArray<TSharedPtr<IMessageHandler, ESPMode::ThreadSafe>> Handlers;
 
    FOnBusNotification NotificationDelegate;
 
    const FGuid Id;
 
    TQueue<TSharedPtr<IMessageContext, ESPMode::ThreadSafe>, EQueueMode::Mpsc> Inbox;
 
    bool InboxEnabled;
 
    const FName Name;
 
    ENamedThreads::Type RecipientThread;
 
private:
 
    FOnMessageEndpointError ErrorDelegate;
 
    FCriticalSection        HandlersCS;
};