Ensuring message deduplication in a distributed system is crucial for maintaining data integrity and preventing unintended side effects. In this blog post, we'll explore how to implement a deduplication mechanism for RabbitMQ messages using C#, ASP.NET Core, EF Core, MassTransit, and the custom IDeduplicationService interface.

Warning

In this article, I skipped many of the basic concepts, this article is talking about an advanced topic, so, you need to be familiar with: RabbitMQ, MassTransit, Message Consumers, EF Core, Event-Driven Architecture.

Prerequisites

Before we begin, make sure you have the following components installed:

• Visual Studio or Visual Studio Code

• .NET 5.0 or later

• RabbitMQ server

  We talked before about:

  How to Setting Up RabbitMQ on Windows?

  Setting up and Using MassTransit with RabbitMQ in ASP.Net Core

• Entity Framework Core

Defining the Deduplication Service Interface

Create the IDeduplicationService interface to manage the deduplication logic.

// IDeduplicationService.cs

public interface IDeduplicationService
{
    Task<bool> IsDuplicateAsync(string messageId);
    Task MarkAsProcessedAsync(string messageId);
}

Implementing Deduplication with Entity Framework Core

Now, let's implement the deduplication service using Entity Framework Core.

// DeduplicationService.cs

public class DeduplicationService : IDeduplicationService
{
    private readonly YourDbContext _dbContext;

    public DeduplicationService(YourDbContext dbContext)
    {
        _dbContext = dbContext;
    }

    public async Task<bool> IsDuplicateAsync(string messageId)
    {
        return await _dbContext.DeduplicationEntries
            .AnyAsync(entry => entry.MessageId == messageId);
    }

    public async Task MarkAsProcessedAsync(string messageId)
    {
        _dbContext.DeduplicationEntries.Add(new DeduplicationEntry { MessageId = messageId });
        await _dbContext.SaveChangesAsync();
    }
}

Don't forget to create the DeduplicationEntry model and the YourDbContext class.

Integrating Deduplication in MassTransit Consumer

Finally, integrate the deduplication service in your MassTransit consumer.

// YourConsumer.cs

public class YourConsumer : IConsumer<YourMessage>
{
    private readonly IDeduplicationService _deduplicationService;

    public YourConsumer(IDeduplicationService deduplicationService)
    {
        _deduplicationService = deduplicationService;
    }

    public async Task Consume(ConsumeContext<YourMessage> context)
    {
        var messageId = context.Message.MessageId;

        if (await _deduplicationService.IsDuplicateAsync(messageId))
        {
            // Log or handle duplicate message
            return;
        }

        // Process the message

        await _deduplicationService.MarkAsProcessedAsync(messageId);
    }
}

Real-World Example: Order Processing System

Let's consider a real-world example of implementing deduplication in an order processing system. We'll create a scenario where orders are submitted through RabbitMQ messages, and we want to ensure that each order is processed exactly once.

1. Define the Order Message

Start by defining the message structure for order processing.

// OrderMessage.cs

public class OrderMessage
{
    public string OrderId { get; set; }
    // Other order details
}

2. Update Consumer to Handle Order Messages

Modify the consumer to handle OrderMessage instead of YourMessage.

// OrderConsumer.cs

public class OrderConsumer : IConsumer<OrderMessage>
{
    private readonly IDeduplicationService _deduplicationService;

    public OrderConsumer(IDeduplicationService deduplicationService)
    {
        _deduplicationService = deduplicationService;
    }

    public async Task Consume(ConsumeContext<OrderMessage> context)
    {
        var orderId = context.Message.OrderId;

        if (await _deduplicationService.IsDuplicateAsync(orderId))
        {
            // Log or handle duplicate order
            return;
        }

        // Process the order
        Console.WriteLine($"Processing Order: {orderId}");

        await _deduplicationService.MarkAsProcessedAsync(orderId);
    }
}