MessageRouter.h

Go to the documentation of this file.

// Copyright Epic Games, Inc. All Rights Reserved.
 
#pragma once
 
#include "CoreMinimal.h"
#include "Containers/Queue.h"
#include "HAL/Runnable.h"
#include "Misc/SingleThreadRunnable.h"
#include "Templates/Atomic.h"
#include "Containers/MpscQueue.h"
 
#include "IMessageContext.h"
#include "IMessageTracer.h"
#include "Bus/MessageTracer.h"
 
class IMessageInterceptor;
class IMessageReceiver;
class IMessageSubscription;
class IBusListener;
 
enum class EMessageBusNotification : uint8;
 
class FMessageRouter
    : public FRunnable
    , private FSingleThreadRunnable
{
    DECLARE_DELEGATE(CommandDelegate)
 
public:
 
    FMessageRouter();
 
    ~FMessageRouter();
 
public:
 
    FORCEINLINE void AddInterceptor(const TSharedRef<IMessageInterceptor, ESPMode::ThreadSafe>& Interceptor, const FTopLevelAssetPath& MessageType)
    {
        EnqueueCommand(FSimpleDelegate::CreateRaw(this, &FMessageRouter::HandleAddInterceptor, Interceptor, MessageType));
    }
 
    FORCEINLINE void AddRecipient(const FMessageAddress& Address, const TSharedRef<IMessageReceiver, ESPMode::ThreadSafe>& Recipient)
    {
        EnqueueCommand(FSimpleDelegate::CreateRaw(this, &FMessageRouter::HandleAddRecipient, Address, TWeakPtr<IMessageReceiver, ESPMode::ThreadSafe>(Recipient)));
    }
 
    FORCEINLINE void AddSubscription(const TSharedRef<IMessageSubscription, ESPMode::ThreadSafe>& Subscription)
    {
        EnqueueCommand(FSimpleDelegate::CreateRaw(this, &FMessageRouter::HandleAddSubscriber, Subscription));
    }
 
    FORCEINLINE TSharedRef<IMessageTracer, ESPMode::ThreadSafe> GetTracer()
    {
        return Tracer;
    }
 
    FORCEINLINE void RemoveInterceptor(const TSharedRef<IMessageInterceptor, ESPMode::ThreadSafe>& Interceptor, const FTopLevelAssetPath& MessageType)
    {
        EnqueueCommand(FSimpleDelegate::CreateRaw(this, &FMessageRouter::HandleRemoveInterceptor, Interceptor, MessageType));
    }
 
    FORCEINLINE void RemoveRecipient(const FMessageAddress& Address)
    {
        EnqueueCommand(FSimpleDelegate::CreateRaw(this, &FMessageRouter::HandleRemoveRecipient, Address));
    }
 
    FORCEINLINE void RemoveSubscription(const TSharedRef<IMessageReceiver, ESPMode::ThreadSafe>& Subscriber, const FTopLevelAssetPath& MessageType)
    {
        EnqueueCommand(FSimpleDelegate::CreateRaw(this, &FMessageRouter::HandleRemoveSubscriber, TWeakPtr<IMessageReceiver, ESPMode::ThreadSafe>(Subscriber), MessageType));
    }
 
    FORCEINLINE void RouteMessage(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context)
    {
        Tracer->TraceSentMessage(Context);
        EnqueueCommand(FSimpleDelegate::CreateRaw(this, &FMessageRouter::HandleRouteMessage, Context));
    }
 
    FORCEINLINE void AddNotificationListener(const TSharedRef<IBusListener, ESPMode::ThreadSafe>& Listener)
    {
        EnqueueCommand(FSimpleDelegate::CreateRaw(this, &FMessageRouter::HandleAddListener, TWeakPtr<IBusListener, ESPMode::ThreadSafe>(Listener)));
    }
 
    FORCEINLINE void RemoveNotificationListener(const TSharedRef<IBusListener, ESPMode::ThreadSafe>& Listener)
    {
        EnqueueCommand(FSimpleDelegate::CreateRaw(this, &FMessageRouter::HandleRemoveListener, TWeakPtr<IBusListener, ESPMode::ThreadSafe>(Listener)));
    }
 
public:
 
    //~ FRunnable interface
 
    virtual FSingleThreadRunnable* GetSingleThreadInterface() override;
    virtual bool Init() override;
    virtual uint32 Run() override;
    virtual void Stop() override;
    virtual void Exit() override;
 
protected:
 
    FTimespan CalculateWaitTime();
 
    FORCEINLINE bool EnqueueCommand(CommandDelegate Command)
    {
        Commands.Enqueue(Command);
        WorkEvent->Trigger();
 
        return true;
    }
 
    void FilterSubscriptions(
        TArray<TSharedPtr<IMessageSubscription, ESPMode::ThreadSafe>>& Subscriptions,
        const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context,
        TArray<TSharedPtr<IMessageReceiver, ESPMode::ThreadSafe>>& OutRecipients);
 
    void FilterRecipients(
        const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Context,
        TArray<TSharedPtr<IMessageReceiver, ESPMode::ThreadSafe>>& OutRecipients);
 
    void DispatchMessage(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& Message);
 
    void ProcessCommands();
 
    void ProcessDelayedMessages();
 
protected:
 
    //~ FSingleThreadRunnable interface
 
    virtual void Tick() override;
 
private:
 
    struct FDelayedMessage
    {
        TSharedPtr<IMessageContext, ESPMode::ThreadSafe> Context;
 
        int64 Sequence;
 
 
        FDelayedMessage() { }
 
        FDelayedMessage(const TSharedRef<IMessageContext, ESPMode::ThreadSafe>& InContext, int64 InSequence)
            : Context(InContext)
            , Sequence(InSequence)
        { }
 
        bool operator<(const FDelayedMessage& Other) const
        {
            const FTimespan Difference = Other.Context->GetTimeSent() - Context->GetTimeSent();
 
            if (Difference.IsZero())
            {
                return (Sequence < Other.Sequence);
            }
 
            return (Difference > FTimespan::Zero());
        }
    };
 
private:
 
    void HandleAddInterceptor(TSharedRef<IMessageInterceptor, ESPMode::ThreadSafe> Interceptor, FTopLevelAssetPath MessageType);
 
    void HandleAddRecipient(FMessageAddress Address, TWeakPtr<IMessageReceiver, ESPMode::ThreadSafe> RecipientPtr);
 
    void HandleAddSubscriber(TSharedRef<IMessageSubscription, ESPMode::ThreadSafe> Subscription);
 
    void HandleRemoveInterceptor(TSharedRef<IMessageInterceptor, ESPMode::ThreadSafe> Interceptor, FTopLevelAssetPath MessageType);
 
    void HandleRemoveRecipient(FMessageAddress Address);
 
    void HandleRemoveSubscriber(TWeakPtr<IMessageReceiver, ESPMode::ThreadSafe> SubscriberPtr, FTopLevelAssetPath MessageType);
 
    void HandleRouteMessage(TSharedRef<IMessageContext, ESPMode::ThreadSafe> Context);
 
    void HandleAddListener(TWeakPtr<IBusListener, ESPMode::ThreadSafe> ListenerPtr);
 
    void HandleRemoveListener(TWeakPtr<IBusListener, ESPMode::ThreadSafe> ListenerPtr);
 
    void NotifyRegistration(const FMessageAddress& Address, EMessageBusNotification Notification);
 
private:
 
    TMap<FTopLevelAssetPath, TArray<TSharedPtr<IMessageInterceptor, ESPMode::ThreadSafe>>> ActiveInterceptors;
 
    TMap<FMessageAddress, TWeakPtr<IMessageReceiver, ESPMode::ThreadSafe>> ActiveRecipients;
 
    TMap<FTopLevelAssetPath, TArray<TSharedPtr<IMessageSubscription, ESPMode::ThreadSafe>>> ActiveSubscriptions;
 
    TArray<TWeakPtr<IBusListener, ESPMode::ThreadSafe>> ActiveRegistrationListeners;
 
    TMpscQueue<CommandDelegate> Commands;
 
    FDateTime CurrentTime;
 
    TArray<FDelayedMessage> DelayedMessages;
 
    int64 DelayedMessagesSequence;
 
    TAtomic<bool> Stopping;
 
    TSharedRef<FMessageTracer, ESPMode::ThreadSafe> Tracer;
 
    FEvent* WorkEvent;
 
    bool bAllowDelayedMessaging;
};