From 9ec54780d475ca56c2c6086039a9bb4d42a4371c Mon Sep 17 00:00:00 2001
From: liangshiwei <liangshiw@outlook.com>
Date: Sun, 9 Jan 2022 21:18:48 +0800
Subject: [PATCH]  Batch publish events from outbox to the rebus event bus

---
 .../EventBus/Rebus/AbpRebusEventBusOptions.cs | 13 +---------
 .../Rebus/RebusDistributedEventBus.cs         | 24 ++++++++++++++++---
 2 files changed, 22 insertions(+), 15 deletions(-)

diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs
index 5238cdd8bb..4c39b9bf1d 100644
--- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs
+++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs
@@ -20,24 +20,13 @@ public class AbpRebusEventBusOptions
     }
     private Action<RebusConfigurer> _configurer;
 
-    [NotNull]
-    public Func<IBus, Type, object, Task> Publish {
-        get => _publish;
-        set => _publish = Check.NotNull(value, nameof(value));
-    }
-    private Func<IBus, Type, object, Task> _publish;
+    public Func<IBus, Type, object, Task> Publish { get; set; }
 
     public AbpRebusEventBusOptions()
     {
-        _publish = DefaultPublish;
         _configurer = DefaultConfigure;
     }
 
-    private async Task DefaultPublish(IBus bus, Type eventType, object eventData)
-    {
-        await bus.Advanced.Routing.Send(InputQueueName, eventData);
-    }
-
     private void DefaultConfigure(RebusConfigurer configure)
     {
         configure.Subscriptions(s => s.StoreInMemory());
diff --git 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
index aad844a4d7..fc6a0667af 100644
--- 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
@@ -148,7 +148,18 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
 
     protected async override Task PublishToEventBusAsync(Type eventType, object eventData)
     {
-        await AbpRebusEventBusOptions.Publish(Rebus, eventType, eventData);
+        await PublishAsync(eventType, eventData);
+    }
+
+    protected virtual async Task PublishAsync(Type eventType, object eventData)
+    {
+        if (AbpRebusEventBusOptions.Publish != null)
+        {
+            await AbpRebusEventBusOptions.Publish(Rebus, eventType, eventData);
+            return;
+        }
+
+        await Rebus.Advanced.Routing.Send(AbpRebusEventBusOptions.InputQueueName, eventData);
     }
 
     protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord)
@@ -210,9 +221,16 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         return PublishToEventBusAsync(eventType, eventData);
     }
 
-    public override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    public async override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
     {
-        throw new NotImplementedException();
+        var outgoingEventArray = outgoingEvents.ToArray();
+
+        foreach (var outgoingEvent in outgoingEventArray)
+        {
+            await PublishFromOutboxAsync(outgoingEvent, outboxConfig);
+        }
+
+        return new MultipleOutgoingEventPublishResult(outgoingEventArray);
     }
 
     public async override Task ProcessFromInboxAsync(