Batch publish events from outbox to the event bus

framework/src/Volo.Abp.EntityFrameworkCore.PostgreSql/Volo/Abp/EntityFrameworkCore/DistributedEvents/PostgreSqlDbContextEventOutbox.cs

@@ -13,7 +13,7 @@ public class PostgreSqlDbContextEventOutbox<TDbContext> : DbContextEventOutbox<T
     }
 
     [UnitOfWork]
-    public override async Task DeleteAsync(Guid id)
+    public async override Task DeleteAsync(Guid id)
     {
         var dbContext = (IHasEventOutbox)await DbContextProvider.GetDbContextAsync();
         var tableName = dbContext.OutgoingEvents.EntityType.GetSchemaQualifiedTableName();

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventOutbox.cs

@@ -56,4 +56,15 @@ public class DbContextEventOutbox<TDbContext> : IDbContextEventOutbox<TDbContext
             dbContext.Remove(outgoingEvent);
         }
     }
+
+    [UnitOfWork]
+    public virtual async Task DeleteManyAsync(IEnumerable<Guid> ids)
+    {
+        var dbContext = (IHasEventOutbox)await DbContextProvider.GetDbContextAsync();
+        var outgoingEvents = await dbContext.OutgoingEvents.Where(x => ids.Contains(x.Id)).ToListAsync();
+        if (outgoingEvents.Any())
+        {
+            dbContext.RemoveRange(outgoingEvents);
+        }
+    }
 }

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

@@ -85,12 +85,17 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         await TriggerHandlersAsync(eventType, eventData);
     }
 
-    public override async Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig)
+    public async override Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig)
     {
         await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData);
     }
 
-    public override async Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig)
+    public override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    {
+        throw new NotImplementedException();
+    }
+
+    public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig)
     {
         var eventType = _eventTypes.GetOrDefault(incomingEvent.EventName);
         if (eventType == null)

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

@@ -162,7 +162,7 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear());
     }
 
-    protected override async Task PublishToEventBusAsync(Type eventType, object eventData)
+    protected async override Task PublishToEventBusAsync(Type eventType, object eventData)
     {
         await PublishAsync(
             eventType,
@@ -196,7 +196,12 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         );
     }
 
-    public override async Task ProcessFromInboxAsync(
+    public override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    {
+        throw new NotImplementedException();
+    }
+
+    public async override Task ProcessFromInboxAsync(
         IncomingEventInfo incomingEvent,
         InboxConfig inboxConfig)
     {

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

@@ -16,225 +16,290 @@ using Volo.Abp.Threading;
 using Volo.Abp.Timing;
 using Volo.Abp.Uow;
 
-namespace Volo.Abp.EventBus.RabbitMq;
-
-/* TODO: How to handle unsubscribe to unbind on RabbitMq (may not be possible for)
- */
-[Dependency(ReplaceServices = true)]
-[ExposeServices(typeof(IDistributedEventBus), typeof(RabbitMqDistributedEventBus))]
-public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDependency
+namespace Volo.Abp.EventBus.RabbitMq
 {
-    protected AbpRabbitMqEventBusOptions AbpRabbitMqEventBusOptions { get; }
-    protected IConnectionPool ConnectionPool { get; }
-    protected IRabbitMqSerializer Serializer { get; }
-
-    //TODO: Accessing to the List<IEventHandlerFactory> may not be thread-safe!
-    protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; }
-    protected ConcurrentDictionary<string, Type> EventTypes { get; }
-    protected IRabbitMqMessageConsumerFactory MessageConsumerFactory { get; }
-    protected IRabbitMqMessageConsumer Consumer { get; private set; }
-
-    public RabbitMqDistributedEventBus(
-        IOptions<AbpRabbitMqEventBusOptions> options,
-        IConnectionPool connectionPool,
-        IRabbitMqSerializer serializer,
-        IServiceScopeFactory serviceScopeFactory,
-        IOptions<AbpDistributedEventBusOptions> distributedEventBusOptions,
-        IRabbitMqMessageConsumerFactory messageConsumerFactory,
-        ICurrentTenant currentTenant,
-        IUnitOfWorkManager unitOfWorkManager,
-        IGuidGenerator guidGenerator,
-        IClock clock)
-        : base(
-            serviceScopeFactory,
-            currentTenant,
-            unitOfWorkManager,
-            distributedEventBusOptions,
-            guidGenerator,
-            clock)
+    /* TODO: How to handle unsubscribe to unbind on RabbitMq (may not be possible for)
+     */
+    [Dependency(ReplaceServices = true)]
+    [ExposeServices(typeof(IDistributedEventBus), typeof(RabbitMqDistributedEventBus))]
+    public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDependency
     {
-        ConnectionPool = connectionPool;
-        Serializer = serializer;
-        MessageConsumerFactory = messageConsumerFactory;
-        AbpRabbitMqEventBusOptions = options.Value;
-
-        HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>();
-        EventTypes = new ConcurrentDictionary<string, Type>();
-    }
+        protected AbpRabbitMqEventBusOptions AbpRabbitMqEventBusOptions { get; }
+        protected IConnectionPool ConnectionPool { get; }
+        protected IRabbitMqSerializer Serializer { get; }
+
+        //TODO: Accessing to the List<IEventHandlerFactory> may not be thread-safe!
+        protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; }
+        protected ConcurrentDictionary<string, Type> EventTypes { get; }
+        protected IRabbitMqMessageConsumerFactory MessageConsumerFactory { get; }
+        protected IRabbitMqMessageConsumer Consumer { get; private set; }
+
+        public RabbitMqDistributedEventBus(
+            IOptions<AbpRabbitMqEventBusOptions> options,
+            IConnectionPool connectionPool,
+            IRabbitMqSerializer serializer,
+            IServiceScopeFactory serviceScopeFactory,
+            IOptions<AbpDistributedEventBusOptions> distributedEventBusOptions,
+            IRabbitMqMessageConsumerFactory messageConsumerFactory,
+            ICurrentTenant currentTenant,
+            IUnitOfWorkManager unitOfWorkManager,
+            IGuidGenerator guidGenerator,
+            IClock clock)
+            : base(
+                serviceScopeFactory,
+                currentTenant,
+                unitOfWorkManager,
+                distributedEventBusOptions,
+                guidGenerator,
+                clock)
+        {
+            ConnectionPool = connectionPool;
+            Serializer = serializer;
+            MessageConsumerFactory = messageConsumerFactory;
+            AbpRabbitMqEventBusOptions = options.Value;
 
-    public void Initialize()
-    {
-        Consumer = MessageConsumerFactory.Create(
-            new ExchangeDeclareConfiguration(
-                AbpRabbitMqEventBusOptions.ExchangeName,
-                type: "direct",
-                durable: true
-            ),
-            new QueueDeclareConfiguration(
-                AbpRabbitMqEventBusOptions.ClientName,
-                durable: true,
-                exclusive: false,
-                autoDelete: false
-            ),
-            AbpRabbitMqEventBusOptions.ConnectionName
-        );
-
-        Consumer.OnMessageReceived(ProcessEventAsync);
-
-        SubscribeHandlers(AbpDistributedEventBusOptions.Handlers);
-    }
+            HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>();
+            EventTypes = new ConcurrentDictionary<string, Type>();
+        }
 
-    private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea)
-    {
-        var eventName = ea.RoutingKey;
-        var eventType = EventTypes.GetOrDefault(eventName);
-        if (eventType == null)
+        public void Initialize()
         {
-            return;
+            Consumer = MessageConsumerFactory.Create(
+                new ExchangeDeclareConfiguration(
+                    AbpRabbitMqEventBusOptions.ExchangeName,
+                    type: "direct",
+                    durable: true
+                ),
+                new QueueDeclareConfiguration(
+                    AbpRabbitMqEventBusOptions.ClientName,
+                    durable: true,
+                    exclusive: false,
+                    autoDelete: false
+                ),
+                AbpRabbitMqEventBusOptions.ConnectionName
+            );
+
+            Consumer.OnMessageReceived(ProcessEventAsync);
+
+            SubscribeHandlers(AbpDistributedEventBusOptions.Handlers);
         }
 
-        var eventBytes = ea.Body.ToArray();
+        private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea)
+        {
+            var eventName = ea.RoutingKey;
+            var eventType = EventTypes.GetOrDefault(eventName);
+            if (eventType == null)
+            {
+                return;
+            }
+
+            var eventBytes = ea.Body.ToArray();
+
+            if (await AddToInboxAsync(ea.BasicProperties.MessageId, eventName, eventType, eventBytes))
+            {
+                return;
+            }
+
+            var eventData = Serializer.Deserialize(eventBytes, eventType);
+
+            await TriggerHandlersAsync(eventType, eventData);
+        }
 
-        if (await AddToInboxAsync(ea.BasicProperties.MessageId, eventName, eventType, eventBytes))
+        public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
         {
-            return;
+            var handlerFactories = GetOrCreateHandlerFactories(eventType);
+
+            if (factory.IsInFactories(handlerFactories))
+            {
+                return NullDisposable.Instance;
+            }
+
+            handlerFactories.Add(factory);
+
+            if (handlerFactories.Count == 1) //TODO: Multi-threading!
+            {
+                Consumer.BindAsync(EventNameAttribute.GetNameOrDefault(eventType));
+            }
+
+            return new EventHandlerFactoryUnregistrar(this, eventType, factory);
         }
 
-        var eventData = Serializer.Deserialize(eventBytes, eventType);
+        /// <inheritdoc/>
+        public override void Unsubscribe<TEvent>(Func<TEvent, Task> action)
+        {
+            Check.NotNull(action, nameof(action));
 
-        await TriggerHandlersAsync(eventType, eventData);
-    }
+            GetOrCreateHandlerFactories(typeof(TEvent))
+                .Locking(factories =>
+                {
+                    factories.RemoveAll(
+                        factory =>
+                        {
+                            var singleInstanceFactory = factory as SingleInstanceHandlerFactory;
+                            if (singleInstanceFactory == null)
+                            {
+                                return false;
+                            }
+
+                            var actionHandler = singleInstanceFactory.HandlerInstance as ActionEventHandler<TEvent>;
+                            if (actionHandler == null)
+                            {
+                                return false;
+                            }
+
+                            return actionHandler.Action == action;
+                        });
+                });
+        }
 
-    public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
-    {
-        var handlerFactories = GetOrCreateHandlerFactories(eventType);
+        /// <inheritdoc/>
+        public override void Unsubscribe(Type eventType, IEventHandler handler)
+        {
+            GetOrCreateHandlerFactories(eventType)
+                .Locking(factories =>
+                {
+                    factories.RemoveAll(
+                        factory =>
+                            factory is SingleInstanceHandlerFactory &&
+                            (factory as SingleInstanceHandlerFactory).HandlerInstance == handler
+                    );
+                });
+        }
 
-        if (factory.IsInFactories(handlerFactories))
+        /// <inheritdoc/>
+        public override void Unsubscribe(Type eventType, IEventHandlerFactory factory)
         {
-            return NullDisposable.Instance;
+            GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory));
         }
 
-        handlerFactories.Add(factory);
+        /// <inheritdoc/>
+        public override void UnsubscribeAll(Type eventType)
+        {
+            GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear());
+        }
 
-        if (handlerFactories.Count == 1) //TODO: Multi-threading!
+        protected async override Task PublishToEventBusAsync(Type eventType, object eventData)
         {
-            Consumer.BindAsync(EventNameAttribute.GetNameOrDefault(eventType));
+            await PublishAsync(eventType, eventData, null);
         }
 
-        return new EventHandlerFactoryUnregistrar(this, eventType, factory);
-    }
+        protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord)
+        {
+            unitOfWork.AddOrReplaceDistributedEvent(eventRecord);
+        }
 
-    /// <inheritdoc/>
-    public override void Unsubscribe<TEvent>(Func<TEvent, Task> action)
-    {
-        Check.NotNull(action, nameof(action));
+        public override Task PublishFromOutboxAsync(
+            OutgoingEventInfo outgoingEvent,
+            OutboxConfig outboxConfig)
+        {
+            return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id);
+        }
 
-        GetOrCreateHandlerFactories(typeof(TEvent))
-            .Locking(factories =>
+        public async override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(
+            IEnumerable<OutgoingEventInfo> outgoingEvents,
+            OutboxConfig outboxConfig)
+        {
+            using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel())
             {
-                factories.RemoveAll(
-                    factory =>
+                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 singleInstanceFactory = factory as SingleInstanceHandlerFactory;
-                        if (singleInstanceFactory == null)
+                        var confirmed = pendingConfirms.Where(x => x.Key <= sequenceNumber);
+                        foreach (var entry in confirmed)
                         {
-                            return false;
+                            pendingConfirms.TryRemove(entry.Key, out var eventId);
+
+                            if (!ack)
+                            {
+                                failures.Add(eventId);
+                            }
                         }
+                    }
+                    else
+                    {
+                        pendingConfirms.TryRemove(sequenceNumber, out var eventId);
 
-                        var actionHandler = singleInstanceFactory.HandlerInstance as ActionEventHandler<TEvent>;
-                        if (actionHandler == null)
+                        if (!ack)
                         {
-                            return false;
+                            failures.Add(eventId);
                         }
+                    }
+                }
 
-                        return actionHandler.Action == action;
-                    });
-            });
-    }
+                foreach (var outgoingEvent in outgoingEventArray)
+                {
+                    pendingConfirms.TryAdd(channel.NextPublishSeqNo, outgoingEvent.Id);
+                    await PublishInternalAsync(channel, outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id);
+                }
 
-    /// <inheritdoc/>
-    public override void Unsubscribe(Type eventType, IEventHandler handler)
-    {
-        GetOrCreateHandlerFactories(eventType)
-            .Locking(factories =>
-            {
-                factories.RemoveAll(
-                    factory =>
-                        factory is SingleInstanceHandlerFactory &&
-                        (factory as SingleInstanceHandlerFactory).HandlerInstance == handler
-                );
-            });
-    }
+                channel.BasicAcks += (_, ea) => CleanPendingConfirms(ea.DeliveryTag, ea.Multiple, true);
+                channel.BasicNacks += (_, ea) => CleanPendingConfirms(ea.DeliveryTag, ea.Multiple, false);
 
-    /// <inheritdoc/>
-    public override void Unsubscribe(Type eventType, IEventHandlerFactory factory)
-    {
-        GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory));
-    }
-
-    /// <inheritdoc/>
-    public override void UnsubscribeAll(Type eventType)
-    {
-        GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear());
-    }
+                channel.WaitForConfirms();
 
-    protected override async Task PublishToEventBusAsync(Type eventType, object eventData)
-    {
-        await PublishAsync(eventType, eventData, null);
-    }
-
-    protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord)
-    {
-        unitOfWork.AddOrReplaceDistributedEvent(eventRecord);
-    }
-
-    public override Task PublishFromOutboxAsync(
-        OutgoingEventInfo outgoingEvent,
-        OutboxConfig outboxConfig)
-    {
-        return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id);
-    }
+                return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !failures.Contains(x.Id)).ToList());
+            }
+        }
 
-    public override async Task ProcessFromInboxAsync(
-        IncomingEventInfo incomingEvent,
-        InboxConfig inboxConfig)
-    {
-        var eventType = EventTypes.GetOrDefault(incomingEvent.EventName);
-        if (eventType == null)
+        public async override Task ProcessFromInboxAsync(
+            IncomingEventInfo incomingEvent,
+            InboxConfig inboxConfig)
         {
-            return;
+            var eventType = EventTypes.GetOrDefault(incomingEvent.EventName);
+            if (eventType == null)
+            {
+                return;
+            }
+
+            var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType);
+            var exceptions = new List<Exception>();
+            await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig);
+            if (exceptions.Any())
+            {
+                ThrowOriginalExceptions(eventType, exceptions);
+            }
         }
 
-        var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType);
-        var exceptions = new List<Exception>();
-        await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig);
-        if (exceptions.Any())
+        protected override byte[] Serialize(object eventData)
         {
-            ThrowOriginalExceptions(eventType, exceptions);
+            return Serializer.Serialize(eventData);
         }
-    }
 
-    protected override byte[] Serialize(object eventData)
-    {
-        return Serializer.Serialize(eventData);
-    }
+        public virtual Task PublishAsync(Type eventType, object eventData, IBasicProperties properties, Dictionary<string, object> headersArguments = null)
+        {
+            var eventName = EventNameAttribute.GetNameOrDefault(eventType);
+            var body = Serializer.Serialize(eventData);
 
-    public Task PublishAsync(Type eventType, object eventData, IBasicProperties properties, Dictionary<string, object> headersArguments = null)
-    {
-        var eventName = EventNameAttribute.GetNameOrDefault(eventType);
-        var body = Serializer.Serialize(eventData);
+            return PublishAsync(eventName, body, properties, headersArguments);
+        }
 
-        return PublishAsync(eventName, body, properties, headersArguments);
-    }
+        protected virtual Task PublishAsync(
+            string eventName,
+            byte[] body,
+            IBasicProperties properties,
+            Dictionary<string, object> headersArguments = null,
+            Guid? eventId = null)
+        {
+            using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel())
+            {
+                return PublishInternalAsync(channel, eventName, body, properties, headersArguments, eventId);
+            }
+        }
 
-    protected Task PublishAsync(
-        string eventName,
-        byte[] body,
-        IBasicProperties properties,
-        Dictionary<string, object> headersArguments = null,
-        Guid? eventId = null)
-    {
-        using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel())
+        protected virtual Task PublishInternalAsync(
+            IModel channel,
+            string eventName,
+            byte[] body,
+            IBasicProperties properties,
+            Dictionary<string, object> headersArguments = null,
+            Guid? eventId = null)
         {
             channel.ExchangeDeclare(
                 AbpRabbitMqEventBusOptions.ExchangeName,
@@ -262,68 +327,68 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
                 basicProperties: properties,
                 body: body
             );
-        }
 
-        return Task.CompletedTask;
-    }
-
-    private void SetEventMessageHeaders(IBasicProperties properties, Dictionary<string, object> headersArguments)
-    {
-        if (headersArguments == null)
-        {
-            return;
+            return Task.CompletedTask;
         }
 
-        properties.Headers ??= new Dictionary<string, object>();
-
-        foreach (var header in headersArguments)
+        private void SetEventMessageHeaders(IBasicProperties properties, Dictionary<string, object> headersArguments)
         {
-            properties.Headers[header.Key] = header.Value;
-        }
-    }
-
-    private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType)
-    {
-        return HandlerFactories.GetOrAdd(
-            eventType,
-            type =>
+            if (headersArguments == null)
             {
-                var eventName = EventNameAttribute.GetNameOrDefault(type);
-                EventTypes[eventName] = type;
-                return new List<IEventHandlerFactory>();
+                return;
             }
-        );
-    }
 
-    protected override IEnumerable<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType)
-    {
-        var handlerFactoryList = new List<EventTypeWithEventHandlerFactories>();
+            properties.Headers ??= new Dictionary<string, object>();
 
-        foreach (var handlerFactory in
-            HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key)))
-        {
-            handlerFactoryList.Add(
-                new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value));
+            foreach (var header in headersArguments)
+            {
+                properties.Headers[header.Key] = header.Value;
+            }
         }
 
-        return handlerFactoryList.ToArray();
-    }
-
-    private static bool ShouldTriggerEventForHandler(Type targetEventType, Type handlerEventType)
-    {
-        //Should trigger same type
-        if (handlerEventType == targetEventType)
+        private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType)
         {
-            return true;
+            return HandlerFactories.GetOrAdd(
+                eventType,
+                type =>
+                {
+                    var eventName = EventNameAttribute.GetNameOrDefault(type);
+                    EventTypes[eventName] = type;
+                    return new List<IEventHandlerFactory>();
+                }
+            );
         }
 
-        //TODO: Support inheritance? But it does not support on subscription to RabbitMq!
-        //Should trigger for inherited types
-        if (handlerEventType.IsAssignableFrom(targetEventType))
+        protected override IEnumerable<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType)
         {
-            return true;
+            var handlerFactoryList = new List<EventTypeWithEventHandlerFactories>();
+
+            foreach (var handlerFactory in
+                HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key)))
+            {
+                handlerFactoryList.Add(
+                    new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value));
+            }
+
+            return handlerFactoryList.ToArray();
         }
 
-        return false;
+        private static bool ShouldTriggerEventForHandler(Type targetEventType, Type handlerEventType)
+        {
+            //Should trigger same type
+            if (handlerEventType == targetEventType)
+            {
+                return true;
+            }
+
+            //TODO: Support inheritance? But it does not support on subscription to RabbitMq!
+            //Should trigger for inherited types
+            if (handlerEventType.IsAssignableFrom(targetEventType))
+            {
+                return true;
+            }
+
+            return false;
+        }
     }
 }

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

@@ -146,7 +146,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         await TriggerHandlersAsync(eventType, eventData);
     }
 
-    protected override async Task PublishToEventBusAsync(Type eventType, object eventData)
+    protected async override Task PublishToEventBusAsync(Type eventType, object eventData)
     {
         await AbpRebusEventBusOptions.Publish(Rebus, eventType, eventData);
     }
@@ -210,7 +210,12 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         return PublishToEventBusAsync(eventType, eventData);
     }
 
-    public override async Task ProcessFromInboxAsync(
+    public override Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    {
+        throw new NotImplementedException();
+    }
+
+    public async override Task ProcessFromInboxAsync(
         IncomingEventInfo incomingEvent,
         InboxConfig inboxConfig)
     {

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

@@ -8,156 +8,162 @@ using Volo.Abp.MultiTenancy;
 using Volo.Abp.Timing;
 using Volo.Abp.Uow;
 
-namespace Volo.Abp.EventBus.Distributed;
-
-public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventBus, ISupportsEventBoxes
+namespace Volo.Abp.EventBus.Distributed
 {
-    protected IGuidGenerator GuidGenerator { get; }
-    protected IClock Clock { get; }
-    protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; }
-
-    protected DistributedEventBusBase(
-        IServiceScopeFactory serviceScopeFactory,
-        ICurrentTenant currentTenant,
-        IUnitOfWorkManager unitOfWorkManager,
-        IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions,
-        IGuidGenerator guidGenerator,
-        IClock clock
-    ) : base(
-        serviceScopeFactory,
-        currentTenant,
-        unitOfWorkManager)
-    {
-        GuidGenerator = guidGenerator;
-        Clock = clock;
-        AbpDistributedEventBusOptions = abpDistributedEventBusOptions.Value;
-    }
-
-    public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) where TEvent : class
+    public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventBus, ISupportsEventBoxes
     {
-        return Subscribe(typeof(TEvent), handler);
-    }
+        protected IGuidGenerator GuidGenerator { get; }
+        protected IClock Clock { get; }
+        protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; }
+
+        protected DistributedEventBusBase(
+            IServiceScopeFactory serviceScopeFactory,
+            ICurrentTenant currentTenant,
+            IUnitOfWorkManager unitOfWorkManager,
+            IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions,
+            IGuidGenerator guidGenerator,
+            IClock clock
+        ) : base(
+            serviceScopeFactory,
+            currentTenant,
+            unitOfWorkManager)
+        {
+            GuidGenerator = guidGenerator;
+            Clock = clock;
+            AbpDistributedEventBusOptions = abpDistributedEventBusOptions.Value;
+        }
 
-    public override Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true)
-    {
-        return PublishAsync(eventType, eventData, onUnitOfWorkComplete, useOutbox: true);
-    }
+        public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) where TEvent : class
+        {
+            return Subscribe(typeof(TEvent), handler);
+        }
 
-    public Task PublishAsync<TEvent>(
-        TEvent eventData,
-        bool onUnitOfWorkComplete = true,
-        bool useOutbox = true)
-        where TEvent : class
-    {
-        return PublishAsync(typeof(TEvent), eventData, onUnitOfWorkComplete, useOutbox);
-    }
+        public override Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true)
+        {
+            return PublishAsync(eventType, eventData, onUnitOfWorkComplete, useOutbox: true);
+        }
 
-    public async Task PublishAsync(
-        Type eventType,
-        object eventData,
-        bool onUnitOfWorkComplete = true,
-        bool useOutbox = true)
-    {
-        if (onUnitOfWorkComplete && UnitOfWorkManager.Current != null)
+        public Task PublishAsync<TEvent>(
+            TEvent eventData,
+            bool onUnitOfWorkComplete = true,
+            bool useOutbox = true)
+            where TEvent : class
         {
-            AddToUnitOfWork(
-                UnitOfWorkManager.Current,
-                new UnitOfWorkEventRecord(eventType, eventData, EventOrderGenerator.GetNext(), useOutbox)
-            );
-            return;
+            return PublishAsync(typeof(TEvent), eventData, onUnitOfWorkComplete, useOutbox);
         }
 
-        if (useOutbox)
+        public async Task PublishAsync(
+            Type eventType,
+            object eventData,
+            bool onUnitOfWorkComplete = true,
+            bool useOutbox = true)
         {
-            if (await AddToOutboxAsync(eventType, eventData))
+            if (onUnitOfWorkComplete && UnitOfWorkManager.Current != null)
             {
+                AddToUnitOfWork(
+                    UnitOfWorkManager.Current,
+                    new UnitOfWorkEventRecord(eventType, eventData, EventOrderGenerator.GetNext(), useOutbox)
+                );
                 return;
             }
-        }
 
-        await PublishToEventBusAsync(eventType, eventData);
-    }
+            if (useOutbox)
+            {
+                if (await AddToOutboxAsync(eventType, eventData))
+                {
+                    return;
+                }
+            }
 
-    public abstract Task PublishFromOutboxAsync(
-        OutgoingEventInfo outgoingEvent,
-        OutboxConfig outboxConfig
-    );
+            await PublishToEventBusAsync(eventType, eventData);
+        }
 
-    public abstract Task ProcessFromInboxAsync(
-        IncomingEventInfo incomingEvent,
-        InboxConfig inboxConfig);
+        public abstract Task PublishFromOutboxAsync(
+            OutgoingEventInfo outgoingEvent,
+            OutboxConfig outboxConfig
+        );
 
-    private async Task<bool> AddToOutboxAsync(Type eventType, object eventData)
-    {
-        var unitOfWork = UnitOfWorkManager.Current;
-        if (unitOfWork == null)
-        {
-            return false;
-        }
+        public abstract Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(
+            IEnumerable<OutgoingEventInfo> outgoingEvents,
+            OutboxConfig outboxConfig
+        );
+
+        public abstract Task ProcessFromInboxAsync(
+            IncomingEventInfo incomingEvent,
+            InboxConfig inboxConfig);
 
-        foreach (var outboxConfig in AbpDistributedEventBusOptions.Outboxes.Values)
+        private async Task<bool> AddToOutboxAsync(Type eventType, object eventData)
         {
-            if (outboxConfig.Selector == null || outboxConfig.Selector(eventType))
+            var unitOfWork = UnitOfWorkManager.Current;
+            if (unitOfWork == null)
             {
-                var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType);
-                var eventName = EventNameAttribute.GetNameOrDefault(eventType);
-                await eventOutbox.EnqueueAsync(
-                    new OutgoingEventInfo(
-                        GuidGenerator.Create(),
-                        eventName,
-                        Serialize(eventData),
-                        Clock.Now
-                    )
-                );
-                return true;
+                return false;
             }
-        }
 
-        return false;
-    }
+            foreach (var outboxConfig in AbpDistributedEventBusOptions.Outboxes.Values)
+            {
+                if (outboxConfig.Selector == null || outboxConfig.Selector(eventType))
+                {
+                    var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType);
+                    var eventName = EventNameAttribute.GetNameOrDefault(eventType);
+                    await eventOutbox.EnqueueAsync(
+                        new OutgoingEventInfo(
+                            GuidGenerator.Create(),
+                            eventName,
+                            Serialize(eventData),
+                            Clock.Now
+                        )
+                    );
+                    return true;
+                }
+            }
 
-    protected async Task<bool> AddToInboxAsync(
-        string messageId,
-        string eventName,
-        Type eventType,
-        byte[] eventBytes)
-    {
-        if (AbpDistributedEventBusOptions.Inboxes.Count <= 0)
-        {
             return false;
         }
 
-        using (var scope = ServiceScopeFactory.CreateScope())
+        protected async Task<bool> AddToInboxAsync(
+            string messageId,
+            string eventName,
+            Type eventType,
+            byte[] eventBytes)
         {
-            foreach (var inboxConfig in AbpDistributedEventBusOptions.Inboxes.Values)
+            if (AbpDistributedEventBusOptions.Inboxes.Count <= 0)
             {
-                if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType))
-                {
-                    var eventInbox = (IEventInbox)scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType);
+                return false;
+            }
 
-                    if (!messageId.IsNullOrEmpty())
+            using (var scope = ServiceScopeFactory.CreateScope())
+            {
+                foreach (var inboxConfig in AbpDistributedEventBusOptions.Inboxes.Values)
+                {
+                    if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType))
                     {
-                        if (await eventInbox.ExistsByMessageIdAsync(messageId))
+                        var eventInbox = (IEventInbox) scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType);
+
+                        if (!messageId.IsNullOrEmpty())
                         {
-                            continue;
+                            if (await eventInbox.ExistsByMessageIdAsync(messageId))
+                            {
+                                continue;
+                            }
                         }
-                    }
 
-                    await eventInbox.EnqueueAsync(
-                        new IncomingEventInfo(
-                            GuidGenerator.Create(),
-                            messageId,
-                            eventName,
-                            eventBytes,
-                            Clock.Now
-                        )
-                    );
+                        await eventInbox.EnqueueAsync(
+                            new IncomingEventInfo(
+                                GuidGenerator.Create(),
+                                messageId,
+                                eventName,
+                                eventBytes,
+                                Clock.Now
+                            )
+                        );
+                    }
                 }
             }
+
+            return true;
         }
 
-        return true;
+        protected abstract byte[] Serialize(object eventData);
     }
-
-    protected abstract byte[] Serialize(object eventData);
 }

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

@@ -12,4 +12,6 @@ public interface IEventOutbox
     Task<List<OutgoingEventInfo>> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default);
 
     Task DeleteAsync(Guid id);
+
+    Task DeleteManyAsync(IEnumerable<Guid> ids);
 }

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

@@ -1,3 +1,4 @@
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace Volo.Abp.EventBus.Distributed;
@@ -9,6 +10,11 @@ public interface ISupportsEventBoxes
         OutboxConfig outboxConfig
     );
 
+    Task<MultipleOutgoingEventPublishResult> PublishManyFromOutboxAsync(
+        IEnumerable<OutgoingEventInfo> outgoingEvents,
+        OutboxConfig outboxConfig
+    );
+
     Task ProcessFromInboxAsync(
         IncomingEventInfo incomingEvent,
         InboxConfig inboxConfig

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

@@ -98,7 +98,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
                         break;
                     }
 
-                    Logger.LogInformation($"Found {waitingEvents.Count} events in the inbox.");
+                    //Logger.LogInformation($"Found {waitingEvents.Count} events in the inbox.");
 
                     foreach (var waitingEvent in waitingEvents)
                     {
@@ -113,7 +113,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
                             await uow.CompleteAsync();
                         }
 
-                        Logger.LogInformation($"Processed the incoming event with id = {waitingEvent.Id:N}");
+                        //Logger.LogInformation($"Processed the incoming event with id = {waitingEvent.Id:N}");
                     }
                 }
             }

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

@@ -1,4 +1,6 @@
 using System;
+using System.Diagnostics;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
@@ -76,25 +78,20 @@ public class OutboxSender : IOutboxSender, ITransientDependency
                 while (true)
                 {
                     var waitingEvents = await Outbox.GetWaitingEventsAsync(EventBusBoxesOptions.OutboxWaitingEventMaxCount, StoppingToken);
-                    if (waitingEvents.Count <= 0)
+                    if (waitingEvents.Count < 1000)
                     {
                         break;
                     }
 
                     Logger.LogInformation($"Found {waitingEvents.Count} events in the outbox.");
 
-                    foreach (var waitingEvent in waitingEvents)
-                    {
-                        await DistributedEventBus
-                            .AsSupportsEventBoxes()
-                            .PublishFromOutboxAsync(
-                                waitingEvent,
-                                OutboxConfig
-                            );
+                    var result = await DistributedEventBus
+                        .AsSupportsEventBoxes()
+                        .PublishManyFromOutboxAsync(waitingEvents, OutboxConfig);
 
-                        await Outbox.DeleteAsync(waitingEvent.Id);
-                        Logger.LogInformation($"Sent the event to the message broker with id = {waitingEvent.Id:N}");
-                    }
+                    await Outbox.DeleteManyAsync(result.PublishedOutgoingEvents.Select(x => x.Id).ToArray());
+
+                    Logger.LogInformation($"Sent {result.PublishedOutgoingEvents.Count} events to message broker");
                 }
             }
             else

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

@@ -0,0 +1,14 @@
+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.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs

@@ -68,4 +68,18 @@ public class MongoDbContextEventOutbox<TMongoDbContext> : IMongoDbContextEventOu
             await dbContext.OutgoingEvents.DeleteOneAsync(x => x.Id.Equals(id));
         }
     }
+
+    [UnitOfWork]
+    public async Task DeleteManyAsync(IEnumerable<Guid> ids)
+    {
+        var dbContext = (IHasEventOutbox)await MongoDbContextProvider.GetDbContextAsync();
+        if (dbContext.SessionHandle != null)
+        {
+            await dbContext.OutgoingEvents.DeleteManyAsync(dbContext.SessionHandle, x => ids.Contains(x.Id));
+        }
+        else
+        {
+            await dbContext.OutgoingEvents.DeleteManyAsync(x => ids.Contains(x.Id));
+        }
+    }
 }