feat: MassTransit (#94)

Author: alex289

CleanArchitecture.Api/CleanArchitecture.Api.csproj

@@ -14,16 +14,18 @@
         <PackageReference Include="AspNetCore.HealthChecks.Redis" Version="9.0.0" />
         <PackageReference Include="AspNetCore.HealthChecks.SqlServer" Version="9.0.0" />
         <PackageReference Include="AspNetCore.HealthChecks.UI.Client" Version="9.0.0" />
-        <PackageReference Include="Grpc.AspNetCore.Server.Reflection" Version="2.67.0" />
-        <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="9.0.2" />
-        <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="9.0.2" />
-        <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="9.0.2">
+        <PackageReference Include="Grpc.AspNetCore.Server.Reflection" Version="2.70.0" />
+        <PackageReference Include="MassTransit.Newtonsoft" Version="8.3.7" />
+        <PackageReference Include="MassTransit.RabbitMQ" Version="8.3.7" />
+        <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="9.0.3" />
+        <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="9.0.3" />
+        <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="9.0.3">
             <PrivateAssets>all</PrivateAssets>
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
         </PackageReference>
-        <PackageReference Include="Microsoft.EntityFrameworkCore.Proxies" Version="9.0.2" />
+        <PackageReference Include="Microsoft.EntityFrameworkCore.Proxies" Version="9.0.3" />
         <PackageReference Include="Microsoft.Extensions.Caching.StackExchangeRedis" Version="9.0.2" />
-        <PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks.EntityFrameworkCore" Version="9.0.2" />
+        <PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks.EntityFrameworkCore" Version="9.0.3" />
         <PackageReference Include="Swashbuckle.AspNetCore" Version="7.3.1" />
         <PackageReference Include="Swashbuckle.AspNetCore.Annotations" Version="7.3.1" />
     </ItemGroup>

CleanArchitecture.Api/Extensions/ConfigurationExtensions.cs

@@ -1,5 +1,5 @@
 using System;
-using CleanArchitecture.Domain.Rabbitmq;
+using CleanArchitecture.Domain.Settings;
 using Microsoft.Extensions.Configuration;
 
 namespace CleanArchitecture.Api.Extensions;
@@ -11,15 +11,13 @@ public static RabbitMqConfiguration GetRabbitMqConfiguration(
     {
         var isAspire = configuration["ASPIRE_ENABLED"] == "true";
 
-        var rabbitEnabled = configuration["RabbitMQ:Enabled"];
         var rabbitHost = configuration["RabbitMQ:Host"];
         var rabbitPort = configuration["RabbitMQ:Port"];
         var rabbitUser = configuration["RabbitMQ:Username"];
         var rabbitPass = configuration["RabbitMQ:Password"];
 
         if (isAspire)
         {
-            rabbitEnabled = "true";
             var connectionString = configuration["ConnectionStrings:RabbitMq"];
 
             var rabbitUri = new Uri(connectionString!);
@@ -33,7 +31,6 @@ public static RabbitMqConfiguration GetRabbitMqConfiguration(
         {
             Host = rabbitHost ?? "",
             Port = int.Parse(rabbitPort ?? "0"),
-            Enabled = bool.Parse(rabbitEnabled ?? "false"),
             Username = rabbitUser ?? "",
             Password = rabbitPass ?? ""
         };

CleanArchitecture.Api/Program.cs

@@ -4,19 +4,21 @@
 using CleanArchitecture.Api.Extensions;
 using CleanArchitecture.Application.Extensions;
 using CleanArchitecture.Application.gRPC;
+using CleanArchitecture.Domain.Consumers;
 using CleanArchitecture.Domain.Extensions;
-using CleanArchitecture.Domain.Rabbitmq.Extensions;
 using CleanArchitecture.Infrastructure.Database;
 using CleanArchitecture.Infrastructure.Extensions;
 using CleanArchitecture.ServiceDefaults;
 using HealthChecks.ApplicationStatus.DependencyInjection;
 using HealthChecks.UI.Client;
+using MassTransit;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Diagnostics.HealthChecks;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
 using RabbitMQ.Client;
 
 var builder = WebApplication.CreateBuilder(args);
@@ -33,11 +35,6 @@
     builder.Services.AddZenFirewall();
 }
 
-builder.Services
-    .AddHealthChecks()
-    .AddDbContextCheck<ApplicationDbContext>()
-    .AddApplicationStatus();
-
 var isAspire = builder.Configuration["ASPIRE_ENABLED"] == "true";
 
 var rabbitConfiguration = builder.Configuration.GetRabbitMqConfiguration();
@@ -47,23 +44,22 @@
     ? builder.Configuration["ConnectionStrings:Database"]
     : builder.Configuration["ConnectionStrings:DefaultConnection"];
 
-if (builder.Environment.IsProduction())
-{
-    builder.Services
-        .AddHealthChecks()
-        .AddSqlServer(dbConnectionString!)
-        .AddRedis(redisConnectionString!, "Redis")
-        .AddRabbitMQ(
-            async _ =>
+builder.Services
+    .AddHealthChecks()
+    .AddDbContextCheck<ApplicationDbContext>()
+    .AddApplicationStatus()
+    .AddSqlServer(dbConnectionString!)
+    .AddRedis(redisConnectionString!, "Redis")
+    .AddRabbitMQ(
+        async _ =>
+        {
+            var factory = new ConnectionFactory
             {
-                var factory = new ConnectionFactory
-                {
-                    Uri = new Uri(rabbitConfiguration.ConnectionString),
-                };
-                return await factory.CreateConnectionAsync();
-            },
-            name: "RabbitMQ");
-}
+                Uri = new Uri(rabbitConfiguration.ConnectionString),
+            };
+            return await factory.CreateConnectionAsync();
+        },
+        name: "RabbitMQ");
 
 builder.Services.AddDbContext<ApplicationDbContext>(options =>
 {
@@ -82,7 +78,48 @@
 builder.Services.AddNotificationHandlers();
 builder.Services.AddApiUser();
 
-builder.Services.AddRabbitMqHandler(rabbitConfiguration);
+builder.Services.AddMassTransit(x =>
+{
+    x.AddConsumer<FanoutEventConsumer>();
+    x.AddConsumer<TenantUpdatedEventConsumer>();
+    
+    x.UsingRabbitMq((context, cfg) =>
+    {
+        cfg.ConfigureNewtonsoftJsonSerializer(settings =>
+        {
+            settings.TypeNameHandling = TypeNameHandling.Objects;
+            settings.NullValueHandling = NullValueHandling.Ignore;
+            return settings;
+        });
+        cfg.UseNewtonsoftJsonSerializer();
+        cfg.ConfigureNewtonsoftJsonDeserializer(settings =>
+        {
+            settings.TypeNameHandling = TypeNameHandling.Objects;
+            settings.NullValueHandling = NullValueHandling.Ignore;
+            return settings;
+        });
+        
+        cfg.Host(rabbitConfiguration.Host, (ushort)rabbitConfiguration.Port, "/", h => {
+            h.Username(rabbitConfiguration.Username);
+            h.Password(rabbitConfiguration.Password);
+        });
+
+        // Every instance of the service will receive the message
+        cfg.ReceiveEndpoint("clean-architecture-fanout-event-" + Guid.NewGuid(), e =>
+        {
+            e.Durable = false;
+            e.AutoDelete = true;
+            e.ConfigureConsumer<FanoutEventConsumer>(context);
+            e.DiscardSkippedMessages();
+        });
+        cfg.ReceiveEndpoint("clean-architecture-fanout-events", e =>
+        {
+            e.ConfigureConsumer<TenantUpdatedEventConsumer>(context);
+            e.DiscardSkippedMessages();
+        });
+        cfg.ConfigureEndpoints(context);
+    });
+});
 
 builder.Services.AddHostedService<SetInactiveUsersService>();
 
@@ -148,7 +185,7 @@
 app.MapGrpcService<UsersApiImplementation>();
 app.MapGrpcService<TenantsApiImplementation>();
 
-app.Run();
+await app.RunAsync();
 
 // Needed for integration tests web application factory
 public partial class Program

CleanArchitecture.AppHost/CleanArchitecture.AppHost.csproj

@@ -15,7 +15,7 @@
     <PackageReference Include="Aspire.Hosting.AppHost" Version="9.1.0" />
     <PackageReference Include="Aspire.Hosting.RabbitMQ" Version="9.1.0" />
     <PackageReference Include="Aspire.Hosting.Redis" Version="9.1.0" />
-    <PackageReference Include="Aspire.Hosting.SqlServer" Version="9.0.0" />
+    <PackageReference Include="Aspire.Hosting.SqlServer" Version="9.1.0" />
   </ItemGroup>
 
   <ItemGroup>

CleanArchitecture.Application/CleanArchitecture.Application.csproj

@@ -6,7 +6,7 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="Microsoft.EntityFrameworkCore" Version="9.0.2" />
+        <PackageReference Include="Microsoft.EntityFrameworkCore" Version="9.0.3" />
     </ItemGroup>
 
     <ItemGroup>

CleanArchitecture.Domain/CleanArchitecture.Domain.csproj

@@ -8,11 +8,11 @@
     <ItemGroup>
         <PackageReference Include="BCrypt.Net-Next" Version="4.0.3" />
         <PackageReference Include="FluentValidation" Version="11.11.0" />
+        <PackageReference Include="MassTransit" Version="8.3.7" />
         <PackageReference Include="MediatR" Version="12.4.1" />
-        <PackageReference Include="Microsoft.Extensions.Options" Version="9.0.2" />
+        <PackageReference Include="Microsoft.Extensions.Options" Version="9.0.3" />
         <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
-        <PackageReference Include="RabbitMQ.Client" Version="7.1.1" />
-        <PackageReference Include="System.IdentityModel.Tokens.Jwt" Version="8.6.0" />
+        <PackageReference Include="System.IdentityModel.Tokens.Jwt" Version="8.6.1" />
     </ItemGroup>
 
     <ItemGroup>

CleanArchitecture.Domain/Constants/Messaging.cs

@@ -0,0 +1,22 @@
+using System.Threading.Tasks;
+using CleanArchitecture.Shared.Events;
+using MassTransit;
+using Microsoft.Extensions.Logging;
+
+namespace CleanArchitecture.Domain.Consumers;
+
+public sealed class FanoutEventConsumer : IConsumer<FanoutDomainEvent>
+{
+    private readonly ILogger<FanoutEventConsumer> _logger;
+
+    public FanoutEventConsumer(ILogger<FanoutEventConsumer> logger)
+    {
+        _logger = logger;
+    }
+
+    public Task Consume(ConsumeContext<FanoutDomainEvent> context)
+    {
+        _logger.LogInformation("FanoutDomainEventConsumer: {FanoutDomainEvent}", context.Message);
+        return Task.CompletedTask;
+    }
+}

CleanArchitecture.Domain/Consumers/FanoutEventConsumer.cs

@@ -0,0 +1,22 @@
+using System.Threading.Tasks;
+using CleanArchitecture.Shared.Events.Tenant;
+using MassTransit;
+using Microsoft.Extensions.Logging;
+
+namespace CleanArchitecture.Domain.Consumers;
+
+public sealed class TenantUpdatedEventConsumer : IConsumer<TenantUpdatedEvent>
+{
+    private readonly ILogger<TenantUpdatedEventConsumer> _logger;
+
+    public TenantUpdatedEventConsumer(ILogger<TenantUpdatedEventConsumer> logger)
+    {
+        _logger = logger;
+    }
+
+    public Task Consume(ConsumeContext<TenantUpdatedEvent> context)
+    {
+        _logger.LogInformation("TenantUpdatedEventConsumer: {TenantId}", context.Message.AggregateId);
+        return Task.CompletedTask;
+    }
+}

CleanArchitecture.Domain/Consumers/TenantUpdatedEventConsumer.cs

@@ -1,27 +1,33 @@
 using System.Threading.Tasks;
-using CleanArchitecture.Domain.Constants;
-using CleanArchitecture.Domain.Rabbitmq;
+using CleanArchitecture.Domain.Interfaces;
 using CleanArchitecture.Shared.Events;
+using MassTransit;
 
 namespace CleanArchitecture.Domain.EventHandler.Fanout;
 
 public sealed class FanoutEventHandler : IFanoutEventHandler
 {
-    private readonly RabbitMqHandler _rabbitMqHandler;
+    private readonly IPublishEndpoint _massTransit;
+    private readonly IUser _user;
 
     public FanoutEventHandler(
-        RabbitMqHandler rabbitMqHandler)
+        IPublishEndpoint massTransit, IUser user)
     {
-        _rabbitMqHandler = rabbitMqHandler;
-        _rabbitMqHandler.InitializeExchange(Messaging.ExchangeNameNotifications);
+        _massTransit = massTransit;
+        _user = user;
     }
 
-    public Task<DomainEvent> HandleDomainEventAsync(DomainEvent @event)
+    public async Task<T> HandleDomainEventAsync<T>(T @event) where T : DomainEvent
     {
-        _rabbitMqHandler.EnqueueExchangeMessage(
-            Messaging.ExchangeNameNotifications,
-            @event);
+        var fanoutDomainEvent =
+            new FanoutDomainEvent(
+                @event.AggregateId,
+                @event,
+                _user.GetUserId());
+        
+        await _massTransit.Publish(fanoutDomainEvent);
+        await _massTransit.Publish(@event);
 
-        return Task.FromResult(@event);
+        return @event;
     }
 }