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);
}
}