Source des événements dans ASP.NET Core: Store change comme événements

Apprenez à implémenter le modèle architectural d’approvisionnement en événement dans votre application ASP.NET Core pour enregistrer les changements de système liés à l’événement.

Avez-vous déjà imaginé avoir un historique complet de données dans une application distribuée? Avec l’approvisionnement en événements, chaque changement devient un événement immuable, pour une traçabilité complète dans le noyau ASP.NET.

L’approvisionnement en événements est un modèle architectural largement utilisé dans les systèmes de grande et moyenne taille en raison de sa polyvalence dans le traitement de la complexité à travers une approche différente des approches traditionnelles: au lieu d’économiser l’état actuel, il enregistre chaque changement du système en tant qu’événement.

Dans cet article, nous comprendrons comment fonctionne l’approvisionnement en événements et quels sont ses avantages par rapport aux approches traditionnelles. Nous implémenterons également une application Core ASP.NET en utilisant les principes de l’approvisionnement en événements.

Comprendre l’approvisionnement des événements

L’approvisionnement en événements est un modèle architectural qui vise à gérer les changements grâce à des événements immuables, en stockant l’historique complet de toutes les modifications apportées à un système, permettant la reconstruction de l’État à tout moment.

Imaginez une plateforme de gestion des comptes bancaire. Au lieu d’avoir une entité de compte, qui a un solde qui est mis à jour avec chaque nouvelle transaction, dans Sourcing d’événements, chaque transaction (dépôt, retrait, transfert) est enregistrée comme un événement qui ne change pas. Chaque événement est enregistré exactement tel qu’il a été envoyé, de sorte que le solde actuel du compte est dérivé des montants déposés et retirés dans les événements précédents. Cela permet de suivre l’historique complet des mouvements de compte, de sorte que toutes les informations du premier changement au dernier sont présentes.

L’image ci-dessous montre comment les mouvements d’un compte bancaire sont gérés en utilisant les approches traditionnelles et d’approvisionnement en événements.

Avantages de l’utilisation d’événements

L’utilisation des événements apporte plusieurs avantages, en particulier dans les systèmes qui doivent maintenir la traçabilité (histoire) et être évolutives et résilientes. Voici quelques-uns des principaux aspects qui font de l’utilisation des événements un excellent choix:

Terminer l’historique des données: L’utilisation des événements vous permet d’avoir une histoire immuable de tout ce qui s’est passé. Cela facilite les audits, le débogage et l’analyse du comportement du système pendant son cycle de vie.

Reproduction et reconstruction de l’État: Il est possible de reconstruire l’état actuel de toute entité, simplement en réappliquant les événements passés.

Évolutivité et performance: L’utilisation des événements permet la séparation entre l’écriture (événements) et la lecture (projections), permettant des optimisations pour chacune de manière spécifique. De plus, les systèmes peuvent être conçus pour évoluer horizontalement, car les événements sont stockés de manière asynchrone.

Intégration asynchrone facile: Permet la création de microservices découplés, où les services réagissent aux événements sans avoir besoin d’appels directs.

Résilience et tolérance aux défauts: En cas d’échec, retraitez simplement les événements pour restaurer l’état du système. De plus, il n’y a aucun risque de perte de données, car les événements sont stockés de manière immuable et sont toujours disponibles.

Implémentation de l’approvisionnement des événements dans ASP.NET Core

Nous allons maintenant implémenter une application Core ASP.NET qui utilise le modèle d’approvisionnement d’événements pour envoyer et recevoir des événements de dépôt et de retrait. Pour cela, nous utiliserons deux bibliothèques: RabbitMQ et MassTransit.

RabbitMQ est un courtier de messages qui implémente le protocole de file d’attente de messages avancé (AMQP). Il agit comme un intermédiaire entre les producteurs de messages et les consommateurs, leur permettant de communiquer de manière découplée et asynchrone.

MassTransit est une bibliothèque .NET pour la messagerie asynchrone. Il simplifie l’intégration avec des courtiers de messages comme RabbitMQ ou Azure Service en abstraction des détails complexes et en prenant en charge les modèles de messagerie tels que la publication / s’abonner, la demande / réponse et l’orchestration de saga.

Condition préalable

Vous trouverez ci-dessous quelques conditions préalables que vous devez rencontrer pour reproduire le tutoriel dans le post.

1. Version .NET – Vous aurez besoin de la dernière version de .NET (.NET 9 ou supérieure)
2. Docker – requis pour exécuter le service RabbitMQ local

Création de l’application et installation des dépendances

Pour créer l’application, vous pouvez utiliser la commande ci-dessous:

dotnet new web -o BankStream

Ensuite, dans la racine du projet, vous pouvez utiliser les commandes suivantes pour installer les packages NuGet:

dotnet add package Microsoft.EntityFrameworkCore
dotnet add package Microsoft.EntityFrameworkCore.Sqlite
dotnet add package Microsoft.EntityFrameworkCore.Design
dotnet add package MassTransit
dotnet add package MassTransit.RabbitMQ
dotnet add package MassTransit.AspNetCore

Créer les événements

Créons maintenant les entités qui représentent les événements de dépôt et de retrait. Ils seront simples, car l’objectif est de démontrer l’envoi et la consommation des événements.

Alors, créez un nouveau dossier appelé «Events» et, à l’intérieur, créez les enregistrements suivants:

namespace BankStream.Events;

public record DepositEvent(Guid Id, string AccountId, decimal Amount);

namespace BankStream.Events;

public record WithdrawalEvent(Guid Id, string AccountId, decimal Amount);

Créer les consommateurs

Ensuite, créons les consommateurs. Ils sont responsables de la réception des événements et du traitement des actions qui y sont liées. Dans ce cas, mettez à jour l’état et enregistrez-le dans la base de données.

Tout d’abord, créons les classes et les énumérements utilisés par les consommateurs. Alors, créez un nouveau dossier appelé «modèles» et, à l’intérieur, créez les classes suivantes:

namespace BankStream.Models;
public enum StatusEnum
{
    Pending,
    Completed,
    Failed,
    Canceled,
    Reversed,
    InProgress,
    OnHold
}

namespace BankStream.Models;

public enum TransactionTypeEnum
{
    Deposit,
    Withdrawal
}

namespace BankStream.Models;

public class TransactionStatus
{
    public TransactionStatus()
    {
    }

    public TransactionStatus(Guid id, string accountId, StatusEnum statusEnum, decimal amount, bool isSuccess, string? errorMessage, TransactionTypeEnum type, DateTime createdAt)
    {
        Id = id;
        AccountId = accountId;
        StatusEnum = statusEnum;
        Amount = amount;
        IsSuccess = isSuccess;
        ErrorMessage = errorMessage;
        Type = type;
        CreatedAt = createdAt;
    }

    public Guid Id { get; set; }
    public string AccountId { get; set; }
    public StatusEnum StatusEnum { get; set;}
    public decimal Amount { get; set; }
    public bool IsSuccess { get; set; }
    public string? ErrorMessage { get; set; }
    public TransactionTypeEnum Type { get; set; }
    public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
}

Ensuite, créez un nouveau dossier appelé «données» et à l’intérieur, ajoutez la classe ci-dessous:

using BankStream.Events;
using BankStream.Models;
using Microsoft.EntityFrameworkCore;

namespace BankStream.Data;

public class EventDbContext : DbContext
{
    public DbSet<DepositEvent> Deposits { get; set; }
    public DbSet<WithdrawalEvent> Withdrawals { get; set; }
    public DbSet<TransactionStatus> TransactionStatus { get; set; }

    public EventDbContext(DbContextOptions<EventDbContext> options)
        : base(options) { }
}

Maintenant, créons les classes grand public. Créez un nouveau dossier appelé «consommateurs» et ajoutez-y les classes suivantes:

using BankStream.Data;
using BankStream.Events;
using BankStream.Models;
using MassTransit;

namespace BankStream.Consumers;

public class DepositConsumer : IConsumer<DepositEvent>
{
    private readonly EventDbContext _dbContext;

    public DepositConsumer(EventDbContext dbContext) => _dbContext = dbContext;

    public async Task Consume(ConsumeContext<DepositEvent> context)
    {
        var deposit = context.Message;

        try
        {
            await _dbContext.TransactionStatus.AddAsync(new TransactionStatus(
                Guid.NewGuid(),
                deposit.AccountId,
                StatusEnum.Completed,
                deposit.Amount,
                true,
                string.Empty,
                TransactionTypeEnum.Deposit,
                DateTime.Now)
            );

            await _dbContext.Deposits.AddAsync(deposit);

            await _dbContext.SaveChangesAsync();
        }
        catch (Exception ex)
        {
            await _dbContext.TransactionStatus.AddAsync(new TransactionStatus(
                Guid.NewGuid(),
                deposit.AccountId,
                StatusEnum.Failed,
                deposit.Amount,
                true,
                ex.Message,
                TransactionTypeEnum.Deposit,
                DateTime.Now)
            );
            await _dbContext.SaveChangesAsync();
            throw;
        }
    }
}

using BankStream.Data;
using BankStream.Events;
using BankStream.Models;
using MassTransit;

namespace BankStream.Consumers;

public class WithdrawalConsumer : IConsumer<WithdrawalEvent>
{
    private readonly EventDbContext _dbContext;

    public WithdrawalConsumer(EventDbContext dbContext) => _dbContext = dbContext;

    public async Task Consume(ConsumeContext<WithdrawalEvent> context)
    {
        var withdrawal = context.Message;
        try
        {
            await _dbContext.TransactionStatus.AddAsync(new TransactionStatus(
                Guid.NewGuid(),
                withdrawal.AccountId,
                StatusEnum.Completed,
                withdrawal.Amount,
                true,
                string.Empty,
                TransactionTypeEnum.Withdrawal,
                DateTime.Now)
            );

            await _dbContext.Withdrawals.AddAsync(context.Message);
            await _dbContext.SaveChangesAsync();
        }
        catch (Exception ex)
        {
            await _dbContext.TransactionStatus.AddAsync(new TransactionStatus(
                Guid.NewGuid(),
                withdrawal.AccountId,
                StatusEnum.Failed,
                withdrawal.Amount,
                false,
                ex.Message,
                TransactionTypeEnum.Withdrawal,
                DateTime.Now)
            );
            await _dbContext.SaveChangesAsync();
            throw;
        }
    }
}

Maintenant, analysons ce qui est fait chez les deux consommateurs.

Les classes ci-dessus DepositConsumer et WithdrawalConsumer utiliser Masstransit pour traiter les événements à travers le IConsumer<T> interface.

Lorsqu’un événement de transaction arrive, le Consume() La méthode extrait les données du message reçu et la persiste dans la base de données. Dans les deux classes, un nouvel enregistrement d’état de transaction (TransactionStatus) est créé, ce qui permet de suivre les actions qui se sont produites pendant le traitement.

Une fois l’état de la transaction créé, les événements de dépôt et de retrait sont ajoutés à leurs ensembles de données de base de données (_dbContext.Deposits et _dbContext.Withdrawals). Enfin, les changements sont persistés dans la base de données via le SaveChangesAsync() appel.

Création de la classe de contrôleur

Dans la classe du contrôleur, nous ajouterons les points de terminaison pour le dépôt et le retrait qui appellent les événements. Donc, créez un nouveau dossier appelé «contrôleurs» et, à l’intérieur, ajoutez la classe de contrôleur suivante:

using BankStream.Data;
using BankStream.Events;
using BankStream.Models;
using MassTransit;
using Microsoft.AspNetCore.Mvc;

namespace BankStream.Controllers;

[ApiController]
[Route("api/accounts")]
public class AccountController : ControllerBase
{
    private readonly IBus _bus;
    private readonly EventDbContext _dbContext;

    public AccountController(IBus bus, EventDbContext dbContext)
    {
        _bus = bus;
        _dbContext = dbContext;
    }

    [HttpPost("deposit")]
    public async Task<IActionResult> Deposit([FromBody] DepositEvent deposit)
    {
        await _dbContext.TransactionStatus.AddAsync(new TransactionStatus(
            Guid.NewGuid(),
            deposit.AccountId,
            StatusEnum.Pending,
            deposit.Amount,
            false,
            string.Empty,
            TransactionTypeEnum.Deposit,
            DateTime.Now)
        );
        await _dbContext.SaveChangesAsync();

        await _bus.Publish(deposit);
        return Accepted();
    }

    [HttpPost("withdrawal")]
    public async Task<IActionResult> Withdraw([FromBody] WithdrawalEvent withdrawal)
    {
        await _dbContext.TransactionStatus.AddAsync(new TransactionStatus(
             Guid.NewGuid(),
             withdrawal.AccountId,
             StatusEnum.Pending,
             withdrawal.Amount,
             false,
             string.Empty,
             TransactionTypeEnum.Withdrawal,
             DateTime.Now)
         );
        await _dbContext.SaveChangesAsync();

        await _bus.Publish(withdrawal);
        return Accepted();
    }

   [HttpGet("{accountId}/balance")]
    public IActionResult GetBalance(string accountId)
    {
        var deposits = _dbContext.
            Deposits.
            Where(d => d.AccountId == accountId)
            .Sum(d => d.Amount);

        var withdrawals = _dbContext
            .Withdrawals
            .Where(w => w.AccountId == accountId)
            .Sum(w => w.Amount);

        return Ok(deposits - withdrawals);
    }
}

Ici, le contrôleur utilise le IBus Interface, qui est le bus de message MassTransit utilisé pour publier des événements.

Lorsqu’un client fait une demande de poste HTTP au /api/accounts/deposit point final, le Deposit() la méthode reçoit un DepositEventreprésentant la demande de dépôt. Avant de publier l’événement dans le bus, un état de transaction en attente est enregistré dans la base de données, indiquant que l’opération a été demandée mais n’a pas encore été terminée. Après avoir persisté ce statut, l’événement est publié avec _bus.Publish(deposit)permettant aux consommateurs asynchrones tels que DepositConsumer pour traiter la transaction.

Le même flux s’applique au Withdraw méthode, qui répond au /api/accounts/withdrawal point de terminaison. Il reçoit un WithdrawalEventenregistre la transaction en attente dans la base de données et publie l’événement dans le bus pour le traitement ultérieur par le WithdrawalConsumer.

En plus de ces deux points de terminaison, il y en a un troisième qui est utilisé pour vérifier le solde. Pour ce faire, au lieu de simplement récupérer la valeur de la base de données comme dans les approches traditionnelles, il utilise le principe de l’approvisionnement en événements pour reconstruire l’état actuel à travers l’historique des transactions. Ainsi, d’abord, les totaux des dépôts et des retraits sont obtenus, et la valeur de solde est obtenue en soustrayant les retraits totaux des dépôts totaux.

Configuration de la classe de programme

Dans la classe de programme, nous configurerons le EventDbContext classe pour utiliser la base de données SQLite. De plus, nous configurerons MassTransit et RabbitMQ pour envoyer et consommer des événements.

Alors, ouvrez la classe de programme de votre application et remplacez ce qui y est par le code suivant:

using BankStream.Consumers;
using BankStream.Data;
using MassTransit;
using Microsoft.EntityFrameworkCore;

var builder = WebApplication.CreateBuilder(args);

builder
    .Services
    .AddDbContext<EventDbContext>(options => options.UseSqlite("Data Source=events.db"));
builder
    .Services
    .AddMassTransit(x =>
    {
        x.AddConsumer<DepositConsumer>();
        x.AddConsumer<WithdrawalConsumer>();
        x.UsingRabbitMq(
            (context, cfg) =>
            {
                cfg.Host("rabbitmq://localhost");
                cfg.ReceiveEndpoint(
                    "event-queue",
                    e =>
                    {
                        e.SetQueueArgument("x-message-ttl", 500000);
                        e.ConfigureConsumer<DepositConsumer>(context);
                        e.ConfigureConsumer<WithdrawalConsumer>(context);
                    }
                );
            }
        );
    });

builder.Services.AddControllers();

var app = builder.Build();

app.MapControllers();

app.Run();

Analysons le code ci-dessus.

Après la configuration de SQLite, MassTransit est enregistré, où les deux consommateurs sont configurés: DepositConsumer et WithdrawalConsumer. De plus, la communication avec RabbitMQ est établie pour transporter les messages, pour lesquels la connexion est établie avec un hôte local (rabbitmq://localhost).

Dans la configuration Rabbitmq, un point de terminaison récepteur appelé event-queue est défini. Cette file d’attente est configurée avec le SetQueueArgument() Méthode et un argument spécial, x-message-ttlqui détermine le temps de vivre des messages dans la file d’attente. Cela nous permet de supprimer automatiquement les messages expirés après 500 000 millisecondes (500 secondes). Cela est fait pour que les messages puissent être vérifiés dans le tableau de bord RabbitMQ. Enfin, les consommateurs sont liés à la file d’attente, de sorte que tout dépôt ou retrait reçu par RabbitMQ sera transmis au consommateur approprié pour le traitement.

Exécution des migrations EF

Pour créer la base de données et les tables, exécutez les commandes EF Core:

1. Commande pour installer EF si vous ne l’avez pas encore:

dotnet tool install --global dotnet-ef

2. Générer les fichiers de migration:

dotnet ef migrations add InitialCreate

3. Exécutez la persévérance:

dotnet ef database update

Configuration de Rabbitmq dans Docker

Pour exécuter l’application, vous devez faire fonctionner RabbitMQ localement. Pour ce faire, vous pouvez utiliser la commande ci-dessous pour télécharger l’image Rabbitmq et en exécuter un conteneur Docker:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

Après avoir exécuté la commande, vous pouvez vérifier le conteneur RabbitQM en cours d’exécution sur Docker Desktop:

Envoi et consommation d’événements

Exécutez maintenant l’application et demandez le point de terminaison ci-dessous. Ce message utilise le progrès Telerik Fiddler partout pour faire les demandes.

POSTE – http://localhost:PORT/api/accounts/deposit

{
  "id": "a1a7c1f2-3d4e-42b9-9e77-5f1e7c8b6c2f",
  "accountId": "123456789",
  "amount": 1300.90
}

Maintenant, accédez à la page d’accueil RabbitMQ à l’adresse indiquée dans Docker (vous pouvez utiliser le nom d’utilisateur et le mot de passe par défaut guest). Cliquez ensuite sur le menu «files d’attente et flux», puis sur «Obtenez des messages», et enfin «obtenir des messages».

Ainsi, vous pouvez vérifier les données envoyées dans le message, ainsi que d’autres données telles que le système d’exploitation, etc., comme indiqué dans les images ci-dessous:

Maintenant, exécutons le point de terminaison du retrait, alors faites une nouvelle demande au point final

POSTE – http://localhost:PORT/api/accounts/withdraw

{
"id": "fef6df04-df3f-46a7-99a7-6d9992dd3558",
"accountId": "123456789",
"amount": 200.00
}

Ensuite, faites une autre demande au point final ci-dessous pour récupérer le solde actuel:

OBTENIR – http://localhost:PORT/api/accounts/123456789/balance

Si tout se passe bien, vous obtiendrez le résultat ci-dessous comme indiqué dans l’image.

Notez que la valeur résultante était de 1100,90. Cette valeur n’existe pas dans la base de données mais est le résultat de la soustraction de la valeur de dépôt de 1300,90 à partir de la valeur de retrait de 200,00.

Conclusion

Que ce soit ou non le procès d’événements est un choix qui doit être fait en fonction des besoins du projet. Si vous devez garder un historique de modifications à un objet spécifique, comme un solde de compte bancaire, par exemple, l’approvisionnement en événement peut être un excellent choix. De plus, l’approvisionnement en événement présente d’autres avantages tels que l’évolutivité facile, les performances, le découplage, la résilience et la tolérance aux pannes.

Dans cet article, nous comprenons comment la source d’événements diffère des approches traditionnelles et comment mettre en œuvre un système de gestion du solde bancaire dans la pratique, en utilisant les ressources de RabbitMQ et MassTransit pour envoyer et consommer des messages.

Donc, j’espère que cet article vous aidera à comprendre et à implémenter l’approvisionnement en événements à l’aide de ressources de pointe comme ASP.NET Core, RabbitMQ, MassTransit et Docker.