mirror of https://github.com/abpframework/abp.git
liangshiwei
5 years ago
25 changed files with 44 additions and 614 deletions
@ -1,53 +0,0 @@ |
|||||
using System; |
|
||||
using System.Collections.Generic; |
|
||||
using System.Linq; |
|
||||
using System.Threading.Tasks; |
|
||||
using Confluent.Kafka; |
|
||||
using Microsoft.Extensions.Logging; |
|
||||
using Microsoft.Extensions.Logging.Abstractions; |
|
||||
using Microsoft.Extensions.Options; |
|
||||
using Volo.Abp.Data; |
|
||||
using Volo.Abp.DependencyInjection; |
|
||||
|
|
||||
namespace Volo.Abp.EventBus.Kafka |
|
||||
{ |
|
||||
public class KafkaEventErrorHandler : EventErrorHandlerBase, ISingletonDependency |
|
||||
{ |
|
||||
protected ILogger<KafkaEventErrorHandler> Logger { get; set; } |
|
||||
|
|
||||
public KafkaEventErrorHandler( |
|
||||
IOptions<AbpEventBusOptions> options) : base(options) |
|
||||
{ |
|
||||
Logger = NullLogger<KafkaEventErrorHandler>.Instance; |
|
||||
} |
|
||||
|
|
||||
protected override async Task RetryAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
if (Options.RetryStrategyOptions.IntervalMillisecond > 0) |
|
||||
{ |
|
||||
await Task.Delay(Options.RetryStrategyOptions.IntervalMillisecond); |
|
||||
} |
|
||||
|
|
||||
context.TryGetRetryAttempt(out var retryAttempt); |
|
||||
|
|
||||
await context.EventBus.As<KafkaDistributedEventBus>().PublishAsync( |
|
||||
context.EventType, |
|
||||
context.EventData, |
|
||||
context.GetProperty(HeadersKey).As<Headers>(), |
|
||||
new Dictionary<string, object> {{RetryAttemptKey, ++retryAttempt}}); |
|
||||
} |
|
||||
|
|
||||
protected override async Task MoveToDeadLetterAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
Logger.LogException( |
|
||||
context.Exceptions.Count == 1 ? context.Exceptions.First() : new AggregateException(context.Exceptions), |
|
||||
LogLevel.Error); |
|
||||
|
|
||||
await context.EventBus.As<KafkaDistributedEventBus>().PublishToDeadLetterAsync( |
|
||||
context.EventType, |
|
||||
context.EventData, |
|
||||
context.GetProperty(HeadersKey).As<Headers>(), |
|
||||
new Dictionary<string, object> {{"exceptions", context.Exceptions.Select(x => x.ToString()).ToList()}}); |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -1,47 +0,0 @@ |
|||||
using System; |
|
||||
using System.Collections.Generic; |
|
||||
using System.Linq; |
|
||||
using System.Threading.Tasks; |
|
||||
using Microsoft.Extensions.Options; |
|
||||
using RabbitMQ.Client; |
|
||||
using Volo.Abp.Data; |
|
||||
using Volo.Abp.DependencyInjection; |
|
||||
|
|
||||
namespace Volo.Abp.EventBus.RabbitMq |
|
||||
{ |
|
||||
public class RabbitMqEventErrorHandler : EventErrorHandlerBase, ISingletonDependency |
|
||||
{ |
|
||||
public RabbitMqEventErrorHandler( |
|
||||
IOptions<AbpEventBusOptions> options) |
|
||||
: base(options) |
|
||||
{ |
|
||||
} |
|
||||
|
|
||||
protected override async Task RetryAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
if (Options.RetryStrategyOptions.IntervalMillisecond > 0) |
|
||||
{ |
|
||||
await Task.Delay(Options.RetryStrategyOptions.IntervalMillisecond); |
|
||||
} |
|
||||
|
|
||||
context.TryGetRetryAttempt(out var retryAttempt); |
|
||||
|
|
||||
await context.EventBus.As<RabbitMqDistributedEventBus>().PublishAsync( |
|
||||
context.EventType, |
|
||||
context.EventData, |
|
||||
context.GetProperty(HeadersKey).As<IBasicProperties>(), |
|
||||
new Dictionary<string, object> |
|
||||
{ |
|
||||
{RetryAttemptKey, ++retryAttempt}, |
|
||||
{"exceptions", context.Exceptions.Select(x => x.ToString()).ToList()} |
|
||||
}); |
|
||||
} |
|
||||
|
|
||||
protected override Task MoveToDeadLetterAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
ThrowOriginalExceptions(context); |
|
||||
|
|
||||
return Task.CompletedTask; |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -1,32 +0,0 @@ |
|||||
using System.Threading.Tasks; |
|
||||
using Microsoft.Extensions.Options; |
|
||||
using Volo.Abp.DependencyInjection; |
|
||||
|
|
||||
namespace Volo.Abp.EventBus.Rebus |
|
||||
{ |
|
||||
/// <summary>
|
|
||||
/// Rebus will automatic retries and error handling: https://github.com/rebus-org/Rebus/wiki/Automatic-retries-and-error-handling
|
|
||||
/// </summary>
|
|
||||
public class RebusEventErrorHandler : EventErrorHandlerBase, ISingletonDependency |
|
||||
{ |
|
||||
public RebusEventErrorHandler( |
|
||||
IOptions<AbpEventBusOptions> options) |
|
||||
: base(options) |
|
||||
{ |
|
||||
} |
|
||||
|
|
||||
protected override Task RetryAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
ThrowOriginalExceptions(context); |
|
||||
|
|
||||
return Task.CompletedTask; |
|
||||
} |
|
||||
|
|
||||
protected override Task MoveToDeadLetterAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
ThrowOriginalExceptions(context); |
|
||||
|
|
||||
return Task.CompletedTask; |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -1,22 +0,0 @@ |
|||||
using System; |
|
||||
|
|
||||
namespace Volo.Abp.EventBus |
|
||||
{ |
|
||||
public class AbpEventBusOptions |
|
||||
{ |
|
||||
public bool EnabledErrorHandle { get; set; } |
|
||||
|
|
||||
public Func<Type, bool> ErrorHandleSelector { get; set; } |
|
||||
|
|
||||
public string DeadLetterName { get; set; } |
|
||||
|
|
||||
public AbpEventBusRetryStrategyOptions RetryStrategyOptions { get; set; } |
|
||||
|
|
||||
public void UseRetryStrategy(Action<AbpEventBusRetryStrategyOptions> action = null) |
|
||||
{ |
|
||||
EnabledErrorHandle = true; |
|
||||
RetryStrategyOptions = new AbpEventBusRetryStrategyOptions(); |
|
||||
action?.Invoke(RetryStrategyOptions); |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -1,9 +0,0 @@ |
|||||
namespace Volo.Abp.EventBus |
|
||||
{ |
|
||||
public class AbpEventBusRetryStrategyOptions |
|
||||
{ |
|
||||
public int IntervalMillisecond { get; set; } = 3000; |
|
||||
|
|
||||
public int MaxRetryAttempts { get; set; } = 3; |
|
||||
} |
|
||||
} |
|
||||
@ -1,76 +0,0 @@ |
|||||
using System; |
|
||||
using System.Threading.Tasks; |
|
||||
using Microsoft.Extensions.Options; |
|
||||
|
|
||||
namespace Volo.Abp.EventBus |
|
||||
{ |
|
||||
public abstract class EventErrorHandlerBase : IEventErrorHandler |
|
||||
{ |
|
||||
public const string HeadersKey = "headers"; |
|
||||
public const string RetryAttemptKey = "retryAttempt"; |
|
||||
|
|
||||
protected AbpEventBusOptions Options { get; } |
|
||||
|
|
||||
protected EventErrorHandlerBase(IOptions<AbpEventBusOptions> options) |
|
||||
{ |
|
||||
Options = options.Value; |
|
||||
} |
|
||||
|
|
||||
public virtual async Task HandleAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
if (!await ShouldHandleAsync(context)) |
|
||||
{ |
|
||||
ThrowOriginalExceptions(context); |
|
||||
} |
|
||||
|
|
||||
if (await ShouldRetryAsync(context)) |
|
||||
{ |
|
||||
await RetryAsync(context); |
|
||||
return; |
|
||||
} |
|
||||
|
|
||||
await MoveToDeadLetterAsync(context); |
|
||||
} |
|
||||
|
|
||||
protected abstract Task RetryAsync(EventExecutionErrorContext context); |
|
||||
|
|
||||
protected abstract Task MoveToDeadLetterAsync(EventExecutionErrorContext context); |
|
||||
|
|
||||
protected virtual Task<bool> ShouldHandleAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
if (!Options.EnabledErrorHandle) |
|
||||
{ |
|
||||
return Task.FromResult(false); |
|
||||
} |
|
||||
|
|
||||
return Task.FromResult(Options.ErrorHandleSelector == null || Options.ErrorHandleSelector.Invoke(context.EventType)); |
|
||||
} |
|
||||
|
|
||||
protected virtual Task<bool> ShouldRetryAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
if (Options.RetryStrategyOptions == null) |
|
||||
{ |
|
||||
return Task.FromResult(false); |
|
||||
} |
|
||||
|
|
||||
if (!context.TryGetRetryAttempt(out var retryAttempt)) |
|
||||
{ |
|
||||
return Task.FromResult(false); |
|
||||
} |
|
||||
|
|
||||
return Task.FromResult(Options.RetryStrategyOptions.MaxRetryAttempts > retryAttempt); |
|
||||
} |
|
||||
|
|
||||
protected virtual void ThrowOriginalExceptions(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); |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -1,38 +0,0 @@ |
|||||
using System; |
|
||||
using System.Collections.Generic; |
|
||||
using Volo.Abp.Data; |
|
||||
using Volo.Abp.ObjectExtending; |
|
||||
|
|
||||
namespace Volo.Abp.EventBus |
|
||||
{ |
|
||||
public class EventExecutionErrorContext : ExtensibleObject |
|
||||
{ |
|
||||
public IReadOnlyList<Exception> Exceptions { get; } |
|
||||
|
|
||||
public object EventData { get; set; } |
|
||||
|
|
||||
public Type EventType { get; } |
|
||||
|
|
||||
public IEventBus EventBus { get; } |
|
||||
|
|
||||
public EventExecutionErrorContext(List<Exception> exceptions, Type eventType, IEventBus eventBus) |
|
||||
{ |
|
||||
Exceptions = exceptions; |
|
||||
EventType = eventType; |
|
||||
EventBus = eventBus; |
|
||||
} |
|
||||
|
|
||||
public bool TryGetRetryAttempt(out int retryAttempt) |
|
||||
{ |
|
||||
retryAttempt = 0; |
|
||||
if (!this.HasProperty(EventErrorHandlerBase.RetryAttemptKey)) |
|
||||
{ |
|
||||
return false; |
|
||||
} |
|
||||
|
|
||||
retryAttempt = this.GetProperty<int>(EventErrorHandlerBase.RetryAttemptKey); |
|
||||
return true; |
|
||||
|
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -1,9 +0,0 @@ |
|||||
using System.Threading.Tasks; |
|
||||
|
|
||||
namespace Volo.Abp.EventBus |
|
||||
{ |
|
||||
public interface IEventErrorHandler |
|
||||
{ |
|
||||
Task HandleAsync(EventExecutionErrorContext context); |
|
||||
} |
|
||||
} |
|
||||
@ -1,60 +0,0 @@ |
|||||
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 |
|
||||
{ |
|
||||
[ExposeServices(typeof(LocalEventErrorHandler), typeof(IEventErrorHandler))] |
|
||||
public class LocalEventErrorHandler : EventErrorHandlerBase, ISingletonDependency |
|
||||
{ |
|
||||
protected Dictionary<Guid, int> RetryTracking { get; } |
|
||||
|
|
||||
public LocalEventErrorHandler( |
|
||||
IOptions<AbpEventBusOptions> options) |
|
||||
: base(options) |
|
||||
{ |
|
||||
RetryTracking = new Dictionary<Guid, int>(); |
|
||||
} |
|
||||
|
|
||||
protected override async Task RetryAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
if (Options.RetryStrategyOptions.IntervalMillisecond > 0) |
|
||||
{ |
|
||||
await Task.Delay(Options.RetryStrategyOptions.IntervalMillisecond); |
|
||||
} |
|
||||
|
|
||||
var messageId = context.GetProperty<Guid>(nameof(LocalEventMessage.MessageId)); |
|
||||
|
|
||||
context.TryGetRetryAttempt(out var retryAttempt); |
|
||||
RetryTracking[messageId] = ++retryAttempt; |
|
||||
|
|
||||
await context.EventBus.As<LocalEventBus>().PublishAsync(new LocalEventMessage(messageId, context.EventData, context.EventType)); |
|
||||
|
|
||||
RetryTracking.Remove(messageId); |
|
||||
} |
|
||||
|
|
||||
protected override Task MoveToDeadLetterAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
ThrowOriginalExceptions(context); |
|
||||
|
|
||||
return Task.CompletedTask; |
|
||||
} |
|
||||
|
|
||||
protected override async Task<bool> ShouldRetryAsync(EventExecutionErrorContext context) |
|
||||
{ |
|
||||
var messageId = context.GetProperty<Guid>(nameof(LocalEventMessage.MessageId)); |
|
||||
context.SetProperty(RetryAttemptKey, RetryTracking.GetOrDefault(messageId)); |
|
||||
|
|
||||
if (await base.ShouldRetryAsync(context)) |
|
||||
{ |
|
||||
return true; |
|
||||
} |
|
||||
|
|
||||
RetryTracking.Remove(messageId); |
|
||||
return false; |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -1,72 +0,0 @@ |
|||||
using System; |
|
||||
using System.Threading.Tasks; |
|
||||
using Shouldly; |
|
||||
using Xunit; |
|
||||
|
|
||||
namespace Volo.Abp.EventBus.Local |
|
||||
{ |
|
||||
public class EventBus_Exception_Handler_Tests : EventBusTestBase |
|
||||
{ |
|
||||
[Fact] |
|
||||
public async Task Should_Not_Handle_Exception() |
|
||||
{ |
|
||||
var retryAttempt = 0; |
|
||||
LocalEventBus.Subscribe<MySimpleEventData>(eventData => |
|
||||
{ |
|
||||
retryAttempt++; |
|
||||
throw new Exception("This exception is intentionally thrown!"); |
|
||||
}); |
|
||||
|
|
||||
var appException = await Assert.ThrowsAsync<Exception>(async () => |
|
||||
{ |
|
||||
await LocalEventBus.PublishAsync(new MySimpleEventData(1)); |
|
||||
}); |
|
||||
|
|
||||
retryAttempt.ShouldBe(1); |
|
||||
appException.Message.ShouldBe("This exception is intentionally thrown!"); |
|
||||
} |
|
||||
|
|
||||
[Fact] |
|
||||
public async Task Should_Handle_Exception() |
|
||||
{ |
|
||||
var retryAttempt = 0; |
|
||||
LocalEventBus.Subscribe<MyExceptionHandleEventData>(eventData => |
|
||||
{ |
|
||||
eventData.Value.ShouldBe(0); |
|
||||
retryAttempt++; |
|
||||
if (retryAttempt < 2) |
|
||||
{ |
|
||||
throw new Exception("This exception is intentionally thrown!"); |
|
||||
} |
|
||||
|
|
||||
return Task.CompletedTask; |
|
||||
|
|
||||
}); |
|
||||
|
|
||||
await LocalEventBus.PublishAsync(new MyExceptionHandleEventData(0)); |
|
||||
retryAttempt.ShouldBe(2); |
|
||||
} |
|
||||
|
|
||||
[Fact] |
|
||||
public async Task Should_Throw_Exception_After_Error_Handle() |
|
||||
{ |
|
||||
var retryAttempt = 0; |
|
||||
LocalEventBus.Subscribe<MyExceptionHandleEventData>(eventData => |
|
||||
{ |
|
||||
eventData.Value.ShouldBe(0); |
|
||||
|
|
||||
retryAttempt++; |
|
||||
|
|
||||
throw new Exception("This exception is intentionally thrown!"); |
|
||||
}); |
|
||||
|
|
||||
var appException = await Assert.ThrowsAsync<Exception>(async () => |
|
||||
{ |
|
||||
await LocalEventBus.PublishAsync(new MyExceptionHandleEventData(0)); |
|
||||
}); |
|
||||
|
|
||||
retryAttempt.ShouldBe(4); |
|
||||
appException.Message.ShouldBe("This exception is intentionally thrown!"); |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -1,12 +0,0 @@ |
|||||
namespace Volo.Abp.EventBus |
|
||||
{ |
|
||||
public class MyExceptionHandleEventData |
|
||||
{ |
|
||||
public int Value { get; set; } |
|
||||
|
|
||||
public MyExceptionHandleEventData(int value) |
|
||||
{ |
|
||||
Value = value; |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
Loading…
Reference in new issue