diff --git a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs index b6dc3e9bb5..aeeccef062 100644 --- a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs @@ -49,7 +49,7 @@ namespace Volo.Abp.EventBus.Kafka MessageConsumerFactory = messageConsumerFactory; Serializer = serializer; ProducerPool = producerPool; - DeadLetterTopicName = AbpEventBusOptions.DeadLetterQueue ?? AbpKafkaEventBusOptions.TopicName + "_error"; + DeadLetterTopicName = AbpEventBusOptions.DeadLetterName ?? AbpKafkaEventBusOptions.TopicName + "_dead_letter"; HandlerFactories = new ConcurrentDictionary>(); EventTypes = new ConcurrentDictionary(); diff --git a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaEventErrorHandler.cs b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaEventErrorHandler.cs index 2ee2a23515..b9e46fdd6a 100644 --- a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaEventErrorHandler.cs +++ b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaEventErrorHandler.cs @@ -64,7 +64,7 @@ namespace Volo.Abp.EventBus.Kafka var index = Serializer.Deserialize(headers.GetLastBytes(RetryIndexKey)); - return Options.RetryStrategyOptions.Count < index; + return Options.RetryStrategyOptions.MaxRetryAttempts < index; } } } diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs index 5baeda6d9f..aa1aacc485 100644 --- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs @@ -25,6 +25,7 @@ namespace Volo.Abp.EventBus.RabbitMq { protected AbpRabbitMqEventBusOptions AbpRabbitMqEventBusOptions { get; } protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; } + protected AbpEventBusOptions AbpEventBusOptions { get; } protected IConnectionPool ConnectionPool { get; } protected IRabbitMqSerializer Serializer { get; } @@ -42,12 +43,14 @@ namespace Volo.Abp.EventBus.RabbitMq IOptions distributedEventBusOptions, IRabbitMqMessageConsumerFactory messageConsumerFactory, ICurrentTenant currentTenant, - IEventErrorHandler errorHandler) + IEventErrorHandler errorHandler, + IOptions abpEventBusOptions) : base(serviceScopeFactory, currentTenant, errorHandler) { ConnectionPool = connectionPool; Serializer = serializer; MessageConsumerFactory = messageConsumerFactory; + AbpEventBusOptions = abpEventBusOptions.Value; AbpDistributedEventBusOptions = distributedEventBusOptions.Value; AbpRabbitMqEventBusOptions = options.Value; @@ -57,6 +60,8 @@ namespace Volo.Abp.EventBus.RabbitMq public void Initialize() { + const string suffix = "_dead_letter"; + Consumer = MessageConsumerFactory.Create( new ExchangeDeclareConfiguration( AbpRabbitMqEventBusOptions.ExchangeName, @@ -67,7 +72,12 @@ namespace Volo.Abp.EventBus.RabbitMq AbpRabbitMqEventBusOptions.ClientName, durable: true, exclusive: false, - autoDelete: false + autoDelete: false, + arguments: new Dictionary + { + {"x-dead-letter-exchange", AbpRabbitMqEventBusOptions.ExchangeName + suffix}, + {"x-dead-letter-routing-key", AbpEventBusOptions.DeadLetterName ?? AbpRabbitMqEventBusOptions.ClientName + suffix} + } ), AbpRabbitMqEventBusOptions.ConnectionName ); @@ -197,6 +207,35 @@ namespace Volo.Abp.EventBus.RabbitMq return Task.CompletedTask; } + public Task PublishAsync(Type eventType, object eventData, Dictionary headers) + { + var eventName = EventNameAttribute.GetNameOrDefault(eventType); + var body = Serializer.Serialize(eventData); + + using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel()) + { + channel.ExchangeDeclare( + AbpRabbitMqEventBusOptions.ExchangeName, + "direct", + durable: true + ); + + var properties = channel.CreateBasicProperties(); + properties.DeliveryMode = RabbitMqConsts.DeliveryModes.Persistent; + properties.Headers = headers; + + channel.BasicPublish( + exchange: AbpRabbitMqEventBusOptions.ExchangeName, + routingKey: eventName, + mandatory: true, + basicProperties: properties, + body: body + ); + } + + return Task.CompletedTask; + } + private List GetOrCreateHandlerFactories(Type eventType) { return HandlerFactories.GetOrAdd( diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqEventErrorHandler.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqEventErrorHandler.cs new file mode 100644 index 0000000000..a17bcfcdcc --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqEventErrorHandler.cs @@ -0,0 +1,82 @@ +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.RabbitMq +{ + public class RabbitMqEventErrorHandler : EventErrorHandlerBase, ISingletonDependency + { + public const string HeadersKey = "headers"; + public const string RetryIndexKey = "retryIndex"; + + protected RabbitMqDistributedEventBus EventBus { get; } + + public RabbitMqEventErrorHandler( + IOptions options, + RabbitMqDistributedEventBus eventBus) + : base(options) + { + EventBus = eventBus; + } + + protected override async Task Retry(EventExecutionErrorContext context) + { + if (Options.RetryStrategyOptions.IntervalMillisecond > 0) + { + await Task.Delay(Options.RetryStrategyOptions.IntervalMillisecond); + } + + var headers = context.GetProperty>(HeadersKey) ?? + new Dictionary(); + + var index = 1; + if (headers.ContainsKey(RetryIndexKey)) + { + index = (int) headers[RetryIndexKey]; + headers[RetryIndexKey] = ++index; + } + else + { + headers[RetryIndexKey] = index; + } + + headers["exceptions"] = context.Exceptions; + + await EventBus.PublishAsync(context.EventType, context.EventData, headers); + } + + protected override Task MoveToDeadLetter(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 headers = context.GetProperty>(HeadersKey); + + if (headers == null || !headers.ContainsKey(RetryIndexKey)) + { + return true; + } + + var index = (int) headers[RetryIndexKey]; + + return Options.RetryStrategyOptions.MaxRetryAttempts < index; + } + } +} diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpEventBusRebusModule.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpEventBusRebusModule.cs index 9ebd91a6a5..83b3f84340 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpEventBusRebusModule.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpEventBusRebusModule.cs @@ -1,5 +1,6 @@ using Microsoft.Extensions.DependencyInjection; using Rebus.Handlers; +using Rebus.Retry.Simple; using Rebus.ServiceProvider; using Volo.Abp.Modularity; @@ -11,23 +12,28 @@ namespace Volo.Abp.EventBus.Rebus { public override void ConfigureServices(ServiceConfigurationContext context) { - var options = context.Services.ExecutePreConfiguredActions(); + var abpEventBusOptions = context.Services.ExecutePreConfiguredActions(); context.Services.AddTransient(typeof(IHandleMessages<>), typeof(RebusDistributedEventHandlerAdapter<>)); Configure(rebusOptions => { - rebusOptions.Configurer = options.Configurer; - rebusOptions.Publish = options.Publish; - rebusOptions.InputQueueName = options.InputQueueName; - }); + context.Services.ExecutePreConfiguredActions(rebusOptions); - context.Services.AddRebus(configurer => - { - options.Configurer?.Invoke(configurer); - return configurer; - }); + context.Services.AddRebus(configure => + { + if (abpEventBusOptions.RetryStrategyOptions != null) + { + configure.Options(b => + b.SimpleRetryStrategy( + errorQueueAddress: abpEventBusOptions.DeadLetterName ?? rebusOptions.InputQueueName + "_dead_letter", + maxDeliveryAttempts: abpEventBusOptions.RetryStrategyOptions.MaxRetryAttempts)); + } + rebusOptions.Configurer?.Invoke(configure); + return configure; + }); + }); } public override void OnApplicationInitialization(ApplicationInitializationContext context) diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpEventBusRebusOptionsSetup.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpEventBusRebusOptionsSetup.cs new file mode 100644 index 0000000000..7d26f174fe --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpEventBusRebusOptionsSetup.cs @@ -0,0 +1,12 @@ +using Microsoft.Extensions.Options; + +namespace Volo.Abp.EventBus.Rebus +{ + public class AbpEventBusRebusOptionsSetup : IConfigureOptions + { + public void Configure(AbpEventBusOptions options) + { + throw new System.NotImplementedException(); + } + } +} diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs index 8aaee7bd63..b6343204c0 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs @@ -32,7 +32,7 @@ namespace Volo.Abp.EventBus.Rebus public AbpRebusEventBusOptions() { _publish = DefaultPublish; - _configurer = DefaultConfigurer; + _configurer = DefaultConfigure; } private async Task DefaultPublish(IBus bus, Type eventType, object eventData) @@ -40,10 +40,10 @@ namespace Volo.Abp.EventBus.Rebus await bus.Advanced.Routing.Send(InputQueueName, eventData); } - private void DefaultConfigurer(RebusConfigurer configurer) + private void DefaultConfigure(RebusConfigurer configure) { - configurer.Subscriptions(s => s.StoreInMemory()); - configurer.Transport(t => t.UseInMemoryTransport(new InMemNetwork(), InputQueueName)); + configure.Subscriptions(s => s.StoreInMemory()); + configure.Transport(t => t.UseInMemoryTransport(new InMemNetwork(), InputQueueName)); } } } diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusEventErrorHandler.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusEventErrorHandler.cs new file mode 100644 index 0000000000..c75715ebf8 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusEventErrorHandler.cs @@ -0,0 +1,44 @@ +using System; +using System.Threading.Tasks; +using Microsoft.Extensions.Options; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.EventBus.Rebus +{ + public class RebusEventErrorHandler : EventErrorHandlerBase, ISingletonDependency + { + public RebusEventErrorHandler( + IOptions options) + : base(options) + { + } + + protected override Task Retry(EventExecutionErrorContext context) + { + Throw(context); + + return Task.CompletedTask; + } + + protected override Task MoveToDeadLetter(EventExecutionErrorContext context) + { + Throw(context); + + return Task.CompletedTask; + } + + private void Throw(EventExecutionErrorContext context) + { + // Rebus will automatic retries and error handling: https://github.com/rebus-org/Rebus/wiki/Automatic-retries-and-error-handling + + 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); + } + } +} diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusModule.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusModule.cs index eb9c17b129..014308c8e6 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusModule.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusModule.cs @@ -20,6 +20,14 @@ namespace Volo.Abp.EventBus AddEventHandlers(context.Services); } + public override void ConfigureServices(ServiceConfigurationContext context) + { + Configure(options => + { + context.Services.ExecutePreConfiguredActions(options); + }); + } + private static void AddEventHandlers(IServiceCollection services) { var localHandlers = new List(); diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusOptions.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusOptions.cs index ad3926b334..3ba5c2a548 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusOptions.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusOptions.cs @@ -8,7 +8,7 @@ namespace Volo.Abp.EventBus public Func ErrorHandleSelector { get; set; } - public string DeadLetterQueue { get; set; } + public string DeadLetterName { get; set; } public AbpEventBusRetryStrategyOptions RetryStrategyOptions { get; set; } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusRetryStrategyOptions.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusRetryStrategyOptions.cs index 5762e8d7a0..4b5b722e96 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusRetryStrategyOptions.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusRetryStrategyOptions.cs @@ -2,9 +2,8 @@ { public class AbpEventBusRetryStrategyOptions { - public int IntervalMillisecond { get; set; } = 3000; - public int Count { get; set; } = 3; + public int MaxRetryAttempts { get; set; } = 3; } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventErrorHandler.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventErrorHandler.cs index 5c3c2ba4f1..da7a0d7f71 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventErrorHandler.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventErrorHandler.cs @@ -57,7 +57,7 @@ namespace Volo.Abp.EventBus.Local var index = RetryTracking.GetOrDefault(messageId); - if (Options.RetryStrategyOptions.Count >= index) + if (Options.RetryStrategyOptions.MaxRetryAttempts >= index) { RetryTracking.Remove(messageId); return false; diff --git a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/AbpKafkaOptions.cs b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/AbpKafkaOptions.cs index 26d15ce818..f6679aec3c 100644 --- a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/AbpKafkaOptions.cs +++ b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/AbpKafkaOptions.cs @@ -14,8 +14,6 @@ namespace Volo.Abp.Kafka public Action ConfigureTopic { get; set; } - public bool ReQueue { get; set; } = true; - public AbpKafkaOptions() { Connections = new KafkaConnections(); diff --git a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs index e519ebc87d..279765efdb 100644 --- a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs +++ b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs @@ -27,6 +27,8 @@ namespace Volo.Abp.Kafka protected AbpKafkaOptions Options { get; } + protected AbpAsyncTimer Timer { get; } + protected ConcurrentBag, Task>> Callbacks { get; } protected IConsumer Consumer { get; private set; } @@ -37,22 +39,27 @@ namespace Volo.Abp.Kafka protected string TopicName { get; private set; } - protected string DeadLetterTopicName { get; private set; } public KafkaMessageConsumer( IConsumerPool consumerPool, IExceptionNotifier exceptionNotifier, IOptions options, - IProducerPool producerPool) + IProducerPool producerPool, + AbpAsyncTimer timer) { ConsumerPool = consumerPool; ExceptionNotifier = exceptionNotifier; ProducerPool = producerPool; + Timer = timer; Options = options.Value; Logger = NullLogger.Instance; Callbacks = new ConcurrentBag, Task>>(); + + Timer.Period = 5000; //5 sec. + Timer.Elapsed = Timer_Elapsed; + Timer.RunOnStart = true; } public virtual void Initialize( @@ -68,9 +75,7 @@ namespace Volo.Abp.Kafka DeadLetterTopicName = deadLetterTopicName; ConnectionName = connectionName ?? KafkaConnections.DefaultConnectionName; GroupId = groupId; - - AsyncHelper.RunSync(CreateTopicAsync); - Consume(); + Timer.Start(); } public virtual void OnMessageReceived(Func, Task> callback) @@ -78,6 +83,14 @@ namespace Volo.Abp.Kafka Callbacks.Add(callback); } + protected virtual async Task Timer_Elapsed(AbpAsyncTimer timer) + { + await CreateTopicAsync(); + Consume(); + + Timer.Stop(); + } + protected virtual async Task CreateTopicAsync() { using (var adminClient = new AdminClientBuilder(Options.Connections.GetOrDefault(ConnectionName)).Build()) @@ -158,8 +171,6 @@ namespace Volo.Abp.Kafka } catch (Exception ex) { - await RequeueAsync(consumeResult); - Logger.LogException(ex); await ExceptionNotifier.NotifyAsync(ex); } @@ -169,17 +180,6 @@ namespace Volo.Abp.Kafka } } - protected virtual async Task RequeueAsync(ConsumeResult consumeResult) - { - if (!Options.ReQueue) - { - return; - } - - var producer = ProducerPool.Get(ConnectionName); - await producer.ProduceAsync(consumeResult.Topic, consumeResult.Message); - } - public virtual void Dispose() { if (Consumer == null) diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs index 8cc07b7bb9..17a5c2dfbf 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs @@ -6,8 +6,7 @@ namespace Volo.Abp.RabbitMQ { public class QueueDeclareConfiguration { - [NotNull] - public string QueueName { get; } + [NotNull] public string QueueName { get; } public bool Durable { get; set; } @@ -18,16 +17,17 @@ namespace Volo.Abp.RabbitMQ public IDictionary Arguments { get; } public QueueDeclareConfiguration( - [NotNull] string queueName, - bool durable = true, - bool exclusive = false, - bool autoDelete = false) + [NotNull] string queueName, + bool durable = true, + bool exclusive = false, + bool autoDelete = false, + Dictionary arguments = null) { QueueName = queueName; Durable = durable; Exclusive = exclusive; AutoDelete = autoDelete; - Arguments = new Dictionary(); + Arguments = arguments ?? new Dictionary(); } public virtual QueueDeclareOk Declare(IModel channel) @@ -41,4 +41,4 @@ namespace Volo.Abp.RabbitMQ ); } } -} \ No newline at end of file +} diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs index 894f68d097..b5a165157f 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs @@ -147,6 +147,7 @@ namespace Volo.Abp.RabbitMQ Channel = ConnectionPool .Get(ConnectionName) .CreateModel(); + Channel.ExchangeDeclare( exchange: Exchange.ExchangeName, type: Exchange.Type, @@ -155,6 +156,23 @@ namespace Volo.Abp.RabbitMQ arguments: Exchange.Arguments ); + if (Queue.Arguments.ContainsKey("x-dead-letter-exchange") && + Queue.Arguments.ContainsKey("x-dead-letter-routing-key")) + { + Channel.ExchangeDeclare( + Exchange.Arguments["x-dead-letter-exchange"].ToString(), + Exchange.Type, + Exchange.Durable, + Exchange.AutoDelete + ); + + Channel.QueueDeclare( + Queue.Arguments["x-dead-letter-routing-key"].ToString(), + Queue.Durable, + Queue.Exclusive, + Queue.AutoDelete); + } + Channel.QueueDeclare( queue: Queue.QueueName, durable: Queue.Durable, @@ -194,14 +212,10 @@ namespace Volo.Abp.RabbitMQ { try { - Channel.BasicNack( - basicDeliverEventArgs.DeliveryTag, - multiple: false, - requeue: true - ); + Channel.BasicReject(basicDeliverEventArgs.DeliveryTag, false); } catch { } - + Logger.LogException(ex); await ExceptionNotifier.NotifyAsync(ex); }