Abstract EventErrorHandler

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

@@ -6,8 +6,10 @@ using System.Threading.Tasks;
 using Confluent.Kafka;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
+using Volo.Abp.Data;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.EventBus.Local;
 using Volo.Abp.Kafka;
 using Volo.Abp.MultiTenancy;
 using Volo.Abp.Threading;
@@ -34,8 +36,9 @@ namespace Volo.Abp.EventBus.Kafka
             IKafkaMessageConsumerFactory messageConsumerFactory,
             IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions,
             IKafkaSerializer serializer,
-            IProducerPool producerPool)
+            IProducerPool producerPool,
-            : base(serviceScopeFactory, currentTenant)
+            IEventErrorHandler errorHandler)
+            : base(serviceScopeFactory, currentTenant, errorHandler)
         {
             AbpKafkaEventBusOptions = abpKafkaEventBusOptions.Value;
             AbpDistributedEventBusOptions = abpDistributedEventBusOptions.Value;
@@ -54,6 +57,7 @@ namespace Volo.Abp.EventBus.Kafka
                 AbpKafkaEventBusOptions.GroupId,
                 AbpKafkaEventBusOptions.ConnectionName);
 
+            Consumer.Consume();
             Consumer.OnMessageReceived(ProcessEventAsync);
 
             SubscribeHandlers(AbpDistributedEventBusOptions.Handlers);
@@ -68,9 +72,10 @@ namespace Volo.Abp.EventBus.Kafka
                 return;
             }
 
-            var eventData = Serializer.Deserialize(message.Value, eventType);
+            var eventMessage = Serializer.Deserialize<LocalEventMessage>(message.Value);
 
-            await TriggerHandlersAsync(eventType, eventData);
+            await TriggerHandlersAsync(eventType, eventMessage,
+                context => { context.SetProperty(KafkaEventErrorHandler.HeadersKey, message.Headers); });
         }
 
         public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) where TEvent : class
@@ -147,6 +152,11 @@ namespace Volo.Abp.EventBus.Kafka
         }
 
         public override async Task PublishAsync(Type eventType, object eventData)
+        {
+            await PublishAsync(eventType, eventData, null);
+        }
+
+        public virtual async Task PublishAsync(Type eventType, object eventData, Headers headers)
         {
             var eventName = EventNameAttribute.GetNameOrDefault(eventType);
             var body = Serializer.Serialize(eventData);
@@ -157,7 +167,7 @@ namespace Volo.Abp.EventBus.Kafka
                 AbpKafkaEventBusOptions.TopicName,
                 new Message<string, byte[]>
                 {
-                    Key = eventName, Value = body
+                    Key = eventName, Value = body, Headers = headers
                 });
         }
 

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

@@ -0,0 +1,90 @@
+using System.Threading.Tasks;
+using Confluent.Kafka;
+using Microsoft.Extensions.Options;
+using Volo.Abp.Data;
+using Volo.Abp.DependencyInjection;
+using Volo.Abp.Kafka;
+
+namespace Volo.Abp.EventBus.Kafka
+{
+    public class KafkaEventErrorHandler : EventErrorHandlerBase, ISingletonDependency
+    {
+        public const string HeadersKey = "headers";
+        public const string RetryIndexKey = "retryIndex";
+
+        protected IKafkaSerializer Serializer { get; }
+        protected KafkaDistributedEventBus EventBus { get; }
+        protected IProducerPool ProducerPool { get; }
+        protected AbpKafkaEventBusOptions AbpKafkaEventBusOptions { get; }
+
+        protected string ErrorTopicName { get; }
+
+        public KafkaEventErrorHandler(
+            IOptions<AbpEventBusOptions> options,
+            IKafkaSerializer serializer,
+            KafkaDistributedEventBus eventBus,
+            IKafkaMessageConsumerFactory consumerFactory,
+            IProducerPool producerPool,
+            IOptions<AbpKafkaEventBusOptions> abpKafkaEventBusOptions) : base(options)
+        {
+            Serializer = serializer;
+            EventBus = eventBus;
+            ProducerPool = producerPool;
+            AbpKafkaEventBusOptions = abpKafkaEventBusOptions.Value;
+
+            ErrorTopicName = options.Value.ErrorQueue ?? abpKafkaEventBusOptions.Value.TopicName + "_error";
+            consumerFactory.Create(ErrorTopicName, string.Empty, abpKafkaEventBusOptions.Value.ConnectionName);
+        }
+
+        protected override async Task Retry(EventExecutionErrorContext context)
+        {
+            if (Options.RetryStrategyOptions.IntervalMillisecond > 0)
+            {
+                await Task.Delay(Options.RetryStrategyOptions.IntervalMillisecond);
+            }
+
+            var headers = context.GetProperty<Headers>(HeadersKey) ?? new Headers();
+            var index = Serializer.Deserialize<int>(headers.GetLastBytes(RetryIndexKey));
+
+            headers.Remove(RetryIndexKey);
+            headers.Add(RetryIndexKey, Serializer.Serialize(++index));
+
+            await EventBus.PublishAsync(context.EventType, context.EventData, headers);
+        }
+
+        protected override async Task MoveToErrorQueue(EventExecutionErrorContext context)
+        {
+            var producer = ProducerPool.Get(AbpKafkaEventBusOptions.ConnectionName);
+            var eventName = EventNameAttribute.GetNameOrDefault(context.EventType);
+            var body = Serializer.Serialize(context.EventData);
+
+            await producer.ProduceAsync(
+                AbpKafkaEventBusOptions.TopicName,
+                new Message<string, byte[]>
+                {
+                    Key = eventName, Value = body,
+                    Headers = new Headers {{"exceptions", Serializer.Serialize(context.Exceptions)}}
+                });
+        }
+
+        protected override bool ShouldRetry(EventExecutionErrorContext context)
+        {
+            if (!base.ShouldRetry(context))
+            {
+                return false;
+            }
+
+            var headers = context.GetProperty<Headers>(HeadersKey);
+            var index = 1;
+
+            if (headers == null)
+            {
+                return true;
+            }
+
+            index = Serializer.Deserialize<int>(headers.GetLastBytes(RetryIndexKey));
+
+            return Options.RetryStrategyOptions.Count < index;
+        }
+    }
+}

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

@@ -41,8 +41,9 @@ namespace Volo.Abp.EventBus.RabbitMq
             IServiceScopeFactory serviceScopeFactory,
             IOptions<AbpDistributedEventBusOptions> distributedEventBusOptions,
             IRabbitMqMessageConsumerFactory messageConsumerFactory,
-            ICurrentTenant currentTenant)
+            ICurrentTenant currentTenant,
-            : base(serviceScopeFactory, currentTenant)
+            IEventErrorHandler errorHandler)
+            : base(serviceScopeFactory, currentTenant, errorHandler)
         {
             ConnectionPool = connectionPool;
             Serializer = serializer;

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

@@ -30,8 +30,9 @@ namespace Volo.Abp.EventBus.Rebus
             ICurrentTenant currentTenant,
             IBus rebus,
             IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions,
-            IOptions<AbpRebusEventBusOptions> abpEventBusRebusOptions) :
+            IOptions<AbpRebusEventBusOptions> abpEventBusRebusOptions,
-            base(serviceScopeFactory, currentTenant)
+            IEventErrorHandler errorHandler) :
+            base(serviceScopeFactory, currentTenant, errorHandler)
         {
             Rebus = rebus;
             AbpRebusEventBusOptions = abpEventBusRebusOptions.Value;

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

@@ -0,0 +1,21 @@
+using System;
+
+namespace Volo.Abp.EventBus
+{
+    public class AbpEventBusOptions
+    {
+        public bool EnabledErrorHandle { get; set; }
+
+        public Func<Type, bool> ErrorHandleSelector { get; set; }
+
+        public string ErrorQueue { get; set; }
+
+        public AbpEventBusRetryStrategyOptions RetryStrategyOptions { get; set; }
+
+        public void UseRetryStrategy(Action<AbpEventBusRetryStrategyOptions> action = null)
+        {
+            RetryStrategyOptions = new AbpEventBusRetryStrategyOptions();
+            action?.Invoke(RetryStrategyOptions);
+        }
+    }
+}

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

@@ -0,0 +1,10 @@
+namespace Volo.Abp.EventBus
+{
+    public class AbpEventBusRetryStrategyOptions
+    {
+
+        public int IntervalMillisecond { get; set; } = 3000;
+
+        public int Count { get; set; } = 3;
+    }
+}

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

@@ -8,6 +8,7 @@ using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
 using Volo.Abp.Collections;
 using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.EventBus.Local;
 using Volo.Abp.MultiTenancy;
 using Volo.Abp.Reflection;
 
@@ -19,10 +20,16 @@ namespace Volo.Abp.EventBus
 
         protected ICurrentTenant CurrentTenant { get; }
 
-        protected EventBusBase(IServiceScopeFactory serviceScopeFactory, ICurrentTenant currentTenant)
+        protected IEventErrorHandler ErrorHandler { get; }
+
+        protected EventBusBase(
+            IServiceScopeFactory serviceScopeFactory,
+            ICurrentTenant currentTenant,
+            IEventErrorHandler errorHandler)
         {
             ServiceScopeFactory = serviceScopeFactory;
             CurrentTenant = currentTenant;
+            ErrorHandler = errorHandler;
         }
 
         /// <inheritdoc/>
@@ -89,7 +96,7 @@ namespace Volo.Abp.EventBus
         /// <inheritdoc/>
         public abstract Task PublishAsync(Type eventType, object eventData);
 
-        public virtual async Task TriggerHandlersAsync(Type eventType, object eventData)
+        public virtual async Task TriggerHandlersAsync(Type eventType, object eventData, Action<EventExecutionErrorContext> onErrorAction = null)
         {
             var exceptions = new List<Exception>();
 
@@ -97,16 +104,13 @@ namespace Volo.Abp.EventBus
 
             if (exceptions.Any())
             {
-                if (exceptions.Count == 1)
+                var context = new EventExecutionErrorContext(exceptions, eventData, eventType);
-                {
+                onErrorAction?.Invoke(context);
-                    exceptions[0].ReThrow();
+                await ErrorHandler.Handle(context);
-                }
-
-                throw new AggregateException("More than one error has occurred while triggering the event: " + eventType, exceptions);
             }
         }
 
-        protected virtual async Task TriggerHandlersAsync(Type eventType, object eventData, List<Exception> exceptions)
+        protected virtual async Task TriggerHandlersAsync(Type eventType, object eventData , List<Exception> exceptions)
         {
             await new SynchronizationContextRemover();
 
@@ -217,6 +221,11 @@ namespace Volo.Abp.EventBus
             };
         }
 
+        protected virtual void OnErrorHandle(EventExecutionErrorContext context)
+        {
+
+        }
+
         protected class EventTypeWithEventHandlerFactories
         {
             public Type EventType { get; }

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

@@ -0,0 +1,57 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Options;
+using Volo.Abp.Data;
+
+namespace Volo.Abp.EventBus
+{
+    public abstract class EventErrorHandlerBase : IEventErrorHandler
+    {
+        protected AbpEventBusOptions Options { get; }
+
+        public EventErrorHandlerBase(IOptions<AbpEventBusOptions> options)
+        {
+            Options = options.Value;
+        }
+
+        public virtual async Task Handle(EventExecutionErrorContext context)
+        {
+            if (!ShouldHandle(context))
+            {
+                return;
+            }
+
+            if (ShouldRetry(context))
+            {
+                await Retry(context);
+                return;
+            }
+
+            await MoveToErrorQueue(context);
+        }
+
+        protected abstract Task Retry(EventExecutionErrorContext context);
+
+        protected abstract Task MoveToErrorQueue(EventExecutionErrorContext context);
+
+        protected virtual bool ShouldHandle(EventExecutionErrorContext context)
+        {
+            if (!Options.EnabledErrorHandle)
+            {
+                return false;
+            }
+
+            if (Options.ErrorHandleSelector != null)
+            {
+                return Options.ErrorHandleSelector.Invoke(context.EventType);
+            }
+
+            return false;
+        }
+
+        protected virtual bool ShouldRetry(EventExecutionErrorContext context)
+        {
+            return Options.RetryStrategyOptions == null && false;
+        }
+    }
+}

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

@@ -0,0 +1,23 @@
+using System;
+using System.Collections.Generic;
+using Volo.Abp.EventBus.Local;
+using Volo.Abp.ObjectExtending;
+
+namespace Volo.Abp.EventBus
+{
+    public class EventExecutionErrorContext : ExtensibleObject
+    {
+        public IReadOnlyList<Exception> Exceptions { get; }
+
+        public object EventData { get; }
+
+        public Type EventType { get; }
+
+        public EventExecutionErrorContext(List<Exception> exceptions, object eventData, Type eventType)
+        {
+            Exceptions = exceptions;
+            EventData = eventData;
+            EventType = eventType;
+        }
+    }
+}

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

@@ -0,0 +1,9 @@
+using System.Threading.Tasks;
+
+namespace Volo.Abp.EventBus
+{
+    public interface IEventErrorHandler
+    {
+        Task Handle(EventExecutionErrorContext context);
+    }
+}

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBus.cs

@@ -7,6 +7,7 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
+using Volo.Abp.Data;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.MultiTenancy;
 using Volo.Abp.Threading;
@@ -31,8 +32,9 @@ namespace Volo.Abp.EventBus.Local
         public LocalEventBus(
             IOptions<AbpLocalEventBusOptions> options,
             IServiceScopeFactory serviceScopeFactory,
-            ICurrentTenant currentTenant)
+            ICurrentTenant currentTenant,
-            : base(serviceScopeFactory, currentTenant)
+            IEventErrorHandler errorHandler)
+            : base(serviceScopeFactory, currentTenant, errorHandler)
         {
             Options = options.Value;
             Logger = NullLogger<LocalEventBus>.Instance;
@@ -119,19 +121,15 @@ namespace Volo.Abp.EventBus.Local
 
         public override async Task PublishAsync(Type eventType, object eventData)
         {
-            var exceptions = new List<Exception>();
+            await PublishAsync(new LocalEventMessage(Guid.NewGuid(), eventData, eventType));
-
+        }
-            await TriggerHandlersAsync(eventType, eventData, exceptions);
 
-            if (exceptions.Any())
+        public virtual async Task PublishAsync(LocalEventMessage localEventMessage)
+        {
+            await TriggerHandlersAsync(localEventMessage.EventType, localEventMessage.EventData, errorContext =>
             {
-                if (exceptions.Count == 1)
+                errorContext.SetProperty("messageId", localEventMessage.MessageId);
-                {
+            });
-                    exceptions[0].ReThrow();
-                }
-
-                throw new AggregateException("More than one error has occurred while triggering the event: " + eventType, exceptions);
-            }
         }
 
         protected override IEnumerable<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType)

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventErrorHandler.cs

@@ -0,0 +1,71 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Options;
+using Volo.Abp.Data;
+using Volo.Abp.DependencyInjection;
+
+namespace Volo.Abp.EventBus.Local
+{
+    public class LocalEventErrorHandler : EventErrorHandlerBase, ISingletonDependency
+    {
+        protected ILocalEventBus LocalEventBus { get; }
+        protected Dictionary<Guid, int> RetryTracking { get; }
+
+        public LocalEventErrorHandler(
+            IOptions<AbpEventBusOptions> options,
+            ILocalEventBus localEventBus)
+            : base(options)
+        {
+            LocalEventBus = localEventBus;
+            RetryTracking = new Dictionary<Guid, int>();
+        }
+
+        protected override async Task Retry(EventExecutionErrorContext context)
+        {
+            if (Options.RetryStrategyOptions.IntervalMillisecond > 0)
+            {
+                await Task.Delay(Options.RetryStrategyOptions.IntervalMillisecond);
+            }
+
+            var messageId = context.GetProperty<Guid>("messageId");
+
+            await LocalEventBus.PublishAsync(context.EventType,
+                new LocalEventMessage(messageId, context.EventData, context.EventType));
+        }
+
+        protected override Task MoveToErrorQueue(EventExecutionErrorContext context)
+        {
+            if (context.Exceptions.Count == 1)
+            {
+                context.Exceptions[0].ReThrow();
+            }
+
+            throw new AggregateException(
+                "More than one error has occurred while triggering the event: " + context.EventType,
+                context.Exceptions);
+        }
+
+        protected override bool ShouldRetry(EventExecutionErrorContext context)
+        {
+            if (!base.ShouldRetry(context))
+            {
+                return false;
+            }
+
+            var messageId = context.GetProperty<Guid>("messageId");
+
+            var index = RetryTracking.GetOrDefault(messageId);
+
+            if (Options.RetryStrategyOptions.Count >= index)
+            {
+                RetryTracking.Remove(messageId);
+                return false;
+            }
+
+            RetryTracking[messageId] = ++index;
+
+            return true;
+        }
+    }
+}

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventMessage.cs

@@ -0,0 +1,20 @@
+using System;
+
+namespace Volo.Abp.EventBus.Local
+{
+    public class LocalEventMessage
+    {
+        public Guid MessageId { get; }
+
+        public object EventData { get; }
+
+        public Type EventType { get; }
+
+        public LocalEventMessage(Guid messageId, object eventData, Type eventType)
+        {
+            MessageId = messageId;
+            EventData = eventData;
+            EventType = eventType;
+        }
+    }
+}

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

@@ -7,5 +7,7 @@ namespace Volo.Abp.Kafka
     public interface IKafkaMessageConsumer
     {
         void OnMessageReceived(Func<Message<string, byte[]>, Task> callback);
+
+        void Consume();
     }
 }

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

@@ -7,5 +7,7 @@ namespace Volo.Abp.Kafka
         byte[] Serialize(object obj);
 
         object Deserialize(byte[] value, Type type);
+
+        T Deserialize<T>(byte[] value);
     }
 }

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

@@ -63,7 +63,6 @@ namespace Volo.Abp.Kafka
             GroupId = groupId;
 
             AsyncHelper.RunSync(CreateTopicAsync);
-            Consume();
         }
 
         public virtual void OnMessageReceived(Func<Message<string, byte[]>, Task> callback)
@@ -98,7 +97,7 @@ namespace Volo.Abp.Kafka
             }
         }
 
-        protected virtual void Consume()
+        public virtual void Consume()
         {
             Consumer = ConsumerPool.Get(GroupId, ConnectionName);
 

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

@@ -23,5 +23,10 @@ namespace Volo.Abp.Kafka
         {
             return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value));
         }
+
+        public T Deserialize<T>(byte[] value)
+        {
+            return _jsonSerializer.Deserialize<T>(Encoding.UTF8.GetString(value));
+        }
     }
 }

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqSerializer.cs

@@ -7,5 +7,7 @@ namespace Volo.Abp.RabbitMQ
         byte[] Serialize(object obj);
 
         object Deserialize(byte[] value, Type type);
+
+        T Deserialize<T>(byte[] value);
     }
 }

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/Utf8JsonRabbitMqSerializer.cs

@@ -23,5 +23,10 @@ namespace Volo.Abp.RabbitMQ
         {
             return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value));
         }
+
+        public T Deserialize<T>(byte[] value)
+        {
+            return _jsonSerializer.Deserialize<T>(Encoding.UTF8.GetString(value));
+        }
     }
-}
+}