我如何将 MassTransit 传奇子步骤与替代拓扑集成？

如何解决我如何将 MassTransit 传奇子步骤与替代拓扑集成？

我有一个具有以下状态的 MassTransit 传奇：

Initial -> QueryRequiredData -> CalibrateData -> PerformValuation -> PersistToContext -> RespondAndFinalize

我已经为大多数这些状态实施了活动，并且可以使用默认拓扑。但是，PerformValuation 状态的活动应该通过发布/订阅模式与外部拓扑集成，在该模式中我们使用来自外部系统的响应并对其采取行动。命令消息需要使用protobuf进行序列化。

使用准系统 RabbitMq，发送代码如下所示：

public void PublishRequest(Google.Protobuf.IMessage request,Guid contextId)
{
    using var connection = _connectionFactory.Create();
    using var model = connection.CreateModel();
    var basicProperties = model.CreateBasicProperties();
    basicProperties.Headers = new Dictionary<string,object>
    {
        {"service","my.external.ValuationService"},{"endpoint","PerformValuation"},{"correlationId",contextId.ToString()},{"reply-to","PerformValuation.Response"},{"reply-to-routing-key","PerformValuationResponseMessage"}
    };
    model.BasicPublish("quant.calc","PerformValuationRequestMessage",false,basicProperties,request.ToByteArray());
}

订阅者代码如下所示：

public Subscriber()
{
    _exchangeName = "PerformValuation.Response";
    _queueName = "PerformValuationResponseMessage";
}

public void StartSubscription()
{
    ...
    try
    {
        var connection = _connectionFactory.Create();
        var model = connection.CreateModel();
        CreateTopology(model);
        connection.AutoClose = true;
        _subscription = new Subscription(model,_queueName,false);
        _subscription.Model.BasicQos(0,(ushort) 5,false);
        EnterMessageProcessingLoop();
    }
    ...
}

private void CreateTopology(IModel model)
{
    model.ExchangeDeclare(_exchangeName,"topic");
    model.QueueDeclare(_queueName,true,new Dictionary<string,object>());
    model.QueueBind(_queueName,_exchangeName,"*",object>
    {
        {"x-match","all"}
    });
}

如何将 MassTransit 配置为以类似方式运行？