Merge pull request #30 from StewKI/feature-mq-support-backend

Message Broker (RabbitMQ) support backend

docker/start.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+docker compose -p aips --env-file ../.env up

dotnet/AipsCore/AipsCore.csproj

@@ -16,6 +16,7 @@
       </PackageReference>
       <PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="10.0.2" />
       <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="10.0.0" />
+      <PackageReference Include="RabbitMQ.Client" Version="7.2.0" />
     </ItemGroup>
 
     <ItemGroup>

dotnet/AipsCore/Application/Abstract/MessageBroking/IMessagePublisher.cs

@@ -0,0 +1,6 @@
+namespace AipsCore.Application.Abstract.MessageBroking;
+
+public interface IMessagePublisher
+{
+    Task PublishAsync<T>(T message, CancellationToken cancellationToken = default);
+}

dotnet/AipsCore/Infrastructure/DI/AipsRegistrationExtensions.cs

@@ -1,5 +1,7 @@
 using AipsCore.Application.Abstract;
+using AipsCore.Application.Abstract.MessageBroking;
 using AipsCore.Application.Common.Dispatcher;
+using AipsCore.Infrastructure.MessageBroking.RabbitMQ;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 
@@ -16,6 +18,9 @@ public static class AipsRegistrationExtensions
 
         services.AddUserContext(configuration);
 
+        services.AddSingleton<IRabbitMqConnection, RabbitMqConnection>();
+        services.AddSingleton<IMessagePublisher, RabbitMqPublisher>();
+        
         return services;
     }
 }

dotnet/AipsCore/Infrastructure/DI/Configuration/ConfigurationEnvExtensions.cs

@@ -6,6 +6,9 @@ public static class ConfigurationEnvExtensions
 {
     private const string DbConnStringKey = "DB_CONN_STRING";
     
+    private const string RabbitMqAmqpUriKey = "RABBITMQ_AMQP_URI";
+    private const string RabbitMqExchange = "RABBITMQ_EXCHANGE";
+    
     private const string JwtIssuer = "JWT_ISSUER";
     private const string JwtAudience = "JWT_AUDIENCE";
     private const string JwtKey = "JWT_KEY";
@@ -18,6 +21,16 @@ public static class ConfigurationEnvExtensions
             return configuration.GetEnvForSure(DbConnStringKey);
         }
 
+        public string GetEnvRabbitMqAmqpUri()
+        {
+            return configuration.GetEnvForSure(RabbitMqAmqpUriKey);
+        }
+
+        public string GetEnvRabbitMqExchange()
+        {
+            return configuration.GetEnvForSure(RabbitMqExchange);
+        }
+
         public string GetEnvJwtIssuer()
         {
             return configuration.GetEnvForSure(JwtIssuer);

dotnet/AipsCore/Infrastructure/MessageBroking/RabbitMQ/IRabbitMQConnection.cs

@@ -0,0 +1,8 @@
+using RabbitMQ.Client;
+
+namespace AipsCore.Infrastructure.MessageBroking.RabbitMQ;
+
+public interface IRabbitMqConnection
+{
+    Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default);
+}

dotnet/AipsCore/Infrastructure/MessageBroking/RabbitMQ/RabbitMqConnection.cs

@@ -0,0 +1,44 @@
+using AipsCore.Infrastructure.DI.Configuration;
+using Microsoft.Extensions.Configuration;
+using RabbitMQ.Client;
+
+namespace AipsCore.Infrastructure.MessageBroking.RabbitMQ;
+
+public class RabbitMqConnection : IRabbitMqConnection
+{
+    private readonly IConfiguration _configuration;
+    private IConnection? _connection;
+
+    public RabbitMqConnection(IConfiguration configuration)
+    {
+        _configuration = configuration;
+    }
+    
+    public async Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)
+    {
+        if (_connection is null)
+        {
+            await CreateConnectionAsync();
+            //throw new InvalidOperationException($"RabbitMQ connection not created for {_configuration.GetEnvRabbitMqAmqpUri()}");
+        }
+
+        return await _connection!.CreateChannelAsync(null, cancellationToken);
+    }
+
+    public async Task CreateConnectionAsync()
+    {
+        var factory = CreateConnectionFactory();
+
+        _connection = await factory.CreateConnectionAsync();
+    }
+
+    private IConnectionFactory CreateConnectionFactory()
+    {
+        var factory = new ConnectionFactory
+        {
+            Uri = new Uri(_configuration.GetEnvRabbitMqAmqpUri())
+        };
+
+        return factory;
+    }
+}

dotnet/AipsCore/Infrastructure/MessageBroking/RabbitMQ/RabbitMqPublisher.cs

@@ -0,0 +1,44 @@
+using System.Text;
+using System.Text.Json;
+using AipsCore.Application.Abstract;
+using AipsCore.Application.Abstract.MessageBroking;
+using AipsCore.Infrastructure.DI.Configuration;
+using Microsoft.Extensions.Configuration;
+using RabbitMQ.Client;
+
+namespace AipsCore.Infrastructure.MessageBroking.RabbitMQ;
+
+public class RabbitMqPublisher : IMessagePublisher
+{
+    private readonly IRabbitMqConnection _connection;
+    private readonly IConfiguration _configuration;
+
+    public RabbitMqPublisher(IRabbitMqConnection connection, IConfiguration configuration)
+    {
+        _connection = connection;
+        _configuration = configuration;
+    }
+    
+    public async Task PublishAsync<T>(T message, CancellationToken cancellationToken = default)
+    {
+        var channel = await _connection.CreateChannelAsync(cancellationToken);
+        
+        await channel.ExchangeDeclareAsync(GetExchange(), ExchangeType.Topic, durable: true, cancellationToken: cancellationToken);
+        
+        var bytes = Serialize(message);
+        await channel.BasicPublishAsync(
+            GetExchange(), 
+            typeof(T).Name, 
+            bytes, 
+            cancellationToken);
+
+        await channel.CloseAsync(cancellationToken);
+    }
+
+    private byte[] Serialize<T>(T message)
+    {
+        return JsonSerializer.SerializeToUtf8Bytes(message);
+    }
+
+    private string GetExchange() => _configuration.GetEnvRabbitMqExchange();
+}

dotnet/AipsWebApi/Controllers/UserController.cs

@@ -1,4 +1,5 @@
 using AipsCore.Application.Abstract;
+using AipsCore.Application.Abstract.MessageBroking;
 using AipsCore.Application.Common.Authentication;
 using AipsCore.Application.Models.User.Command.LogIn;
 using AipsCore.Application.Models.User.Command.SignUp;
@@ -34,4 +35,11 @@ public class UserController : ControllerBase
         var result = await _dispatcher.Execute(command, cancellationToken);
         return Ok(result.Value);
     }
+
+    [AllowAnonymous]
+    [HttpPost("test")]
+    public async Task Test(IMessagePublisher publisher)
+    {
+        await publisher.PublishAsync("Test poruka");
+    }
 }