Azure Service Bus Transport

Azure Service Bus is a messaging service from Microsoft Azure that allows for communication between decoupled systems. It offers a reliable and secure platform for asynchronous transfer of data and state. It supports a variety of messaging patterns, including queuing, publish/subscribe, and request/response.

With Service Bus, you can create messaging entities such as queues, topics, and subscriptions. Queues provide one-to-one messaging, where each message is consumed by a single receiver. Topics and subscriptions provide one-to-many messaging, where a message is delivered to multiple subscribers.

Service Bus also provides advanced features such as partitioning and auto-scaling, which allow for high availability and scalability. Additionally, it offers a dead letter queue, which is a special queue that stores undelivered or expired messages.

Topology

The send and publish topologies are extended to support the Azure Service Bus features, and make it possible to configure how topics are created.

Topics

An Azure Service Bus Topic is a messaging entity that allows for one-to-many messaging, where a message is delivered to multiple subscribers. Topics are built on top of Azure Service Bus Queues and provide additional functionality for publish/subscribe messaging patterns.

When a message is sent to a topic, it is automatically broadcast to all subscribers that have a subscription to that topic. Subscriptions are used to filter messages that are delivered to the subscribers. Subscribers can create multiple subscriptions to a topic, each with its own filter, to receive only the messages that are of interest to them.

Topics also provide a feature called Session-based messaging, which allows for guaranteed ordering of messages, and the ability to send and receive messages in a stateful manner.

Topics provide a robust and scalable messaging infrastructure for building distributed systems, where multiple services or systems can subscribe to a topic and receive messages that are relevant to them. Topics also support advanced features such as partitioning and auto-scaling, which allow for high availability and scalability.

To specify properties used when a topic is created, the publish topology can be configured during bus creation:

cfg.Publish<OrderSubmitted>(x =>
{
    x.EnablePartitioning = true;
});

PartitionKey

When publishing messages to an Azure Service Bus topic, you can use the PartitionKey property to specify a value that will be used to partition the messages across multiple topic partitions. This can be useful in situations where you want to ensure that related messages are always delivered to the same partition, and thus will be guaranteed to be processed in the order they were sent.

By setting a PartitionKey, all messages with the same key will be sent to the same partition, and thus will be received by consumers in the order they were sent. This is particularly useful when building distributed systems that require strict ordering of messages, such as event sourcing or stream processing.

Another use case for the PartitionKey is when you have a large number of messages and want to distribute them evenly across multiple partitions for better performance, this way the messages are load balanced across all the partitions.

It's important to note that when you use a PartitionKey, it's important to choose a key that will result in an even distribution of messages across partitions, to avoid overloading a single partition.

The PartitionKey on published/sent messages can be configured by convention, allowing the same method to be used for messages which implement a common interface type. If no common type is shared, each message type may be configured individually using various conventional selectors. Alternatively, developers may create their own convention to fit their needs.

When configuring a bus, the send topology can be used to specify a routing key formatter for a particular message type.

public record SubmitOrder
{
    public string CustomerId { get; init; }
    public Guid TransactionId { get; init; }
}
cfg.Send<SubmitOrder>(x =>
{
    x.UsePartitionKeyFormatter(context => context.Message.CustomerId);
});

SessionId

When publishing messages to an Azure Service Bus Topic, you can use the SessionId property to specify a value that will be used to group messages together in a session. This can be useful in situations where you want to ensure that related messages are always delivered together, and thus will be guaranteed to be processed in the order they were sent.

A session is a logical container for messages, and all messages within a session have a guaranteed order of delivery. This means that messages with the same SessionId will be delivered in the order they were sent, regardless of the order they were received by the topic.

A common use case for sessions is when you have a set of related messages that need to be processed together. For example, if you are sending a series of commands to control a device, you would want to ensure that the commands are delivered in the order they were sent and that all related commands are delivered together.

Another use case for sessions is when you have a large number of messages and want to ensure that each consumer processes the messages in a specific order.

It's important to note that when you use sessions, the consumers must be able to process the messages in the order they were sent, otherwise messages might get stuck in the session and cause delays.

The SessionId on published/sent messages can be configured by convention, allowing the same method to be used for messages which implement a common interface type. If no common type is shared, each message type may be configured individually using various conventional selectors. Alternatively, developers may create their own convention to fit their needs.

When configuring a bus, the send topology can be used to specify a routing key formatter for a particular message type.

public record UpdateUserStatus
{
    public Guid UserId { get; init; }
    public string Status { get; init; }
}
cfg.Send<UpdateUserStatus>(x =>
{
    x.UseSessionIdFormatter(context => context.Message.UserId);
});

Subscriptions

In Azure, topics and topic subscriptions provide a mechanism for one-to-many communication (versus queues that are designed for one-to-one). A topic subscription acts as a virtual queue. To subscribe to a topic subscription directly the SubscriptionEndpoint should be used:

cfg.SubscriptionEndpoint<MessageType>("subscription-name", e =>
{
    e.ConfigureConsumer<MyConsumer>(provider);
})

Note that a topic subscription's messages can be forwarded to a receive endpoint (an Azure Service Bus queue), in the following way. Behind the scenes MassTransit is setting up Service Bus Auto-forwarding between a topic subscription and a queue.

cfg.ReceiveEndpoint("input-queue", e =>
{
    e.Subscribe("topic-name");
    e.Subscribe<MessageType>();
})

The properties of the topic subscription may also be configured:

cfg.ReceiveEndpoint("input-queue", e =>
{
    e.Subscribe("topic-name", x =>
    {
        x.AutoDeleteOnIdle = TimeSpan.FromMinutes(60);
    });
})

Subscription Filters

MassTransit supports the configuration of subscription rules and filters, which can be used to filter messages as they are delivered to either the subscription endpoint or forwarded to the receive endpoint.

To specify a subscription filter:

cfg.ReceiveEndpoint("input-queue", e =>
{
    e.Subscribe("topic-name", x =>
    {
        x.Filter = new SqlRuleFilter("1 = 1");
    });
})

Saga State Machine Event Filter

This is an advanced scenario in which a saga state machine has an event that needs to filter messages from the topic via the subscription.

First, configure the event, which is defined in the saga state machine, so that it does not configure the consume topology.

public class FilteredSagaStateMachine :
    MassTransitStateMachine<FilteredSaga>
{
    public FilteredSagaStateMachine()
    {
        Event(() => FilteredEvent, x => x.ConfigureConsumeTopology = false);
    }
    
    public Event<Filtered> FilteredEvent { get; }
}

Note that this may cause the saga state machine to be difficult to unit test, since events will no longer be automatically routed to the saga's receive endpoint.

Next, add a saga definition for the saga and explicitly subscribe to the event type

public class FilteredSagaDefinition :
    SagaDefinition<FilteredSaga>
{
    protected virtual void ConfigureSaga(IReceiveEndpointConfigurator endpointConfigurator, 
        ISagaConfigurator<FilteredSaga> sagaConfigurator)
    {
        if(endpointConfigurator is IServiceBusReceiveEndpointConfigurator sb)
        {
            sb.Subscribe<Filtered>("subscription-name", x =>
            {
                x.Rule = new CreateRuleOptions("Only47", new SqlRuleFilter("ClientId = 47"));
            });
        }
    }
}

Finally, add the saga state machine and the definition when configuring MassTransit.

services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<FilteredSagaStateMachine, FilteredSaga, FilteredSagaDefinition>();
});

Broker Topology

The topics, queues, and subscriptions configured on Azure Service Bus are shown below.

azure-topology