MassTransit消费者传奇能由它所策划的相同信息发起吗？

Question

7.0中对Event Hub Riders的新支持，加上现有的InmemoryRepository对Sagas的支持，似乎可以提供一种基于相关消息流（例如，在建筑物中的所有传感器之间）创建聚合状态的简单方法。在这个场景中，建筑物的标识符将用作消息的CorrelationId、Saga和发送到事件集线器的EventData消息的PartitionKey，以确保同一消费服务实例在给定时间接收该设备的所有消息。根据Event Hub重新平衡的工作方式，可以假设在该服务运行的某个时刻，管理分区消息的服务实例将转移到一个新主机，该主机将开始读取建筑物中传感器发送的消息。那一刻：

null

这带来的挑战是：在新的服务实例上，我们需要创建一个新的Saga来接管以前的Saga，但唯一知道给定实体不存在Saga的是masstransit：新实例上的任何东西都不知道从构建a读取的传感器是自该服务html" target="_blank">实例接管跟踪聚合构建a状态以来从构建a读取的第一个传感器。我们认为这可以通过在同一消息(datacollected)中同时标记initiatedby和courrates)来处理：

public class BuildingAggregator:
            ISaga,
            InitiatedBy<DataCollected>, //init saga on first DataCollected with a given CorrelationId seen
            Orchestrates<DataCollected> //then keep handling those in that saga
{
     //saga Consume methods
}

但是，当BuildingAggregator收到具有给定GUID的第二条DataCollected消息时，会引发以下异常：

Saga exception on receipt of MassTransitFW_POC.Program+DataCollected: The message cannot be accepted by an existing saga
   at MassTransit.Saga.Policies.NewSagaPolicy`2.MassTransit.Saga.ISagaPolicy<TSaga,TMessage>.Existing(SagaConsumeContext`2 context, IPipe`1 next)
   at MassTransit.Saga.SendSagaPipe`2.<Send>d__5.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
   at MassTransit.Saga.SendSagaPipe`2.<Send>d__5.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at MassTransit.Saga.InMemoryRepository.InMemorySagaRepositoryContextFactory`1.<Send>d__4`1.MoveNext()
--- End of stack trace from previous location where exception was thrown ---

有没有另一种实现这一逻辑的方法？这是应用传奇的“错误方式”吗？

Answer 1

根据Chris Patterson对上述问题的评论，这是可以通过状态机语法实现的：

Initially(
    When(DataCollected)
        .Then(f => _logger.LogInformation("Initiating Network Manager for Network: {NetworkId}", f.Data.NetworkId))
        .TransitionTo(Running));
During(Running,
    When(DataCollected)
        .Then(f => { // activities and state transitions }),
    When(SimulationComplete)
        .Then(f => _logger.LogInformation("Network {NetworkId} shutting down.", f.Instance.CorrelationId))
        .TransitionTo(Final));

请注意在初始状态转换和由初始条件设置的状态转换中如何处理DataCollected事件。