From f36c44fc78203c4f520cfce62d99fce82ecb24c4 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Sun, 9 Jan 2022 20:50:17 +0800 Subject: [PATCH] Batch publish events from outbox to the azure event bus --- .../Azure/AzureDistributedEventBus.cs | 35 +++++++++++++++++-- 1 file changed, 32 insertions(+), 3 deletions(-) diff --git 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 index b253164161..a253a8ba96 100644 --- 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 @@ -90,9 +90,38 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData); } - public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) { - throw new NotImplementedException(); + var outgoingEventArray = outgoingEvents.ToArray(); + var failures = new List(); + + var publisher = await _publisherPool.GetAsync( + _options.TopicName, + _options.ConnectionName); + + using var messageBatch = await publisher.CreateMessageBatchAsync(); + + var failed = false; + foreach (var outgoingEvent in outgoingEventArray) + { + if (failed) + { + failures.Add(outgoingEvent.Id); + continue; + } + + var body = _serializer.Serialize(outgoingEvent.EventData); + + if (!messageBatch.TryAddMessage(new ServiceBusMessage(body) { Subject = outgoingEvent.EventName })) + { + failed = true; + failures.Add(outgoingEvent.Id); + } + } + + await publisher.SendMessagesAsync(messageBatch); + + return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !failures.Contains(x.Id)).ToList()); } public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) @@ -183,7 +212,7 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen .Locking(factories => factories.Clear()); } - protected override async Task PublishToEventBusAsync(Type eventType, object eventData) + protected async override Task PublishToEventBusAsync(Type eventType, object eventData) { await PublishAsync(EventNameAttribute.GetNameOrDefault(eventType), eventData); }