Как правильно настроить деятельность в саге о состоянии машин Masstransit?

Question

Я пару дней много боролся с этой проблемой, но не могу найти способ, который действительно работает.

Мой сценарий таков: у меня есть сага о конечном автомате, где я хочу что-то делатьв той же транзакции (Entity Framework), в которой запускается сага, так что дела и состояние объединяются.

Теперь я понял, что сам конечный автомат не должен иметь никаких зависимостей, и, следовательно,есть это Activity(x => x.OfInstanceType<MyActivity>) действие, которое можно добавить, которое разрешается из контейнера DI и может иметь любые зависимости (службы и т. д.) в нем.Пока все хорошо ...

Моя проблема в том, что, что бы я ни выполнял в упражнении, я просто не могу заставить его работать так, как должно.Он разрешается из контейнера, и вызывается метод Execute, но затем он как-то просто завершается.
Похоже, что выдается какое-то исключение, но это, очевидно, не доходит до тестового жгута.

Я использую библиотеку Dependency Injection от Microsoft.

Вот код

public class MyStateMachine : MassTransitStateMachine<SagaInstance>
{
    public MyStateMachine()
    {
        InstanceState(instance => instance.CurrentState);

        Event(() => Start, x => x.CorrelateBy(saga => saga.CorrelationId, context => context.Message.CorrelationId));

        Initially(
            When(Start)
                .Activity(c => c.OfType<MyActivity>())
                .TransitionTo(Running)
                .Publish(new Started())
        );
    }   

    Event<Run> Start { get; set;} 
    State Running { get; set; }
}

public class MyActivity : Activity<SagaInstance, Run>
{
    private readonly IMyService _service;
    public MyActivity(IMyService service)
    {
        _service = service;
    }

    public async Task Execute(BehaviorContext<SagaInstance, Run> context, Behavior<SagaInstance, Run> next)
    {   
        //throw new Exception("BOO"); // uncommenting this line doesn't throw anywhere
        _service.DoThatThing();
        await next.Execute(context).ConfigureAwait(false);
    }
}

[Fact]
public async Task RunMessageSent_NonexistingSaga_StateIsRunning()
{
    var harness = new InMemoryTestHarness();
    var machine = new MyStateMachine();

    var collection = new ServiceCollection();
    collection.AddMassTransit();
    collection.RegisterInMemorySagaRepository<SagaInstance>();
    collection.RegisterSagaStateMachine<MyStateMachine, SagaInstance>();
    collection.AddScoped<MyActivity>();
    collection.AddScoped<MyService>();

    var provider = collection.BuildServiceProvider();

    harness.OnConfigureInMemoryReceiveEndpoint += cfg => cfg.StateMachineSaga(machine, provider);

    await harness.Start();

    try
    {
        var guid = CorrelationId = Guid.NewGuid();
        await harness.InputQueueSendEndpoint.Send(new Run({CorrelationId = guid}));

        var repo = provider.GetService<ISagaRepository<SagaInstance>>() as InMemorySagaRepository<SagaInstance>;
        var saga = await repo.ShouldContainSaga(s => s.Serial == serial, timeout: TimeSpan.FromSeconds(1));

        Assert.NotNull(saga);
        // this Equal fails saying the saga is in state Initial
        Assert.Equal(machine.Running.Name, repo[saga.Value].Instance.CurrentState);
    }
    finally
    {
        await harness.Stop();
    }   
}

Answer 1

Я работал на dotnet core 2.1, и после долгих поисков в Google репозитория Masstransit я понял, что в последней (5.3) Masstransit есть более совершенные API для инфраструктуры Dependency Injection, которые я не хотел обновлять чтобы сначала избежать необходимости обновлять зависимость времени выполнения dotnet до 2.2.

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

.

Так что, используя эту строку, я получил ее на работу!

var saga = await repo.ShouldContainSagaInState(guid, machine, x => x.Running, TimeSpan.FromSeconds(1));

Кроме того, при выполнении кода так, как я намеревался сделать, используя шаблон Запрос / Ответ, потенциальные исключения отправляются вызывающим абонентам в виде сообщений об ошибках. В тестовом жгуте это сообщение каким-то образом доступно, но я пока не нашел правильного способа проверить его.