MassTransit：如何让发布过滤器访问PublishContext回调设置的消息Header？

使用MassTransit v8.1.3的Scoped Publish Filter时，希望根据发布时指定的条件为消息添加Header，但发现通过IPublishEndpoint.Publish回调设置的Header无法在过滤器中访问——因为回调在管道的最后执行，过滤器执行时Header尚未被设置。

发布代码示例：

await publishEndpoint.Publish(message, (PublishContext context) =>
{
    context.Headers.Set(MessageHeaderNames.SourceSystem, "iothub");
});

过滤器代码示例：

public class MyPublishFilter<T> : IFilter<PublishContext<T>> where T : class
{
    public async Task Send(PublishContext<T> context, IPipe<PublishContext<T>> next)
    {
        // 此处无法获取到回调中设置的SourceSystem Header
        context.Headers.TryGetHeader(MessageHeaderNames.SourceSystem, out var sourceSystem); 

        await next.Send(context);
    }
}

目前尝试过包装消息重复发布的变通方案，但引发了其他问题；也考虑过引入Scoped消息上下文，但希望找到更简便的实现方式。

核心原因说明 #

MassTransit的发布管道执行顺序为：先执行所有注册的Publish过滤器 → 执行用户提供的PublishContext回调 → 发送消息。因此回调中设置的Header在过滤器执行阶段还未生效，无法直接读取。

以下是两种可靠的实现方式：

1. 使用PublishContext.Items传递临时数据 #

PublishContext.Items是MassTransit内置的键值对集合，用于在管道各阶段共享临时数据，可直接在发布回调中写入，过滤器中读取。

调整发布代码：

await publishEndpoint.Publish(message, context =>
{
    // 将数据存入Items集合，而非直接设置Header
    context.Items[MessageHeaderNames.SourceSystem] = "iothub";
    // 若最终仍需设置Header，可同时保留此处的Header设置
    // context.Headers.Set(MessageHeaderNames.SourceSystem, "iothub");
});

调整过滤器代码：

public class MyPublishFilter<T> : IFilter<PublishContext<T>> where T : class
{
    public async Task Send(PublishContext<T> context, IPipe<PublishContext<T>> next)
    {
        if (context.Items.TryGetValue(MessageHeaderNames.SourceSystem, out var sourceSystem))
        {
            // 使用传递的SourceSystem值执行逻辑，比如添加其他Header
            context.Headers.Set("DerivedHeader", $"Processed_{sourceSystem}");
        }

        await next.Send(context);
    }
}

2. 使用Scoped上下文类共享数据 #

如果需要传递复杂结构的数据，或多个服务/过滤器需要共享上下文信息，可定义Scoped生命周期的上下文类，通过依赖注入实现数据共享。

1. 定义Scoped上下文类：

public class PublishMessageContext
{
    public string SourceSystem { get; set; }
    // 可添加更多需要共享的字段
}

2. 注册Scoped服务：

services.AddScoped<PublishMessageContext>();

3. 发布时填充上下文：

public class MessagePublisher
{
    private readonly IPublishEndpoint _publishEndpoint;
    private readonly PublishMessageContext _messageContext;

    public MessagePublisher(IPublishEndpoint publishEndpoint, PublishMessageContext messageContext)
    {
        _publishEndpoint = publishEndpoint;
        _messageContext = messageContext;
    }

    public async Task PublishMyMessage()
    {
        // 填充上下文数据
        _messageContext.SourceSystem = "iothub";
        await _publishEndpoint.Publish(new MyMessage());
    }
}

4. 过滤器中读取上下文：

public class MyPublishFilter<T> : IFilter<PublishContext<T>> where T : class
{
    private readonly PublishMessageContext _messageContext;

    public MyPublishFilter(PublishMessageContext messageContext)
    {
        _messageContext = messageContext;
    }

    public async Task Send(PublishContext<T> context, IPipe<PublishContext<T>> next)
    {
        if (!string.IsNullOrEmpty(_messageContext.SourceSystem))
        {
            context.Headers.Set(MessageHeaderNames.SourceSystem, _messageContext.SourceSystem);
        }

        await next.Send(context);
    }
}

• Items集合：轻量、无需额外DI配置，适合简单数据传递，是优先推荐的方案。
• Scoped上下文类：适合复杂数据结构或多场景共享的需求，扩展性更强。

避免使用重复发布消息的变通方案，会引发重复消息、性能损耗等问题。

内容的提问来源于stack exchange，提问作者Thomas U.