Manipulating message headers

Manipulating custom headers can be useful for storing infrastructure-level information not directly related to the business message. Instead of forcing all message types to inherit from a base class or implement a specific interface in order to force the existence of certain properties, instead consider moving this information into message headers.

Message headers are best manipulated through pipeline behaviors, however they can be accessed and modified from message handlers and saga handlers as well.

Depending on the message transport, headers are stored with the message either as native headers (if supported) or via a serialized collection of key/value pairs within the message body itself.

This article covers the various ways of manipulating the message headers.

Reading incoming headers

Headers can be read for an incoming message.

From a behavior

Copy code|Copy usings|Edit

public class IncomingBehavior :
    Behavior<IIncomingPhysicalMessageContext>
{
    public override Task Invoke(IIncomingPhysicalMessageContext context, Func<Task> next)
    {
        var headers = context.Message.Headers;
        var nsbVersion = headers[Headers.NServiceBusVersion];
        var customHeader = headers["MyCustomHeader"];
        return next();
    }
}

From a mutator

Copy code|Copy usings|Edit

public class MutateIncomingTransportMessages :
    IMutateIncomingTransportMessages
{
    public Task MutateIncoming(MutateIncomingTransportMessageContext context)
    {
        var headers = context.Headers;
        var nsbVersion = headers[Headers.NServiceBusVersion];
        var customHeader = headers["MyCustomHeader"];
        return Task.CompletedTask;
    }
}

From a handler

Copy code|Copy usings|Edit

public class ReadHandler :
    IHandleMessages<MyMessage>
{
    public Task Handle(MyMessage message, IMessageHandlerContext context)
    {
        var headers = context.MessageHeaders;
        var nsbVersion = headers[Headers.NServiceBusVersion];
        var customHeader = headers["MyCustomHeader"];
        return Task.CompletedTask;
    }
}

From a saga

Copy code|Copy usings|Edit

public class ReadSaga :
    Saga<ReadSagaData>,
    IHandleMessages<MyMessage>
{
    public Task Handle(MyMessage message, IMessageHandlerContext context)
    {
        var headers = context.MessageHeaders;
        var nsbVersion = headers[Headers.NServiceBusVersion];
        var customHeader = headers["MyCustomHeader"];
        return Task.CompletedTask;
    }

Writing outgoing headers

Headers can be written for an outgoing message.

From a behavior

Copy code|Copy usings|Edit

public class OutgoingBehavior :
    Behavior<IOutgoingPhysicalMessageContext>
{
    public override Task Invoke(IOutgoingPhysicalMessageContext context, Func<Task> next)
    {
        var headers = context.Headers;
        headers["MyCustomHeader"] = "My custom value";
        return next();
    }
}

From a mutator

Copy code|Copy usings|Edit

public class MutateOutgoingTransportMessages :
    IMutateOutgoingTransportMessages
{
    public Task MutateOutgoing(MutateOutgoingTransportMessageContext context)
    {
        context.OutgoingHeaders["MyCustomHeader"] = "My custom value";
        return Task.CompletedTask;
    }
}

From a handler

Copy code|Copy usings|Edit

public class WriteHandler :
    IHandleMessages<MyMessage>
{
    public async Task Handle(MyMessage message, IMessageHandlerContext context)
    {
        var sendOptions = new SendOptions();
        sendOptions.SetHeader("MyCustomHeader", "My custom value");
        await context.Send(new SomeOtherMessage(), sendOptions)
            .ConfigureAwait(false);

        var replyOptions = new ReplyOptions();
        replyOptions.SetHeader("MyCustomHeader", "My custom value");
        await context.Reply(new SomeOtherMessage(), replyOptions)
            .ConfigureAwait(false);

        var publishOptions = new PublishOptions();
        publishOptions.SetHeader("MyCustomHeader", "My custom value");
        await context.Publish(new SomeOtherMessage(), publishOptions)
            .ConfigureAwait(false);
    }
}

From a saga

Copy code|Copy usings|Edit

public class WriteSaga :
    Saga<WriteSagaData>,
    IHandleMessages<MyMessage>
{
    public async Task Handle(MyMessage message, IMessageHandlerContext context)
    {
        var sendOptions = new SendOptions();
        sendOptions.SetHeader("MyCustomHeader", "My custom value");
        await context.Send(new SomeOtherMessage(), sendOptions)
            .ConfigureAwait(false);

        var replyOptions = new ReplyOptions();
        replyOptions.SetHeader("MyCustomHeader", "My custom value");
        await context.Reply(new SomeOtherMessage(), replyOptions)
            .ConfigureAwait(false);

        var publishOptions = new PublishOptions();
        publishOptions.SetHeader("MyCustomHeader", "My custom value");
        await context.Publish(new SomeOtherMessage(), publishOptions)
            .ConfigureAwait(false);
    }

For all outgoing messages

NServiceBus supports registering headers at configuration time that are then added to all outgoing messages for the endpoint.

Copy code|Copy usings|Edit

endpointConfiguration.AddHeaderToAllOutgoingMessages("MyGlobalHeader", "some static value");

Reading incoming headers

From a behavior

From a mutator

From a handler

From a saga

Writing outgoing headers

From a behavior

From a mutator

From a handler

From a saga

For all outgoing messages

Samples

Related Articles