RabbitMQ Consumer

src/Directory.Packages.props

@@ -78,5 +78,7 @@
     <PackageVersion Include="WorkflowCore.Persistence.SqlServer" Version="3.9.0" />
     <PackageVersion Include="xunit" Version="2.5.1" />
     <PackageVersion Include="xunit.runner.visualstudio" Version="2.5.1" />
+    <PackageVersion Include="RabbitMQ.Client" Version="6.6.0" />
+    <PackageVersion Include="Polly" Version="8.0.0" />
   </ItemGroup>
 </Project>

src/Framework.RabbitMq.Consumer/BackgroundServices/RabbitMqBackgroundService.cs

@@ -0,0 +1,125 @@
+using System.Text;
+
+using Framework.RabbitMq.Consumer.Enums;
+using Framework.RabbitMq.Consumer.Interfaces;
+using Framework.RabbitMq.Consumer.Settings;
+using Framework.RabbitMq.Interfaces;
+using Framework.RabbitMq.Settings;
+
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+using RabbitMQ.Client;
+
+namespace Framework.RabbitMq.Consumer.BackgroundServices;
+
+public class RabbitMqBackgroundService : BackgroundService
+{
+    private readonly IRabbitMqClient _client;
+
+    private readonly IRabbitMqConsumerInitializer _consumerInitializer;
+
+    private readonly RabbitMqConsumerSettings _consumerSettings;
+
+    private readonly ILogger<RabbitMqBackgroundService> _logger;
+
+    private readonly IRabbitMqMessageProcessor _messageProcessor;
+
+    private readonly RabbitMqServerSettings _serverSettings;
+
+    private IModel? _channel;
+
+    private IConnection? _connection;
+
+    public RabbitMqBackgroundService(
+        IRabbitMqClient client,
+        IRabbitMqConsumerInitializer consumerInitializer,
+        IRabbitMqMessageProcessor messageProcessor,
+        IOptions<RabbitMqServerSettings> serverOptions,
+        IOptions<RabbitMqConsumerSettings> consumerOptions,
+        ILogger<RabbitMqBackgroundService> logger)
+    {
+        this._client = client;
+        this._consumerInitializer = consumerInitializer;
+        this._messageProcessor = messageProcessor;
+        this._logger = logger;
+        this._serverSettings = serverOptions.Value;
+        this._consumerSettings = consumerOptions.Value;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        this._connection = await this._client.TryConnectAsync(this._consumerSettings.ConnectionAttemptCount, stoppingToken);
+        if (this._connection == null)
+        {
+            this._logger.LogInformation("Listening RabbitMQ events wasn't started");
+            return;
+        }
+
+        this._channel = this._connection!.CreateModel();
+        this._consumerInitializer.Initialize(this._channel);
+
+        await this.Listen(stoppingToken);
+    }
+
+    private async Task Listen(CancellationToken token)
+    {
+        this._logger.LogInformation(
+            "Listening RabbitMQ events has started on {Address}. Queue name is {Queue}",
+            this._serverSettings.Address,
+            this._consumerSettings.Queue);
+
+        while (!token.IsCancellationRequested)
+        {
+            var result = this._channel!.BasicGet(this._consumerSettings.Queue, false);
+            if (result is null)
+            {
+                await Delay(this._consumerSettings.ReceiveMessageDelayMilliseconds, token);
+                continue;
+            }
+
+            await this.ProcessAsync(result, token);
+        }
+    }
+
+    public override void Dispose()
+    {
+        this._channel?.Close();
+        this._connection?.Close();
+        base.Dispose();
+        GC.SuppressFinalize(this);
+    }
+
+    private async Task ProcessAsync(BasicGetResult result, CancellationToken token)
+    {
+        try
+        {
+            var message = Encoding.UTF8.GetString(result.Body.Span);
+            if (result.Redelivered && result.DeliveryTag > this._consumerSettings.FailedMessageRetryCount)
+            {
+                var behaviour = await this._messageProcessor.ProcessDeadLetterAsync(
+                                    result.BasicProperties,
+                                    result.RoutingKey,
+                                    message,
+                                    token);
+                if (behaviour == DeadLetterBehaviour.Skip)
+                {
+                    this._channel!.BasicAck(result.DeliveryTag, false);
+                    return;
+                }
+            }
+
+            await this._messageProcessor.ProcessAsync(result.BasicProperties, result.RoutingKey, message, token);
+            this._channel!.BasicAck(result.DeliveryTag, false);
+        }
+        catch (Exception ex)
+        {
+            this._logger.LogError(ex, "There was some problem with processing message. Routing key:'{RoutingKey}'", result.RoutingKey);
+            this._channel!.BasicNack(result.DeliveryTag, false, true);
+            await Delay(this._consumerSettings.RejectMessageDelayMilliseconds, token);
+        }
+    }
+
+    private static Task Delay(int value, CancellationToken token) => Task.Delay(TimeSpan.FromMilliseconds(value), token);
+}

src/Framework.RabbitMq.Consumer/DependencyInjection.cs

@@ -0,0 +1,20 @@
+using Framework.RabbitMq.Consumer.BackgroundServices;
+using Framework.RabbitMq.Consumer.Interfaces;
+using Framework.RabbitMq.Consumer.Services;
+using Framework.RabbitMq.Consumer.Settings;
+
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Framework.RabbitMq.Consumer;
+
+public static class DependencyInjection
+{
+    public static IServiceCollection AddRabbitMqConsumer<TMessageProcessor>(this IServiceCollection services, IConfiguration configuration)
+        where TMessageProcessor : class, IRabbitMqMessageProcessor =>
+        services
+            .Configure<RabbitMqConsumerSettings>(configuration.GetSection("RabbitMQ:Consumer"))
+            .AddSingleton<IRabbitMqMessageProcessor, TMessageProcessor>()
+            .AddSingleton<IRabbitMqConsumerInitializer, RabbitMqConsumerInitializer>()
+            .AddHostedService<RabbitMqBackgroundService>();
+}

src/Framework.RabbitMq.Consumer/Enums/DeadLetterBehaviour.cs

@@ -0,0 +1,8 @@
+namespace Framework.RabbitMq.Consumer.Enums;
+
+public enum DeadLetterBehaviour
+{
+    ForeverRetry = 1,
+
+    Skip = 2
+}

src/Framework.RabbitMq.Consumer/Framework.RabbitMq.Consumer.csproj

@@ -0,0 +1,9 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <PackageId>Luxoft.Framework.RabbitMq.Consumer</PackageId>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\Framework.RabbitMq\Framework.RabbitMq.csproj" />
+  </ItemGroup>
+</Project>

src/Framework.RabbitMq.Consumer/Interfaces/IRabbitMqConsumerInitializer.cs

@@ -0,0 +1,8 @@
+using RabbitMQ.Client;
+
+namespace Framework.RabbitMq.Consumer.Interfaces;
+
+public interface IRabbitMqConsumerInitializer
+{
+    void Initialize(IModel model);
+}

src/Framework.RabbitMq.Consumer/Interfaces/IRabbitMqMessageProcessor.cs

@@ -0,0 +1,17 @@
+using Framework.RabbitMq.Consumer.Enums;
+
+using RabbitMQ.Client;
+
+namespace Framework.RabbitMq.Consumer.Interfaces;
+
+public interface IRabbitMqMessageProcessor
+{
+    Task ProcessAsync(IBasicProperties properties, string routingKey, string message, CancellationToken token);
+
+    Task<DeadLetterBehaviour> ProcessDeadLetterAsync(
+        IBasicProperties properties,
+        string routingKey,
+        string message,
+        CancellationToken token) =>
+        Task.FromResult(DeadLetterBehaviour.ForeverRetry);
+}

src/Framework.RabbitMq.Consumer/Services/RabbitMqConsumerInitializer.cs

@@ -0,0 +1,22 @@
+using Framework.RabbitMq.Consumer.Interfaces;
+using Framework.RabbitMq.Consumer.Settings;
+
+using Microsoft.Extensions.Options;
+
+using RabbitMQ.Client;
+
+namespace Framework.RabbitMq.Consumer.Services;
+
+public record RabbitMqConsumerInitializer(IOptions<RabbitMqConsumerSettings> Options) : IRabbitMqConsumerInitializer
+{
+    public void Initialize(IModel model)
+    {
+        var consumerSettings = this.Options.Value;
+
+        model.ExchangeDeclare(consumerSettings.Exchange, ExchangeType.Topic, true);
+        model.QueueDeclare(consumerSettings.Queue, true, false, false, null);
+
+        foreach (var routingKey in consumerSettings.RoutingKeys)
+            model.QueueBind(consumerSettings.Queue, consumerSettings.Exchange, routingKey);
+    }
+}

src/Framework.RabbitMq.Consumer/Settings/RabbitMqConsumerSettings.cs

@@ -0,0 +1,18 @@
+namespace Framework.RabbitMq.Consumer.Settings;
+
+public class RabbitMqConsumerSettings
+{
+    public int ReceiveMessageDelayMilliseconds { get; set; } = 1000;
+
+    public int RejectMessageDelayMilliseconds { get; set; } = 3000;
+
+    public ulong FailedMessageRetryCount { get; set; } = 3;
+
+    public int ConnectionAttemptCount { get; set; } = 3;
+
+    public string Exchange { get; set; } = default!;
+
+    public string Queue { get; set; } = default!;
+
+    public string[] RoutingKeys { get; set; } = Array.Empty<string>();
+}

src/Framework.RabbitMq/DependencyInjection.cs

@@ -0,0 +1,16 @@
+using Framework.RabbitMq.Interfaces;
+using Framework.RabbitMq.Services;
+using Framework.RabbitMq.Settings;
+
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Framework.RabbitMq;
+
+public static class DependencyInjection
+{
+    public static IServiceCollection AddRabbitMqClient(this IServiceCollection services, IConfiguration configuration) =>
+        services
+            .Configure<RabbitMqServerSettings>(configuration.GetSection("RabbitMQ:Server"))
+            .AddSingleton<IRabbitMqClient, RabbitMqClient>();
+}

src/Framework.RabbitMq/Framework.RabbitMq.csproj

@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <PackageId>Luxoft.Framework.RabbitMq</PackageId>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" />
+    <PackageReference Include="Polly" />
+    <PackageReference Include="RabbitMQ.Client" />
+  </ItemGroup>
+
+</Project>

src/Framework.RabbitMq/Interfaces/IRabbitMqClient.cs

@@ -0,0 +1,8 @@
+using RabbitMQ.Client;
+
+namespace Framework.RabbitMq.Interfaces;
+
+public interface IRabbitMqClient
+{
+    Task<IConnection?> TryConnectAsync(int attempts, CancellationToken token = default);
+}

src/Framework.RabbitMq/Services/RabbitMqClient.cs

@@ -0,0 +1,51 @@
+using Framework.RabbitMq.Interfaces;
+using Framework.RabbitMq.Settings;
+
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+using Polly;
+using Polly.Retry;
+
+using RabbitMQ.Client;
+
+namespace Framework.RabbitMq.Services;
+
+public record RabbitMqClient(IOptions<RabbitMqServerSettings> Options, ILogger<RabbitMqClient> Logger) : IRabbitMqClient
+{
+    private const int RetryConnectDelay = 5000;
+
+    public async Task<IConnection?> TryConnectAsync(int attempts, CancellationToken token = default)
+    {
+        var serverSettings = this.Options.Value;
+        var factory = new ConnectionFactory
+                      {
+                          HostName = serverSettings.Host,
+                          Port = serverSettings.Port,
+                          UserName = serverSettings.UserName,
+                          Password = serverSettings.Secret,
+                          VirtualHost = serverSettings.VirtualHost,
+                          AutomaticRecoveryEnabled = true
+                      };
+
+        var policy = this.CreateRetryPolicy(attempts);
+        try
+        {
+            return await policy.ExecuteAsync(_ => Task.FromResult(factory.CreateConnection()), token);
+        }
+        catch (Exception ex)
+        {
+            this.LogConnectionError(ex);
+            return null;
+        }
+    }
+
+    private AsyncRetryPolicy CreateRetryPolicy(int attempts) =>
+        Policy.Handle<Exception>()
+              .WaitAndRetryAsync(
+                  attempts,
+                  _ => TimeSpan.FromMilliseconds(RetryConnectDelay),
+                  (ex, _) => this.LogConnectionError(ex));
+
+    private void LogConnectionError(Exception exception) => this.Logger.LogError(exception, "Could not connect to RabbitMQ server");
+}

src/Framework.RabbitMq/Settings/RabbitMqServerSettings.cs

@@ -0,0 +1,16 @@
+namespace Framework.RabbitMq.Settings;
+
+public class RabbitMqServerSettings
+{
+    public string Host { get; set; } = default!;
+
+    public int Port { get; set; } = 5672;
+
+    public string UserName { get; set; } = default!;
+
+    public string Secret { get; set; } = default!;
+
+    public string VirtualHost { get; set; } = default!;
+
+    public Uri Address => new($"{this.Host}:{this.Port}", UriKind.Absolute);
+}