Aggiunta implementazione RabbitMQ

Author: AngeloDotNet

src/NET6CustomLibrary/RabbitMQ/MessageManager.cs

@@ -0,0 +1,84 @@
+namespace NET6CustomLibrary.RabbitMQ;
+
+internal class MessageManager : IMessageSender, IDisposable
+{
+    private const string MaxPriorityHeader = "x-max-priority";
+
+    internal IConnection Connection { get; private set; }
+    internal IModel Channel { get; private set; }
+
+    private readonly MessageManagerSettings messageManagerSettings;
+    private readonly QueueSettings queueSettings;
+
+    public MessageManager(MessageManagerSettings messageManagerSettings, QueueSettings queueSettings)
+    {
+        var factory = new ConnectionFactory { Uri = new Uri(messageManagerSettings.ConnectionString) };
+        Connection = factory.CreateConnection();
+
+        Channel = Connection.CreateModel();
+
+        if (messageManagerSettings.QueuePrefetchCount > 0)
+        {
+            Channel.BasicQos(0, messageManagerSettings.QueuePrefetchCount, false);
+        }
+
+        Channel.ExchangeDeclare(messageManagerSettings.ExchangeName, ExchangeType.Direct, durable: true);
+
+        foreach (var queue in queueSettings.Queues)
+        {
+            var args = new Dictionary<string, object>
+            {
+                [MaxPriorityHeader] = 10
+            };
+
+            Channel.QueueDeclare(queue.Name, durable: true, exclusive: false, autoDelete: false, args);
+            Channel.QueueBind(queue.Name, messageManagerSettings.ExchangeName, queue.Name, null);
+        }
+
+        this.messageManagerSettings = messageManagerSettings;
+        this.queueSettings = queueSettings;
+    }
+
+    public Task PublishAsync<T>(T message, int priority = 1) where T : class
+    {
+        var sendBytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize<object>(message, messageManagerSettings.JsonSerializerOptions ?? JsonOptions.Default));
+
+        var routingKey = queueSettings.Queues.First(q => q.Type == typeof(T)).Name;
+        return PublishAsync(sendBytes.AsMemory(), routingKey, priority);
+    }
+
+    private Task PublishAsync(ReadOnlyMemory<byte> body, string routingKey, int priority = 1)
+    {
+        var properties = Channel.CreateBasicProperties();
+        properties.Persistent = true;
+        properties.Priority = Convert.ToByte(priority);
+
+        Channel.BasicPublish(messageManagerSettings.ExchangeName, routingKey, properties, body);
+        return Task.CompletedTask;
+    }
+
+    public void MarkAsComplete(BasicDeliverEventArgs message) => Channel.BasicAck(message.DeliveryTag, false);
+
+    public void MarkAsRejected(BasicDeliverEventArgs message) => Channel.BasicReject(message.DeliveryTag, false);
+
+    public void Dispose()
+    {
+        try
+        {
+            if (Channel.IsOpen)
+            {
+                Channel.Close();
+            }
+
+            if (Connection.IsOpen)
+            {
+                Connection.Close();
+            }
+        }
+        catch
+        {
+        }
+
+        GC.SuppressFinalize(this);
+    }
+}

src/NET6CustomLibrary/RabbitMQ/MessageManagerSettings.cs

@@ -0,0 +1,9 @@
+namespace NET6CustomLibrary.RabbitMQ;
+
+public class MessageManagerSettings
+{
+    public string ConnectionString { get; set; }
+    public string ExchangeName { get; set; }
+    public ushort QueuePrefetchCount { get; set; }
+    public JsonSerializerOptions JsonSerializerOptions { get; set; } = JsonOptions.Default;
+}

src/NET6CustomLibrary/RabbitMQ/QueueListener.cs

@@ -0,0 +1,77 @@
+namespace NET6CustomLibrary.RabbitMQ;
+
+internal class QueueListener<T> : BackgroundService where T : class
+{
+    private readonly MessageManager messageManager;
+    private readonly MessageManagerSettings messageManagerSettings;
+    private readonly ILogger logger;
+    private readonly IServiceProvider serviceProvider;
+    private readonly string queueName;
+
+    public QueueListener(MessageManager messageManager, MessageManagerSettings messageManagerSettings, QueueSettings settings, ILogger<QueueListener<T>> logger, IServiceProvider serviceProvider)
+    {
+        this.messageManager = messageManager;
+        this.messageManagerSettings = messageManagerSettings;
+        this.logger = logger;
+        this.serviceProvider = serviceProvider;
+
+        queueName = settings.Queues.First(q => q.Type == typeof(T)).Name;
+    }
+
+    public override Task StartAsync(CancellationToken cancellationToken)
+    {
+        logger.LogDebug("RabbitMQ Listener for {QueueName} started", queueName);
+
+        return base.StartAsync(cancellationToken);
+    }
+
+    public override Task StopAsync(CancellationToken cancellationToken)
+    {
+        logger.LogDebug("RabbitMQ Listener for {QueueName} stopped", queueName);
+
+        return base.StopAsync(cancellationToken);
+    }
+
+    protected override Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        stoppingToken.ThrowIfCancellationRequested();
+
+        var consumer = new EventingBasicConsumer(messageManager.Channel);
+        consumer.Received += async (_, message) =>
+        {
+            try
+            {
+                logger.LogDebug("Messaged received: {Request}", Encoding.UTF8.GetString(message.Body.Span));
+
+                using var scope = serviceProvider.CreateScope();
+
+                var receiver = scope.ServiceProvider.GetRequiredService<IMessageReceiver<T>>();
+                var response = JsonSerializer.Deserialize<T>(message.Body.Span, messageManagerSettings.JsonSerializerOptions ?? JsonOptions.Default);
+                await receiver.ReceiveAsync(response, stoppingToken);
+
+                messageManager.MarkAsComplete(message);
+
+                logger.LogDebug("Message processed");
+            }
+            catch (Exception ex)
+            {
+                messageManager.MarkAsRejected(message);
+                logger.LogError(ex, "Unexpected error while processing message");
+            }
+
+            stoppingToken.ThrowIfCancellationRequested();
+        };
+
+        messageManager.Channel.BasicConsume(queueName, autoAck: false, consumer);
+
+        return Task.CompletedTask;
+    }
+
+    public override void Dispose()
+    {
+        messageManager.Dispose();
+        base.Dispose();
+
+        GC.SuppressFinalize(this);
+    }
+}

src/NET6CustomLibrary/RabbitMQ/RabbitMqExtensions.cs

@@ -0,0 +1,29 @@
+namespace NET6CustomLibrary.RabbitMQ;
+
+public static class RabbitMQExtensions
+{
+    public static IMessagingBuilder AddRabbitMq(this IServiceCollection services, Action<MessageManagerSettings> messageManagerConfiguration, Action<QueueSettings> queuesConfiguration)
+    {
+        services.AddSingleton<MessageManager>();
+        services.AddSingleton<IMessageSender>(provider => provider.GetService<MessageManager>());
+
+        var messageManagerSettings = new MessageManagerSettings();
+        messageManagerConfiguration.Invoke(messageManagerSettings);
+        services.AddSingleton(messageManagerSettings);
+
+        var queueSettings = new QueueSettings();
+        queuesConfiguration.Invoke(queueSettings);
+        services.AddSingleton(queueSettings);
+
+        return new DefaultMessagingBuilder(services);
+    }
+
+    public static IMessagingBuilder AddReceiver<TObject, TReceiver>(this IMessagingBuilder builder) where TObject : class
+        where TReceiver : class, IMessageReceiver<TObject>
+    {
+        builder.Services.AddHostedService<QueueListener<TObject>>();
+        builder.Services.AddTransient<IMessageReceiver<TObject>, TReceiver>();
+
+        return builder;
+    }
+}