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; /// /// Handles errors that occur during RabbitMQ message consumption and publishes them using the configured publisher service. /// public class RabbitConsumerErrorHandler(IPublisherService publisherService) { private const int MaxRetries = 2; private static readonly ILogger Logger = Log.ForContext(); /// /// Processes a received message by invoking the next handler in the pipeline, and on failure logs the error, /// serializes the message body to a string, and delegates to the retry handler using the message properties, /// received info, and the captured exception. /// /// The deserialized message whose body is inspected or re-serialized for retry handling. /// Information about the message receipt, passed to the retry handler. /// The asynchronous delegate representing the next step in the processing pipeline. 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"); } } /// /// Retrieves the retry count from the message properties headers. Returns the deserialized integer value when a "retries" header containing a byte array is present, or 0 when the headers are missing, the key is not found, or the value is not a byte array. /// /// The message properties whose headers are inspected for the "retries" entry. /// The number of retries parsed from the "retries" header, or 0 if the header is absent or not a byte array. 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; } /// /// Constructs a new error message that preserves the original message properties and attaches exception details for downstream processing or dead-letter handling. /// /// The metadata of the original received message, including exchange, routing key, and queue information used to identify the error source. /// The properties of the original message from which delivery mode, content type, correlation id, and message id are copied. /// The raw body of the original message that caused the error. /// The exception that was raised, whose message is included in the error payload. /// The headers to associate with the resulting error message. /// A containing the constructed error and the propagated message properties. 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); } }