Asynchronous Communication Between Microservices Using .NET Core API, RabbitMQ, and Docker

namespace ProductOwner.Microservice.Model
{
    public class ProductDetails
    {
        public int Id { get; set; }
        public string ProductName { get; set; }
        public int ProductPrice { get; set; }
    }
}

namespace ProductOwner.Microservice.Model
{
    public class ProductOfferDetail
    {
        public int Id { get; set; }
        public int ProductID { get; set; }    
        public string ProductName { get; set; }
        public string ProductOfferDetails { get; set; }
    }
}

using Microsoft.EntityFrameworkCore;
using ProductOwner.Microservice.Model;

namespace ProductOwner.Microservice.Data
{
    public class DbContextClass : DbContext
    {
        protected readonly IConfiguration Configuration;

        public DbContextClass(IConfiguration configuration)
        {
            Configuration = configuration;
        }
        protected override void OnConfiguring(DbContextOptionsBuilder options)
        {
            options.UseSqlServer(Configuration.GetConnectionString("DefaultConnection"));
        }

        public DbSet<ProductDetails> Products { get; set; }
    }
}

namespace ProductOwner.Microservice.Utility
{
    static class StaticConfigurationManager
    {
        public static IConfiguration AppSetting
        {
            get;
        }
        static StaticConfigurationManager()
        {
            AppSetting = new ConfigurationBuilder().SetBasePath(Directory.GetCurrentDirectory()).AddJsonFile("appsettings.json").Build();
        }
    }
}

using ProductOwner.Microservice.Model;

namespace ProductOwner.Microservice.Services
{
    public interface IProductService
    {
        public Task<IEnumerable<ProductDetails>> GetProductListAsync();
        public Task<ProductDetails> GetProductByIdAsync(int id);
        public Task<ProductDetails> AddProductAsync(ProductDetails product);
        public bool SendProductOffer(ProductOfferDetail productOfferDetails);
    }
}

using Microsoft.EntityFrameworkCore;
using Newtonsoft.Json;
using ProductOwner.Microservice.Data;
using ProductOwner.Microservice.Model;
using ProductOwner.Microservice.Utility;
using RabbitMQ.Client;
using System.Text;

namespace ProductOwner.Microservice.Services
{
    public class ProductService : IProductService
    {
        private readonly DbContextClass _dbContext;

        public ProductService(DbContextClass dbContext)
        {
            _dbContext = dbContext;
        }

        public async Task<IEnumerable<ProductDetails>> GetProductListAsync()
        {
            return await _dbContext.Products.ToListAsync();
        }
        public async Task<ProductDetails> GetProductByIdAsync(int id)
        {
            return await _dbContext.Products.Where(x => x.Id == id).FirstOrDefaultAsync();
        }

        public async Task<ProductDetails> AddProductAsync(ProductDetails product)
        {
            var result = _dbContext.Products.Add(product);
            await _dbContext.SaveChangesAsync();
            return result.Entity;
        }

        public bool SendProductOffer(ProductOfferDetail productOfferDetails)
        {
            var RabbitMQServer = "";
            var RabbitMQUserName = "";
            var RabbutMQPassword = "";

            if (Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT") == "Production")
            {
                RabbitMQServer = Environment.GetEnvironmentVariable("RABBIT_MQ_SERVER");
                RabbitMQUserName = Environment.GetEnvironmentVariable("RABBIT_MQ_USERNAME");
                RabbutMQPassword = Environment.GetEnvironmentVariable("RABBIT_MQ_PASSWORD");

            }
            else
            {
                RabbitMQServer = StaticConfigurationManager.AppSetting["RabbitMQ:RabbitURL"];
                RabbitMQUserName = StaticConfigurationManager.AppSetting["RabbitMQ:Username"];
                RabbutMQPassword = StaticConfigurationManager.AppSetting["RabbitMQ:Password"];
            }

            try
            {
                var factory = new ConnectionFactory()
                { HostName = RabbitMQServer, UserName = RabbitMQUserName, Password = RabbutMQPassword };
                using (var connection = factory.CreateConnection())
                using (var channel = connection.CreateModel())
                {
                    //Direct Exchange Details like name and type of exchange
                    channel.ExchangeDeclare(StaticConfigurationManager.AppSetting["RabbitMqSettings:ExchangeName"], StaticConfigurationManager.AppSetting["RabbitMqSettings:ExchhangeType"]);

                    //Declare Queue with Name and a few property related to Queue like durabality of msg, auto delete and many more
                    channel.QueueDeclare(queue: StaticConfigurationManager.AppSetting["RabbitMqSettings:QueueName"],
                                         durable: true,
                                         exclusive: false,
                                         autoDelete: false,
                                         arguments: null);

                    //Bind Queue with Exhange and routing details
                    channel.QueueBind(queue: StaticConfigurationManager.AppSetting["RabbitMqSettings:QueueName"], exchange: StaticConfigurationManager.AppSetting["RabbitMqSettings:ExchangeName"], routingKey: StaticConfigurationManager.AppSetting["RabbitMqSettings:RouteKey"]);

                    //Seriliaze object using Newtonsoft library
                    string productDetail = JsonConvert.SerializeObject(productOfferDetails);
                    var body = Encoding.UTF8.GetBytes(productDetail);

                    var properties = channel.CreateBasicProperties();
                    properties.Persistent = true;

                    //publish msg 
                    channel.BasicPublish(exchange: StaticConfigurationManager.AppSetting["RabbitMqSettings:ExchangeName"],
                                         routingKey: StaticConfigurationManager.AppSetting["RabbitMqSettings:RouteKey"],
                                         basicProperties: properties,
                                         body: body);

                    return true;
                }
            }
       
            catch (Exception)
            {
            }
            return false;
        }
    }
}

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.AspNetCore": "Warning"
    }
  },
  "AllowedHosts": "*",
  "ConnectionStrings": {
    "DefaultConnection": "Data Source=DESKTOP;Initial Catalog=ProductOwnerServiceDB;User Id=sa;Password=database;"
  },
  "RabbitMQ": {
    "RabbitURL": "localhost",
    "Username": "guest",
    "Password": "guest"
  },
  "RabbitMqSettings": {
    "ExchangeName": "OfferExchange",
    "ExchhangeType": "direct",
    "QueueName": "offer_queue",
    "RouteKey": "offer_route"
  }
}

using ProductOwner.Microservice.Data;
using ProductOwner.Microservice.Services;

var builder = WebApplication.CreateBuilder(args);

// Add services to the container.
builder.Services.AddScoped<IProductService, ProductService>();
builder.Services.AddDbContext<DbContextClass>();

builder.Services.AddControllers();
// Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();

var app = builder.Build();

// Configure the HTTP request pipeline.

app.UseSwagger();
app.UseSwaggerUI();


app.UseHttpsRedirection();

app.UseAuthorization();

app.MapControllers();

app.Run();

using Microsoft.AspNetCore.Mvc;
using ProductOwner.Microservice.Model;
using ProductOwner.Microservice.Services;

namespace ProductOwner.Microservice.Controllers
{
    [Route("api/[controller]")]
    [ApiController]
    public class ProductsController : ControllerBase
    {
        private readonly IProductService productService;

        public ProductsController(IProductService _productService)
        {
            productService = _productService;
        }

        [HttpGet("list")]
        public Task<IEnumerable<ProductDetails>> ProductListAsync()
        {
            var productList = productService.GetProductListAsync();
            return productList;

        }
        [HttpGet("filterlist")]
        public Task<ProductDetails> GetProductByIdAsync(int Id)
        {
            return productService.GetProductByIdAsync(Id);
        }

        [HttpPost("addproduct")]
        public Task<ProductDetails> AddProductAsync(ProductDetails product)
        {
            var productData = productService.AddProductAsync(product);
            return productData;
        }

        [HttpPost("sendoffer")]
        public bool SendProductOfferAsync(ProductOfferDetail productOfferDetails)
        {
            bool isSent = false;
            if (productOfferDetails != null)
            {
                isSent = productService.SendProductOffer(productOfferDetails);

                return isSent;
            }
            return isSent;
        }
    }
}

Step 16

Apply migration and update database using the following command

add-migration "Initial"

update-database

Let’s start the Implementation of Product User Service

Step 1

Install the following NuGet Packages

Step 2

Create Product Offer Details class inside model

namespace ProductUser.Microservice.Model
{
    public class ProductOfferDetail
    {
        public int Id { get; set; }
        public int ProductID { get; set; }
        public string ProductName { get; set; }
        public string ProductOfferDetails { get; set; }
    }
}

Step 3

Next, add DbContextClass inside the Data folder for data manipulation

using Microsoft.EntityFrameworkCore;
using ProductUser.Microservice.Model;

namespace ProductUser.Microservice.Data
{
    public class DbContextClass : DbContext
    {
        protected readonly IConfiguration Configuration;

        public DbContextClass(IConfiguration configuration)
        {
            Configuration = configuration;
        }
        protected override void OnConfiguring(DbContextOptionsBuilder options)
        {
            options.UseSqlServer(Configuration.GetConnectionString("DefaultConnection"));
        }

        public DbSet<ProductOfferDetail> ProductOffers { get; set; }
    }
}

Step 4

Later on, create Static Configuration Manager to configure the appsettings.json file to retrieve environmental variables

namespace ProductUser.Microservice.Utility
{
    static class StaticConfigurationManager
    {
        public static IConfiguration AppSetting
        {
            get;
        }
        static StaticConfigurationManager()
        {
            AppSetting = new ConfigurationBuilder().SetBasePath(Directory.GetCurrentDirectory()).AddJsonFile("appsettings.json").Build();
        }
    }
}

Step 5

Create IUserService and UserService class for data manipulation

IUserService

using ProductUser.Microservice.Model;

namespace ProductUser.Microservice.Services
{
    public interface IUserService
    {
        public Task<IEnumerable<ProductOfferDetail>> GetProductListAsync();
        public Task<ProductOfferDetail> GetProductByIdAsync(int id);
        public Task<ProductOfferDetail> AddProductAsync(ProductOfferDetail product);
    }
}

UserService

using Microsoft.EntityFrameworkCore;
using ProductUser.Microservice.Data;
using ProductUser.Microservice.Model;

namespace ProductUser.Microservice.Services
{
    public class UserService : IUserService
    {
        private readonly DbContextClass _dbContext;

        public UserService(DbContextClass dbContext)
        {
            _dbContext = dbContext;
        }

        public async Task<ProductOfferDetail> GetProductByIdAsync(int id)
        {
            return await _dbContext.ProductOffers.Where(x => x.Id == id).FirstOrDefaultAsync();
        }

        public async Task<IEnumerable<ProductOfferDetail>> GetProductListAsync()
        {
            return await _dbContext.ProductOffers.ToListAsync();
        }

        public async Task<ProductOfferDetail> AddProductAsync(ProductOfferDetail product)
        {
            var result = _dbContext.ProductOffers.Add(product);
            await _dbContext.SaveChangesAsync();
            return result.Entity;
        }
    }
}

Step 6

Add the connection string and some RabbitMQ server settings related to configuration

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.AspNetCore": "Warning"
    }
  },
  "AllowedHosts": "*",
  "ConnectionStrings": {
    "DefaultConnection": "Data Source=DESKTOP;Initial Catalog=ProductUserServiceDB;User Id=sa;Password=database;"
  },
  "RabbitMQ": {
    "RabbitURL": "localhost",
    "Username": "guest",
    "Password": "guest"
  },
  "RabbitMqSettings": {
    "ExchangeName": "OfferExchange",
    "ExchhangeType": "direct",
    "QueueName": "offer_queue",
    "RouteKey": "offer_route"
  }
}

Step 7

Register a few services inside Program.cs class

using ProductUser.Microservice.BackgroundServices;
using ProductUser.Microservice.Data;
using ProductUser.Microservice.Services;

var builder = WebApplication.CreateBuilder(args);

// Add services to the container.
builder.Services.AddScoped<IUserService, UserService>();
builder.Services.AddDbContext<DbContextClass>();
builder.Services.AddHostedService<RabbitMQBackgroundConsumerService>();

builder.Services.AddControllers();
// Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();

var app = builder.Build();

// Configure the HTTP request pipeline.

app.UseSwagger();
app.UseSwaggerUI();

app.UseHttpsRedirection();

app.UseAuthorization();

app.MapControllers();

app.Run();

Step 8

Add RabbitMQ Background Consumer Service which is run continuously in the background and checks the product offer message in the message queue which is produced by the producer

using Newtonsoft.Json;
using ProductUser.Microservice.Data;
using ProductUser.Microservice.Model;
using ProductUser.Microservice.Utility;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

namespace ProductUser.Microservice.BackgroundServices
{
    public class RabbitMQBackgroundConsumerService : BackgroundService
    {
        private IConnection _connection;
        private IModel _channel;
        private IServiceScopeFactory serviceScopeFactory;

        public RabbitMQBackgroundConsumerService(IServiceScopeFactory _serviceScopeFactory)
        {
            serviceScopeFactory = _serviceScopeFactory;
            InitRabbitMQ();
        }

        private void InitRabbitMQ()
        {
            var RabbitMQServer = "";
            var RabbitMQUserName = "";
            var RabbutMQPassword = "";

            if (Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT") == "Production")
            {
                RabbitMQServer = Environment.GetEnvironmentVariable("RABBIT_MQ_SERVER");
                RabbitMQUserName = Environment.GetEnvironmentVariable("RABBIT_MQ_USERNAME");
                RabbutMQPassword = Environment.GetEnvironmentVariable("RABBIT_MQ_PASSWORD");

            }
            else
            {
                RabbitMQServer = StaticConfigurationManager.AppSetting["RabbitMQ:RabbitURL"];
                RabbitMQUserName = StaticConfigurationManager.AppSetting["RabbitMQ:Username"];
                RabbutMQPassword = StaticConfigurationManager.AppSetting["RabbitMQ:Password"];
            }

            var factory = new ConnectionFactory()
            { HostName = RabbitMQServer, UserName = RabbitMQUserName, Password = RabbutMQPassword };

            // create connection  
            _connection = factory.CreateConnection();

            // create channel  
            _channel = _connection.CreateModel();

            //Direct Exchange Details like name and type of exchange
            _channel.ExchangeDeclare(StaticConfigurationManager.AppSetting["RabbitMqSettings:ExchangeName"], StaticConfigurationManager.AppSetting["RabbitMqSettings:ExchhangeType"]);

            //Declare Queue with Name and a few property related to Queue like durabality of msg, auto delete and many more
            _channel.QueueDeclare(queue: StaticConfigurationManager.AppSetting["RabbitMqSettings:QueueName"],
                        durable: true,
                        exclusive: false,
                        autoDelete: false,
                        arguments: null);


            _channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

            _channel.QueueBind(queue: StaticConfigurationManager.AppSetting["RabbitMqSettings:QueueName"], exchange: StaticConfigurationManager.AppSetting["RabbitMqSettings:ExchangeName"], routingKey: StaticConfigurationManager.AppSetting["RabbitMqSettings:RouteKey"]);

            _connection.ConnectionShutdown += RabbitMQ_ConnectionShutdown;
        }

        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            stoppingToken.ThrowIfCancellationRequested();

            var consumer = new EventingBasicConsumer(_channel);
            consumer.Received += (ch, ea) =>
            {
                // received message  
                var content = System.Text.Encoding.UTF8.GetString(ea.Body.ToArray());

                // acknowledge the received message  
                _channel.BasicAck(ea.DeliveryTag, false);

                //Deserilized Message
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                var productDetails = JsonConvert.DeserializeObject<ProductOfferDetail>(message);

                //Stored Offer Details into the Database
                using (var scope = serviceScopeFactory.CreateScope())
                {
                    var _dbContext = scope.ServiceProvider.GetRequiredService<DbContextClass>();
                    var result = _dbContext.ProductOffers.Add(productDetails);
                    _dbContext.SaveChanges();
                }

            };

            consumer.Shutdown += OnConsumerShutdown;
            consumer.Registered += OnConsumerRegistered;
            consumer.Unregistered += OnConsumerUnregistered;
            consumer.ConsumerCancelled += OnConsumerConsumerCancelled;

            _channel.BasicConsume(StaticConfigurationManager.AppSetting["RabbitMqSettings:QueueName"], false, consumer);
            return Task.CompletedTask;
        }

        private void OnConsumerConsumerCancelled(object sender, ConsumerEventArgs e) { }
        private void OnConsumerUnregistered(object sender, ConsumerEventArgs e) { }
        private void OnConsumerRegistered(object sender, ConsumerEventArgs e) { }
        private void OnConsumerShutdown(object sender, ShutdownEventArgs e) { }
        private void RabbitMQ_ConnectionShutdown(object sender, ShutdownEventArgs e) { }

        public override void Dispose()
        {
            _channel.Close();
            _connection.Close();
            base.Dispose();
        }
    }
}

Step 9

Next, create User Offers Controller

using Microsoft.AspNetCore.Mvc;
using ProductUser.Microservice.Model;
using ProductUser.Microservice.Services;

namespace ProductUser.Microservice.Controllers
{
    [Route("api/[controller]")]
    [ApiController]
    public class UserOffersController : ControllerBase
    {
        private readonly IUserService userService;

        public UserOffersController(IUserService _userService)
        {
           userService = _userService;
        }

        [HttpGet("offerlist")]
        public Task<IEnumerable<ProductOfferDetail>> ProductListAsync()
        {
            var productList = userService.GetProductListAsync();
            return productList;

        }

        [HttpGet("getofferbyid")]
        public Task<ProductOfferDetail> GetProductByIdAsync(int Id)
        {
            return userService.GetProductByIdAsync(Id);
        }
    }
}

Step 10

Add migration and update database using the following command

add-migration "Initial"

update-database

RabbitMQ Message Queue Setup using Docker

Install RabbitMQ docker image using the following command (Note- docker desktop is in running mode):

docker pull rabbitmq:3-management

Next, create a container and start using the RabbitMQ Image that we downloaded:

docker run --rm -it -p 15672:15672 -p 5672:5672 rabbitmq:3-management

Open the following URL to open the RabbitMQ dashboard on the port we set while running docker:

http://localhost:15672/

When you click the URL, the login page will open.

Login using the default username and password guest

Here you can see the offer queue when you run the producer and consumer applications. Also, details about the message which are ready and all things related to that.

Containerization of Producer and Consumer using Docker

(Note- I’m not using Visual Studio to create docker related things)

Step 1

Create a Docker file for Product Owner Microservice inside the root folder.

FROM mcr.microsoft.com/dotnet/sdk:6.0 AS build
WORKDIR /app

EXPOSE 443
EXPOSE 80 

# copy project csproj file and restore it in docker directory
COPY ./*.csproj ./
RUN dotnet restore

# Copy everything into the docker directory and build
COPY . .
RUN dotnet publish -c Release -o out

# Build runtime final image
FROM mcr.microsoft.com/dotnet/aspnet:6.0
WORKDIR /app
COPY --from=build /app/out .
ENTRYPOINT ["dotnet", "ProductOwner.Microservice.dll"]

Step 2

Next, create a Docker file for Product User Microservice inside the root folder.

FROM mcr.microsoft.com/dotnet/sdk:6.0 AS build
WORKDIR /app

EXPOSE 443
EXPOSE 80 

# copy project csproj file and restore it in docker directory
COPY ./*.csproj ./
RUN dotnet restore

# Copy everything into the docker directory and build
COPY . .
RUN dotnet publish -c Release -o out

# Build runtime final image
FROM mcr.microsoft.com/dotnet/aspnet:6.0
WORKDIR /app
COPY --from=build /app/out .
ENTRYPOINT ["dotnet", "ProductUser.Microservice.dll"]

Step 3

Create a docker-compose.yml file inside the main solution folder.

version: '3.5'
services:
  ProductOwner.Microservice:
   image: ${DOCKER_REGISTRY-}ownerservice:v1
   build:
    context: ./ProductOwner.Microservice
    dockerfile: Dockerfile
   environment: 
    - ASPNETCORE_ENVIRONMENT=Production
    - CONNECTIONSTRINGS__DEFAULTCONNECTION=Data Source=192.168.2.1,1433;Initial Catalog=ProductOwnerServiceDB;User Id=sa;Password=database
    - RABBIT_MQ_SERVER=192.168.2.1
    - RABBIT_MQ_USERNAME=guest
    - RABBIT_MQ_PASSWORD=guest
    - RABBITMQSETTINGS__EXCHANGENAME=OfferExchange
    - RABBITMQSETTINGS__EXCHHANGETYPE=direct
    - RABBITMQSETTINGS__QUEUENAME=offer_queue
    - RABBITMQSETTINGS__ROUTEKEY=offer_route
   ports:
    - "4201:80"
  ProductUser.Microservice:
   image: ${DOCKER_REGISTRY-}userservice:v1
   build:
    context: ./ProductUser.Microservice
    dockerfile: Dockerfile
   environment:
    - ASPNETCORE_ENVIRONMENT=Production   
    - CONNECTIONSTRINGS__DEFAULTCONNECTION=Data Source=192.168.2.1,1433;Initial Catalog=ProductUserServiceDB;User Id=sa;Password=database
    - RABBIT_MQ_SERVER=192.168.2.1
    - RABBIT_MQ_USERNAME=guest
    - RABBIT_MQ_PASSWORD=guest
    - RABBITMQSETTINGS__EXCHANGENAME=OfferExchange
    - RABBITMQSETTINGS__EXCHHANGETYPE=direct
    - RABBITMQSETTINGS__QUEUENAME=offer_queue
    - RABBITMQSETTINGS__ROUTEKEY=offer_route
   ports:
    - "4202:80"
  

• Here you can see that we used the environment section to override the connection string that is present in the appsetting.json file and some environmental variables
• Also, we put our machine IP Address over there in the connection string and port on which our SQL Server is running mode. Because if you put the server’s name, it will show some error while you are navigating your application which is run in a docker container.
• You can get your IP address using the ipconfig command through CMD.

Step 4

Open CMD inside the main solution folder where the docker-compose.yml file is present and execute the following command.

docker-compose build

docker-compose up

Step 5

Open Docker Desktop and you can see inside that there are three images created and is in running mode.

Step 6

Open the Container section and there you will see your two images running inside one container and another one related to RabbitMQ is on another conatiner with their individual port number.

Step 7

Use the following URLs to check your microservices functionality which is running inside a docker container.

http://localhost:4201/swagger/index.html

http://localhost:4202/swagger/index.html

Step 8

Now when we send some offers using product owner microservice then it will immediately listen by consumer user background service from the RabbitMQ Message Queue and stored in the database of User’s Services

Step 9)

Here you will see offer details which is sent by producer and our background services stored that message inside database

This is a simple scenario for understanding purposes there are many real-time scenarios in which RabbitMQ plays a very important role. Suppose consumer service is offline and the producer produces some message related to offers that persist inside the message queue and when the user is online and then uses services is in a running mode that will take a message from the message queue and store it inside the database. RabbitMQ helps to decouple applications in many cases like this.

Conclusion

Here we discussed RabbitMQ and real-time scenarios with the help of Message Queue. Also, the containerization of microservices using Docker and asynchronous communication between them.

Happy Learning!