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
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
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
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
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
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
public class MutateOutgoingTransportMessages :
IMutateOutgoingTransportMessages
{
public Task MutateOutgoing(MutateOutgoingTransportMessageContext context)
{
context.OutgoingHeaders["MyCustomHeader"] = "My custom value";
return Task.CompletedTask;
}
}
From a handler
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
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.
endpointConfiguration.AddHeaderToAllOutgoingMessages("MyGlobalHeader", "some static value");