如何通过 InMemoryTestHarness 成功驱动 MassTransitStateMachine？

跟进：如何编写 MassTransitStateMachine 单元测试？ https://stackoverflow.com/questions/49763600/how-to-write-masstransitstatemachine-unit-tests

这是一个简单的测试类（使用 MS Test），用于一个名为的简单状态机ProcedureStateMachine（注意：这对我们来说不是一个真正的生产状态机......只是我曾经玩过的一个实验MassTransitStateMachine不久前..这似乎是一个方便的独立场所，可以尝试进行单元测试）：

[TestClass]
public class ProcedureStateMachineTests
{
    private ProcedureStateMachine _machine;
    private InMemoryTestHarness _harness;
    private StateMachineSagaTestHarness<ProcedureContext, ProcedureStateMachine> _saga;

    [TestInitialize]
    public void SetUp()
    {
        _machine = new ProcedureStateMachine();
        _harness = new InMemoryTestHarness();
        _saga = _harness.StateMachineSaga<ProcedureContext, ProcedureStateMachine>(_machine);

        _harness.Start().Wait();
    }

    [TestCleanup]
    public void TearDown()
    {
        _harness.Stop().Wait();
    }

    [TestMethod]
    public async Task That_Can_Start()
    {
        // Arrange
        // Act
        await _harness.InputQueueSendEndpoint.Send(new BeginProcessing
        {
            ProcedureId = Guid.NewGuid(),
            Steps = new List<string> {"A", "B", "C" }
        });

        // Assert
        var sagaContext = _saga.Created.First();
        sagaContext.Saga.RemainingSteps.ShouldHaveCountOf(2);
    }
}

这是状态机类本身：

public class ProcedureStateMachine : MassTransitStateMachine<ProcedureContext>
{
    public State Processing { get; private set; }
    public State Cancelling { get; private set; }
    public State CompleteOk { get; private set; }
    public State CompleteError { get; private set; }
    public State CompleteCancelled { get; private set; }

    public Event<BeginProcessing> Begin { get; private set; }
    public Event<StepCompleted> StepDone { get; private set; }
    public Event<CancelProcessing> Cancel { get; private set; }
    public Event<FinalizeProcessing> Finalize { get; private set; }

    public ProcedureStateMachine()
    {
        InstanceState(x => x.CurrentState);

        Event(() => Begin);
        Event(() => StepDone);
        Event(() => Cancel);
        Event(() => Finalize);

        BeforeEnterAny(binder => binder
            .ThenAsync(context => Console.Out.WriteLineAsync(
                $"ENTERING STATE [{context.Data.Name}]")));

        Initially(
            When(Begin)
                .Then(context =>
                {
                    context.Instance.RemainingSteps = new Queue<string>(context.Data.Steps);
                })
                .ThenAsync(context => Console.Out.WriteLineAsync(
                    $"EVENT [{nameof(Begin)}]: Procedure [{context.Data.ProcedureId}] Steps [{string.Join(",", context.Data.Steps)}]"))
                .Publish(context => new ExecuteStep
                {
                    ProcedureId = context.Instance.CorrelationId,
                    StepId = context.Instance.RemainingSteps.Dequeue()
                })
                .Publish(context => new SomeFunMessage
                {
                    CorrelationId = context.Data.CorrelationId,
                    TheMessage = $"Procedure [{context.Data.CorrelationId} has begun..."
                })
                .TransitionTo(Processing)
            );

        During(Processing,
            When(StepDone)
                .Then(context =>
                {
                    if (null == context.Instance.AccumulatedResults)
                    {
                        context.Instance.AccumulatedResults = new List<StepResult>();
                    }
                    context.Instance.AccumulatedResults.Add(
                        new StepResult
                        {
                            CorrelationId = context.Instance.CorrelationId,
                            StepId = context.Data.StepId,
                            WhatHappened = context.Data.WhatHappened
                        });
                })
                .ThenAsync(context => Console.Out.WriteLineAsync(
                    $"EVENT [{nameof(StepDone)}]: Procedure [{context.Data.ProcedureId}] Step [{context.Data.StepId}] Result [{context.Data.WhatHappened}] RemainingSteps [{string.Join(",", context.Instance.RemainingSteps)}]"))
                .If(context => !context.Instance.RemainingSteps.Any(),
                    binder => binder.TransitionTo(CompleteOk))
                .If(context => context.Instance.RemainingSteps.Any(),
                    binder => binder.Publish(context => new ExecuteStep
                    {
                        ProcedureId = context.Instance.CorrelationId,
                        StepId = context.Instance.RemainingSteps.Dequeue()
                    })),
            When(Cancel)
                .Then(context =>
                {
                    context.Instance.RemainingSteps.Clear();
                })
                .ThenAsync(context => Console.Out.WriteLineAsync(
                    $"EVENT [{nameof(Cancel)}]: Procedure [{context.Data.ProcedureId}] will be cancelled with following steps remaining [{string.Join(",", context.Instance.RemainingSteps)}]"))
                .TransitionTo(Cancelling)
            );

        During(Cancelling,
            When(StepDone)
                .Then(context =>
                {
                    context.Instance.SomeStringValue = "Booo... we cancelled...";
                })
                .ThenAsync(context => Console.Out.WriteLineAsync(
                    $"EVENT [{nameof(StepDone)}]: Procedure [{context.Data.ProcedureId}] Step [{context.Data.StepId}] completed while cancelling."))
                .TransitionTo(CompleteCancelled));

        During(CompleteOk, When(Finalize).Finalize());
        During(CompleteCancelled, When(Finalize).Finalize());
        During(CompleteError, When(Finalize).Finalize());

        // The "SetCompleted*" thing is what triggers purging of the state context info from the store (eg. Redis)...  without that, the 
        // old completed state keys will gradually accumulate and dominate the Redis store.
        SetCompletedWhenFinalized();
    }
}

调试此测试时，_harness有BeginProcessing消息中的Sent集合，但是里面什么都没有_saga.Created收藏。看来我缺少一些管道，导致发送消息时线束实际驱动状态机？

====

删除.Wait()来电来自SetUp() and TearDown()并将测试更新为以下内容不会改变行为：

    [TestMethod]
    public async Task That_Can_Start()
    {
        try
        {
            await _harness.Start();
            // Arrange

            // Act
            await _harness.InputQueueSendEndpoint.Send(new BeginProcessing
            {
                ProcedureId = Guid.NewGuid(),
                Steps = new List<string> {"A", "B", "C"}
            });

            // Assert
            var sagaContext = _saga.Created.First();
            sagaContext.Saga.RemainingSteps.ShouldHaveCountOf(3);
        }
        finally
        {
            await _harness.Stop();
        }
    }

事实证明，如上所示的测试代码受到了之间的竞争条件的影响。_harness.InputQueueSendEndpoint.Send操作和一些异步（超出了等待的范围）Send等待）中的行为StateMachineSagaTestHarness。因此，测试代码的“断言”阶段在创建传奇并允许处理发送的消息之前执行。

深入挖掘SagaTestHarness编写了一些代码后，我发现了一些辅助方法，我可以使用它们来等待传奇的某些条件得到满足。方法有：

/// <summary>
/// Waits until a saga exists with the specified correlationId
/// </summary>
/// <param name="sagaId"></param>
/// <param name="timeout"></param>
/// <returns></returns>
public async Task<Guid?> Exists(Guid sagaId, TimeSpan? timeout = null)

/// <summary>
/// Waits until at least one saga exists matching the specified filter
/// </summary>
/// <param name="filter"></param>
/// <param name="timeout"></param>
/// <returns></returns>
public async Task<IList<Guid>> Match(Expression<Func<TSaga, bool>> filter, TimeSpan? timeout = null)

/// <summary>
/// Waits until the saga matching the specified correlationId does NOT exist
/// </summary>
/// <param name="sagaId"></param>
/// <param name="timeout"></param>
/// <returns></returns>
public async Task<Guid?> NotExists(Guid sagaId, TimeSpan? timeout = null)

所以我决定使用类似的东西await _saga.Match(s => null != s.RemainingSteps);这样可以有效地复制我后来的断言，并等待直到超时（默认为 30 秒）或后来断言的条件变为真（因此可以安全地断言）..以先到者为准。

这会让我暂时摆脱困境，直到我能想出更好的方法来知道安全带何时“被抓住”并准备好接受审讯。