Efficient Event Communication RabbitMQ: Implementing Event Bus and Outbox Patterns with CAP in .NET Core Microservices
Tags: webdev, programming, dotnet, opensource
Published: 2023-08-12 06:30:58

## Software's needed

1. Visual Studio 2022
2. Docker Desktop with RabbitMQ image.

##Introduction

In the process of building an SOA or MicroService system, we usually need to use the event to integrate each service. In the process, simple use of message queue does not guarantee reliability. CAP adopts local message table program integrated with the current database to solve exceptions that may occur in the process of the distributed system calling each other. It can ensure that the event messages are not lost in any case.

You can also use CAP as an EventBus. CAP provides a simpler way to implement event publishing and subscriptions. You do not need to inherit or implement any interface during subscription and sending process.
##what is CAP?
CAP is a library based on .net standard, which is a solution to deal with distributed transactions, also has the function of EventBus, it is lightweight, easy to use, and efficient.

**CAP implements the Outbox Pattern described in the eShop ebook**
![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/1gd5j4zd4wnfuty3tm96.png)

Learn how to build a microservices event bus architecture using CAP, which offers advantages over direct integration of message queues, and what out-of-the-box features it provides.

<img width="100%" style="width:100%" src="https://media.giphy.com/media/3aGZA6WLI9Jde/giphy.gif">

Create 2 new .NET core 8.0 web api projects name **Consumer** and **Publisher**.

![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/t44klaoi67o5p49yldk8.png)
Add these Nuget packages in **Publisher** web api project.
```csharp
 <PackageReference Include="DotNetCore.CAP" Version="7.1.4" />
 <PackageReference Include="DotNetCore.CAP.Dashboard" Version="7.1.4" />
 <PackageReference Include="DotNetCore.CAP.RabbitMQ" Version="7.1.4" />
 <PackageReference Include="DotNetCore.CAP.SqlServer" Version="7.1.4" />
 <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="8.0.0-preview.5.23302.2" />
 <PackageReference Include="Microsoft.EntityFrameworkCore" Version="6.0.20" />
 <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="6.0.20">
```
**Adding Customer Entity**
Create a new folder data and a new class named  **Customer**
that inherits from customerinsert classs.
 ```csharp
 public class Customer: CustomerInsert
 {
  
     [Key]
     public Guid Id { get; set; }= Guid.NewGuid();
 }
 public class CustomerInsert
 {
     public string FirstName { get; set; }
     public string LastName { get; set; }
     public string MobilNumber { get; set; }
 }
```
**Creating Db Context**
Create **ServiceDbContext** class that inherits from DbContext.
```csharp

    public class ServiceDbContext : DbContext
    {
        public DbSet<Customer> customers { get; set; }
        public ServiceDbContext(DbContextOptions<ServiceDbContext> options)
            : base(options)
        {
        }
        protected override void OnModelCreating(ModelBuilder modelBuilder)
        {
            base.OnModelCreating(modelBuilder);
        }

    }
```

Now right click controller folder and add a new controller name Customer Controller with this code.
```csharp
 [Route("api/[controller]")]
 [ApiController]
 public class CustomerController : ControllerBase
 {
 
     private readonly ICustomerService _customerService;


     public CustomerController(ICustomerService customerService)
     {
         _customerService = customerService;
  
     }

     // POST api/<OrderController>
     [HttpPost]
     public async Task<IActionResult> Post([FromBody] CustomerInsert customer)
     {
         try
         {
             return Ok( await _customerService.AddCustomer(customer));
         }
         catch (Exception ex)
         {

             throw;
         }
     }


 }
```
Create a new folder Services where we will create our **ICustomerService** and **CusomterService**

ICustomerService
```csharp
    public interface ICustomerService
    {
        Task<bool> AddCustomer(CustomerInsert order);

    }
```
Customer Service
```csharp
    public class CustomerService : ICustomerService
    {
        private readonly ServiceDbContext _dbContext;
        private readonly ICapPublisher _capPublisher;

        public CustomerService(ServiceDbContext dbContext, ICapPublisher capPublisher)
        {
            _dbContext = dbContext;
            _capPublisher = capPublisher;
        }

        public async Task<bool> AddCustomer(CustomerInsert customerInsert)
        {
            Customer customer = new Customer
            {
                FirstName = customerInsert.FirstName,
                LastName = customerInsert.LastName,
                MobilNumber = customerInsert.MobilNumber,
            };

            await _dbContext.AddAsync(customer);
            await _dbContext.SaveChangesAsync();
            var content = JsonSerializer.Serialize(customer);
            await _capPublisher.PublishAsync<string>("CustomerAdded", content);
            return true;
        }
    }

```
In our customer service we are injecting **ICapPublisher** and interface provided by **DotNetCore.CAP** package.
in our add customer method we are adding a new customer in our database and informing other services using cappublished event handler that an event is being published. That event that is being published will be stored in our database which we will see later.

Let Configure our **program.cs** and **appdevelopment.json** for using database and configuration of our services and events.
**appsettings**
```json
  "ConnectionStrings": {
    "DefaultConnection": "Data Source=.;Initial Catalog=Publisher;Trusted_Connection=True;Encrypt=False;"
  }
```
**Program.cs**
```csharp
#region Creating Db Context
builder.Services.AddDbContext<ServiceDbContext>(options => options.UseSqlServer(builder.Configuration.GetConnectionString("DefaultConnection")));
#endregion
#region Add Cap library
builder.Services.AddCap(options =>
{
    options.UseEntityFramework<ServiceDbContext>();
    options.UseSqlServer(builder.Configuration.GetConnectionString("DefaultConnection"));
    options.UseDashboard(path => path.PathMatch = "/cap-dashboard");
    options.UseRabbitMQ(options =>
    {
        options.ConnectionFactoryOptions = options =>
        {
            options.Ssl.Enabled = false;
            options.HostName = "localhost";
            options.UserName = "guest";
            options.Password = "guest";
            options.Port = 5672;
        };
    });
});

#endregion
///adding services
builder.Services.AddScoped<ICustomerService, CustomerService>();
```
Create New Database name **Publisher** in Sql Server Managment Studio.
Generate Migrations in **Publisher** web api project using .net CLI
`dotnet ef migrations add InitialCreate`
and update our database and schema using this command  
`dotnet ef database update`
> now our **publisher** project has been completed lets go to our **consumer** project and add the implementation for consuming our events that are being published.

add these Nuget packages in the project.
```csharp
	<PackageReference Include="DotNetCore.CAP" Version="7.1.4" />
	<PackageReference Include="DotNetCore.CAP.RabbitMQ" Version="7.1.4" />
	<PackageReference Include="DotNetCore.CAP.SqlServer" Version="7.1.4" />
	<PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="6.0.20" />
```
Create a new folder named event and add a class named **CustomerAddedEventSubscriber** that inherits from **ICapSubscribe** because we need to need listen to events that being published in this class. 
```csharp
    public class CustomerAddedEventSubscriber : ICapSubscribe
    {
        [CapSubscribe("CustomerAdded")]
        public void Consumer(JsonElement customerData)
        {
            Console.WriteLine(customerData);
        }
    }
```
we have added a cosnumer class that receives a json element as a pramater.

adding connection string in appsettings.json
> note the connection string needs to be same as our **publisher api project** since events are stored in database table.
```json
  "ConnectionStrings": {
    "DefaultConnection": "Data Source=.;Initial Catalog=Publisher;Trusted_Connection=True;Encrypt=False;"
  }
```
Create **DbContext** and Configure **Program.cs** file.

## DbContext
```csharp
    public class MainDbContext:DbContext
    {
     //   public DbSet<CustomerData> customers { get; set; }
        public MainDbContext(DbContextOptions<MainDbContext> options)
            : base(options)
        {
        }
        protected override void OnModelCreating(ModelBuilder modelBuilder)
        {
            base.OnModelCreating(modelBuilder);
        }

    }
```
## Program.cs
```csharp
// Add dbcontext.
builder.Services.AddDbContext<MainDbContext>(options => options.UseSqlServer(builder.Configuration.GetConnectionString("DefaultConnection")));
//add cap implementation
builder.Services.AddCap(options =>
{
    options.UseEntityFramework<MainDbContext>();
    options.UseSqlServer(builder.Configuration.GetConnectionString("DefaultConnection"));
  
    options.UseRabbitMQ(options =>
    {
        options.ConnectionFactoryOptions = options =>
        {
            options.Ssl.Enabled = false;
            options.HostName = "localhost";
            options.UserName = "guest";
            options.Password = "guest";
            options.Port = 5672;
        };
    });
});
//add event handler service
builder.Services.AddSingleton<CustomerAddedEventSubscriber>();
```
### Our both project are being configured 

let install **RabbitMQ** image in docker desktop using this command.

> docker pull rabbitmq:3.10-management

Run Docker desktop and run start our RabbitMq image present in images section.

![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/jd9s4vh7ycxzyl9bgkcx.png)
RabbitMQ image is running run lets run our project.

Right click your solution and click select startup project use multiple projects select consumer and publisher project.
Image Taken From RabbitMQ
![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/tvldrsy3pzbfhursm3xw.png)
Start the project 
<img width="100%" style="width:100%" src="https://media.giphy.com/media/uurtMtTKqkJda4dk8Y/giphy-downsized-large.gif">
 Go to the database first we can see CAP has Created 2 new tables for us **CAP.Published** and **CAP.Received** lets examine them.

![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/xt4vntfli10g3um23t72.png)

Both table has same number of column's. when we publish you event it will be saved in CAP.Published table and when our consumer consume the event the it will be added in CAP.Received table.
Go the Customer Controller and add a new customer so we can publish some events too.

![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/jeklfidbah0lncs7gq0i.png)



![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/wupvjzyr17uud5xj157a.png)
In Publisher app console the event is being added in database for publishing 

![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/5f4toxn5n3jcyusyczjp.png)


In Consumer Console we can see the output of our data the we published from Publisher api received successfully in consumer app.

![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/jcd2j4z2y8u2ykj9if3u.png)


Examine the database and select cap.published and received table

```sql
select * From cap.Published
select * From cap.Received 
```

![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/0zzjqd5ywilive1gfqhp.png)
**Name:** Event Name this is being published.
**Group:** Used to group Events
**Content:** This the main column where our data is present.
**Status Name:** Identifying that event is Succeeded or Failed.
In Content column we have a Json Element parsed and saved in our column from the event examine it and see your data there.
We can also see our data in **CAP Dashboard** that we added in publisher project by going to published and received tabs there. using the url the we configured for dashboard access in **Program.cs** file.
`https://localhost:{yourport}/cap-dashboard`

![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/vkat54xy84fuhblzu18t.png)


![Image](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/6xse6axt5mzgnzezgw61.png)


<img width="100%" style="width:100%" src="https://media.giphy.com/media/zTyOZBq1WUY78bmqUv/giphy.gif">

### Key Take Away
That’s the simplest example. And it’s pretty simple and hopefully gives you an idea of how CAP works with implementing the outbox pattern. There are various message brokers it supports such as Kafka, RabbitMQ, and Azure Service Bus. On the database side it supports SQL Server, MySQL, PostgreSQL, and MongoDB.

> In this project we are using EventBus implementation in our publish class but most of the time in real world application scenarios we implement EventBus implementation using generic classes in separate dll and that can be used in our overall microservices project. this is just simple demo project for beginners.