Native message access

This document describes how to access native message information with Azure Service Bus transport.

  • Incoming message access is available from version 1.4.0 and above.
  • Outgoing message access is available from version 1.7.0 and above.

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:

1.4 NServiceBus.Transport.AzureServiceBus
class DoNotAttemptMessageProcessingIfMessageIsNotLocked : Behavior<ITransportReceiveContext>
{
    public override Task Invoke(ITransportReceiveContext context, Func<Task> next)
    {
        var lockedUntilUtc = context.Extensions.Get<Message>().SystemProperties.LockedUntilUtc;

        if (lockedUntilUtc <= DateTime.UtcNow)
        {
            return next();
        }

        context.AbortReceiveOperation();

        return Task.CompletedTask;
    }
}
1.4 NServiceBus.Transport.AzureServiceBus
class DoNotAttemptMessageProcessingIfMessageIsNotLocked : Behavior<ITransportReceiveContext>
{
    public override Task Invoke(ITransportReceiveContext context, Func<Task> next)
    {
        var lockedUntilUtc = context.Extensions.Get<Message>().SystemProperties.LockedUntilUtc;

        if (lockedUntilUtc <= DateTime.UtcNow)
        {
            return next();
        }

        context.AbortReceiveOperation();

        return Task.CompletedTask;
    }
}

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. The message with the lost lock will not go through the recoverability process, skipping attempted 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.

Customizing an outgoing message from a message handler:

1.7 NServiceBus.Transport.AzureServiceBus
// send a command
var sendOptions = new SendOptions();
sendOptions.CustomizeNativeMessage(context, m => m.Label = "custom-label");
await context.Send(new MyCommand(), sendOptions).ConfigureAwait(false);

// publish an event
var publishOptions = new PublishOptions();
publishOptions.CustomizeNativeMessage(context, m => m.Label = "custom-label");
await context.Publish(new MyEvent(), publishOptions).ConfigureAwait(false);
1.7 NServiceBus.Transport.AzureServiceBus
// send a command
var sendOptions = new SendOptions();
sendOptions.CustomizeNativeMessage(context, m => m.Label = "custom-label");
await context.Send(new MyCommand(), sendOptions).ConfigureAwait(false);

// publish an event
var publishOptions = new PublishOptions();
publishOptions.CustomizeNativeMessage(context, m => m.Label = "custom-label");
await context.Publish(new MyEvent(), publishOptions).ConfigureAwait(false);

Customizing an outgoing message using IMessageSession:

1.7 NServiceBus.Transport.AzureServiceBus
// send a command
var sendOptions = new SendOptions();
sendOptions.CustomizeNativeMessage(m => m.Label = "custom-label");
await messageSession.Send(new MyCommand(), sendOptions).ConfigureAwait(false);

// publish an event
var publishOptions = new PublishOptions();
publishOptions.CustomizeNativeMessage(m => m.Label = "custom-label");
await messageSession.Publish(new MyEvent(), publishOptions).ConfigureAwait(false);
1.7 NServiceBus.Transport.AzureServiceBus
// send a command
var sendOptions = new SendOptions();
sendOptions.CustomizeNativeMessage(m => m.Label = "custom-label");
await messageSession.Send(new MyCommand(), sendOptions).ConfigureAwait(false);

// publish an event
var publishOptions = new PublishOptions();
publishOptions.CustomizeNativeMessage(m => m.Label = "custom-label");
await messageSession.Publish(new MyEvent(), publishOptions).ConfigureAwait(false);

Customizing an outgoing message from a physical behavior:

1.7 NServiceBus.Transport.AzureServiceBus
public class PhysicalBehavior : Behavior<IIncomingPhysicalMessageContext>
{
    public override async Task Invoke(IIncomingPhysicalMessageContext context, Func<Task> next)
    {
        var sendOptions = new SendOptions();
        sendOptions.CustomizeNativeMessage(context, m => m.Label = "custom-label");

        await context.Send(new MyCommand(), sendOptions);

        await next();
    }
}
1.7 NServiceBus.Transport.AzureServiceBus
public class PhysicalBehavior : Behavior<IIncomingPhysicalMessageContext>
{
    public override async Task Invoke(IIncomingPhysicalMessageContext context, Func<Task> next)
    {
        var sendOptions = new SendOptions();
        sendOptions.CustomizeNativeMessage(context, m => m.Label = "custom-label");

        await context.Send(new MyCommand(), sendOptions);

        await next();
    }
}

Customizing an outgoing message from a logical behavior:

1.7 NServiceBus.Transport.AzureServiceBus
public class LogicalBehavior : Behavior<IIncomingLogicalMessageContext>
{
    public override async Task Invoke(IIncomingLogicalMessageContext context, Func<Task> next)
    {
        var publishOptions = new PublishOptions();
        publishOptions.CustomizeNativeMessage(context, m => m.Label = "custom-label");

        await context.Publish(new MyEvent(), publishOptions);

        await next();
    }
}
1.7 NServiceBus.Transport.AzureServiceBus
public class LogicalBehavior : Behavior<IIncomingLogicalMessageContext>
{
    public override async Task Invoke(IIncomingLogicalMessageContext context, Func<Task> next)
    {
        var publishOptions = new PublishOptions();
        publishOptions.CustomizeNativeMessage(context, m => m.Label = "custom-label");

        await context.Publish(new MyEvent(), publishOptions);

        await next();
    }
}

Last modified