MassTransit, Saga и RabbitMQ для реализации диспетчера процессов

Что из себя представляет Saga?

Saga — это реализация шаблона "Диспетчер процессов" из книги «Шаблоны интеграции корпоративных приложений», который позволяет описать процесс в виде конечного автомата. На вход прилетает какое-то событие, Saga выполняет последовательность действий. При этом на любом из этапов Saga может потребоваться решение человека. Тогда она создает задачу в трекере и «засыпает» на неопределённое время, ожидая новых событий. Saga реализована на базе Automatonymous. Она декларативно описывается в классе, унаследованном от MassTransitStateMachine<>. Для Saga нужно описать все состояния, принимаемые события и совершаемые действия при наступлении определенных событий. Текущее состояние сохраняется в базе.

Для начала описываем все состояния и события Saga и даём им понятные имена. Выглядит это следующим образом:

public sealed partial class StateMachine
{
	public State AwaitingTaskCreated { get; set; }
	public State AwaitingTaskTakedToWork { get; set; }
	public State AwaitingDecisionAboutTask { get; set; }
	public State Rejected { get; set; }

	public Event StartWorkflowCommandReceived { get; set; }
	public Event TaskCreated { get; set; }
	public Event TaskTakedToWork { get; set; }
	public Event TaskDeclined { get; set; }
	public Event TaskApproved { get; set; }

	private void BuildStateMachine()
	{
		InstanceState(x => x.CurrentState);
		Event(() => StartWorkflowCommandReceived, x => x.CorrelateById(ctx => 
                        ctx.Message.CorrelationId)
			.SelectId(context => context.Message.CorrelationId));
		Event(() => TaskCreated, x => x.CorrelateById(ctx => 
                        ctx.Message.CorrelationId));
		Event(() => TaskTakedToWork, x => x.CorrelateById(ctx => 
                        ctx.Message.CorrelationId));
		Event(() => TaskDeclined, x => x.CorrelateById(ctx => 
                        ctx.Message.CorrelationId));
		Event(() => TaskApproved, x => x.CorrelateById(ctx => 
                        ctx.Message.CorrelationId));
	}
}

Мы завели partial class, где декларируем списком все состояния и ивенты, и метод BuildStateMachine, в котором описывается корреляция ивентов с Saga. Для этого в каждом ивенте передается специальный параметр CorrelationId — это Guid, который курсирует между всеми связанными системами и в системах мониторинга.

Таким образом, при возникновении каких-либо проблем мы можем восстановить всю картину происходящего по логам из всех связанных систем. Мы отправляем CorrelationId в сообщениях из Saga, его же отправляют системы обратно в нотификациях, чтобы мы могли соотнести сообщение с конкретной Saga.

А вот пример самого класса стейт машины:

public sealed partial class StateMachine : MassTransitStateMachine
{
	public StateMachine()
	{
		BuildStateMachine();
		Initially(WhenStartWorkflowCommandReceived());
		During(AwaitingTaskCreatedInPlanner, WhenTaskCreated());
		During(AwaitingTaskTakedToWork, WhenTaskTakedToWork());
		During(AwaitingDecisionAboutTask,
			WhenTaskApproved(),
			WhenTaskDeclined());
	}

	private EventActivityBinder
                  WhenStartWorkflowCommandReceived()
	{
		return When(StartWorkflowCommandReceived)
			.Then(ctx => ctx.Instance.SaveConfigurationRequestInfo(ctx.Data))
			.Send(TaskManagerQueue, ctx => new CreateTaskCommand(ctx.Instance))
			.TransitionTo(AwaitingTaskCreated);
	}

	private EventActivityBinder 
                   WhenTaskCreated()
	{
		return When(DPORPApproveTaskCreatedInPlanner)
			.Then(ctx => ctx.Instance.SaveCreatedTaskInfo(ctx.Data))
			.Send(MailServiceQueue, ctx => new
                        NotifyRequestAuthorThatWorkflowStarted(ctx.Instance))
			.TransitionTo(AwaitingTaskTakedToWork);
	}

	private EventActivityBinder 
                   WhenTaskTakedToWork()
	{
		return When(TaskTakedToWork)
			.Then(ctx => ctx.Instance.MarkTaskAsTakedToWork(ctx.Data))
			.TransitionTo(AwaitingDecisionAboutTask);
	}

	private EventActivityBinder 
                   WhenTaskApproved()
	{
		return When(TaskApproved)
			.Then(ctx => ctx.Instance.MarkTaskAsApproved(ctx.Data))
			.Finalize();
	}

	private EventActivityBinder 
                   WhenTaskDeclined()
	{
		return When(TaskDeclined)
			.Then(ctx => ctx.Instance.MarkTaskAsDeclined(ctx.Data))
			.TransitionTo(Rejected);
	}
}

В конструкторе описываются состояния. Прием каждого ивента вынесен в отдельный метод, чтобы сохранить читаемость. Вся логика конструирования сообщений вынесена в сами сообщения, иначе с возрастанием сложности системы Saga довольно быстро распухает.

Стоит внимательно отнестись к выработке конвенций и сохранению читабельности. Из-за императивности C# в нем очень сложно декларировать описание состояний и действий. Даже для простеньких машин состояний начинается настоящий ад.

Теперь несколько слов про SagaInstance. SagaInstance — это класс, унаследованный от SagaStateMachineInstance. Он состоит из объектов и полей, которые характеризуют машину состояний. Грубо говоря, это память Saga. Мы храним там все данные Saga, которые ей понадобятся на протяжении всей жизни. Также в этом классе описана логика изменений этих данных по ходу работы.

Приведём пример:

public class WorkflowSaga 
: SagaStateMachineInstance
, ISagaWithState
, ICreatedOnOffset
, IModifiedOnOffset
, ICreatedBy
, IModifiedBy
{
	public Guid CorrelationId { get; set; }
	public string CurrentState { get; set; }
	public string InitialRequestViewUrl { get; set; }
	public string RequestNumber { get; set; }
	public string RequestAuthor { get; set; }
	public string RequestText { get; set; }
	public byte[] RowVersion { get; set; }
	public string CreatedBy { get; set; }
	public string ModifiedBy { get; set; }
	public DateTimeOffset CreatedOn { get; set; }
	public DateTimeOffset ModifiedOn { get; set; }
	public DateTimeOffset CompletedOn { get; set; }

	public virtual ICollection RelatedTasks { get; set; }

	public void SaveGabrielConfigurationRequestInfo(
        ICreateGabrielConfigurationRequestCommand command)
	{
		CorrelationId = command.CorrelationId;
		RequestNumber = command.RequestNumber;
		RequestAuthor = command.Author;
		RequestText = command.RequestText;
		InitialRequestViewUrl = command.InitialRequestViewUrl;
		CreatedOn = RuntimeContext.Current.DateTimeOffset.Now;
	}

	public void SaveCreatedTaskInfo(ITaskCreationInfo taskCreationInfo)
	{
		RelatedPlannerTasks.Add(new RelatedPlannerTask(taskCreationInfo));
	}

	public void MarkTaskAsTakedToWork(ITaskUpdatedInfo taskInfo)
	{
		UpdateTaskInfo(taskInfo, TaskStatus.TakedToWork);
	}

	public void MarkTaskAsApproved(TaskApprovedNotification taskInfo)
	{
		UpdateTaskInfo(taskInfo, TaskStatus.Completed, taskInfo.Comment);
		CompletedOn = RuntimeContext.Current.DateTimeOffset.Now;
	}

	public void MarkTaskAsDeclined(TaskDeclinedNotification taskInfo)
	{
        UpdateTaskInfo(taskInfo, TaskStatus.Declined, taskInfo.Comment);
		CompletedOn = RuntimeContext.Current.DateTimeOffset.Now;
	}

	private void UpdateTaskInfo(ITaskUpdatedInfo taskInfo, 
    TaskStatus taskStatus, string comment = null)
	{
		var task = RelatedTasks.Single(t => t.Number == taskInfo.Number);
		task.ModifiedBy = taskInfo.TaskModifiedBy;
		task.Comment = comment;
		task.Status = taskStatus;
	}
}

  

На примере видно, что в SagaInstance сохраняется CorrelationId для корреляции ивентов с Saga и CurrentState для хранения текущего состояния Saga.

[TestFixture]
public class SagaTests : TestFixtureBase
{
	protected const string HostName = "HostName";

	protected InMemoryTestHarness Harness;
	protected StateMachine StateMachine;
	protected StateMachineSagaTestHarness Saga;

	[SetUp]
	public void SetUp()
	{
		StateMachine = (StateMachine)Kernel.
        Get>();

		Harness = new InMemoryTestHarness(HostName);
		Saga = Harness
        .StateMachineSaga(StateMachine);
	}

	[TearDown]
	public async Task TearDown()
	{
		await Harness.Stop();
	}


	protected async Task InitializeSaga()
	{
		await Harness.Start();

		var command = new TestStartWorkflowCommand
		{
			CorrelationId = SagaId,
			Author = RequestAuthor,
			InitialRequestViewUrl = InitialRequestViewUrl,
			RequestText = RequestText,
			RequestNumber = RequestNumber,
		};

		await Harness.InputQueueSendEndpoint
        .Send(command);
		// Эта строчка нам нужна, поскольку consume срабатывает не сразу и, 
		// соответственно, и Saga не будет, пока не законсюмим
		Assert.IsTrue(Harness.Consumed
        .Select().Any());

		var currentSaga = Saga.Created.Contains(SagaId);
		currentSaga.RelatedPlannerTasks = new List();
		return currentSaga;
	}

	[Test]
	public async Task CheckCurrntStateWhenStartWorkflowCommand()
	{
		var saga = await InitializeSaga();

		Assert.IsNotNull(saga);
		Assert.AreEqual(StateMachine
        .AwaitingORDTApproveTaskCreatedInPlanner.Name, saga.CurrentState);
	}
}

public class WhenTaskCreated : SagaTestsBase
{
	private async Task InitializeState()
	{
		var saga = await InitializeSaga(true);

		saga.CurrentState = StateMachine.AwaitingTaskCreated.Name;
		InitializeRelatedTask(saga);

		await SendTaskCreatedNotification();

		Assert.IsTrue(Harness.Consumed
        .Select().Any());
		return saga;
	}

	[Test]
	public async Task SaveWorkflowDataWhenTaskCreated()
	{
		var saga = await InitializeState();

		var taskInfo = saga.RelatedPlannerTasks
        .First(task => 
        task.PlannerTaskType == PlannerTaskType.DPORPApprove);

		Assert.AreEqual(TaskNumber, taskInfo.Number);
		Assert.AreEqual(TaskUrl, taskInfo.TaskUrl);
		Assert.AreEqual(SagaId, taskInfo.SagaCorrelationId);
		Assert.AreEqual(TaskStatus.Created, taskInfo.Status);
		Assert.AreEqual(User, taskInfo.ModifiedBy);

		Assert.AreEqual(saga.CurrentState, 
        StateMachine.AwaitingTaskTakedToWork.Name);
	}


	[Test]
	public async Task SendsMailWhenTaskCreated()
	{
		var mailConsumer =	Harness
            .Consumer>
            (RabbitMqRouting.QueueNames
				.SendEmailsQueueName);

		await InitializeState();

		Assert.IsTrue(mailConsumer.Consumed
        .Select().Any());
	}

	private async Task SendTaskCreatedNotification()
	{
		await Harness.InputQueueSendEndpoint
        .Send(new TaskCreatedNotification
		{
			TaskUrl = TaskUrl,
			Number = TaskNumber,
			TaskModifiedBy = User,
			CorrelationId = SagaId
		});
	}
}

В заключении приводим список полезных советов, основанных на нашем опыте.

1. Контракты и схемы общения через шину лучше всего поместить в приватный nuget. Так у вас не будет различий в именованиях на отправляющей и принимающей сторонах. Также в nuget можно поместить константы с именованием очередей и хостов. Nuget настраивается за день. А также некоторые source controls поддерживают nuget, есть платные частные фиды.

2. Разберитесь в различиях между Send и Publish. Используйте Send, если у вас один подписчик и вы точно знаете имя очереди, в которую отправляете команду. Publish предназначен для отправки оповещений broadcast-ом. Подробности по ссылке.

3. Если вам нужно построить Request/Response сообщение, то лучше добавить в контракт имя очереди для ответа, чем использовать схему Request/Response от MassTransit, которую и сам MassTransit предлагает избегать. Так как это сильно уменьшает надежность. Вы теряете все преимущества асинхронности. Но если вам всё же нужно получить ответ в ограниченное время, то лучше использовать прямой вызов. Лучше всего об этом написано во всё той же книге «Шаблоны интеграции корпоративных приложений».

4. Saga должна быть тонкой. Попытайтесь унести всю тяжелую логику в другие системы. А Saga должна скакать по состояниям и разбрасывать сообщения налево и направо.

5. Добавьте во все сообщения CorrelationId, который будет курсировать между системами. Так намного проще потом анализировать логи и связывать все сообщения в единую картину. Также поступает и сам Masstransit. CorrelationId добавляется в сообщения при наследовании от интерфейса CorrelatedBy.

Настройте логи и мониторинг в своих системах, это никогда не повредит. 

Оригинал опубликован на Habr.com