Subscriber abstraction + RabbitMQ consumer service

src/Aspire/NCafe.AppHost/Program.cs

@@ -12,17 +12,16 @@
     .WithReference(eventStore)
     .WaitFor(eventStore);
 
-var baristaProject = builder.AddProject<Projects.NCafe_Barista_Api>("barista-api")
+var cashierProject = builder.AddProject<Projects.NCafe_Cashier_Api>("cashier-api")
     .WithReference(eventStore)
     .WithReference(rabbitMq)
     .WaitFor(eventStore)
     .WaitFor(rabbitMq);
 
-var cashierProject = builder.AddProject<Projects.NCafe_Cashier_Api>("cashier-api")
+var baristaProject = builder.AddProject<Projects.NCafe_Barista_Api>("barista-api")
     .WithReference(eventStore)
     .WithReference(rabbitMq)
-    .WaitFor(eventStore)
-    .WaitFor(rabbitMq);
+    .WaitFor(cashierProject);
 
 var webUiProject = builder.AddProject<Projects.NCafe_Web>("web-ui")
     .WithExternalHttpEndpoints();

src/Barista/NCafe.Barista.Api/Program.cs

@@ -1,19 +1,24 @@
 using Microsoft.AspNetCore.ResponseCompression;
+using Microsoft.AspNetCore.SignalR;
 using NCafe.Barista.Api.Hubs;
-using NCafe.Barista.Api.MessageBus;
 using NCafe.Barista.Api.Projections;
 using NCafe.Barista.Domain.Commands;
 using NCafe.Barista.Domain.Queries;
 using NCafe.Barista.Domain.ReadModels;
 using NCafe.Core.Commands;
+using NCafe.Core.MessageBus.Events;
 using NCafe.Core.Queries;
 using NCafe.Infrastructure;
+using NCafe.Shared.Hubs;
 
 var builder = WebApplication.CreateBuilder(args);
 
 builder.AddServiceDefaults();
 
-builder.AddRabbitMQClient("rabbitmq");
+builder.AddRabbitMQClient("rabbitmq", configureConnectionFactory: config =>
+{
+    config.DispatchConsumersAsync = true;
+});
 
 // Add services to the container.
 builder.Services.AddEventStoreRepository(builder.Configuration)
@@ -25,7 +30,7 @@
                 .AddEventStoreProjectionService<BaristaOrder>()
                 .AddHostedService<BaristaOrderProjectionService>();
 
-builder.Services.AddHostedService<OrdersConsumerService>();
+builder.Services.AddRabbitMqConsumerService(builder.Configuration);
 
 builder.Services.AddEndpointsApiExplorer()
                 .AddSwaggerGen();
@@ -51,6 +56,21 @@
 
 var app = builder.Build();
 
+app.UseMessageSubscriber()
+   .Subscribe<OrderPlaced>(async (serviceProvider, message) =>
+   {
+       var commandDispatcher = serviceProvider.GetRequiredService<ICommandDispatcher>();
+       var hubContext = serviceProvider.GetRequiredService<IHubContext<OrderHub>>();
+
+       // Dispatch domain command
+       await commandDispatcher.DispatchAsync(new PlaceOrder(message.Id, message.ProductId, message.Quantity));
+
+       // Notify clients
+       await hubContext.Clients.All.SendAsync(
+           "ReceiveOrder",
+           new Order(message.Id, message.ProductId, message.Quantity));
+   });
+
 app.MapDefaultEndpoints();
 
 app.UseResponseCompression();

src/Barista/NCafe.Barista.Api/appsettings.json

@@ -9,5 +9,8 @@
   "ConnectionStrings": {
     "EventStore": "esdb://localhost:2113?tls=false",
     "RabbitMq": "host=localhost;username=myuser;password=mypassword"
+  },
+  "RabbitMq": {
+    "queuePrefix": "barista-service"
   }
 }

src/Cashier/NCafe.Cashier.Api/appsettings.json

@@ -9,5 +9,8 @@
   "ConnectionStrings": {
     "EventStore": "esdb://localhost:2113?tls=false",
     "RabbitMq": "host=localhost;username=myuser;password=mypassword"
+  },
+  "RabbitMq": {
+    "exchangeName": "cashier"
   }
 }

src/Cashier/NCafe.Cashier.Domain.Tests/Commands/PlaceOrderTests.cs

@@ -75,7 +75,7 @@ public async Task GivenOrderSaved_ShouldPublishToMessageBus()
 
         // Assert
         exception.ShouldBeNull();
-        A.CallTo(() => _publisher.Publish("orders_queue", A<OrderPlaced>.That.Matches(o => o.ProductId == productId && o.Quantity == 1)))
+        A.CallTo(() => _publisher.Publish(A<OrderPlaced>.That.Matches(o => o.ProductId == productId && o.Quantity == 1)))
             .MustHaveHappenedOnceExactly();
     }
 }

src/Cashier/NCafe.Cashier.Domain/Commands/PlaceOrder.cs

@@ -20,15 +20,13 @@ internal sealed class PlaceOrderHandler(
     private readonly IReadModelRepository<Product> _productReadRepository = productReadRepository;
     private readonly IPublisher _publisher = publisher;
 
-    private const string Queue = "orders_queue";
-
     public async Task HandleAsync(PlaceOrder command)
     {
         var product = _productReadRepository.GetById(command.ProductId) ?? throw new ProductNotFoundException(command.ProductId);
         var order = new Order(Guid.NewGuid(), product.Id, command.Quantity);
 
         await _repository.Save(order);
 
-        await _publisher.Publish(Queue, new OrderPlaced(order.Id, order.ProductId, order.Quantity));
+        await _publisher.Publish(new OrderPlaced(order.Id, order.ProductId, order.Quantity));
     }
 }

src/Common/NCafe.Core/MessageBus/Events/OrderPlaced.cs

@@ -1,3 +1,4 @@
 namespace NCafe.Core.MessageBus.Events;
 
+[Message("cashier")]
 public sealed record OrderPlaced(Guid Id, Guid ProductId, int Quantity) : IBusMessage;

src/Common/NCafe.Core/MessageBus/IMessageSubscriber.cs

@@ -0,0 +1,7 @@
+namespace NCafe.Core.MessageBus;
+
+public interface IMessageSubscriber
+{
+    IReadOnlyDictionary<Type, Func<IServiceProvider, object, Task>> Handlers { get; }
+    IMessageSubscriber Subscribe<T>(Func<IServiceProvider, T, Task> handle) where T : class, IBusMessage;
+}

src/Common/NCafe.Core/MessageBus/IPublisher.cs

@@ -2,5 +2,5 @@
 
 public interface IPublisher
 {
-    Task Publish<T>(string topicName, T message) where T : class, IBusMessage;
+    Task Publish<T>(T message) where T : class, IBusMessage;
 }

src/Common/NCafe.Core/MessageBus/MessageAttribute.cs

@@ -0,0 +1,7 @@
+namespace NCafe.Core.MessageBus;
+
+[AttributeUsage(AttributeTargets.Class, Inherited = false, AllowMultiple = false)]
+public sealed class MessageAttribute(string exchangeName) : Attribute
+{
+    public string ExchangeName { get; } = exchangeName;
+}

src/Common/NCafe.Infrastructure/DependencyRegistration.cs

@@ -1,6 +1,8 @@
 using EventStore.Client;
+using Microsoft.AspNetCore.Builder;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
 using NCafe.Core.Commands;
 using NCafe.Core.MessageBus;
 using NCafe.Core.Projections;
@@ -10,9 +12,10 @@
 using NCafe.Infrastructure.Commands;
 using NCafe.Infrastructure.EventStore;
 using NCafe.Infrastructure.Logging;
-using NCafe.Infrastructure.MessageBus;
+using NCafe.Infrastructure.MessageBrokers.RabbitMQ;
 using NCafe.Infrastructure.Queries;
 using NCafe.Infrastructure.ReadModels;
+using RabbitMQ.Client;
 
 namespace NCafe.Infrastructure;
 
@@ -35,9 +38,7 @@ public static IServiceCollection AddEventStoreRepository(this IServiceCollection
     public static IServiceCollection AddEventStoreProjectionService<TModel>(this IServiceCollection services)
         where TModel : ReadModel
     {
-        services.AddSingleton<IProjectionService<TModel>, EventStoreProjectionService<TModel>>();
-
-        return services;
+        return services.AddSingleton<IProjectionService<TModel>, EventStoreProjectionService<TModel>>();
     }
 
     public static IServiceCollection AddCommandHandlers<T>(this IServiceCollection services)
@@ -79,11 +80,50 @@ public static IServiceCollection AddRabbitMqPublisher(this IServiceCollection se
             throw new InvalidOperationException("Invalid RabbitMq configuration.");
         }
 
+        services.AddOptions<RabbitMqSettings>()
+                .Bind(configuration.GetSection(RabbitMqSettings.SectionName));
+
         services.AddSingleton<IPublisher, RabbitMqPublisher>();
 
+        InitializeRabbitMqExchange(services);
+
         return services;
     }
 
+    public static IServiceCollection AddRabbitMqConsumerService(this IServiceCollection services, IConfiguration configuration)
+    {
+        services.AddOptions<RabbitMqSettings>()
+                .Bind(configuration.GetSection(RabbitMqSettings.SectionName));
+
+        return services.AddSingleton<IMessageSubscriber, RabbitMqMessageSubscriber>()
+                       .AddHostedService<RabbitMqConsumerService>();
+    }
+
+    public static IMessageSubscriber UseMessageSubscriber(this IApplicationBuilder app)
+    {
+        var messageSubscriber = app.ApplicationServices.GetService<IMessageSubscriber>();
+
+        if (messageSubscriber is null)
+        {
+            throw new InvalidOperationException("Message Subscriber is not registered. Make sure to call Services.AddRabbitMqConsumerService.");
+        }
+
+        return messageSubscriber;
+    }
+
+    private static void InitializeRabbitMqExchange(IServiceCollection services)
+    {
+        var scope = services.BuildServiceProvider().CreateScope();
+        var connection = scope.ServiceProvider.GetRequiredService<IConnection>();
+        var settings = scope.ServiceProvider.GetRequiredService<IOptions<RabbitMqSettings>>();
+
+        var channel = connection.CreateModel();
+        channel.ExchangeDeclare(exchange: ExchangeNameProvider.Get(settings.Value.ExchangeName),
+                                type: ExchangeType.Topic,
+                                durable: true,
+                                autoDelete: false);
+    }
+
     public static IServiceCollection AddInMemoryReadModelRepository<T>(this IServiceCollection services) where T : ReadModel
     {
         services.AddSingleton<IReadModelRepository<T>, InMemoryReadModelRepository<T>>();

src/Common/NCafe.Infrastructure/MessageBrokers/RabbitMQ/ExchangeNameProvider.cs

@@ -0,0 +1,11 @@
+using System.Reflection;
+
+namespace NCafe.Infrastructure.MessageBrokers.RabbitMQ;
+
+internal static class ExchangeNameProvider
+{
+    public static string Get(string exchange)
+    {
+        return !string.IsNullOrWhiteSpace(exchange) ? exchange : Assembly.GetEntryAssembly().GetName().Name;
+    }
+}