Merge pull request #16333 from abpframework/DistributedEventSentAndReceived

Inform when a new distributed message received or sent.
3 years ago · 034e451e6a
16 changed files with 459 additions and 100 deletions
--- a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs
+++ b/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs
@ -21,6 +21,7 @@ public class AbpAspNetCoreMvcDaprEventsController : AbpController
        var daprSerializer = HttpContext.RequestServices.GetRequiredService<IDaprSerializer>();
        var body = (await JsonDocument.ParseAsync(HttpContext.Request.Body));

+        var id = body.RootElement.GetProperty("id").GetString();
        var pubSubName = body.RootElement.GetProperty("pubsubname").GetString();
        var topic = body.RootElement.GetProperty("topic").GetString();
        var data = body.RootElement.GetProperty("data").GetRawText();
@ -32,7 +33,7 @@ public class AbpAspNetCoreMvcDaprEventsController : AbpController

        var distributedEventBus = HttpContext.RequestServices.GetRequiredService<DaprDistributedEventBus>();
        var eventData = daprSerializer.Deserialize(data, distributedEventBus.GetEventType(topic));
-        await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(topic), eventData);
+        await distributedEventBus.TriggerHandlersAsync(id, distributedEventBus.GetEventType(topic), eventData);
        return Ok();
    }
 }
--- a/framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/DistributedEventReceived.cs
+++ b/framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/DistributedEventReceived.cs
@ -0,0 +1,10 @@
+namespace Volo.Abp.EventBus.Distributed;
+
+public class DistributedEventReceived
+{
+    public DistributedEventSource Source { get; set; }
+
+    public string EventName { get; set; }
+
+    public object EventData { get; set; }
+}
--- a/framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/DistributedEventSent.cs
+++ b/framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/DistributedEventSent.cs
@ -0,0 +1,10 @@
+namespace Volo.Abp.EventBus.Distributed;
+
+public class DistributedEventSent
+{
+    public DistributedEventSource Source { get; set; }
+
+    public string EventName { get; set; }
+
+    public object EventData { get; set; }
+}
--- a/framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/DistributedEventSource.cs
+++ b/framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/DistributedEventSource.cs
@ -0,0 +1,10 @@
+namespace Volo.Abp.EventBus.Distributed;
+
+public enum DistributedEventSource
+{
+    Direct,
+
+    Inbox,
+
+    Outbox
+}
--- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs
+++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs
@ -9,6 +9,7 @@ using Microsoft.Extensions.Options;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.EventBus.Distributed;
 using Volo.Abp.AzureServiceBus;
+using Volo.Abp.EventBus.Local;
 using Volo.Abp.Guids;
 using Volo.Abp.MultiTenancy;
 using Volo.Abp.Threading;
@ -21,13 +22,13 @@ namespace Volo.Abp.EventBus.Azure;
 [ExposeServices(typeof(IDistributedEventBus), typeof(AzureDistributedEventBus))]
 public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDependency
 {
-    private readonly AbpAzureEventBusOptions _options;
-    private readonly IAzureServiceBusMessageConsumerFactory _messageConsumerFactory;
-    private readonly IPublisherPool _publisherPool;
-    private readonly IAzureServiceBusSerializer _serializer;
-    private readonly ConcurrentDictionary<Type, List<IEventHandlerFactory>> _handlerFactories;
-    private readonly ConcurrentDictionary<string, Type> _eventTypes;
-    private IAzureServiceBusMessageConsumer _consumer;
+    protected AbpAzureEventBusOptions Options { get; }
+    protected IAzureServiceBusMessageConsumerFactory MessageConsumerFactory { get; }
+    protected IPublisherPool PublisherPool { get; }
+    protected IAzureServiceBusSerializer Serializer { get; }
+    protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; }
+    protected ConcurrentDictionary<string, Type> EventTypes  { get; }
+    protected IAzureServiceBusMessageConsumer Consumer  { get; private set; }

    public AzureDistributedEventBus(
        IServiceScopeFactory serviceScopeFactory,
@ -40,31 +41,33 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        IAzureServiceBusSerializer serializer,
        IAzureServiceBusMessageConsumerFactory messageConsumerFactory,
        IPublisherPool publisherPool,
-        IEventHandlerInvoker eventHandlerInvoker)
+        IEventHandlerInvoker eventHandlerInvoker,
+        ILocalEventBus localEventBus)
        : base(serviceScopeFactory,
            currentTenant,
            unitOfWorkManager,
            abpDistributedEventBusOptions,
            guidGenerator,
            clock,
-            eventHandlerInvoker)
+            eventHandlerInvoker,
+            localEventBus)
    {
-        _options = abpAzureEventBusOptions.Value;
-        _serializer = serializer;
-        _messageConsumerFactory = messageConsumerFactory;
-        _publisherPool = publisherPool;
-        _handlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>();
-        _eventTypes = new ConcurrentDictionary<string, Type>();
+        Options = abpAzureEventBusOptions.Value;
+        Serializer = serializer;
+        MessageConsumerFactory = messageConsumerFactory;
+        PublisherPool = publisherPool;
+        HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>();
+        EventTypes = new ConcurrentDictionary<string, Type>();
    }

    public void Initialize()
    {
-        _consumer = _messageConsumerFactory.CreateMessageConsumer(
-            _options.TopicName,
-            _options.SubscriberName,
-            _options.ConnectionName);
+        Consumer = MessageConsumerFactory.CreateMessageConsumer(
+            Options.TopicName,
+            Options.SubscriberName,
+            Options.ConnectionName);

-        _consumer.OnMessageReceived(ProcessEventAsync);
+        Consumer.OnMessageReceived(ProcessEventAsync);
        SubscribeHandlers(AbpDistributedEventBusOptions.Handlers);
    }

@ -75,24 +78,31 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        {
            return;
        }
-        var eventType = _eventTypes.GetOrDefault(eventName);
+        var eventType = EventTypes.GetOrDefault(eventName);
        if (eventType == null)
        {
            return;
        }

-        if (await AddToInboxAsync(message.MessageId, eventName, eventType, message.Body.ToArray()))
+        var eventData = Serializer.Deserialize(message.Body.ToArray(), eventType);
+
+        if (await AddToInboxAsync(message.MessageId, eventName, eventType, eventData))
        {
            return;
        }

-        var eventData = _serializer.Deserialize(message.Body.ToArray(), eventType);
-
-        await TriggerHandlersAsync(eventType, eventData);
+        await TriggerHandlersDirectAsync(eventType, eventData);
    }

    public async override Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig)
    {
+        await TriggerDistributedEventSentAsync(new DistributedEventSent()
+        {
+            Source = DistributedEventSource.Outbox,
+            EventName = outgoingEvent.EventName,
+            EventData = outgoingEvent.EventData
+        });
+
        await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, outgoingEvent.Id);
    }

@ -100,9 +110,9 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
    {
        var outgoingEventArray =  outgoingEvents.ToArray();

-        var publisher = await _publisherPool.GetAsync(
-            _options.TopicName,
-            _options.ConnectionName);
+        var publisher = await PublisherPool.GetAsync(
+            Options.TopicName,
+            Options.ConnectionName);

        using var messageBatch = await publisher.CreateMessageBatchAsync();

@ -120,6 +130,13 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
                throw new AbpException(
                    "The message is too large to fit in the batch. Set AbpEventBusBoxesOptions.OutboxWaitingEventMaxCount to reduce the number");
            }
+
+            await TriggerDistributedEventSentAsync(new DistributedEventSent()
+            {
+                Source = DistributedEventSource.Outbox,
+                EventName = outgoingEvent.EventName,
+                EventData = outgoingEvent.EventData
+            });
        }

        await publisher.SendMessagesAsync(messageBatch);
@ -127,15 +144,15 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen

    public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig)
    {
-        var eventType = _eventTypes.GetOrDefault(incomingEvent.EventName);
+        var eventType = EventTypes.GetOrDefault(incomingEvent.EventName);
        if (eventType == null)
        {
            return;
        }

-        var eventData = _serializer.Deserialize(incomingEvent.EventData, eventType);
+        var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType);
        var exceptions = new List<Exception>();
-        await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig);
+        await TriggerHandlersFromInboxAsync(eventType, eventData, exceptions, inboxConfig);
        if (exceptions.Any())
        {
            ThrowOriginalExceptions(eventType, exceptions);
@ -144,7 +161,7 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen

    protected override byte[] Serialize(object eventData)
    {
-        return _serializer.Serialize(eventData);
+        return Serializer.Serialize(eventData);
    }

    public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
@ -225,7 +242,7 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen

    protected virtual Task PublishAsync(string eventName, object eventData)
    {
-        var body = _serializer.Serialize(eventData);
+        var body = Serializer.Serialize(eventData);

        return PublishAsync(eventName, body, null);
    }
@ -245,16 +262,16 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
            message.MessageId = (eventId ?? GuidGenerator.Create()).ToString("N");
        }

-        var publisher = await _publisherPool.GetAsync(
-            _options.TopicName,
-            _options.ConnectionName);
+        var publisher = await PublisherPool.GetAsync(
+            Options.TopicName,
+            Options.ConnectionName);

        await publisher.SendMessageAsync(message);
    }

    protected override IEnumerable<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType)
    {
-        return _handlerFactories
+        return HandlerFactories
            .Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key))
            .Select(handlerFactory =>
                new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value))
@ -266,14 +283,20 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        return handlerEventType == targetEventType || handlerEventType.IsAssignableFrom(targetEventType);
    }

+    protected override Task OnAddToOutboxAsync(string eventName, Type eventType, object eventData)
+    {
+        EventTypes.GetOrAdd(eventName, eventType);
+        return base.OnAddToOutboxAsync(eventName, eventType, eventData);
+    }
+
    private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType)
    {
-        return _handlerFactories.GetOrAdd(
+        return HandlerFactories.GetOrAdd(
            eventType,
            type =>
            {
                var eventName = EventNameAttribute.GetNameOrDefault(type);
-                _eventTypes[eventName] = type;
+                EventTypes.GetOrAdd(eventName, eventType);
                return new List<IEventHandlerFactory>();
            }
        );
--- a/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs
+++ b/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs
@ -8,6 +8,7 @@ using Microsoft.Extensions.Options;
 using Volo.Abp.Dapr;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.EventBus.Local;
 using Volo.Abp.Guids;
 using Volo.Abp.MultiTenancy;
 using Volo.Abp.Threading;
@ -37,8 +38,9 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend
        IEventHandlerInvoker eventHandlerInvoker,
        IDaprSerializer serializer,
        IOptions<AbpDaprEventBusOptions> daprEventBusOptions,
-        IAbpDaprClientFactory daprClientFactory)
-        : base(serviceScopeFactory, currentTenant, unitOfWorkManager, abpDistributedEventBusOptions, guidGenerator, clock, eventHandlerInvoker)
+        IAbpDaprClientFactory daprClientFactory,
+        ILocalEventBus localEventBus)
+        : base(serviceScopeFactory, currentTenant, unitOfWorkManager, abpDistributedEventBusOptions, guidGenerator, clock, eventHandlerInvoker, localEventBus)
    {
        Serializer = serializer;
        DaprEventBusOptions = daprEventBusOptions.Value;
@ -141,6 +143,13 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend

    public override async Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig)
    {
+        await TriggerDistributedEventSentAsync(new DistributedEventSent()
+        {
+            Source = DistributedEventSource.Outbox,
+            EventName = outgoingEvent.EventName,
+            EventData = outgoingEvent.EventData
+        });
+
        await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName)));
    }

@ -150,10 +159,27 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend

        foreach (var outgoingEvent in outgoingEventArray)
        {
+            await TriggerDistributedEventSentAsync(new DistributedEventSent()
+            {
+                Source = DistributedEventSource.Outbox,
+                EventName = outgoingEvent.EventName,
+                EventData = outgoingEvent.EventData
+            });
+
            await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName)));
        }
    }

+    public virtual async Task TriggerHandlersAsync(string messageId, Type eventType, object eventData)
+    {
+        if (await AddToInboxAsync(messageId, EventNameAttribute.GetNameOrDefault(eventType), eventType, eventData))
+        {
+            return;
+        }
+
+        await TriggerHandlersDirectAsync(eventType, eventData);
+    }
+
    public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig)
    {
        var eventType = EventTypes.GetOrDefault(incomingEvent.EventName);
@ -164,7 +190,7 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend

        var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType);
        var exceptions = new List<Exception>();
-        await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig);
+        await TriggerHandlersFromInboxAsync(eventType, eventData, exceptions, inboxConfig);
        if (exceptions.Any())
        {
            ThrowOriginalExceptions(eventType, exceptions);
@ -176,6 +202,12 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend
        return Serializer.Serialize(eventData);
    }

+    protected override Task OnAddToOutboxAsync(string eventName, Type eventType, object eventData)
+    {
+        EventTypes.GetOrAdd(eventName, eventType);
+        return base.OnAddToOutboxAsync(eventName, eventType, eventData);
+    }
+
    private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType)
    {
        return HandlerFactories.GetOrAdd(
@ -183,7 +215,7 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend
            type =>
            {
                var eventName = EventNameAttribute.GetNameOrDefault(type);
-                EventTypes[eventName] = type;
+                EventTypes.GetOrAdd(eventName, eventType);
                return new List<IEventHandlerFactory>();
            }
        );
--- a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs
+++ b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs
@ -8,6 +8,7 @@ using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.EventBus.Local;
 using Volo.Abp.Guids;
 using Volo.Abp.Kafka;
 using Volo.Abp.MultiTenancy;
@ -40,7 +41,8 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        IProducerPool producerPool,
        IGuidGenerator guidGenerator,
        IClock clock,
-        IEventHandlerInvoker eventHandlerInvoker)
+        IEventHandlerInvoker eventHandlerInvoker,
+        ILocalEventBus localEventBus)
        : base(
            serviceScopeFactory,
            currentTenant,
@ -48,7 +50,8 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
            abpDistributedEventBusOptions,
            guidGenerator,
            clock,
-            eventHandlerInvoker)
+            eventHandlerInvoker,
+            localEventBus)
    {
        AbpKafkaEventBusOptions = abpKafkaEventBusOptions.Value;
        MessageConsumerFactory = messageConsumerFactory;
@ -80,15 +83,14 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        }

        var messageId = message.GetMessageId();
+        var eventData = Serializer.Deserialize(message.Value, eventType);

-        if (await AddToInboxAsync(messageId, eventName, eventType, message.Value))
+        if (await AddToInboxAsync(messageId, eventName, eventType, eventData))
        {
            return;
        }

-        var eventData = Serializer.Deserialize(message.Value, eventType);
-
-        await TriggerHandlersAsync(eventType, eventData);
+        await TriggerHandlersDirectAsync(eventType, eventData);
    }

    public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
@ -177,11 +179,18 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        unitOfWork.AddOrReplaceDistributedEvent(eventRecord);
    }

-    public override Task PublishFromOutboxAsync(
+    public override async Task PublishFromOutboxAsync(
        OutgoingEventInfo outgoingEvent,
        OutboxConfig outboxConfig)
    {
-        return PublishAsync(
+        await TriggerDistributedEventSentAsync(new DistributedEventSent()
+        {
+            Source = DistributedEventSource.Outbox,
+            EventName = outgoingEvent.EventName,
+            EventData = outgoingEvent.EventData
+        });
+
+        await PublishAsync(
            AbpKafkaEventBusOptions.TopicName,
            outgoingEvent.EventName,
            outgoingEvent.EventData,
@ -192,19 +201,26 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        );
    }

-    public override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
    {
        var producer = ProducerPool.Get(AbpKafkaEventBusOptions.ConnectionName);
        var outgoingEventArray = outgoingEvents.ToArray();
-        
+
        foreach (var outgoingEvent in outgoingEventArray)
        {
            var messageId = outgoingEvent.Id.ToString("N");
-            var headers = new Headers 
+            var headers = new Headers
            {
                { "messageId", System.Text.Encoding.UTF8.GetBytes(messageId)}
            };
-            
+
+            await TriggerDistributedEventSentAsync(new DistributedEventSent()
+            {
+                Source = DistributedEventSource.Outbox,
+                EventName = outgoingEvent.EventName,
+                EventData = outgoingEvent.EventData
+            });
+
            producer.Produce(
                AbpKafkaEventBusOptions.TopicName,
                new Message<string, byte[]>
@ -214,8 +230,6 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
                    Headers = headers
                });
        }
-        
-        return Task.CompletedTask;
    }

    public async override Task ProcessFromInboxAsync(
@ -230,7 +244,7 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen

        var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType);
        var exceptions = new List<Exception>();
-        await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig);
+        await TriggerHandlersFromInboxAsync(eventType, eventData, exceptions, inboxConfig);
        if (exceptions.Any())
        {
            ThrowOriginalExceptions(eventType, exceptions);
@ -251,9 +265,9 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
    }

    private Task<DeliveryResult<string, byte[]>> PublishAsync(
-        string topicName, 
+        string topicName,
        string eventName,
-        byte[] body, 
+        byte[] body,
        Headers headers)
    {
        var producer = ProducerPool.Get(AbpKafkaEventBusOptions.ConnectionName);
@ -268,6 +282,12 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
            });
    }

+    protected override Task OnAddToOutboxAsync(string eventName, Type eventType, object eventData)
+    {
+        EventTypes.GetOrAdd(eventName, eventType);
+        return base.OnAddToOutboxAsync(eventName, eventType, eventData);
+    }
+
    private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType)
    {
        return HandlerFactories.GetOrAdd(
@ -275,7 +295,7 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
            type =>
            {
                var eventName = EventNameAttribute.GetNameOrDefault(type);
-                EventTypes[eventName] = type;
+                EventTypes.GetOrAdd(eventName, eventType);
                return new List<IEventHandlerFactory>();
            }
        );
--- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs
+++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs
@ -9,6 +9,7 @@ using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.EventBus.Local;
 using Volo.Abp.Guids;
 using Volo.Abp.MultiTenancy;
 using Volo.Abp.RabbitMQ;
@ -47,7 +48,8 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
        IUnitOfWorkManager unitOfWorkManager,
        IGuidGenerator guidGenerator,
        IClock clock,
-        IEventHandlerInvoker eventHandlerInvoker)
+        IEventHandlerInvoker eventHandlerInvoker,
+        ILocalEventBus localEventBus)
        : base(
            serviceScopeFactory,
            currentTenant,
@ -55,7 +57,8 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
            distributedEventBusOptions,
            guidGenerator,
            clock,
-            eventHandlerInvoker)
+            eventHandlerInvoker,
+            localEventBus)
    {
        ConnectionPool = connectionPool;
        Serializer = serializer;
@ -98,16 +101,14 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
            return;
        }

-        var eventBytes = ea.Body.ToArray();
+        var eventData = Serializer.Deserialize(ea.Body.ToArray(), eventType);

-        if (await AddToInboxAsync(ea.BasicProperties.MessageId, eventName, eventType, eventBytes))
+        if (await AddToInboxAsync(ea.BasicProperties.MessageId, eventName, eventType, eventData))
        {
            return;
        }

-        var eventData = Serializer.Deserialize(eventBytes, eventType);
-
-        await TriggerHandlersAsync(eventType, eventData);
+        await TriggerHandlersDirectAsync(eventType, eventData);
    }

    public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
@ -193,11 +194,18 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
        unitOfWork.AddOrReplaceDistributedEvent(eventRecord);
    }

-    public override Task PublishFromOutboxAsync(
+    public override async Task PublishFromOutboxAsync(
        OutgoingEventInfo outgoingEvent,
        OutboxConfig outboxConfig)
    {
-        return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id);
+        await TriggerDistributedEventSentAsync(new DistributedEventSent()
+        {
+            Source = DistributedEventSource.Outbox,
+            EventName = outgoingEvent.EventName,
+            EventData = outgoingEvent.EventData
+        });
+
+        await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id);
    }

    public async override Task PublishManyFromOutboxAsync(
@ -211,10 +219,17 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe

            foreach (var outgoingEvent in outgoingEventArray)
            {
+                await TriggerDistributedEventSentAsync(new DistributedEventSent()
+                {
+                    Source = DistributedEventSource.Outbox,
+                    EventName = outgoingEvent.EventName,
+                    EventData = outgoingEvent.EventData
+                });
+
                await PublishAsync(
                    channel,
-                    outgoingEvent.EventName, 
-                    outgoingEvent.EventData,  
+                    outgoingEvent.EventName,
+                    outgoingEvent.EventData,
                    properties: null,
                    eventId: outgoingEvent.Id);
            }
@ -235,7 +250,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe

        var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType);
        var exceptions = new List<Exception>();
-        await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig);
+        await TriggerHandlersFromInboxAsync(eventType, eventData, exceptions, inboxConfig);
        if (exceptions.Any())
        {
            ThrowOriginalExceptions(eventType, exceptions);
@ -249,7 +264,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe

    public Task PublishAsync(
        Type eventType,
-        object eventData, 
+        object eventData,
        IBasicProperties properties,
        Dictionary<string, object> headersArguments = null)
    {
@ -343,6 +358,12 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
        }
    }

+    protected override Task OnAddToOutboxAsync(string eventName, Type eventType, object eventData)
+    {
+        EventTypes.GetOrAdd(eventName, eventType);
+        return base.OnAddToOutboxAsync(eventName, eventType, eventData);
+    }
+
    private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType)
    {
        return HandlerFactories.GetOrAdd(
@ -350,7 +371,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
            type =>
            {
                var eventName = EventNameAttribute.GetNameOrDefault(type);
-                EventTypes[eventName] = type;
+                EventTypes.GetOrAdd(eventName, eventType);
                return new List<IEventHandlerFactory>();
            }
        );
--- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs
+++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs
@ -11,6 +11,7 @@ using Rebus.Pipeline;
 using Rebus.Transport;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.EventBus.Local;
 using Volo.Abp.Guids;
 using Volo.Abp.MultiTenancy;
 using Volo.Abp.Threading;
@ -41,7 +42,8 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        IRebusSerializer serializer,
        IGuidGenerator guidGenerator,
        IClock clock,
-        IEventHandlerInvoker eventHandlerInvoker) :
+        IEventHandlerInvoker eventHandlerInvoker,
+        ILocalEventBus localEventBus) :
        base(
            serviceScopeFactory,
            currentTenant,
@ -49,7 +51,8 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
            abpDistributedEventBusOptions,
            guidGenerator,
            clock,
-            eventHandlerInvoker)
+            eventHandlerInvoker,
+            localEventBus)
    {
        Rebus = rebus;
        Serializer = serializer;
@ -142,12 +145,12 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        var messageId = MessageContext.Current.TransportMessage.GetMessageId();
        var eventName = EventNameAttribute.GetNameOrDefault(eventType);

-        if (await AddToInboxAsync(messageId, eventName, eventType, MessageContext.Current.TransportMessage.Body))
+        if (await AddToInboxAsync(messageId, eventName, eventType, eventData))
        {
            return;
        }

-        await TriggerHandlersAsync(eventType, eventData);
+        await TriggerHandlersDirectAsync(eventType, eventData);
    }

    protected async override Task PublishToEventBusAsync(Type eventType, object eventData)
@ -181,6 +184,12 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        unitOfWork.AddOrReplaceDistributedEvent(eventRecord);
    }

+    protected override Task OnAddToOutboxAsync(string eventName, Type eventType, object eventData)
+    {
+        EventTypes.GetOrAdd(eventName, eventType);
+        return base.OnAddToOutboxAsync(eventName, eventType, eventData);
+    }
+
    private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType)
    {
        return HandlerFactories.GetOrAdd(
@ -188,7 +197,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
            type =>
            {
                var eventName = EventNameAttribute.GetNameOrDefault(type);
-                EventTypes[eventName] = type;
+                EventTypes.GetOrAdd(eventName, eventType);
                return new List<IEventHandlerFactory>();
            }
        );
@ -225,14 +234,21 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        return false;
    }

-    public override Task PublishFromOutboxAsync(
+    public override async Task PublishFromOutboxAsync(
        OutgoingEventInfo outgoingEvent,
        OutboxConfig outboxConfig)
    {
        var eventType = EventTypes.GetOrDefault(outgoingEvent.EventName);
        var eventData = Serializer.Deserialize(outgoingEvent.EventData, eventType);

-        return PublishAsync(eventType, eventData, eventId: outgoingEvent.Id);
+        await TriggerDistributedEventSentAsync(new DistributedEventSent()
+        {
+            Source = DistributedEventSource.Outbox,
+            EventName = outgoingEvent.EventName,
+            EventData = outgoingEvent.EventData
+        });
+
+        await PublishAsync(eventType, eventData, eventId: outgoingEvent.Id);
    }

    public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
@ -243,9 +259,16 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
        {
            foreach (var outgoingEvent in outgoingEventArray)
            {
+                await TriggerDistributedEventSentAsync(new DistributedEventSent()
+                {
+                    Source = DistributedEventSource.Outbox,
+                    EventName = outgoingEvent.EventName,
+                    EventData = outgoingEvent.EventData
+                });
+
                await PublishFromOutboxAsync(outgoingEvent, outboxConfig);
            }
-            
+
            await scope.CompleteAsync();
        }
    }
@ -262,7 +285,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen

        var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType);
        var exceptions = new List<Exception>();
-        await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig);
+        await TriggerHandlersFromInboxAsync(eventType, eventData, exceptions, inboxConfig);
        if (exceptions.Any())
        {
            ThrowOriginalExceptions(eventType, exceptions);
--- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs
+++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs
@ -4,6 +4,7 @@ using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
+using Volo.Abp.EventBus.Local;
 using Volo.Abp.Guids;
 using Volo.Abp.MultiTenancy;
 using Volo.Abp.Timing;
@ -16,6 +17,7 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
    protected IGuidGenerator GuidGenerator { get; }
    protected IClock Clock { get; }
    protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; }
+    protected ILocalEventBus LocalEventBus { get; }

    protected DistributedEventBusBase(
        IServiceScopeFactory serviceScopeFactory,
@ -24,8 +26,8 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
        IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions,
        IGuidGenerator guidGenerator,
        IClock clock,
-        IEventHandlerInvoker eventHandlerInvoker
-    ) : base(
+        IEventHandlerInvoker eventHandlerInvoker,
+        ILocalEventBus localEventBus) : base(
        serviceScopeFactory,
        currentTenant,
        unitOfWorkManager,
@ -34,6 +36,7 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
        GuidGenerator = guidGenerator;
        Clock = clock;
        AbpDistributedEventBusOptions = abpDistributedEventBusOptions.Value;
+        LocalEventBus = localEventBus;
    }

    public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) where TEvent : class
@ -78,6 +81,13 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
            }
        }

+        await TriggerDistributedEventSentAsync(new DistributedEventSent()
+        {
+            Source = DistributedEventSource.Direct,
+            EventName = EventNameAttribute.GetNameOrDefault(eventType),
+            EventData = eventData
+        });
+
        await PublishToEventBusAsync(eventType, eventData);
    }

@ -95,7 +105,7 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
        IncomingEventInfo incomingEvent,
        InboxConfig inboxConfig);

-    private async Task<bool> AddToOutboxAsync(Type eventType, object eventData)
+    protected virtual async Task<bool> AddToOutboxAsync(Type eventType, object eventData)
    {
        var unitOfWork = UnitOfWorkManager.Current;
        if (unitOfWork == null)
@ -107,9 +117,18 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
        {
            if (outboxConfig.Selector == null || outboxConfig.Selector(eventType))
            {
-                var eventOutbox =
-                    (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType);
+                var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType);
                var eventName = EventNameAttribute.GetNameOrDefault(eventType);
+
+                await OnAddToOutboxAsync(eventName, eventType, eventData);
+
+                await TriggerDistributedEventSentAsync(new DistributedEventSent()
+                {
+                    Source = DistributedEventSource.Direct,
+                    EventName = eventName,
+                    EventData = eventData
+                });
+
                await eventOutbox.EnqueueAsync(
                    new OutgoingEventInfo(
                        GuidGenerator.Create(),
@ -125,11 +144,16 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
        return false;
    }

+    protected virtual Task OnAddToOutboxAsync(string eventName, Type eventType, object eventData)
+    {
+        return Task.CompletedTask;
+    }
+
    protected async Task<bool> AddToInboxAsync(
        string messageId,
        string eventName,
        Type eventType,
-        byte[] eventBytes)
+        object eventData)
    {
        if (AbpDistributedEventBusOptions.Inboxes.Count <= 0)
        {
@ -153,12 +177,19 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
                        }
                    }

+                    await TriggerDistributedEventReceivedAsync(new DistributedEventReceived
+                    {
+                        Source = DistributedEventSource.Direct,
+                        EventName = EventNameAttribute.GetNameOrDefault(eventType),
+                        EventData = eventData
+                    });
+
                    await eventInbox.EnqueueAsync(
                        new IncomingEventInfo(
                            GuidGenerator.Create(),
                            messageId,
                            eventName,
-                            eventBytes,
+                            Serialize(eventData),
                            Clock.Now
                        )
                    );
@ -170,4 +201,52 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
    }

    protected abstract byte[] Serialize(object eventData);
+
+    protected virtual async Task TriggerHandlersDirectAsync(Type eventType, object eventData)
+    {
+        await TriggerDistributedEventReceivedAsync(new DistributedEventReceived
+        {
+            Source = DistributedEventSource.Direct,
+            EventName = EventNameAttribute.GetNameOrDefault(eventType),
+            EventData = eventData
+        });
+
+        await TriggerHandlersAsync(eventType, eventData);
+    }
+
+    protected virtual async Task TriggerHandlersFromInboxAsync(Type eventType, object eventData, List<Exception> exceptions, InboxConfig inboxConfig = null)
+    {
+        await TriggerDistributedEventReceivedAsync(new DistributedEventReceived
+        {
+            Source = DistributedEventSource.Inbox,
+            EventName = EventNameAttribute.GetNameOrDefault(eventType),
+            EventData = eventData
+        });
+
+        await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig);
+    }
+
+    public virtual async Task TriggerDistributedEventSentAsync(DistributedEventSent distributedEvent)
+    {
+        try
+        {
+            await LocalEventBus.PublishAsync(distributedEvent);
+        }
+        catch (Exception _)
+        {
+            // ignored
+        }
+    }
+
+    public virtual async Task TriggerDistributedEventReceivedAsync(DistributedEventReceived distributedEvent)
+    {
+        try
+        {
+            await LocalEventBus.PublishAsync(distributedEvent);
+        }
+        catch (Exception _)
+        {
+            // ignored
+        }
+    }
 }
--- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs
+++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs
@ -28,6 +28,30 @@ public class LocalDistributedEventBus : IDistributedEventBus, ISingletonDependen
        ServiceScopeFactory = serviceScopeFactory;
        AbpDistributedEventBusOptions = distributedEventBusOptions.Value;
        Subscribe(distributedEventBusOptions.Value.Handlers);
+
+        // For unit testing
+        if (localEventBus is LocalEventBus eventBus)
+        {
+            eventBus.OnEventHandleInvoking = async (eventType, eventData) =>
+            {
+                await localEventBus.PublishAsync(new DistributedEventReceived()
+                {
+                    Source = DistributedEventSource.Direct,
+                    EventName = EventNameAttribute.GetNameOrDefault(eventType),
+                    EventData = eventData
+                }, onUnitOfWorkComplete: false);
+            };
+
+            eventBus.OnPublishing = async (eventType, eventData) =>
+            {
+                await localEventBus.PublishAsync(new DistributedEventSent()
+                {
+                    Source = DistributedEventSource.Direct,
+                    EventName = EventNameAttribute.GetNameOrDefault(eventType),
+                    EventData = eventData
+                }, onUnitOfWorkComplete: false);
+            };
+        }
    }

    public virtual void Subscribe(ITypeList<IEventHandler> handlers)
@ -132,7 +156,7 @@ public class LocalDistributedEventBus : IDistributedEventBus, ISingletonDependen
    {
        return _localEventBus.PublishAsync(eventType, eventData, onUnitOfWorkComplete);
    }
-    
+
    public Task PublishAsync<TEvent>(TEvent eventData, bool onUnitOfWorkComplete = true, bool useOutbox = true) where TEvent : class
    {
        return _localEventBus.PublishAsync(eventData, onUnitOfWorkComplete);
@ -142,4 +166,4 @@ public class LocalDistributedEventBus : IDistributedEventBus, ISingletonDependen
    {
        return _localEventBus.PublishAsync(eventType, eventData, onUnitOfWorkComplete);
    }
-}
+}
--- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusBase.cs
+++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusBase.cs
@ -9,7 +9,6 @@ using Microsoft.Extensions.DependencyInjection;
 using Volo.Abp.Collections;
 using Volo.Abp.EventBus.Distributed;
 using Volo.Abp.MultiTenancy;
-using Volo.Abp.Reflection;
 using Volo.Abp.Uow;

 namespace Volo.Abp.EventBus;
@ -214,7 +213,7 @@ public abstract class EventBusBase : IEventBus

                using (CurrentTenant.Change(GetEventDataTenantId(eventData)))
                {
-                    await EventHandlerInvoker.InvokeAsync(eventHandlerWrapper.EventHandler, eventData, eventType);
+                    await InvokeEventHandlerAsync(eventHandlerWrapper.EventHandler, eventData, eventType);
                }
            }
            catch (TargetInvocationException ex)
@ -228,6 +227,11 @@ public abstract class EventBusBase : IEventBus
        }
    }

+    protected virtual Task InvokeEventHandlerAsync(IEventHandler eventHandler, object eventData, Type eventType)
+    {
+        return EventHandlerInvoker.InvokeAsync(eventHandler, eventData, eventType);
+    }
+
    protected virtual Guid? GetEventDataTenantId(object eventData)
    {
        return eventData switch
--- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBus.cs
+++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBus.cs
@ -8,6 +8,7 @@ using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
 using Volo.Abp.DependencyInjection;
+using Volo.Abp.EventBus.Distributed;
 using Volo.Abp.MultiTenancy;
 using Volo.Abp.Threading;
 using Volo.Abp.Uow;
@ -168,4 +169,45 @@ public class LocalEventBus : EventBusBase, ILocalEventBus, ISingletonDependency

        return false;
    }
+
+    // Internal for unit testing
+    internal Func<Type, object, Task> OnEventHandleInvoking { get; set; }
+
+    // Internal for unit testing
+    protected async override Task InvokeEventHandlerAsync(IEventHandler eventHandler, object eventData, Type eventType)
+    {
+        if (OnEventHandleInvoking != null && eventType != typeof(DistributedEventSent) && eventType != typeof(DistributedEventReceived))
+        {
+            await OnEventHandleInvoking(eventType, eventData);
+        }
+
+        await base.InvokeEventHandlerAsync(eventHandler, eventData, eventType);
+    }
+
+    // Internal for unit testing
+    internal Func<Type, object, Task> OnPublishing { get; set; }
+
+    // For unit testing
+    public async override Task PublishAsync(
+        Type eventType,
+        object eventData,
+        bool onUnitOfWorkComplete = true)
+    {
+        if (onUnitOfWorkComplete && UnitOfWorkManager.Current != null)
+        {
+            AddToUnitOfWork(
+                UnitOfWorkManager.Current,
+                new UnitOfWorkEventRecord(eventType, eventData, EventOrderGenerator.GetNext())
+            );
+            return;
+        }
+
+        // For unit testing
+        if (OnPublishing != null && eventType != typeof(DistributedEventSent) && eventType != typeof(DistributedEventReceived))
+        {
+            await OnPublishing(eventType, eventData);
+        }
+
+        await PublishToEventBusAsync(eventType, eventData);
+    }
 }
--- a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/AbpMongoDbDateTimeSerializer.cs
+++ b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/AbpMongoDbDateTimeSerializer.cs
@ -34,7 +34,7 @@ public class AbpMongoDbDateTimeSerializer : DateTimeSerializer
        return (dateTime - BsonConstants.UnixEpoch).Ticks / 10000L;
    }

-    // For unit testing.
+    // For unit testing
    internal void SetDateTimeKind(DateTimeKind dateTimeKind)
    {
        DateTimeKind = dateTimeKind;
--- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/DistributedEventHandles.cs
+++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/DistributedEventHandles.cs
@ -0,0 +1,22 @@
+using System.Threading.Tasks;
+
+namespace Volo.Abp.EventBus.Distributed;
+
+public class DistributedEventHandles : ILocalEventHandler<DistributedEventSent>, ILocalEventHandler<DistributedEventReceived>
+{
+    public static int SentCount { get; set; }
+
+    public static int ReceivedCount { get; set; }
+
+    public Task HandleEventAsync(DistributedEventSent eventData)
+    {
+        SentCount++;
+        return Task.CompletedTask;
+    }
+
+    public Task HandleEventAsync(DistributedEventReceived eventData)
+    {
+        ReceivedCount++;
+        return Task.CompletedTask;
+    }
+}
--- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus_Test.cs
+++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus_Test.cs
@ -1,7 +1,8 @@
 using System;
 using System.Threading.Tasks;
 using Volo.Abp.Domain.Entities.Events.Distributed;
-using Volo.Abp.MultiTenancy;
+using Volo.Abp.EventBus.Local;
+using Volo.Abp.Uow;
 using Xunit;

 namespace Volo.Abp.EventBus.Distributed;
@ -49,7 +50,7 @@ public class LocalDistributedEventBus_Test : LocalDistributedEventBusTestBase
    public async Task Should_Get_TenantId_From_EventEto_Extra_Property()
    {
        var tenantId = Guid.NewGuid();
-        
+
        DistributedEventBus.Subscribe<MySimpleEto>(GetRequiredService<MySimpleDistributedSingleInstanceEventHandler>());

        await DistributedEventBus.PublishAsync(new MySimpleEto
@ -59,7 +60,44 @@ public class LocalDistributedEventBus_Test : LocalDistributedEventBusTestBase
                {"TenantId", tenantId.ToString()}
            }
        });
-        
+
        Assert.Equal(tenantId, MySimpleDistributedSingleInstanceEventHandler.TenantId);
    }
+
+    [Fact]
+    public async Task DistributedEventSentAndReceived_Test()
+    {
+        GetRequiredService<ILocalEventBus>().Subscribe<DistributedEventSent, DistributedEventHandles>();
+        GetRequiredService<ILocalEventBus>().Subscribe<DistributedEventReceived, DistributedEventHandles>();
+
+        DistributedEventBus.Subscribe<MyEventDate, MyEventHandle>();
+
+        using (var uow = GetRequiredService<IUnitOfWorkManager>().Begin())
+        {
+            await DistributedEventBus.PublishAsync(new MyEventDate(), onUnitOfWorkComplete: false);
+
+            Assert.Equal(1, DistributedEventHandles.SentCount);
+            Assert.Equal(1, DistributedEventHandles.ReceivedCount);
+
+            await DistributedEventBus.PublishAsync(new MyEventDate(), onUnitOfWorkComplete: true);
+
+            await uow.CompleteAsync();
+
+            Assert.Equal(2, DistributedEventHandles.SentCount);
+            Assert.Equal(2, DistributedEventHandles.ReceivedCount);
+        }
+    }
+
+    class MyEventDate
+    {
+
+    }
+
+    class MyEventHandle : IDistributedEventHandler<MyEventDate>
+    {
+        public Task HandleEventAsync(MyEventDate eventData)
+        {
+            return Task.CompletedTask;
+        }
+    }
 }