rebus - 使用 Rebus 将错误消息从错误队列移回原始队列

Question

是否可以以编程方式或通过 UI 将错误消息从错误队列移动到其原始队列？

更新

关于下面代码的问题:

1 以下代码是否适用于发布者或订阅者或两者？

下面的代码:

 Configure.With(activator)
    .Transport(t => (...)) //< use queue "error" here
    .Routing(r =>
    {
        r.AddTransportMessageForwarder(async transportMessage =>
        {
            var sourceQueue = transportMessage.Headers.TryGetValue(Headers.SourceQueue, out var result)
                ? result
                : throw new ArgumentException($"Could not find '{Headers.SourceQueue}' header");

            return ForwardAction.ForwardTo(sourceQueue);
        });
    })
    .Start();

2 Transport下面的方法适用于我的代码。但是，上面的代码建议使用 error队列名称，它会起作用吗？

如果使用上面的代码，下面指定的发布者和订阅者队列名称在哪里？

请提供 pub sub 模式的代码。

出版商:

.Transport(t => t.UseAzureServiceBus(Consts.ServiceBusConnectionString, Consts.Publisher))

订户:

.Transport(t=>t.UseAzureServiceBus(Consts.ServiceBusConnectionString, Consts.Subscriber1))

https://github.com/rebus-org/Rebus/wiki/Transport-message-forwarding

Answer 1

由于 Rebus 使用普通队列作为其死信队列，因此很容易使用 error 启动总线实例。作为输入队列 - 那么你可以例如使用 Rebus 的内置 transport message forwarding能够对消息执行您想要的操作 - 例如将它们转发到它们的源队列:

Configure.With(activator)
    .Transport(t => (...)) //< use queue "error" here
    .Routing(r =>
    {
        r.AddTransportMessageForwarder(async transportMessage =>
        {
            var sourceQueue = transportMessage.Headers.TryGetValue(Headers.SourceQueue, out var result)
                ? result
                : throw new ArgumentException($"Could not find '{Headers.SourceQueue}' header");

            return ForwardAction.ForwardTo(sourceQueue);
        });
    })
    .Start();

或者你想要的任何东西。

还有一个 UI， Fleet Manager ，可以做到这一点 - 它完全取代了死信队列的需要，因为它将失败的消息存储在其数据库中，并可以将失败的消息返回到它们的源队列(或另一个队列，如果这是你想要的)，但仅当您是 Rebus Pro 时才可用订户。

更新(包含更新中问题的答案):

1) AddTransportMessageForwarder 仅与接收消息的端点相关。

2) 它是指定为 .Useblablabla 的参数的“队列名称”方法。例如，使用 Azure 服务总线，它将读取

.Transport(t => t.UseAzureServiceBus(Consts.ServiceBusConnectionString, "error"))