Added `virtual GetWaitingEventsAsync` to `InboxProcessor/OutboxSender`.

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<List<IncomingEventInfo>> GetWaitingEventsAsync()
+    {
+        return await Inbox.GetWaitingEventsAsync(EventBusBoxesOptions.InboxWaitingEventMaxCount, EventBusBoxesOptions.InboxProcessorFilter, StoppingToken);
+    }
+
     protected virtual async Task DeleteOldEventsAsync()
     {
         if (LastCleanTime != null && LastCleanTime + EventBusBoxesOptions.CleanOldEventTimeIntervalSpan > Clock.Now)

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<List<OutgoingEventInfo>> GetWaitingEventsAsync()
+    {
+        return await Outbox.GetWaitingEventsAsync(EventBusBoxesOptions.OutboxWaitingEventMaxCount, EventBusBoxesOptions.OutboxProcessorFilter, StoppingToken);
+    }
+
     protected virtual async Task PublishOutgoingMessagesAsync(List<OutgoingEventInfo> waitingEvents)
     {
         foreach (var waitingEvent in waitingEvents)