Refactor

framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs

@@ -92,10 +92,9 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, outgoingEvent.Id);
     }
 
-    public async override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
     {
         var outgoingEventArray =  outgoingEvents.ToArray();
-        var failures = new List<Guid>();
 
         var publisher = await _publisherPool.GetAsync(
             _options.TopicName,
@@ -103,25 +102,15 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
 
         using var messageBatch = await publisher.CreateMessageBatchAsync();
 
-        var failed = false;
         foreach (var outgoingEvent in outgoingEventArray)
         {
-            if (failed)
-            {
-                failures.Add(outgoingEvent.Id);
-                continue;
-            }
-
             if (!messageBatch.TryAddMessage(new ServiceBusMessage(outgoingEvent.EventData) { Subject = outgoingEvent.EventName }))
             {
-                failed = true;
-                failures.Add(outgoingEvent.Id);
+                throw new AbpException("The message is too large to fit in the batch. Set AbpEventBusBoxesOptions.OutboxWaitingEventMaxCount to reduce the number");
             }
         }
 
         await publisher.SendMessagesAsync(messageBatch);
-
-        return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !failures.Contains(x.Id)).ToList());
     }
 
     public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig)

framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs

@@ -193,38 +193,40 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         );
     }
 
-    public async override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    public override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
     {
-        var pendingConfirms = new ConcurrentDictionary<string, Guid>();
+        var producer = ProducerPool.Get();
         var outgoingEventArray = outgoingEvents.ToArray();
-
-        var tasks = new List<Task>();
-        foreach (var outgoingEvent in outgoingEventArray)
+        producer.BeginTransaction();
+        try
         {
-            var messageId = outgoingEvent.Id.ToString("N");
-            pendingConfirms.TryAdd(messageId, outgoingEvent.Id);
-
-            var task = PublishAsync(
-                AbpKafkaEventBusOptions.TopicName,
-                outgoingEvent.EventName,
-                outgoingEvent.EventData,
-                new Headers { { "messageId", System.Text.Encoding.UTF8.GetBytes(messageId)} },
-                null
-            );
-
-            tasks.Add(task.ContinueWith(t =>
-           {
-               if (!t.IsFaulted)
-               {
-                   var message = t.Result.Message;
-                   pendingConfirms.TryRemove(message.GetMessageId(), out _);
-               }
-           }));
+            foreach (var outgoingEvent in outgoingEventArray)
+            {
+                var messageId = outgoingEvent.Id.ToString("N");
+                var headers = new Headers 
+                {
+                    { "messageId", System.Text.Encoding.UTF8.GetBytes(messageId)}
+                };
+
+                 producer.Produce(
+                    AbpKafkaEventBusOptions.TopicName,
+                    new Message<string, byte[]>
+                    {
+                        Key = outgoingEvent.EventName,
+                        Value = outgoingEvent.EventData,
+                        Headers = headers
+                    });
+            }
+            
+            producer.CommitTransaction();
         }
-
-        await Task.WhenAll(tasks);
-
-        return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !pendingConfirms.Select(p => p.Value).Contains(x.Id)).ToList());
+        catch (Exception e)
+        {
+            producer.AbortTransaction();
+            throw;
+        }
+        
+        return Task.CompletedTask;
     }
 
     public async override Task ProcessFromInboxAsync(
@@ -270,10 +272,26 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         return PublishAsync(topicName, eventName, body, headers, headersArguments);
     }
 
-    private Task<DeliveryResult<string, byte[]>> PublishAsync(string topicName, string eventName, byte[] body, Headers headers, Dictionary<string, object> headersArguments)
+    private Task<DeliveryResult<string, byte[]>> PublishAsync(
+        string topicName, 
+        string eventName,
+        byte[] body, 
+        Headers headers,
+        Dictionary<string, object> headersArguments)
     {
         var producer = ProducerPool.Get(AbpKafkaEventBusOptions.ConnectionName);
 
+        return PublishAsync(producer, topicName, eventName, body, headers, headersArguments);
+    }
+
+    private Task<DeliveryResult<string, byte[]>> PublishAsync(
+        IProducer<string, byte[]> producer,
+        string topicName,
+        string eventName,
+        byte[] body,
+        Headers headers,
+        Dictionary<string, object> headersArguments)
+    {
         SetEventMessageHeaders(headers, headersArguments);
 
         return producer.ProduceAsync(

framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs

@@ -34,6 +34,8 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
     protected IRabbitMqMessageConsumerFactory MessageConsumerFactory { get; }
     protected IRabbitMqMessageConsumer Consumer { get; private set; }
 
+    private bool _exchangeCreated;
+
     public RabbitMqDistributedEventBus(
         IOptions<AbpRabbitMqEventBusOptions> options,
         IConnectionPool connectionPool,
@@ -84,6 +86,8 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
         Consumer.OnMessageReceived(ProcessEventAsync);
 
         SubscribeHandlers(AbpDistributedEventBusOptions.Handlers);
+
+        
     }
 
     private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea)
@@ -197,7 +201,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
         return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id);
     }
 
-    public async override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(
+    public async override Task PublishManyFromOutboxAsync(
         IEnumerable<OutgoingEventInfo> outgoingEvents,
         OutboxConfig outboxConfig)
     {
@@ -206,48 +210,17 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
             var outgoingEventArray = outgoingEvents.ToArray();
             channel.ConfirmSelect();
 
-            var pendingConfirms = new ConcurrentDictionary<ulong, Guid>();
-            var failures = new ConcurrentBag<Guid>();
-
-            void CleanPendingConfirms(ulong sequenceNumber, bool multiple, bool ack)
-            {
-                if (multiple)
-                {
-                    var confirmed = pendingConfirms.Where(x => x.Key <= sequenceNumber);
-                    foreach (var entry in confirmed)
-                    {
-                        pendingConfirms.TryRemove(entry.Key, out var eventId);
-
-                        if (!ack)
-                        {
-                            failures.Add(eventId);
-                        }
-                    }
-                }
-                else
-                {
-                    pendingConfirms.TryRemove(sequenceNumber, out var eventId);
-
-                    if (!ack)
-                    {
-                        failures.Add(eventId);
-                    }
-                }
-            }
-
             foreach (var outgoingEvent in outgoingEventArray)
             {
-                pendingConfirms.TryAdd(channel.NextPublishSeqNo, outgoingEvent.Id);
-                await PublishAsync(channel, outgoingEvent.EventName, outgoingEvent.EventData, null,
+                await PublishAsync(
+                    channel,
+                    outgoingEvent.EventName, 
+                    outgoingEvent.EventData,  
+                    properties: null,
                     eventId: outgoingEvent.Id);
             }
 
-            channel.BasicAcks += (_, ea) => CleanPendingConfirms(ea.DeliveryTag, ea.Multiple, true);
-            channel.BasicNacks += (_, ea) => CleanPendingConfirms(ea.DeliveryTag, ea.Multiple, false);
-
-            channel.WaitForConfirms();
-
-            return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !failures.Contains(x.Id)).ToList());
+            channel.WaitForConfirmsOrDie();
         }
     }
 
@@ -308,11 +281,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
         Dictionary<string, object> headersArguments = null,
         Guid? eventId = null)
     {
-        channel.ExchangeDeclare(
-            AbpRabbitMqEventBusOptions.ExchangeName,
-            "direct",
-            durable: true
-        );
+        EnsureExchangeExists(channel);
 
         if (properties == null)
         {
@@ -338,6 +307,20 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
         return Task.CompletedTask;
     }
 
+    private void EnsureExchangeExists(IModel channel)
+    {
+        if (_exchangeCreated)
+        {
+            return;
+        }
+        
+        channel.ExchangeDeclare(
+            AbpRabbitMqEventBusOptions.ExchangeName,
+            "direct",
+            durable: true
+        );
+    }
+
     private void SetEventMessageHeaders(IBasicProperties properties, Dictionary<string, object> headersArguments)
     {
         if (headersArguments == null)

framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs

@@ -162,7 +162,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
             return;
         }
 
-        await Rebus.Advanced.Routing.Send(AbpRebusEventBusOptions.InputQueueName, eventData);
+        await Rebus.Publish(eventData);
     }
 
     protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord)
@@ -224,7 +224,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         return PublishToEventBusAsync(eventType, eventData);
     }
 
-    public async override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
     {
         var outgoingEventArray = outgoingEvents.ToArray();
 
@@ -234,11 +234,9 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
             {
                 await PublishFromOutboxAsync(outgoingEvent, outboxConfig);
             }
-
+            
             await scope.CompleteAsync();
         }
-
-        return new MultipleOutgoingEventPublishResult(outgoingEventArray);
     }
 
     public async override Task ProcessFromInboxAsync(

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs

@@ -34,6 +34,11 @@ public class AbpEventBusBoxesOptions
     /// Default: 2 hours
     /// </summary>
     public TimeSpan WaitTimeToDeleteProcessedInboxEvents { get; set; }
+    
+    /// <summary>
+    /// Default: false
+    /// </summary>
+    public bool OutboxPublishInBatch { get; set; }
 
     public AbpEventBusBoxesOptions()
     {
@@ -43,5 +48,6 @@ public class AbpEventBusBoxesOptions
         PeriodTimeSpan = TimeSpan.FromSeconds(2);
         DistributedLockWaitDuration = TimeSpan.FromSeconds(15);
         WaitTimeToDeleteProcessedInboxEvents = TimeSpan.FromHours(2);
+        OutboxPublishInBatch = false;
     }
 }

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs

@@ -85,7 +85,7 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
         OutboxConfig outboxConfig
     );
 
-    public abstract Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(
+    public abstract Task PublishManyFromOutboxAsync(
         IEnumerable<OutgoingEventInfo> outgoingEvents,
         OutboxConfig outboxConfig
     );

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs

@@ -10,7 +10,7 @@ public interface ISupportsEventBoxes
         OutboxConfig outboxConfig
     );
 
-    Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(
+    Task PublishManyFromOutboxAsync(
         IEnumerable<OutgoingEventInfo> outgoingEvents,
         OutboxConfig outboxConfig
     );

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
 using System.Threading;
@@ -84,14 +85,15 @@ public class OutboxSender : IOutboxSender, ITransientDependency
                     }
 
                     Logger.LogInformation($"Found {waitingEvents.Count} events in the outbox.");
-
-                    var result = await DistributedEventBus
-                        .AsSupportsEventBoxes()
-                        .PublishManyFromOutboxAsync(waitingEvents, OutboxConfig);
-
-                    await Outbox.DeleteManyAsync(result.PublishedOutgoingEvents.Select(x => x.Id).ToArray());
-
-                    Logger.LogInformation($"Sent {result.PublishedOutgoingEvents.Count} events to message broker");
+                    
+                    if (EventBusBoxesOptions.OutboxPublishInBatch)
+                    {
+                        await PublishOutgoingMessagesInBatchAsync(waitingEvents);
+                    }
+                    else
+                    {
+                        await PublishOutgoingMessagesAsync(waitingEvents);
+                    }
                 }
             }
             else
@@ -105,4 +107,32 @@ public class OutboxSender : IOutboxSender, ITransientDependency
             }
         }
     }
+
+    protected virtual async Task PublishOutgoingMessagesAsync(List<OutgoingEventInfo> waitingEvents)
+    {
+        foreach (var waitingEvent in waitingEvents)
+        {
+            await DistributedEventBus
+                .AsSupportsEventBoxes()
+                .PublishFromOutboxAsync(
+                    waitingEvent,
+                    OutboxConfig
+                );
+
+            await Outbox.DeleteAsync(waitingEvent.Id);
+            
+            Logger.LogInformation($"Sent the event to the message broker with id = {waitingEvent.Id:N}");
+        }
+    }
+
+    protected virtual async Task PublishOutgoingMessagesInBatchAsync(List<OutgoingEventInfo> waitingEvents)
+    {
+        await DistributedEventBus
+            .AsSupportsEventBoxes()
+            .PublishManyFromOutboxAsync(waitingEvents, OutboxConfig);
+                    
+        await Outbox.DeleteManyAsync(waitingEvents.Select(x => x.Id).ToArray());
+        
+        Logger.LogInformation($"Sent {waitingEvents.Count} events to message broker");
+    }
 }

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/MultipleOutgoingEventPublishResult.cs

@@ -1,14 +0,0 @@
-using System.Collections.Generic;
-using Volo.Abp.EventBus.Distributed;
-
-namespace Volo.Abp.EventBus;
-
-public class MultipleOutgoingEventPublishResult
-{
-    public IReadOnlyList<OutgoingEventInfo> PublishedOutgoingEvents { get; }
-
-    public MultipleOutgoingEventPublishResult(IReadOnlyList<OutgoingEventInfo> outgoingEvents)
-    {
-        PublishedOutgoingEvents = outgoingEvents;
-    }
-}

framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ConsumerPool.cs

@@ -76,6 +76,7 @@ public class ConsumerPool : IConsumerPool, ISingletonDependency
 
             try
             {
+                consumer.Value.Unsubscribe();
                 consumer.Value.Close();
                 consumer.Value.Dispose();
             }

framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs

@@ -80,9 +80,11 @@ public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency,
 
     protected virtual async Task Timer_Elapsed(AbpAsyncTimer timer)
     {
-        await CreateTopicAsync();
-        Consume();
-        Timer.Stop();
+        if (Consumer == null)
+        {
+            await CreateTopicAsync();
+            Consume();
+        }
     }
 
     protected virtual async Task CreateTopicAsync()
@@ -164,12 +166,21 @@ public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency,
 
     public virtual void Dispose()
     {
+        Timer.Stop();
         if (Consumer == null)
         {
             return;
         }
 
-        Consumer.Close();
-        Consumer.Dispose();
+        try
+        {
+            Consumer.Unsubscribe();
+            Consumer.Close();
+            Consumer.Dispose();
+            Consumer = null;
+        }
+        catch (ObjectDisposedException)
+        {
+        }
     }
 }

framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ProducerPool.cs

@@ -17,6 +17,8 @@ public class ProducerPool : IProducerPool, ISingletonDependency
     protected ConcurrentDictionary<string, Lazy<IProducer<string, byte[]>>> Producers { get; }
 
     protected TimeSpan TotalDisposeWaitDuration { get; set; } = TimeSpan.FromSeconds(10);
+    
+    protected TimeSpan DefaultTransactionsWaitDuration { get; set; } = TimeSpan.FromSeconds(30);
 
     public ILogger<ProducerPool> Logger { get; set; }
 
@@ -37,11 +39,18 @@ public class ProducerPool : IProducerPool, ISingletonDependency
         return Producers.GetOrAdd(
             connectionName, connection => new Lazy<IProducer<string, byte[]>>(() =>
             {
-                var config = Options.Connections.GetOrDefault(connection);
-
-                Options.ConfigureProducer?.Invoke(new ProducerConfig(config));
-
-                return new ProducerBuilder<string, byte[]>(config).Build();
+                var producerConfig = new ProducerConfig(Options.Connections.GetOrDefault(connection));
+                Options.ConfigureProducer?.Invoke(producerConfig);
+
+                if (producerConfig.TransactionalId.IsNullOrWhiteSpace())
+                {
+                    producerConfig.TransactionalId = Guid.NewGuid().ToString();
+                }
+                
+                var producer = new ProducerBuilder<string, byte[]>(producerConfig).Build();
+                producer.InitTransactions(DefaultTransactionsWaitDuration);
+                
+                return producer;
             })).Value;
     }
 
@@ -69,7 +78,7 @@ public class ProducerPool : IProducerPool, ISingletonDependency
         foreach (var producer in Producers.Values)
         {
             var poolItemDisposeStopwatch = Stopwatch.StartNew();
-
+        
             try
             {
                 producer.Value.Dispose();
@@ -77,19 +86,19 @@ public class ProducerPool : IProducerPool, ISingletonDependency
             catch
             {
             }
-
+        
             poolItemDisposeStopwatch.Stop();
-
+        
             remainingWaitDuration = remainingWaitDuration > poolItemDisposeStopwatch.Elapsed
                 ? remainingWaitDuration.Subtract(poolItemDisposeStopwatch.Elapsed)
                 : TimeSpan.Zero;
         }
-
+        
         poolDisposeStopwatch.Stop();
-
+        
         Logger.LogInformation(
             $"Disposed Kafka Producer Pool ({Producers.Count} producers in {poolDisposeStopwatch.Elapsed.TotalMilliseconds:0.00} ms).");
-
+        
         if (poolDisposeStopwatch.Elapsed.TotalSeconds > 5.0)
         {
             Logger.LogWarning(