using adas_core.Application.Services.Interfaces; using EasyNetQ; using EasyNetQ.SystemMessages; using Newtonsoft.Json; using Serilog; using System.Text; using ILogger = Serilog.ILogger; namespace adas_core.Infrastructure.Utils; public class RabbitConsumerErrorHandler(IPublisherService publisherService) { private const int MaxRetries = 2; private static readonly ILogger Logger = Log.ForContext(); public async Task HandleAsync( Message message, MessageReceivedInfo receivedInfo, Func next) { try { await next(); } catch (Exception exception) { Logger.Error(exception, "Consumer error {Message}", exception.Message); var properties = message.Properties; var body = Encoding.UTF8.GetString( message.Body is byte[] bytes ? bytes : Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message.Body))); HandleRetries(receivedInfo, properties, body, exception); } } private void HandleRetries( MessageReceivedInfo receivedInfo, MessageProperties properties, string body, Exception exception) { try { var headers = properties.Headers != null ? new Dictionary(properties.Headers) : []; var retries = GetRetries(properties); // MAX RETRIES → ERROR QUEUE if (retries > MaxRetries) { if (!receivedInfo.Queue.StartsWith("Error")) { headers["retries"] = BitConverter.GetBytes(MaxRetries + 1); var errorMsg = CreateErrorMessage( receivedInfo, properties, body, exception, headers ); _ = publisherService.SendMessageError( errorMsg, $"Error{receivedInfo.Queue}"); } return; } // RETRY NORMAL headers["retries"] = BitConverter.GetBytes(retries + 1); var newProps = new MessageProperties { DeliveryMode = 2, Headers = headers, ContentType = properties.ContentType, CorrelationId = properties.CorrelationId, MessageId = properties.MessageId }; var message = new Message(body, newProps); var result = publisherService .SendMessage(message, receivedInfo.Queue) .GetAwaiter() .GetResult(); if (!result) throw new Exception("Requeue failed"); } catch (Exception e) { Logger.Error(e, "Exception processing rabbit message"); } } private int GetRetries(MessageProperties properties) { if (properties.Headers != null && properties.Headers.TryGetValue("retries", out var retriesObj) && retriesObj is byte[] bytes) { return BitConverter.ToInt32(bytes, 0); } return 0; } private static Message CreateErrorMessage( MessageReceivedInfo receivedInfo, MessageProperties originalProperties, string body, Exception exception, Dictionary headers) { var props = new MessageProperties { Headers = headers, DeliveryMode = originalProperties.DeliveryMode, ContentType = originalProperties.ContentType, CorrelationId = originalProperties.CorrelationId, MessageId = originalProperties.MessageId }; var error = new Error( body, exception.Message, receivedInfo.Exchange, receivedInfo.RoutingKey, receivedInfo.Queue, DateTime.UtcNow, props ); return new Message(error, props); } }