Skip to content
CodeDesignPlus

Overview

La arquitectura orientada a eventos (Event-Driven) es un estilo de arquitectura de software que promueve la producción, detección, consumo y reacción a eventos. Un evento es un cambio de estado o una actualización significativa en el sistema. Esta arquitectura es ideal para aplicaciones que requieren alta escalabilidad, flexibilidad y desacoplamiento entre componentes.

Este tipo de arquitectura se basa en patrón de publicación y suscripción (Pub/Sub), donde los componentes del sistema se comunican entre sí mediante eventos. Los eventos son mensajes que representan un cambio en el estado de una entidad o un cambio significativo en el sistema. Los eventos son generados por productores de eventos y consumidos por consumidores de eventos.

Propósito y alcance

La librería CodeDesignPlus.Net.PubSub tiene como objetivo estandarizar la implementación y el uso de un sistema de publicación y suscripción (Pub/Sub) en aplicaciones .NET. Esta librería proporciona una serie de interfaces y clases base que facilitan la implementación de clientes de eventos y manejadores de eventos como RabbitMQ, Kafka, Redis Pub/Sub, Event Store entre otros.

Principales características

  • Estandariza la implementación de un sistema de publicación y suscripción (Pub/Sub) en aplicaciones .NET.
  • Facilita la integración con brokers de eventos como RabbitMQ, Kafka, Redis Pub/Sub, Event Store, entre otros.
  • Desacoplamiento: Los publicadores y suscriptores no necesitan conocerse entre sí.
  • Escalabilidad: Maneja fácilmente un gran número de mensajes y suscriptores.
  • Asincronía: Los eventos se procesan de manera asíncrona.
  • Flexibilidad: Se pueden añadir o eliminar suscriptores sin afectar al sistema.
  • Filtrado: Los suscriptores reciben solo los mensajes de su interés.
  • Reactividad: Los componentes reaccionan a eventos en tiempo real.
  • Estado Descentralizado: El estado se distribuye entre eventos y consumidores.
  • Desempeño Mejorado: Reduce la latencia al procesar eventos de inmediato.
  • Interoperabilidad: Facilita la integración entre sistemas heterogéneos.
  • Resiliencia> Mejora la tolerancia a fallos mediante la redundancia y el manejo de errores.

Casos de uso típicos

  • Notificaciones en Tiempo Real: En sistemas de mensajería instantánea y redes sociales, donde los usuarios reciben notificaciones sobre nuevas publicaciones, mensajes o eventos.
  • Monitorización de Sistemas: Herramientas que envían alertas cuando se detectan problemas en la infraestructura, como picos de tráfico o caídas de servicio.
  • Aplicaciones de E-Commerce: Para actualizar el stock en tiempo real y enviar notificaciones sobre cambios en precios o promociones.
  • Sistemas de Logs: Para la recolección y análisis de logs distribuidos en tiempo real.
  • Microservicios: En arquitecturas de microservicios, donde los componentes independientes reaccionan a eventos sin necesidad de llamadas directas.
  • Procesamiento de Datos en Tiempo Real: Análisis de flujos de datos en tiempo real, como procesamiento de transacciones financieras o análisis de datos de sensores IoT.
  • Aplicaciones de Juegos: Sincronización de eventos de juego entre múltiples jugadores en tiempo real.
  • Automatización de Procesos Empresariales: Donde varios sistemas reaccionan automáticamente a eventos, como en la gestión de inventarios o la orquestación de workflows.

Componentes Principales

  • IMessage: Interfaz que define el contrato para los servicios de mensajería que se utilizan para publicar eventos a un broker externo como RabbitMQ, Kafka o Redis Pub/Sub.
  • IEventHandler<TEvent>: Interfaz que define el contrato para los manejadores de eventos. Los manejadores de eventos son responsables de procesar los eventos recibidos.
  • IPubSub: Interfaz que define el contrato para la publicación de eventos de dominio y es usada por los clientes para publicar eventos.
  • IEventQueue: Interfaz que define el contrato para el servicio de cola de eventos. Este servicio se utiliza para almacenar eventos en una cola en memoria antes de ser procesados.
  • IActivityService: Interfaz que define el contrato para el servicio de actividad. Este servicio se utiliza para registrar y rastrear actividades dentro del sistema que serán expuestas usando OpenTelemetry.
  • EventQueueBackgroundService: Clase que implementa BackgroundService y se encarga de procesar los eventos en segundo plano cuando se activa el sistema de queue en memoria.
  • Directorysrc
    • DirectoryCodeDesignPlus.Net.PubSub
      • DirectoryDiagnostics
        • ActivitySourceService.cs
      • DirectoryExceptions
        • EventHandlerAlreadyRegisteredException.cs
        • EventIsNotRegisteredException.cs
        • EventNotExistException.cs
        • EventNotImplementedException.cs
        • PubSubException.cs
      • DirectoryExtensions
        • PubSubExtensions.cs
        • ServiceCollectionExtensions.cs
      • DirectoryServices
        • EventQueueBackgroundService.cs
        • EventQueueService.cs
        • PubSubService.cs
        • RegisterEventHandlerBackgroundService.cs
    • DirectoryCodeDesignPlus.Net.PubSub.Abstractions
      • IActivityService.cs
      • IEventHandler.cs
      • IEventQueue.cs
      • IMessage.cs
      • IPubSub.cs
      • IQueueService.cs
      • Subscription.cs
      • DirectoryOptions
        • PubSubOptions.cs

Primeros Pasos

En esta sección, aprenderás a instalar y configurar la librería CodeDesignPlus.Net.PubSub en tu aplicación .NET. Sigue los siguientes pasos para comenzar a utilizar la librería en tu proyecto.

Requisitos previos

  • .NET 8 o superior.
  • Conocimientos básicos de patrones de diseño y sistemas distribuidos.
  • Conocimientos sobre la arquitectura orientada a eventos (Event-Driven).
  • Inyección de dependencias en aplicaciones .NET.

Instalación

Para instalar la librería CodeDesignPlus.Net.PubSub en tu proyecto de .NET, puedes utilizar el administrador de paquetes NuGet o la interfaz de línea de comandos de .NET. A continuación, se muestran los comandos para instalar la librería en tu proyecto:

Terminal window
dotnet add package CodeDesignPlus.Net.PubSub

Configuración básica

  1. Asignar las opciones de configuración en el appsettings.json:

    {
        "Core": {
            "Business": "CodeDesignPlus",
            "AppName": "sample-pubsub",
            "Version": "v1",
            "Description": "Sample of CodeDesignPlus.Net.Core",
            "Contact": {
                "Name": "CodeDesignPlus",
                "Email": "custom@outlook.com"
            }
        },
        "PubSub": {
            "UseQueue": true,
            "SecondsWaitQueue": 2,
            "EnableDiagnostic": true,
            "RegisterAutomaticHandlers": true
        }
    }

  2. Registra los servicios el contenedor de dependencias de tu aplicación:

    // ...
    services.AddPubSub(configuration);

Ejemplo rápido

El proyecto de ejemplo CodeDesignPlus.Net.PubSub.Sample muestra cómo publicar y suscribir eventos de dominio. De igual forma se muestra cómo implementar un broker en memoria para el ejemplo.

// See https://aka.ms/new-console-template for more information


using CodeDesignPlus.Net.PubSub.Abstractions;
using CodeDesignPlus.Net.PubSub.Extensions;
using CodeDesignPlus.Net.PubSub.Sample.Handlers;
using CodeDesignPlus.Net.PubSub.Sample.Server;


var builder = Host.CreateApplicationBuilder(args);


builder.Services.AddLogging();
builder.Services.AddPubSub(builder.Configuration);
builder.Services.AddSingleton<IMessage, InMemoryBroker>();


var host = builder.Build();


_ = Task.Run(() => host.Run());


await Task.Delay(5000);


var pubSub = host.Services.GetRequiredService<IPubSub>();


var userCreatedEvent = new UserCreatedEvent(Guid.NewGuid(), "John Doe");


await pubSub.PublishAsync(userCreatedEvent, CancellationToken.None);


Console.ReadLine();

  1. Creamos la entidad UserEntity que sera utilizada en el evento UserCreatedEvent.

    using System;
    using CodeDesignPlus.Net.Core.Abstractions;
    

    namespace CodeDesignPlus.Net.PubSub.Sample.Entities;
    

    public class UserEntity : IEntity
    {
        public bool IsActive { get; set; }
        public Instant CreatedAt { get; set; }
        public Guid CreatedBy { get; set; }
        public Instant? UpdatedAt { get; set; }
        public Guid? UpdatedBy { get; set; }
        public required string Name { get; set; }
    

        public Guid Id { get; set; }
    }

  2. Creamos el evento UserCreatedEvent que hereda de DomainEvent.

    using System;
    using CodeDesignPlus.Net.Core.Abstractions;
    using CodeDesignPlus.Net.Core.Abstractions.Attributes;
    using CodeDesignPlus.Net.PubSub.Sample.Entities;
    

    namespace CodeDesignPlus.Net.PubSub.Sample.Handlers;
    

    [EventKey<UserEntity>(1, "created")]
    public class UserCreatedEvent(Guid aggregateId, string name, Guid? eventId = null, Instant? occurredAt = null, Dictionary<string, object>? metadata = null) : DomainEvent(aggregateId, eventId, occurredAt, metadata)
    {
        public string Name { get; set; } = name;
    

        public static UserCreatedEvent Create(Guid aggregateId, string name, Guid? eventId = null, Instant? occurredAt = null, Dictionary<string, object>? metadata = null)
        {
            return new UserCreatedEvent(aggregateId, name, eventId, occurredAt, metadata);
        }
    }

  3. Creamos el manejador UserCreatedEventHandler que implementa IEventHandler<UserCreatedEvent>.

    using System;
    using CodeDesignPlus.Net.PubSub.Abstractions;
    using Microsoft.Extensions.Logging;
    

    namespace CodeDesignPlus.Net.PubSub.Sample.Handlers;
    

    public class UserCreatedEventHandler(ILogger<UserCreatedEventHandler> logger) : IEventHandler<UserCreatedEvent>
    {
        public Task HandleAsync(UserCreatedEvent data, CancellationToken token)
        {
            logger.LogInformation("Handling event {@data}", data);
    

            return Task.CompletedTask;
        }
    }

  4. Creamos el servidor InMemoryBroker que implementa IMessage.

    using System;
    using CodeDesignPlus.Net.Core.Abstractions;
    using CodeDesignPlus.Net.PubSub.Abstractions;
    using Microsoft.Extensions.Logging;
    

    namespace CodeDesignPlus.Net.PubSub.Sample.Server;
    

    public class InMemoryBroker(IServiceProvider serviceProvider, ILogger<InMemoryBroker> logger) : IMessage
    {
        private Action<object>? onMessage = null;
    

        public Task PublishAsync(IDomainEvent @event, CancellationToken cancellationToken)
        {
            logger.LogInformation("Publishing event {@event}", @event);
    

            onMessage?.Invoke(@event);
    

            return Task.CompletedTask;
        }
    

        public Task PublishAsync(IReadOnlyList<IDomainEvent> @event, CancellationToken cancellationToken)
        {
            foreach (var item in @event)
            {
                logger.LogInformation("Publishing event {@event}", item);
    

                onMessage?.Invoke(@event);
            }
    

            return Task.CompletedTask;
        }
    

        public Task SubscribeAsync<TEvent, TEventHandler>(CancellationToken cancellationToken)
            where TEvent : IDomainEvent
            where TEventHandler : IEventHandler<TEvent>
        {
            logger.LogInformation("Subscribing to event {event} with handler {handler}", typeof(TEvent).Name, typeof(TEventHandler).Name);
    

            onMessage = (message) =>
            {
                if (message is TEvent @event)
                {
                    var handler = serviceProvider.GetRequiredService<TEventHandler>();
    

                    handler.HandleAsync(@event, cancellationToken);
                }
            };
    

            return Task.CompletedTask;
        }
    

        public Task UnsubscribeAsync<TEvent, TEventHandler>(CancellationToken cancellationToken)
            where TEvent : IDomainEvent
            where TEventHandler : IEventHandler<TEvent>
        {
            logger.LogInformation("Unsubscribing to event {event} with handler {handler}", typeof(TEvent).Name, typeof(TEventHandler).Name);
    

            onMessage = null;
    

            return Task.CompletedTask;
        }
    }

  5. Registraremos los servicios necesarios en el contenedor de dependencias.

    builder.Services.AddPubSub(builder.Configuration);

  6. Obtenemos una instancia de IPubSub del contenedor de dependencias.

    var pubSub = host.Services.GetRequiredService<IPubSub>();

  7. Publicamos un evento de dominio UserCreatedEvent.

    var userCreatedEvent = new UserCreatedEvent(Guid.NewGuid(), "John Doe");
    

    await pubSub.PublishAsync(userCreatedEvent, CancellationToken.None);

Métodos de extensión

La librería CodeDesignPlus.Net.PubSub proporciona una serie de métodos de extensión para facilitar la integración de la librería en aplicaciones .NET. Estos métodos permiten registrar automáticamente los servicios necesarios en el sistema de inyección de dependencias de .NET. A continuación se listan los métodos de extensión disponibles:

ServiceCollectionExtensions

Esta clase proporciona métodos de extensión para registrar los servicios de CodeDesignPlus.Net.PubSub en el sistema de inyección de dependencias de .NET.

Explora ServiceCollectionExtensions

PubSubExtensions

Esta clase proporciona métodos de extensión para facilitar las operaciones relacionadas con la publicación y suscripción de eventos.

Explora PubSubExtensions

Opciones de configuración

CodeDesignPlus.Net.PubSub proporciona una clase de opciones de configuración para personalizar el comportamiento de la librería. Estas opciones permiten a los desarrolladores configurar parámetros como el uso de un sistema de queue en memoria, el tiempo de espera para procesar eventos y el tiempo de espera para publicar eventos. A continuación se listan las opciones de configuración disponibles:

Explora PubSubOptions

Servicios

CodeDesignPlus.Net.PubSub proporciona interfaces e implementaciones esenciales para implementar patrones de publicación y suscripción en aplicaciones .NET. A continuación se listan los servicios disponibles en la librería:

IPubSub

Esta interfaz define el contrato para la publicación de eventos de dominio.

Explora IPubSub

IMessage

Esta interfaz define el contrato para los servicios de mensajería que se utilizan para publicar eventos a un broker externo como RabbitMQ, Kafka o Redis Pub/Sub.

Explora IMessage

IEventHandler<TEvent>

Esta interfaz define el contrato para los manejadores

Explora IEventHandler

IEventQueue

Esta interfaz define el contrato para el servicio de cola de eventos. Este servicio se utiliza para almacenar eventos en una cola en memoria antes de ser procesados.

Explora IEventQueue

IActivityService

Esta interfaz define el contrato para el servicio de actividad. Este servicio se utiliza para registrar y rastrear actividades dentro del sistema que seran expuestas usando OpenTelemetry.

Explora IActivityService

Conclusiones

La librería CodeDesignPlus.Net.PubSub es una herramienta poderosa para los desarrolladores que buscan implementar aplicaciones robustas, escalables y mantenibles, alineadas con las mejores prácticas de desarrollo. Esta librería facilita la implementación de un sistema de publicación y suscripción (Pub/Sub) en aplicaciones .NET, permitiendo a los desarrolladores crear aplicaciones desacopladas mediante el uso de eventos y manejadores de eventos.

Referencias