修改传出 NServiceBus 订阅消息的标头以将信息传递给 IAuthorizeSubscriptions答案

【问题标题】：Modifying headers of outgoing NServiceBus subscription messages to pass information to IAuthorizeSubscriptions修改传出 NServiceBus 订阅消息的标头以将信息传递给 IAuthorizeSubscriptions
【发布时间】：2014-06-18 08:40:20
【问题描述】：

我正在使用 NServiceBus 4.4.2，并且我创建了一个实现 IAuthorizeSubscriptions 的类，以便接受/拒绝订阅。我需要发布者中有关订阅者的一些额外信息，所以我尝试创建一个实现IMutateOutgoingTransportMessages 的类并使用ConfigureComponent<MyMutator>(DependencyLifecycle.InstancePerCall) 以正常方式注册它。这个类所做的只是为传出的消息添加一个标题。

不幸的是，从订阅者发送的订阅控制消息中没有设置标题。我已经检查了订阅者是否执行了 Bus.Send()，我的自定义标头附加到传输消息的 Headers 集合中，所以我认为这不是我的 mutator 的问题，除非我需要以不同的方式注册。

有没有办法将标头附加到订阅消息中，或者我是否以错误的方式处理此问题，应该通过其他方式传递信息？

编辑：

突变代码：

public class MyMutator : IMutateOutgoingTransportMessages
{
    public void MutateOutgoing(object[] messages, TransportMessage transportMessage)
    {
        transportMessage.Headers.Add("CustomHeader", "Test");
    }
}

IAuthorizeSubscriptions 代码：

我可以在AuthorizeSubscribe 方法内设置断点，它会被命中，但headers 不包含我的自定义标头。

public class SubscriptionAuthoriser : IAuthorizeSubscriptions
{
    public bool AuthorizeSubscribe(string messageType, string clientEndpoint, IDictionary<string, string> headers)
    {
        // headers doesn't contain the "CustomHeader" header

        return true;
    }

    public bool AuthorizeUnsubscribe(string messageType, string clientEndpoint, IDictionary<string, string> headers)
    {
        return true;
    }
}

【问题讨论】：

根据此对话 (groups.google.com/forum/#!topic/particularsoftware/XVLQkCouKCk)，从 v4.3 开始，IMutateOutgoingTransportMessages 实现不再用于传出订阅消息。
很好的发现。感谢张贴的链接。尽管这确实让我感到难过，但它不再可能。如果你单独发，我可以标记为答案。

标签： nservicebus

【解决方案1】：

从 NServiceBus 版本 4.3 开始，不再对传出订阅消息调用 InvokeOutgoingTransportMessagesMutators，因此消息头似乎不再可能。不久前我问了一个类似的问题@@https://groups.google.com/forum/#!topic/particularsoftware/XVLQkCouKCk。在等待上述线程的响应时，我实施了一个临时解决方法，即手动订阅方法（主要使用 NServiceBus 单播总线源构建）。这可能不被推荐，但我想我会发布它，以防万一它对某人有帮助。

public void Subscribe(Type eventType, Address targetAddress)
{
    var transportMessage = new TransportMessage
        {
            ReplyToAddress = Address.PublicReturnAddress,
            Recoverable = true
        };
        transportMessage.Headers.Add("NServiceBus.ControlMessage", true.ToString());
        transportMessage.Headers["SubscriptionMessageType"] = eventType.AssemblyQualifiedName;
        transportMessage.MessageIntent = MessageIntentEnum.Subscribe;

        var fdqn = OutgoingTransportMessageMutator.GetLocalhostFqdn();
        transportMessage.ReplyToAddress = new Address(transportMessage.ReplyToAddress.Queue, fdqn);

        string fullPath = MsmqUtilities.GetFullPath(targetAddress);
        try
        {
            using (var messageQueue = new MessageQueue(fullPath, false, false, QueueAccessMode.Send))
            {
                using (var message1 = MsmqUtilities.Convert(transportMessage))
                {
                    message1.UseDeadLetterQueue = false;
                    message1.UseJournalQueue = false;
                    message1.ResponseQueue = new MessageQueue(MsmqUtilities.GetReturnAddress(transportMessage.ReplyToAddress.ToString(), targetAddress.ToString()));
                    messageQueue.Send(message1, MessageQueueTransactionType.Single);
                }
            }
        }
        catch (MessageQueueException ex)
        {
            if (ex.MessageQueueErrorCode == MessageQueueErrorCode.QueueNotFound)
            {
                string message1 = targetAddress == (Address)null ? "Failed to send message. Target address is null." : string.Format("Failed to send message to address: [{0}]", targetAddress);
                throw new QueueNotFoundException(targetAddress, message1, ex);
            }
            throw;
        }
        catch (Exception ex)
        {
            _log.Error(ex);
            throw;
         }
     }

【讨论】：