Getting Started With MassTransit And RabbitMq

We will now proceed to declare the message format to be sent across. We will now create a Class Library project and add the following record to it. This library would be shared between the producer and consumer for the Message type definition.

public record CommandMessage(long Id,string MessageString);

Let us now create an endpoint in our producer Web API which would be raising the request to send the message to the consumer.

app.MapPost("/sendmessage", (long id, string message, IPublishEndpoint publishEndPoint) => {
    publishEndPoint.Publish(new CommandMessage(id, message));;
});

Consumer Code Sample

With the producer code in place, it is time to concentrate on the consumer application. The consumer application would read any messages from the designated queue and process them depending on the message types. Each message type that needs to be processed by the consumer application would require an implementation of IConsumer<TMessageType>. This would trigger the required actions that need to happen when a message of the particular type arrives.

The IConsumer<TMessageType> is defined as:

public interface IConsumer < in TMessage > : IConsumer
where TMessage: class {
    Task Consume(ConsumeContext < TMessage > context);
}
public interface IConsumer {}

As you can see, there is only one method exposed by the interface - Consume. The Consume method receives the message wrapped in an IConsumeContext interface. This allows it to access the incoming message, but also the details surrounding the message, including the headers.

Let us begin by writing our handler for the message type CommandMessage.

public class CommandMessageConsumer: IConsumer < CommandMessage > {
    public async Task Consume(ConsumeContext < CommandMessage > context) {
        var message = context.Message;
        await Console.Out.WriteLineAsync($ "Message from Producer : {message.MessageString}");
        // Do something useful with the message
    }
}

As you can see from the code above, for the demo purpose, we are not doing anything useful with the incoming message. This is to keep the example code simple, but the code could perform any business logic required by the application.

Now that we have the CommandMessageConsumer ready, it is important to enlist it to the MassTransit pipeline. This could be achieved easily with the following code:

var rabbitMqSettings = builder.Configuration.GetSection(nameof(RabbitMqSettings)).Get < RabbitMqSettings > ();
builder.Services.AddMassTransit(mt => mt.UsingRabbitMq((cntxt, cfg) => {
    cfg.Host(rabbitMqSettings.Uri, "/", c => {
        c.Username(rabbitMqSettings.UserName);
        c.Password(rabbitMqSettings.Password);
    });
    cfg.ReceiveEndpoint("samplequeue", (c) => {
        c.Consumer < CommandMessageConsumer > ();
    });
}));

In the above code sample, any message of type CommandMessage arriving on the samplequeue queue will be consumed by the implementation of CommandMessageConsumer.

That's all that is required to run the first example of MassTransit with RabbitMq. Let us now test the sample application by running our Producer and Consumer applications.

Producer

Consumer

info: MassTransit[0]
      Bus started: rabbitmq://localhost/
Message from Producer : Hello World !!

You could test the asynchronous nature of message delivery by taking your Consumer application offline, while sending a message using your Producer code. The message could be verified within the RabbitMq using the Management tool. Later, when the Consumer application is online again, it would recieve the message.

Summary

As seen in the example codes above, MassTransit provides a simple, easy-to0-use abstraction over a different transport mechanism. In this example, we use RabbitMq, but it would be as simple to integrate an Azure Bus or an ActiveMq. The abstraction also allows the developer to focus on the business logic irrespective of the underlying implementation of transport layer.

• Complete source code of the demo application