Mass Transit project is a well known abstraction layer for .NET over most popular message brokers, covering providers like RabbitMQ or Azure Service Bus. It will not only configure underlying message broker via friendly API, but will also address issues like error handling or concurrency. It also introduces a clean implementation of saga pattern and couple of other patterns useful in distributed systems.
This article is a brief introduction to MassTransit for Azure Service Bus users.
Sending a message
When sending a message we must specify so called “send endpoint” name. Under the hood, send endpoint is Azure Service Bus queue name. When send method is called, queue is automatically created.
ISendEndpointProvider needs to be injected to call Send method. See producers docs for other ways on how to send a message.
Queue name can be specified by creating a mapping from message type to queue name: EndpointConvention.Map<IMyCommand>(new Uri("queue:my-command"));
Publishing an event
When publishing an event we do not have to specify any endpoint name to which we send the message. MassTransit by convention creates a topic corresponding to published massage full name (including namespace). So under the hood we have a concrete topic as we have concrete queue when sending a message, but in case of events we do not specify that topic explicitly. I find it a bit inconsistent, but I understand the idea – conceptually on the abstraction level, publishing an event does not have a target receiver. It is up to subscribers to subscribe for it. Publisher just throws the event into the air.
To publish an endpoint we simply call Publish method on injected IPublishEndpoint
await _publishEndpoint.Publish<IMyEventDone>(new {
MyProperty = "some value"
});
Event subscribers
It is important to understand how topics and subscriptions work in Azure Service Bus. Each subscription is a queue. When topic does not have any subscriptions, then events published to this topic are simply lost. This is a by-design behaviour in pub/sub pattern.
Consumes connect to subscriptions to process the messages. If there is no active consumers, then messages will be left in the subscription queue until there is a consumer or until a timeout. Subscription is a persistent entity, but consumers are dynamic processes. There may be multiple competing consumers for given subscription to scale-out message processing. Usually different subscriptions are created by different services/sub-systems interested in given events.
cfg.SubscriptionEndpoint<IMyEventDone>("my-subscription", c => {
c.ConfigureConsumer<Consumer>(context);
});
Worth to mention that if we use MassTransit and we subscribe to a subscription endpoint but we will not register any consumers, then messages sent to this endpoint will be automatically moved to _skipped queue created for IMyEventDone type.
There is also an alternative way of creating subscriptions, which will use additional queue that will have messages from the subscription auto-forwarded, see docs for details.
Anonymous types for messages
It is recommended by MassTransit author to use interfaces for massage contracts. MassTransit comes with a useful Roslyn analysers package which simplifies using anonymous types as interface implementations. After installing analyzers: Install-Package MassTransit.Analyzers we can automatically add missing properties with Alt+Enter:
