Micro-services communication: Rabbitmq and Asp.net core

When building micro-services it is very important to take into consideration which messaging mechanism you use. There are several ways in which micro-services communicate using different messaging protocols. Today we will focus on the AMQP messaging protocol with rabbitmq and asp.net core.

What is Rabbitmq ?

Rabbitmq is a messaging system that permits your applications to communicate together using messages and queues. Rabbitmq uses the AMQP (Advanced Message Queuing Protocol) messaging protocol. It is open-source and is very robust. Rabbitmq is written in Erlang, a functional programming language created for telecommunication systems by Ericsson. Erlang is a concurrent programming language with several other features which permit it to be an ideal choice for building highly robust and resilient messaging systems.

How Rabbitmq Works

As we mentioned earlier, Rabbitmq implements the AMQP protocol. Here is a brief overview of how messages are sent via rabbitmq. An application willing to send messages (Producer) to another application, does so through a broker (Rabbitmq). The broker then receives the message via an Exchange which is then responsible for routing that message to appropriate Queues using routing keys and rules called bindings. The message is then received by another application that subscribes to the given queue (Consumer). You can learn more about this process here.

Some other attributes of this communication process could be configured, like sending acknowledgments when a message is received, persisting a queue in a database e.t.c. There are several types of exchanges in Rabbitmq and each determines the way in which your message is routed to corresponding queues.

Rabbitmq and ASP.net Core

Though rabbitmq seems a little bit complicated in the beginning, there are awesome libraries that abstract all of these AMQP notions and make it easier to use when building your ASP.net core microservices. We will use one of these libraries (Masstransit).

Installing Rabbitmq

It is easy to install rabbitmq, you can download it directly from their website and install it on your operating system. Or you could use a docker container. We will be using the latter approach. The image’s name is:

rabbitmq:3-management

1	rabbitmq:3-management

Here is the command to run rabbitmq on docker. You can configure the password and username to what you wish.

docker run -d -t -it --hostname my-rabbitmq --name rabbitmq3-server -p 15672:15672 -p 5672:5672 -e RABBITMQ_DEFAULT_USER=user -e RABBITMQ_DEFAULT_PASS=password -e RABBITMQ_DEFAULT_VHOST=my_vhost rabbitmq:3-management

1	docker run -d -t -it --hostname my-rabbitmq --name rabbitmq3-server -p 15672:15672 -p 5672:5672 -e RABBITMQ_DEFAULT_USER=user -e RABBITMQ_DEFAULT_PASS=password -e RABBITMQ_DEFAULT_VHOST=my_vhost rabbitmq:3-management

What we will build

To demonstrate micro-services communication, we will build two simple rest APIs which will communicate with each other. One will be the user’s service and the other will be the companies service. The users’ service will receive API calls to subscribe to specific companies. Once it receives a call, it asynchronously tells the companies service to add the user to the list of subscribers in the specified company “Command”. Thereafter, the company service broadcasts a message telling the subscription process is over “Event”. All these are through rabbitmq and asp.net core.

Masstransit and its Rabbitmq abstraction layer

Rabbitmq has a library that permits us to communicate in a raw fashion. Using the notions of queues and exchanges, listening for messages with loops, and all the like. Though we can perfectly use this when communicating between our micro-services, it will require much work, and what if in the future we want to change the messaging system from Rabbitmq to Service bus or something else? What if we want to do things in a clean and maintainable way ?.

MassTransit is a library that provides a good layer of abstraction above rabbitmq and other messaging tools. Switching from rabbitmq to another tool will just be a matter of changing configurations.

Messages are sent using Events and Commands. A command is a message sent to a specific endpoint. And events are sent to whosoever is willing to subscribe to them. To receive a message, you need to subscribe a consumer, this processes the message. When a message is sent with MassTransit via rabbitmq, it is sent to an exchange with the name corresponding to the type of the message’s class. Messages sent contain metadata about the message.

If you like this post, don’t hesitate to follow me on Twitter or Gi t hub and subscribe to this page’s notifications to stay updated with new articles or like my page on Facebook.

Let’s code

The source code for this post is found here.

First, we need to add mass-transit to our services. The code bellow will not only initialize configurations for rabbitmq and masstransit, but also set up our consumers and start the bus service.


            services.AddMassTransit(x =>
            {
                var configSections = Configuration.GetSection("Rabbitmq");
                var host = configSections["Host"];
                var userName = configSections["UserName"];
                var password = configSections["Password"];
                var virtualHost = configSections["VirtualHost"];
                var port = Convert.ToUInt16(configSections["Port"]);

                x.AddConsumer<SubscriptionSuccessfulConsumer>();

                x.AddBus(provider =>
                {
                    var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
                    {
                        cfg.Host(host, port, virtualHost, host =>
                        {
                            host.Username(userName);
                            host.Password(password);
                        });

                        cfg.ReceiveEndpoint(configSections["Endpoint"], ep =>
                        {
                            ep.PrefetchCount = Convert.ToUInt16(configSections["PrefetchCount"]);

                            ep.ConfigureConsumer<SubscriptionSuccessfulConsumer>(provider);
                        });
                    });

                    bus.Start();

                    return bus;
                });
            });

            services.Configure<HealthCheckPublisherOptions>(options =>
            {
                options.Delay = TimeSpan.FromSeconds(2);
                options.Predicate = (check) => check.Tags.Contains("ready");
            });

            services.AddMassTransitHostedService();

services.AddMassTransit(x =>

{

var configSections = Configuration.GetSection("Rabbitmq");

var host = configSections["Host"];

var userName = configSections["UserName"];

var password = configSections["Password"];

var virtualHost = configSections["VirtualHost"];

var port = Convert.ToUInt16(configSections["Port"]);

x.AddConsumer<SubscriptionSuccessfulConsumer>();

x.AddBus(provider =>

{

var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>

{

cfg.Host(host, port, virtualHost, host =>

{

host.Username(userName);

host.Password(password);

});

cfg.ReceiveEndpoint(configSections["Endpoint"], ep =>

{

ep.PrefetchCount = Convert.ToUInt16(configSections["PrefetchCount"]);

ep.ConfigureConsumer<SubscriptionSuccessfulConsumer>(provider);

});

bus.Start();

return bus;

});

services.Configure<HealthCheckPublisherOptions>(options =>

{

options.Delay = TimeSpan.FromSeconds(2);

options.Predicate = (check) => check.Tags.Contains("ready");

});

services.AddMassTransitHostedService();

As mentioned earlier, we will send messages via events/commands. These are simple classes containing the data we want to send.

We use only asynchronous communication here. Commands are sent while targeting specific endpoints as seen below.

public async Task SendMessage<T>(T message, string targetEndPoint)
        {
            var endpoint = $"rabbitmq://{_rabbitConfig.Host}:{_rabbitConfig.Port}/{targetEndPoint}?durable={_rabbitConfig.DurableQueue}";
            var finalEndpoint = await _sendEndpoint.GetSendEndpoint(new Uri(endpoint));
            await finalEndpoint.Send(message);
        }

public async Task SendMessage<T>(T message, string targetEndPoint)

{

var endpoint = $"rabbitmq://{_rabbitConfig.Host}:{_rabbitConfig.Port}/{targetEndPoint}?durable={_rabbitConfig.DurableQueue}";

var finalEndpoint = await _sendEndpoint.GetSendEndpoint(new Uri(endpoint));

await finalEndpoint.Send(message);

}

As you can see above, you can set an option (?durable=) specifying if you want Rabbitmq to persist a queue. Note that, an endpoint so that it persists messages it receives.

Sending events doesn’t require you to specify any endpoint. You just publish it to who so ever wants to listen to it. As shown below:


                _publishEndPoint.Publish(new SubscriptionSuccessfulEvent
                {
                    CompanyId = context.Message.CompanyId,
                    UserId = context.Message.UserId
                });

_publishEndPoint.Publish(new SubscriptionSuccessfulEvent

{

CompanyId = context.Message.CompanyId,

UserId = context.Message.UserId

});

Now receiving messages is just as easy. You just need to create a consumer for the specific message you are willing to receive and register it as we can see above in the startup class. You can see below.


    public class SubscribeToCompanyConsumer : IConsumer<SubscribeToCompanyCommand>
    {
        RabbitmqConfig _rabbitConfig;
        ISendEndpointProvider _sendEndpoint;
        IPublishEndpoint _publishEndPoint;
        FakeStore _store;

        public SubscribeToCompanyConsumer(IOptions<RabbitmqConfig> rabbitConfig, 
            ISendEndpointProvider sendEndpoint, IPublishEndpoint publish,
            FakeStore store)
        {
            _store = store;
            _rabbitConfig = rabbitConfig.Value;
            _sendEndpoint = sendEndpoint;
            _publishEndPoint = publish;
        }

        public Task Consume(ConsumeContext<SubscribeToCompanyCommand> context)
        {
            try
            {
                var company = _store.Companies.SingleOrDefault(c => c.Id == context.Message.CompanyId);
                var f = company.FollowersIds;
                company.FollowersIds.Add(context.Message.UserId);

                return _publishEndPoint.Publish(new SubscriptionSuccessfulEvent
                {
                    CompanyId = context.Message.CompanyId,
                    UserId = context.Message.UserId
                });
            }
            catch (Exception e)
            {
                throw;
            }
        }
    }

public class SubscribeToCompanyConsumer : IConsumer<SubscribeToCompanyCommand>

{

RabbitmqConfig _rabbitConfig;

ISendEndpointProvider _sendEndpoint;

IPublishEndpoint _publishEndPoint;

FakeStore _store;

public SubscribeToCompanyConsumer(IOptions<RabbitmqConfig> rabbitConfig,

ISendEndpointProvider sendEndpoint, IPublishEndpoint publish,

FakeStore store)

{

_store = store;

_rabbitConfig = rabbitConfig.Value;

_sendEndpoint = sendEndpoint;

_publishEndPoint = publish;

}

public Task Consume(ConsumeContext<SubscribeToCompanyCommand> context)

{

try

{

var company = _store.Companies.SingleOrDefault(c => c.Id == context.Message.CompanyId);

var f = company.FollowersIds;

company.FollowersIds.Add(context.Message.UserId);

return _publishEndPoint.Publish(new SubscriptionSuccessfulEvent

{

CompanyId = context.Message.CompanyId,

UserId = context.Message.UserId

});

}

catch (Exception e)

{

throw;

}

– For messaging between the two microservices to work, you need to first run rabbit mq in a docker container or anywhere on a network you have access to.
– Then get RabbitMq’s password and username, and put them in the appsettings.json of the microservices
– Then as shown above in the startup.cs, configure your microservices with MassTransit,
– To communicate between Microservices, I mentioned in the post there is a method named “SendMessage”
If you look into the sample, you will find the “SendMessage” method is called inside “SubscribeToCompany” put request of the user controller.
– Making a postman, curl, etc to this method will fire a message to your companies controller.
– Your companies controller will receive the message and perform the required logic inside the “SubscribeToCompanyConsumer” class.
– Then this class will send another message to the users’ microservice confirming the user’s subscription.

Conclusion

Above we have a quick start guide to implement micro-services communication with rabbitmq and asp.net core and MassTransit. You can get the source code from here. You might be interested in this article about setting optional form body in asp.net core web API.

Follow me on social media and stay updated

2 comments

Kaushik Roy Chowdhury
May 28, 2021
Hi Dear Doumer, I have tried to run the solution by cloning your Github repo. But sorry to say that I was unable to run the solution with both the projects as startup first and then tried to make one of them startup and run that. Unfortunaltely, after waiting for 10-15 in either case the server didn’t fetch any response and the browser would try loading indefinitely without a response. I am not sure if you put the code in the repo after testing and if it was working why is it not working now? Can you please shed some light as how to test the app if it needs to be done in some way other than usual. I ahd even had the Docker container running and logged on to RabbitMQ admin dashboard with the given credentials. It was a very disappointing experience.
Cheers
Kaushik
Loading...
- Damien Doumer
  May 28, 2021
  Hello Roy,
  First Thanks, I appreciate you read my blog thanks.
  Please What issue did you have while running the sample ?
  Maybe I didn’t explain the process of running the sample in detail ? If yes, then let me do that now if you find this useful, then I’ll update the post to make it clearer for other readers;
  – For messaging between the two microservices to work, you need to first run rabbit mq in a docker container or anywhere on a network you have access to,
  – Then get RabbitMq’s password and username, and put it in the appsettings.json of the microservices
  – Then as shown above in the startup.cs, configure your microservices with masstransit,
  – To communicate between Microservices, I mentioned in the post there is a method named “SendMessage”
  If you look into the sample, you will find the “SendMessage” method is called inside “SubscribeToCompany” put request of the userscontroller.
  – Making a postman, curl etc to this method will fire a message to your companies controller.
  – Your companies controller will receive the message and perform the required logic inside the “SubscribeToCompanyConsumer” class.
  – Then this class will send another message to the users microservice confirming the user’s subscription.
  Loading...