diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs index f12c3e32d9..1eb7689aba 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Reflection; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; @@ -6,6 +6,7 @@ using Microsoft.Extensions.Options; using Volo.Abp.Collections; using Volo.Abp.DependencyInjection; using Volo.Abp.EventBus.Local; +using Volo.Abp.Uow; namespace Volo.Abp.EventBus.Distributed; @@ -15,16 +16,19 @@ public class LocalDistributedEventBus : IDistributedEventBus, ISingletonDependen { private readonly ILocalEventBus _localEventBus; + protected IUnitOfWorkManager UnitOfWorkManager { get; } protected IServiceScopeFactory ServiceScopeFactory { get; } protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; } public LocalDistributedEventBus( ILocalEventBus localEventBus, + IUnitOfWorkManager unitOfWorkManager, IServiceScopeFactory serviceScopeFactory, IOptions distributedEventBusOptions) { _localEventBus = localEventBus; + UnitOfWorkManager = unitOfWorkManager; ServiceScopeFactory = serviceScopeFactory; AbpDistributedEventBusOptions = distributedEventBusOptions.Value; Subscribe(distributedEventBusOptions.Value.Handlers); @@ -122,24 +126,56 @@ public class LocalDistributedEventBus : IDistributedEventBus, ISingletonDependen _localEventBus.UnsubscribeAll(eventType); } - public Task PublishAsync(TEvent eventData, bool onUnitOfWorkComplete = true) + public async Task PublishAsync(TEvent eventData, bool onUnitOfWorkComplete = true) where TEvent : class { - return _localEventBus.PublishAsync(eventData, onUnitOfWorkComplete); + await PublishAsync(typeof(TEvent), eventData, onUnitOfWorkComplete); } - public Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true) + public async Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true) { - return _localEventBus.PublishAsync(eventType, eventData, onUnitOfWorkComplete); + if (onUnitOfWorkComplete && UnitOfWorkManager.Current != null) + { + AddToUnitOfWork( + UnitOfWorkManager.Current, + new UnitOfWorkEventRecord(eventType, eventData, EventOrderGenerator.GetNext()) + ); + return; + } + + await _localEventBus.PublishAsync(eventType, eventData, onUnitOfWorkComplete: false); + } + + public async Task PublishAsync(TEvent eventData, bool onUnitOfWorkComplete = true, bool useOutbox = true) + where TEvent : class + { + await PublishAsync(typeof(TEvent), eventData, onUnitOfWorkComplete, useOutbox); } - public Task PublishAsync(TEvent eventData, bool onUnitOfWorkComplete = true, bool useOutbox = true) where TEvent : class + public async Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true, bool useOutbox = true) { - return _localEventBus.PublishAsync(eventData, onUnitOfWorkComplete); + if (onUnitOfWorkComplete && UnitOfWorkManager.Current != null) + { + AddToUnitOfWork( + UnitOfWorkManager.Current, + new UnitOfWorkEventRecord(eventType, eventData, EventOrderGenerator.GetNext(), useOutbox) + ); + return; + } + + if (useOutbox && UnitOfWorkManager.Current != null) + { + UnitOfWorkManager.Current.OnCompleted(async() => { + await _localEventBus.PublishAsync(eventType, eventData, onUnitOfWorkComplete: false); + }); + return; + } + + await _localEventBus.PublishAsync(eventType, eventData, onUnitOfWorkComplete: false); } - public Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true, bool useOutbox = true) + protected virtual void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord) { - return _localEventBus.PublishAsync(eventType, eventData, onUnitOfWorkComplete); + unitOfWork.AddOrReplaceDistributedEvent(eventRecord); } } diff --git 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 index 8a98a6a94c..497503680d 100644 --- 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,9 @@ using System; using System.Threading.Tasks; +using Shouldly; using Volo.Abp.Domain.Entities.Events.Distributed; using Volo.Abp.MultiTenancy; +using Volo.Abp.Uow; using Xunit; namespace Volo.Abp.EventBus.Distributed; @@ -55,11 +57,75 @@ public class LocalDistributedEventBus_Test : LocalDistributedEventBusTestBase await DistributedEventBus.PublishAsync(new MySimpleEto { Properties = - { - {"TenantId", tenantId.ToString()} - } + { + {"TenantId", tenantId.ToString()} + } }); Assert.Equal(tenantId, MySimpleDistributedSingleInstanceEventHandler.TenantId); } + + [Fact] + public async Task Event_Should_Published_On_UnitOfWorkComplete() + { + var id = 0; + DistributedEventBus.Subscribe(data => + { + id = data.Value; + return Task.CompletedTask; + }); + + var unitOfWorkManager = GetRequiredService(); + using (var uow = unitOfWorkManager.Begin()) + { + await DistributedEventBus.PublishAsync(new MySimpleEventData(3), onUnitOfWorkComplete: true, useOutbox: false); + } + id.ShouldBe(0); + + using (var uow = unitOfWorkManager.Begin()) + { + await DistributedEventBus.PublishAsync(new MySimpleEventData(3), onUnitOfWorkComplete: true, useOutbox: false); + await uow.CompleteAsync(); + } + id.ShouldBe(3); + + id = 0; + using (var uow = unitOfWorkManager.Begin()) + { + await DistributedEventBus.PublishAsync(new MySimpleEventData(3), onUnitOfWorkComplete: false, useOutbox: false); + } + id.ShouldBe(3); + } + + [Fact] + public async Task Event_Should_Published_On_UnitOfWorkComplete_UseOutbox() + { + var id = 0; + DistributedEventBus.Subscribe(data => + { + id = data.Value; + return Task.CompletedTask; + }); + + var unitOfWorkManager = GetRequiredService(); + using (var uow = unitOfWorkManager.Begin()) + { + await DistributedEventBus.PublishAsync(new MySimpleEventData(3), onUnitOfWorkComplete: false, useOutbox: true); + } + id.ShouldBe(0); + + using (var uow = unitOfWorkManager.Begin()) + { + await DistributedEventBus.PublishAsync(new MySimpleEventData(3), onUnitOfWorkComplete: false, useOutbox: true); + await uow.CompleteAsync(); + } + id.ShouldBe(3); + + id = 0; + using (var uow = unitOfWorkManager.Begin()) + { + await DistributedEventBus.PublishAsync(new MySimpleEventData(3), onUnitOfWorkComplete: false, useOutbox: false); + } + id.ShouldBe(3); + } }