‫بررسی الگوی SAGA در میکروسرویس‌ها: رویکردهای Choreography و Orchestration و استفاده از MassTransit Automatonymous

// Implementing the “ISagaVersion” interface is required to persist the data in MongoDB.
public class OrderState : SagaStateMachineInstance,ISagaVersion
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; }
    public DateTime OrderStartDate { get; set; }
    public int Version { get; set; } 
}

using MassTransit;
using StateMachineSample.Events;

namespace StateMachineSample.StateMachine;

public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public OrderStateMachine()
    {
    
    }

    #region Events

    public Event<OrderProcessInitializationEvent> OrderProcessInitializationEvent { get; }
    public Event<Fault<OrderProcessInitializationEvent>> OrderProcessInitializationFaultEvent { get; }

    public Event<CheckProductStockEvent> CheckProductStockEvent { get; }
    public Event<Fault<CheckProductStockEvent>> CheckProductStockFaultEvent { get; }

    public Event<TakePaymentEvent> TakePaymentEvent { get; }
    public Event<Fault<TakePaymentEvent>> TakePaymentEventFaultEvent { get; }

    public Event<CreateOrderEvent> CreateOrderEvent { get; }
    public Event<Fault<CreateOrderEvent>> CreateOrderFaultEvent { get; }

    public Event<OrderProcessFailedEvent> OrderProcessFailedEvent { get; }

    #endregion
}

using MassTransit;
using StateMachineSample.Events;

namespace StateMachineSample.StateMachine;

public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public OrderStateMachine()
    {
    }

    #region States

    public State OrderProcessInitializedState { get; }
    public State OrderProcessInitializedFaultedState { get; }

    public State CheckProductStockState { get; }
    public State CheckProductStockFaultedState { get; }

    public State TakePaymentState { get; }
    public State TakePaymentFaultedState { get; }

    public State CreateOrderState { get; }
    public State CreateOrderFaultedState { get; }

    public State OrderProcessFailedState { get; }

    #endregion
}

using MassTransit;
using StateMachineSample.Events;

namespace StateMachineSample.StateMachine;

public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public OrderStateMachine()
    {
        InstanceState(x => x.CurrentState);
    }
}

using MassTransit;
using StateMachineSample.Events;

namespace StateMachineSample.StateMachine;

public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public OrderStateMachine()
    {
        #region EventsDefinitions

        Event(() => OrderProcessInitializationEvent);
        Event(() => OrderProcessInitializationFaultEvent,
            x => x.CorrelateById(context => context.InitiatorId ?? context.Message.Message.OrderId));

        Event(() => CheckProductStockEvent);
        Event(() => CheckProductStockFaultEvent,
            x => x.CorrelateById(context => context.InitiatorId ?? context.Message.Message.OrderId));

        Event(() => TakePaymentEvent);
        Event(() => TakePaymentEventFaultEvent,
            x => x.CorrelateById(context => context.InitiatorId ?? context.Message.Message.OrderId));

        Event(() => CreateOrderEvent);
        Event(() => CreateOrderFaultEvent,
            x => x.CorrelateById(context => context.InitiatorId ?? context.Message.Message.OrderId));

        Event(() => OrderProcessFailedEvent);

        #endregion
    }
}

• OrderProcessInitializationEvent: این رویداد برای شروع فرآیند پردازش سفارش تعریف شده است.
• OrderProcessInitializationFaultEvent: این رویداد زمانی که خطایی در فرآیند پردازش سفارش اتفاق می‌افتد، فعال می‌شود. همچنین از CorrelateById برای مطابقت پیام‌های ورودی با یک شناسه خاص استفاده می‌شود که از InitiatorId یا OrderId پیام استخراج می‌شود.
• CheckProductStockEvent: این رویداد زمانی فعال می‌شود که نیاز به بررسی موجودی محصولات باشد.
• CheckProductStockFaultEvent: این رویداد زمانی فعال می‌شود که در حین بررسی موجودی محصولات خطایی رخ دهد.
• TakePaymentEvent: این رویداد برای انجام عملیات پرداخت سفارش است.
• TakePaymentEventFaultEvent: این رویداد زمانی که در حین پردازش پرداخت خطا رخ می‌دهد فعال می‌شود.
• CreateOrderEvent: این رویداد زمانی که می‌خواهیم سفارش را ایجاد کنیم، فعال می‌شود.
• CreateOrderFaultEvent: این رویداد زمانی که خطا در ایجاد سفارش رخ می‌دهد، فعال می‌شود.
• OrderProcessFailedEvent: این رویداد برای مدیریت خطاهای کلی در فرآیند پردازش سفارش است.
• مفهوم CorrelateById: در بیشتر رویدادها از متد CorrelateById استفاده شده است تا اطمینان حاصل شود که رویدادهای مختلف به یک سفارش خاص مرتبط هستند. این کار با استفاده از شناسه‌های InitiatorId یا OrderId انجام می‌شود. این شناسه‌ها برای ارتباط دادن پیام‌ها و رویدادها در طول ماشین وضعیت به کار می‌روند.

using MassTransit;
using StateMachineSample.Events;

namespace StateMachineSample.StateMachine;

public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public OrderStateMachine()
    {
        #region Flow

        During(Initial,
            When(OrderProcessInitializationEvent)
                .Then(x => x.Saga.OrderStartDate = DateTime.Now)
                .TransitionTo(OrderProcessInitializedState));

        During(OrderProcessInitializedState,
            When(CheckProductStockEvent)
                .TransitionTo(CheckProductStockState));

        During(CheckProductStockState,
            When(TakePaymentEvent)
                .TransitionTo(TakePaymentState));

        During(TakePaymentState,
            When(CreateOrderEvent)
                .TransitionTo(CreateOrderState));

        #endregion

    }
}

During(Initial,
    When(OrderProcessInitializationEvent)
        .Then(x => x.Saga.OrderStartDate = DateTime.Now)
        .TransitionTo(OrderProcessInitializedState));

• Initial: وضعیت اولیه ماشین است که معمولاً وقتی سفارش تازه ایجاد می‌شود شروع می‌شود.
• When(OrderProcessInitializationEvent): وقتی رویداد OrderProcessInitializationEvent رخ می‌دهد، ماشین حالت باید تغییرات زیر را انجام دهد:
• Then(x => x.Saga.OrderStartDate = DateTime.Now): این خط تاریخ و زمان شروع پردازش سفارش را در OrderStartDate ذخیره می‌کند.
• TransitionTo(OrderProcessInitializedState): پس از انجام این کار، وضعیت ماشین به OrderProcessInitializedState تغییر می‌کند.

During(OrderProcessInitializedState,
    When(CheckProductStockEvent)
        .TransitionTo(CheckProductStockState));

• OrderProcessInitializedState: وضعیت پس از راه‌اندازی پردازش سفارش است.
• When(CheckProductStockEvent): زمانی که رویداد CheckProductStockEvent رخ دهد، ماشین وضعیت به CheckProductStockState تغییر می‌کند.

During(CheckProductStockState,
    When(TakePaymentEvent)
        .TransitionTo(TakePaymentState));

• CheckProductStockState: وضعیت‌ای است که در آن موجودی محصولات بررسی می‌شود.
• When(TakePaymentEvent): زمانی که رویداد TakePaymentEvent رخ دهد، وضعیت ماشین به TakePaymentState تغییر می‌کند.

During(TakePaymentState,
    When(CreateOrderEvent)
        .TransitionTo(CreateOrderState));

• TakePaymentState: وضعیت‌ای است که در آن پردازش پرداخت انجام می‌شود.
• When(CreateOrderEvent): زمانی که رویداد CreateOrderEvent رخ دهد، ماشین وضعیت به CreateOrderState تغییر می‌کند.

using MassTransit;
using StateMachineSample.Events;

namespace StateMachineSample.StateMachine;

public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public OrderStateMachine()
    {
        #region Fault-Companse State

        DuringAny(When(CreateOrderFaultEvent)
            .TransitionTo(CreateOrderFaultedState)
            .Then(context => context.Publish<Fault<TakePaymentEvent>>(new {context.Message})));


        DuringAny(When(TakePaymentEventFaultEvent)
            .TransitionTo(TakePaymentFaultedState)
            .Then(context => context.Publish<Fault<CheckProductStockEvent>>(new {context.Message})));

        DuringAny(When(CheckProductStockFaultEvent)
            .TransitionTo(CheckProductStockFaultedState)
            .Then(context => context.Publish<Fault<OrderProcessInitializationEvent>>(new {context.Message})));

        DuringAny(When(OrderProcessInitializationFaultEvent)
            .TransitionTo(OrderProcessInitializedFaultedState)
            .Then(context => context.Publish<OrderProcessFailedEvent>(new {OrderId = context.Saga.CorrelationId})));

        DuringAny(When(OrderProcessFailedEvent)
            .TransitionTo(OrderProcessFailedState));

        #endregion
    }
}

• CreateOrderFaultEvent: زمانی که خطا در ایجاد سفارش رخ می‌دهد، وضعیت به CreateOrderFaultedState تغییر می‌کند و سپس یک رویداد Fault<TakePaymentEvent> ارسال می‌شود.
• TakePaymentEventFaultEvent: زمانی که خطا در پرداخت رخ می‌دهد، وضعیت به TakePaymentFaultedState تغییر می‌کند و سپس یک رویداد Fault<CheckProductStockEvent> ارسال می‌شود.
• CheckProductStockFaultEvent: زمانی که خطا در بررسی موجودی محصول رخ می‌دهد، وضعیت به CheckProductStockFaultedState تغییر می‌کند و سپس یک رویداد Fault<OrderProcessInitializationEvent> ارسال می‌شود.
• OrderProcessInitializationFaultEvent: زمانی که خطا در فرآیند اولیه سفارش رخ می‌دهد، وضعیت به OrderProcessInitializedFaultedState تغییر می‌کند و سپس یک رویداد OrderProcessFailedEvent ارسال می‌شود که شامل OrderId است.
• OrderProcessFailedEvent: زمانی که فرآیند سفارش به شکست می‌انجامد، وضعیت به OrderProcessFailedState تغییر می‌کند.

using System.Runtime.CompilerServices;
using MassTransit;

namespace StateMachineSample.Events;

public class OrderProcessInitializationEvent
{
    public Guid OrderId { get; set; }
    
    [ModuleInitializer]
    internal static  void Init()
    {
        GlobalTopology.Send.UseCorrelationId<OrderProcessInitializationEvent>(x=> x.OrderId);
    }
}

• این کلاس نماینده یک رویداد (event) است که در یک سیستم مبتنی بر state machine یا معماری event-driven ارسال شود. این کلاس دارای پراپرتی OrderId است که یک شناسه (UUID) است که به طور منحصر به فرد، هر سفارش (Order) را مشخص می‌کند.

• این ویژگی (attribute) در فضای نام System.Runtime.CompilerServices معرفی شده و به شما اجازه می‌دهد کدی را که باید قبل از استفاده از کلاس یا ماژول اجرا شود، مشخص کنید.
• متد Init که با ویژگی ModuleInitializer تزئین شده است، به صورت خودکار هنگام بارگذاری ماژول فراخوانی می‌شود.

• این خط از کد در متد Init استفاده شده و مربوط به تنظیمات MassTransit برای استفاده از یک CorrelationId است.
• CorrelationId به MassTransit کمک می‌کند تا پیام‌ها را با یک شناسه مشترک مرتبط کند. در اینجا، پیام‌هایی که از نوع OrderProcessInitializationEvent هستند، بر اساس مقدار OrderId به هم مرتبط خواهند شد.

using MassTransit;
using StateMachineSample.StateMachine;
using StateMachineSample.API.Consumers;
using StateMachineSample.StateMachine.Settings;
using StateMachineSample.API.Settings;

var builder = WebApplication.CreateBuilder(args);

var messageBrokerQueueSettings = builder.Configuration.GetSection("MessageBroker:QueueSettings").Get<MessageBrokerQueueSettings>();
var messageBrokerPersistenceSettings = builder.Configuration.GetSection("MessageBroker:StateMachinePersistence").Get<MessageBrokerPersistenceSettings>();

builder.Services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<OrderStateMachine, OrderState>().MongoDbRepository(r =>
    {
        r.Connection = messageBrokerPersistenceSettings.Connection;
        r.DatabaseName = messageBrokerPersistenceSettings.DatabaseName;
        r.CollectionName = messageBrokerPersistenceSettings.CollectionName;
    });

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(messageBrokerQueueSettings.HostName, messageBrokerQueueSettings.VirtualHost, h =>
        {
            h.Username(messageBrokerQueueSettings.UserName);
            h.Password(messageBrokerQueueSettings.Password);
        });

        cfg.ConfigureEndpoints(context);
    });

    x.AddConsumer<OrderProcessInitializationEventConsumer>();
    x.AddRequestClient<OrderProcessInitializationEventConsumer>();

    x.AddConsumer<OrderProcessInitializationFaultEventConsumer>();
    x.AddRequestClient<OrderProcessInitializationFaultEventConsumer>();

    x.AddConsumer<CheckProductStockEventConsumer>();
    x.AddRequestClient<CheckProductStockEventConsumer>();

    x.AddConsumer<CheckProductStockFaultEventConsumer>();
    x.AddRequestClient<CheckProductStockFaultEventConsumer>();

    x.AddConsumer<TakePaymentEventConsumer>();
    x.AddRequestClient<TakePaymentEventConsumer>();

    x.AddConsumer<TakePaymentFaultEventConsumer>();
    x.AddRequestClient<TakePaymentFaultEventConsumer>();

    x.AddConsumer<CreateOrderEventConsumer>();
    x.AddRequestClient<CreateOrderEventConsumer>();

    x.AddConsumer<CreateOrderFaultEventConsumer>();
    x.AddRequestClient<CreateOrderFaultEventConsumer>();

    x.AddConsumer<OrderProcessFailedEventConsumer>();
    x.AddRequestClient<OrderProcessFailedEventConsumer>();
});

builder.Services.AddControllers();
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();

var app = builder.Build();

if (app.Environment.IsDevelopment())
{
    app.UseSwagger();
    app.UseSwaggerUI();
}

app.UseHttpsRedirection();
app.UseAuthorization();
app.MapControllers();

app.Run();

• پیکربندی‌های MessageBrokerQueueSettings و MessageBrokerPersistenceSettings از فایل‌های پیکربندی خوانده می‌شوند و برای اتصال به Message Broker و State Machine Persistence استفاده می‌شوند.
• سپس، MassTransit پیکربندی می‌شود.
• <AddSagaStateMachine<OrderStateMachine, OrderState(): این کد یک State Machine برای مدیریت وضعیت‌ها و تغییرات موجودیت OrderStateMachine (که نوعی از SagaStateMachine است) ثبت می‌کند.
• در بخش MongoDbRepository(r => { ... })، اطلاعات مربوط به MongoDB برای ذخیره وضعیت‌های State Machine تنظیم می‌شود (اتصال به پایگاه داده، نام پایگاه داده و مجموعه مربوط به وضعیت‌ها).
• UsingRabbitMq: این بخش، پیکربندی مربوط به RabbitMQ (Message Broker) را انجام می‌دهد.
• cfg.Host: آدرس سرور RabbitMQ و اطلاعات مربوط به اتصال مانند نام کاربری و رمز عبور را تنظیم می‌کند.
• cfg.ConfigureEndpoints: این بخش به MassTransit می‌گوید که نقاط پایانی RabbitMQ را پیکربندی کند.
• Consumers: در این بخش، چندین Consumer به MassTransit اضافه می‌شود که مسئولیت دریافت رویدادها (Events) و درخواست‌های مختلف مانند شروع پردازش سفارش، بررسی موجودی کالا، انجام پرداخت و غیره را بر عهده دارند.
• RequestClient: برای هر مصرف‌کننده، RequestClient مربوطه نیز اضافه می‌شود که به سیستم امکان ارسال درخواست‌های همزمان به سایر سرویس‌ها را می‌دهد.

using MassTransit;
using Microsoft.AspNetCore.Mvc;
using StateMachineSample.API.Requests;
using StateMachineSample.Events;
using StateMachineSample.Events.Responses;

namespace StateMachineSample.API.Controllers;

public class EventsController
{
    private readonly IPublishEndpoint _publishEndpoint;
    private readonly IRequestClient<OrderProcessInitializationEvent> _orderProcessInitializationEventRequestClient;
    private readonly IRequestClient<CheckProductStockEvent> _checkProductStockEventRequestClient;
    private readonly IRequestClient<TakePaymentEvent> _takePaymentEventRequestClient;
    private readonly IRequestClient<CreateOrderEvent> _createOrderEventRequestClient;
        
    public EventsController(
        IPublishEndpoint publishEndpoint,
        IRequestClient<OrderProcessInitializationEvent> orderProcessInitializationEventRequestClient,
        IRequestClient<CheckProductStockEvent> checkProductStockEventRequestClient,
        IRequestClient<TakePaymentEvent> takePaymentEventRequestClient,
        IRequestClient<CreateOrderEvent> createOrderEventRequestClient)
    {
        _publishEndpoint = publishEndpoint;
        _orderProcessInitializationEventRequestClient = orderProcessInitializationEventRequestClient;
        _checkProductStockEventRequestClient = checkProductStockEventRequestClient;
        _takePaymentEventRequestClient = takePaymentEventRequestClient;
        _createOrderEventRequestClient = createOrderEventRequestClient;
    }
}

• برای استفاده از MassTransit در این کنترلر، از چندین وابستگی (dependency) استفاده شده است که به‌طور خاص برای ارسال و دریافت رویدادها و درخواست‌ها استفاده می شوند.
• IPublishEndpoint: این وابستگی برای ارسال پیام‌ها به سیستم‌های دیگر استفاده می‌شود. به عبارتی، از این سرویس برای انتشار رویدادها به‌طور عمومی استفاده می‌شود تا سایر سیستم‌ها از آن اطلاع پیدا کنند.
• <IRequestClient<T: این وابستگی برای ارسال درخواست‌ها به سیستم‌های دیگر و دریافت پاسخ استفاده می‌شود. در اینجا به‌طور خاص از آن برای ارسال رویدادهای مختلف مانند OrderProcessInitializationEvent، CheckProductStockEvent و غیره استفاده شده است. این روش به شما این امکان را می‌دهد که منتظر پاسخ سرویس مقصد باشید و بتوانید عملیات بعدی را انجام دهید.

using MassTransit;
using Microsoft.AspNetCore.Mvc;
using StateMachineSample.API.Requests;
using StateMachineSample.Events;
using StateMachineSample.Events.Responses;

namespace StateMachineSample.API.Controllers;

public class EventsController
{

    [HttpPost("initialize/order")]
    public async Task<IActionResult> OrderProcessInitializedEvent([FromBody] EventCommonRequest request)
    {
        if (!request.IsRequestResponsePattern)
        {
            await _publishEndpoint.Publish<OrderProcessInitializationEvent>(new {request.OrderId});
            return new NoContentResult();
        }

        var result = await _orderProcessInitializationEventRequestClient.GetResponse<OrderProcessInitiazationDto>(new {request.OrderId});

        return new NoContentResult();
    }
}

using MassTransit;

namespace StateMachineSample.API.Consumers.Base;

public abstract class ConsumerBase<T> : IConsumer<T> where T : class
{
    public async Task Consume(ConsumeContext<T> context)
    {
        try
        {
            await ConsumeInternal(context);
        }
        catch (Exception e)
        {
            await context.Publish<Fault<T>>(context);
            
            // global exception handling
            throw;
        }
    }

    protected abstract Task ConsumeInternal(ConsumeContext<T> context);
}

• متد Consume مسئول پردازش پیام‌ها است و در صورت بروز هر گونه استثنا، ایونت خطا (<Fault<T) منتشر می‌شود.
• متد ConsumeInternal که انتزاعی است، در کلاس‌های فرزند پیاده‌سازی خواهد شد و منطق خاص برای هر نوع ایونت در آن قرار می‌گیرد.

using MassTransit;
using StateMachineSample.API.Consumers.Base;
using StateMachineSample.Events;
using StateMachineSample.Events.Responses;

namespace StateMachineSample.API.Consumers;

public class OrderProcessInitializationEventConsumer : ConsumerBase<OrderProcessInitializationEvent>
{
    protected override Task ConsumeInternal(ConsumeContext<OrderProcessInitializationEvent> context)
    {
        Console.WriteLine("Order process Initialized");

        context.RespondAsync(new OrderProcessInitiazationDto
        {
            OrderId = context.Message.OrderId
        });
        
        return Task.CompletedTask;
    }
}

• منطق خاص برای ایونت OrderProcessInitializationEvent در متد ConsumeInternal پیاده‌سازی شده است.
• پس از پردازش ایونت، با استفاده از متد RespondAsync، یک شیء پاسخ به ناشر ارسال می‌شود (اگر از الگوی درخواست/پاسخ استفاده می‌کنیم).