From b4b4738e425a68a7ec3c07475507b5d0dc2a285d Mon Sep 17 00:00:00 2001 From: maliming Date: Tue, 24 Dec 2024 15:11:14 +0800 Subject: [PATCH] Added `virtual GetWaitingEventsAsync` to `InboxProcessor/OutboxSender`. --- .../Volo/Abp/EventBus/Distributed/InboxProcessor.cs | 12 +++++++++--- .../Volo/Abp/EventBus/Distributed/OutboxSender.cs | 7 ++++++- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs index 896264b16c..6ce431daba 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; @@ -60,7 +61,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency await RunAsync(); } - public Task StartAsync(InboxConfig inboxConfig, CancellationToken cancellationToken = default) + public virtual Task StartAsync(InboxConfig inboxConfig, CancellationToken cancellationToken = default) { InboxConfig = inboxConfig; Inbox = (IEventInbox)ServiceProvider.GetRequiredService(inboxConfig.ImplementationType); @@ -69,7 +70,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency return Task.CompletedTask; } - public Task StopAsync(CancellationToken cancellationToken = default) + public virtual Task StopAsync(CancellationToken cancellationToken = default) { StoppingTokenSource.Cancel(); Timer.Stop(cancellationToken); @@ -92,7 +93,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency while (true) { - var waitingEvents = await Inbox.GetWaitingEventsAsync(EventBusBoxesOptions.InboxWaitingEventMaxCount, EventBusBoxesOptions.InboxProcessorFilter, StoppingToken); + var waitingEvents = await GetWaitingEventsAsync(); if (waitingEvents.Count <= 0) { break; @@ -129,6 +130,11 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency } } + protected virtual async Task> GetWaitingEventsAsync() + { + return await Inbox.GetWaitingEventsAsync(EventBusBoxesOptions.InboxWaitingEventMaxCount, EventBusBoxesOptions.InboxProcessorFilter, StoppingToken); + } + protected virtual async Task DeleteOldEventsAsync() { if (LastCleanTime != null && LastCleanTime + EventBusBoxesOptions.CleanOldEventTimeIntervalSpan > Clock.Now) diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs index a398dc8478..bbc41e3f86 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs @@ -77,7 +77,7 @@ public class OutboxSender : IOutboxSender, ITransientDependency { while (true) { - var waitingEvents = await Outbox.GetWaitingEventsAsync(EventBusBoxesOptions.OutboxWaitingEventMaxCount, EventBusBoxesOptions.OutboxProcessorFilter, StoppingToken); + var waitingEvents = await GetWaitingEventsAsync(); if (waitingEvents.Count <= 0) { break; @@ -107,6 +107,11 @@ public class OutboxSender : IOutboxSender, ITransientDependency } } + protected virtual async Task> GetWaitingEventsAsync() + { + return await Outbox.GetWaitingEventsAsync(EventBusBoxesOptions.OutboxWaitingEventMaxCount, EventBusBoxesOptions.OutboxProcessorFilter, StoppingToken); + } + protected virtual async Task PublishOutgoingMessagesAsync(List waitingEvents) { foreach (var waitingEvent in waitingEvents)