Feature/saga pattern implementation

Test.RabbitMq.Shop.Api/Controllers/OrderController.cs

@@ -1,5 +1,6 @@
 using MassTransit;
 using Microsoft.AspNetCore.Mvc;
+using Newtonsoft.Json;
 using Test.RabbitMq.Shop.Api.Helpers;
 using Test.RabbitMq.Shop.Api.Models;
 using Test.RabbitMq.Shop.Common.Messages;
@@ -16,13 +17,17 @@ public class OrderController : ControllerBase
     private readonly IOrderRepository _orderRepository;
     private readonly IProductRepository _productRepository;
     private readonly IPublishEndpoint _publishEndpoint;
+    private readonly IBus _bus;
+    private readonly ISendEndpointProvider _sendEndpointProvider;
 
-    public OrderController(ILogger<OrderController> logger, IOrderRepository orderRepository, IProductRepository productRepository, IPublishEndpoint publishEndpoint)
+    public OrderController(ILogger<OrderController> logger, IOrderRepository orderRepository, IProductRepository productRepository, IPublishEndpoint publishEndpoint, ISendEndpointProvider sendEndpointProvider, IBus bus)
     {
         _logger = logger;
         _orderRepository = orderRepository;
         _productRepository = productRepository;
         _publishEndpoint = publishEndpoint;
+        _sendEndpointProvider = sendEndpointProvider;
+        _bus = bus;
     }
 
     [HttpGet]
@@ -38,7 +43,7 @@ public ActionResult<IEnumerable<OrderModel>> Get()
     }
 
     [HttpPost]
-    public ActionResult Post(CreateOrderModel model)
+    public async Task<ActionResult> Post(CreateOrderModel model)
     {
         var product = _productRepository.GetProduct(model.ProductId);
         if (product == null)
@@ -57,15 +62,16 @@ public ActionResult Post(CreateOrderModel model)
         _orderRepository.AddOrder(newOrder);
 
         _logger.LogInformation($"Order {newOrder.Id} added");
-
-        _publishEndpoint.Publish(new OrderCreatedEvent(
-            Guid.NewGuid(), 
+
+        var message = new OrderCreatedEvent(
             newOrder.Id,
-            product.Id, 
-            model.ProductQuantity, 
-            newOrder.OrderPrice));
+            product.Id,
+            model.ProductQuantity,
+            newOrder.OrderPrice);
+
+        await _publishEndpoint.Publish(message);
 
-        _logger.LogInformation($"OrderCreatedEvent message published");
+        _logger.LogWarning($"OrderCreatedEvent message published: {JsonConvert.SerializeObject(message)}");
 
         return Ok();
     }

Test.RabbitMq.Shop.Api/Program.cs

@@ -9,13 +9,16 @@
 builder.Services.AddMassTransit(x =>
 {
     x.SetKebabCaseEndpointNameFormatter();
-    x.UsingRabbitMq((context, cfg) =>
+
+    x.UsingRabbitMq((ctx, cfg) =>
     {
         cfg.Host("localhost", "/", h =>
         {
             h.Username("guest");
             h.Password("guest");
         });
+
+        cfg.UseMessageRetry(r => r.Interval(3, 1000));
     });
 });
 

Test.RabbitMq.Shop.Api/Test.RabbitMq.Shop.Api.csproj

@@ -9,6 +9,7 @@
 
     <ItemGroup>
         <PackageReference Include="MassTransit" Version="8.1.0" />
+        <PackageReference Include="MassTransit.AspNetCore" Version="7.3.1" />
         <PackageReference Include="MassTransit.RabbitMQ" Version="8.1.0" />
         <PackageReference Include="Swashbuckle.AspNetCore" Version="6.5.0"/>
     </ItemGroup>
@@ -20,6 +21,7 @@
     </ItemGroup>
 
     <ItemGroup>
+      <ProjectReference Include="..\Test.RabbitMq.Shop.Common.NotificationService\Test.RabbitMq.Shop.Common.NotificationService.csproj" />
       <ProjectReference Include="..\Test.RabbitMq.Shop.Common\Test.RabbitMq.Shop.Common.csproj" />
       <ProjectReference Include="..\Test.RabbitMq.Shop.Core\Test.RabbitMq.Shop.Core.csproj" />
       <ProjectReference Include="..\Test.RabbitMq.Shop.Data\Test.RabbitMq.Shop.Data.csproj" />

Test.RabbitMq.Shop.Common.NotificationService/Consumers/NotificationSentConsumer.cs → Test.RabbitMq.Shop.Common.NotificationService/Consumers/CheckNotificationConsumer.cs

@@ -4,23 +4,23 @@
 
 namespace Test.RabbitMq.Shop.Common.NotificationService.Consumers;
 
-public class NotificationSentConsumer : IConsumer<NotificationSentEvent>
+public class CheckNotificationConsumer : IConsumer<ICheckNotificationEvent>
 {
-    private readonly ILogger<NotificationSentConsumer> _logger;
+    private readonly ILogger<CheckNotificationConsumer> _logger;
 
-    public NotificationSentConsumer(ILogger<NotificationSentConsumer> logger)
+    public CheckNotificationConsumer(ILogger<CheckNotificationConsumer> logger)
     {
         _logger = logger;
     }
 
-    public Task Consume(ConsumeContext<NotificationSentEvent> context)
+    public Task Consume(ConsumeContext<ICheckNotificationEvent> context)
     {
         var message = context.Message;
         var jsonMessage = JsonConvert.SerializeObject(message);
 
-        _logger.LogInformation($"NotificationSentConsumer message: {jsonMessage}");
+        _logger.LogInformation($"ICheckNotificationEvent message: {jsonMessage}");
 
-        // simulate receiving an email
+        // simulated email verification
         _logger.LogWarning("Notification message was received");
 
         return Task.CompletedTask;

Test.RabbitMq.Shop.Common.NotificationService/Consumers/SendNotificationConsumer.cs

@@ -0,0 +1,52 @@
+using MassTransit;
+using Newtonsoft.Json;
+using Test.RabbitMq.Shop.Common.Messages;
+
+namespace Test.RabbitMq.Shop.Common.NotificationService.Consumers;
+
+public class SendNotificationConsumer : IConsumer<ISendNotificationEvent>
+{
+    private readonly ISendEndpointProvider _sendEndpointProvider;
+    private readonly ILogger<SendNotificationConsumer> _logger;
+
+    public SendNotificationConsumer(ILogger<SendNotificationConsumer> logger, ISendEndpointProvider sendEndpointProvider)
+    {
+        _logger = logger;
+        _sendEndpointProvider = sendEndpointProvider;
+    }
+
+    public async Task Consume(ConsumeContext<ISendNotificationEvent> context)
+    {
+        var message = context.Message;
+        var jsonMessage = JsonConvert.SerializeObject(message);
+
+        _logger.LogInformation($"ISendNotificationEvent message: {jsonMessage}");
+
+        // imitation of sending an email
+        _logger.LogWarning($"Notification on email: {DateTime.Now:G} - Order {message.OrderId}");
+
+        var sendEndpoint = await _sendEndpointProvider.GetSendEndpoint(
+            new Uri($"queue:{QueueNames.OrderSagaQueueName}"));
+
+        if (IsSuccessUsingDnD())
+        {
+            await sendEndpoint.Send(new NotificationSentEvent(message.CorrelationId, message.OrderId));
+        }
+        else
+        {
+            throw new Exception("Failed because DnD");
+        }
+    }
+
+    // gamification of successful notification sending
+    private bool IsSuccessUsingDnD()
+    {
+        const int check = 11;
+        var d20dice = new Random();
+        var roll = d20dice.Next(1, 21);
+
+        _logger.LogWarning($"DND: Notification success ({check}) roll - {roll}");
+
+        return roll >= check;
+    }
+}

Test.RabbitMq.Shop.Common.NotificationService/Program.cs

@@ -1,14 +1,15 @@
 using MassTransit;
+using Test.RabbitMq.Shop.Common;
 using Test.RabbitMq.Shop.Common.NotificationService.Consumers;
 
 var builder = WebApplication.CreateBuilder(args);
 
 builder.Services.AddMassTransit(x =>
 {
     x.SetKebabCaseEndpointNameFormatter();
-    
-    x.AddConsumer<OrderCreatedConsumer>();
-    x.AddConsumer<NotificationSentConsumer>();
+
+    x.AddConsumer<SendNotificationConsumer>();
+    x.AddConsumer<CheckNotificationConsumer>();
 
     x.UsingRabbitMq((context, cfg) =>
     {
@@ -17,11 +18,13 @@
             h.Username("guest");
             h.Password("guest");
         });
-                
-        cfg.ReceiveEndpoint("order_saga", c =>
+
+        cfg.ReceiveEndpoint(QueueNames.NotificationQueueName, c =>
         {
-            c.ConfigureConsumer<OrderCreatedConsumer>(context);
-            c.ConfigureConsumer<NotificationSentConsumer>(context);
+            c.UseMessageRetry(r => r.Interval(3, 1000));
+
+            c.ConfigureConsumer<SendNotificationConsumer>(context);
+            c.ConfigureConsumer<CheckNotificationConsumer>(context);
         });
     });
 });

Test.RabbitMq.Shop.Common.NotificationService/Test.RabbitMq.Shop.Common.NotificationService.csproj

@@ -15,8 +15,8 @@
     </ItemGroup>
 
     <ItemGroup>
-      <PackageReference Include="MassTransit" Version="8.1.1-develop.1522" />
-      <PackageReference Include="MassTransit.RabbitMQ" Version="8.1.1-develop.1522" />
+      <PackageReference Include="MassTransit" Version="8.1.0" />
+      <PackageReference Include="MassTransit.RabbitMQ" Version="8.1.0" />
       <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
     </ItemGroup>
 

Test.RabbitMq.Shop.Common.StateMachineService/Dockerfile

@@ -0,0 +1,18 @@
+FROM mcr.microsoft.com/dotnet/runtime:6.0 AS base
+WORKDIR /app
+
+FROM mcr.microsoft.com/dotnet/sdk:6.0 AS build
+WORKDIR /src
+COPY ["Test.RabbitMq.Shop.Common.StateMachineService/Test.RabbitMq.Shop.Common.StateMachineService.csproj", "Test.RabbitMq.Shop.Common.StateMachineService/"]
+RUN dotnet restore "Test.RabbitMq.Shop.Common.StateMachineService/Test.RabbitMq.Shop.Common.StateMachineService.csproj"
+COPY . .
+WORKDIR "/src/Test.RabbitMq.Shop.Common.StateMachineService"
+RUN dotnet build "Test.RabbitMq.Shop.Common.StateMachineService.csproj" -c Release -o /app/build
+
+FROM build AS publish
+RUN dotnet publish "Test.RabbitMq.Shop.Common.StateMachineService.csproj" -c Release -o /app/publish /p:UseAppHost=false
+
+FROM base AS final
+WORKDIR /app
+COPY --from=publish /app/publish .
+ENTRYPOINT ["dotnet", "Test.RabbitMq.Shop.Common.StateMachineService.dll"]

Test.RabbitMq.Shop.Common.StateMachineService/OrderState.cs

@@ -0,0 +1,10 @@
+using MassTransit;
+
+namespace Test.RabbitMq.Shop.Common.StateMachineService;
+
+public class OrderState : SagaStateMachineInstance
+{
+    public Guid CorrelationId { get; set; }
+    public Guid OrderId { get; set; }
+    public string CurrentState { get; set; }
+}

Test.RabbitMq.Shop.Common.StateMachineService/OrderStateMachine.cs

@@ -0,0 +1,57 @@
+using MassTransit;
+using Newtonsoft.Json;
+using Test.RabbitMq.Shop.Common.Messages;
+
+namespace Test.RabbitMq.Shop.Common.StateMachineService;
+
+public class OrderStateMachine : MassTransitStateMachine<OrderState>
+{
+    private readonly ILogger<OrderStateMachine> _logger;
+
+    public Event<IOrderCreatedEvent> OrderCreatedEvent { get; set; }
+    public Event<INotificationSentEvent> NotificationSentEvent { get; set; }
+    public Event<Fault<ISendNotificationEvent>> SendNotificationFaultEvent { get; set; }
+
+    public State OrderCreated { get; set; }
+    public State SendNotificationFault { get; set; }
+    public State NotificationSent { get; set; }
+
+    public OrderStateMachine(ILogger<OrderStateMachine> logger)
+    {
+        _logger = logger;
+        InstanceState(x => x.CurrentState);
+
+        Event(() => OrderCreatedEvent);
+        Event(() => NotificationSentEvent);
+        Event(() => SendNotificationFaultEvent,
+            x => x.CorrelateById(
+                ctx => ctx.InitiatorId ?? ctx.Message.Message.CorrelationId));
+
+        Initially(
+            When(OrderCreatedEvent)
+                .Then(ctx => 
+                    _logger.LogWarning($"OrderCreatedEvent message: {JsonConvert.SerializeObject(ctx.Message)}"))
+                .Then(ctx => ctx.Saga.OrderId = ctx.Message.OrderId)
+                .Send(new Uri($"queue:{QueueNames.NotificationQueueName}"),
+                    ctx => 
+                        new SendNotificationEvent(ctx.Saga.CorrelationId, ctx.Message.OrderId))
+                .TransitionTo(OrderCreated));
+
+        During(OrderCreated,
+            When(SendNotificationFaultEvent)
+                .Then(ctx =>
+                    _logger.LogError(
+                        $"SendNotificationFaultEvent message: {JsonConvert.SerializeObject(ctx.Message)}"))
+                .TransitionTo(SendNotificationFault));
+
+        During(OrderCreated,
+            When(NotificationSentEvent)
+                .Then(ctx => 
+                    _logger.LogWarning($"NotificationSentEvent message: {JsonConvert.SerializeObject(ctx.Message)}"))
+                .Send(new Uri($"queue:{QueueNames.NotificationQueueName}"),
+                    ctx => 
+                        new CheckNotificationEvent(ctx.Saga.CorrelationId, ctx.Message.OrderId))
+                .TransitionTo(NotificationSent)
+                .Finalize());
+    }
+}

Test.RabbitMq.Shop.Common.StateMachineService/Program.cs

@@ -0,0 +1,36 @@
+using MassTransit;
+using Test.RabbitMq.Shop.Common;
+using Test.RabbitMq.Shop.Common.StateMachineService;
+
+var builder = WebApplication.CreateBuilder(args);
+
+builder.WebHost.UseUrls("http://localhost:5002/");
+
+builder.Services.AddMassTransit(x =>
+{
+    x.SetKebabCaseEndpointNameFormatter();
+
+    x.AddSagaStateMachine<OrderStateMachine, OrderState>()
+        .Endpoint(e => { e.Name = QueueNames.OrderSagaQueueName; })
+        .InMemoryRepository();
+
+    x.UsingRabbitMq((context, cfg) =>
+    {
+        cfg.Host("localhost", "/", h =>
+        {
+            h.Username("guest");
+            h.Password("guest");
+        });
+
+        cfg.ReceiveEndpoint(QueueNames.OrderSagaQueueName, c =>
+        {
+            c.UseMessageRetry(r => r.Interval(3, 1000));
+
+            c.StateMachineSaga<OrderState>(context);
+        });
+    });
+});
+
+var app = builder.Build();
+
+app.Run();