Access to the native Azure Service Bus incoming message
It can sometimes be useful to access the native Service Bus incoming message from behaviors and handlers. When a message is received, the transport adds the native Service Bus Message to the message processing context. Use the code below to access the message details from a pipeline behavior:
class DoNotAttemptMessageProcessingIfMessageIsNotLocked
: Behavior<ITransportReceiveContext>
{
public override Task Invoke(ITransportReceiveContext context, Func<Task> next)
{
var nativeMessage = context.Extensions.Get<ServiceBusReceivedMessage>();
if (nativeMessage.LockedUntil <= DateTime.UtcNow)
{
return next();
}
throw new Exception($"Message lock lost for MessageId {context.Message.MessageId} and it cannot be processed.");
}
}
The behavior above uses the native message's LockedUntilUtc system property to determine where the message lost its lock as a result of aggressive prefetching and slow processing. If desired, a custom recoverability policy can be used so that the message will skip attempted retry processing that otherwise would be guaranteed to fail due to the message's lost lock.
Access to the native Azure Service Bus outgoing message
It can also be useful to access the native Service Bus outgoing message from behaviors and handlers for customizations.
// send a command
var sendOptions = new SendOptions();
sendOptions.CustomizeNativeMessage(m => m.Subject = "custom-label");
await context.Send(new MyCommand(), sendOptions);
// publish an event
var publishOptions = new PublishOptions();
publishOptions.CustomizeNativeMessage(m => m.Subject = "custom-label");
await context.Publish(new MyEvent(), publishOptions);
Or on a more global level directly on the transport
transport.OutgoingNativeMessageCustomization = (operation, message) =>
{
// Customize the outgoing message based on the operation
};
Native outgoing messages cannot be customized when using the outbox as customizations are not persistent.