Indiko.Blocks.EventBus.Abstractions 2.1.2

Indiko.Blocks.EventBus.Abstractions

Core abstractions for event-driven architecture and publish-subscribe messaging in the Indiko framework.

Overview

This package defines the fundamental contracts for implementing event bus systems, enabling decoupled communication between application components through asynchronous event publishing and handling.

Features

• IEventBus Interface: Core contract for event bus implementations
• IEvent Interface: Marker interface for event messages
• IEventHandler<TEvent>: Generic event handler interface
• Event Registration: Dynamic event handler registration
• Publish-Subscribe Pattern: Decoupled event-driven communication
• Async/Await Support: Full asynchronous event handling
• Multiple Handlers: Support for multiple handlers per event type
• Builder Pattern: Fluent API for event bus configuration

Installation

dotnet add package Indiko.Blocks.EventBus.Abstractions

Key Interfaces

IEventBus

Main interface for event bus operations.

public interface IEventBus
{
    // Connection management
    Task<bool> ConnectAsync(CancellationToken cancellationToken = default);
    Task DisconnectAsync(CancellationToken cancellationToken = default);
    
    // Publish events
    void Publish<TEvent>(TEvent eventMessage) where TEvent : IEvent;
    ValueTask PublishAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default) 
        where TEvent : IEvent;
    
    // Handler registration
    void RegisterEventHandler<TEvent>(IEventHandler<TEvent> eventHandler) 
        where TEvent : IEvent;
    void RegisterEventHandlers<TEvent>(params IEventHandler<TEvent>[] eventHandlers) 
        where TEvent : IEvent;
    void UnRegisterEventHandler<TEvent>() where TEvent : IEvent;
    void UnRegisterAllEventHandlers();
}

IEvent

Marker interface for event messages.

public interface IEvent
{
    // Marker interface - events inherit from this
}

IEventHandler<TEvent>

Interface for handling specific event types.

public interface IEventHandler<TEvent> where TEvent : IEvent
{
    Task HandleAsync(TEvent @event, CancellationToken cancellationToken = default);
}

Usage Examples

Define Events

using Indiko.Blocks.EventBus.Abstractions.Interfaces;

// Simple event
public class UserCreatedEvent : IEvent
{
    public Guid UserId { get; set; }
    public string Email { get; set; }
    public DateTime CreatedAt { get; set; }
}

// Event with rich data
public class OrderPlacedEvent : IEvent
{
    public Guid OrderId { get; set; }
    public Guid UserId { get; set; }
    public List<OrderItem> Items { get; set; }
    public decimal TotalAmount { get; set; }
    public DateTime OrderDate { get; set; }
}

public class OrderItem
{
    public Guid ProductId { get; set; }
    public string ProductName { get; set; }
    public int Quantity { get; set; }
    public decimal Price { get; set; }
}

Implement Event Handlers

// Single event handler
public class SendWelcomeEmailHandler : IEventHandler<UserCreatedEvent>
{
    private readonly IEmailService _emailService;

    public SendWelcomeEmailHandler(IEmailService emailService)
    {
        _emailService = emailService;
    }

    public async Task HandleAsync(UserCreatedEvent @event, CancellationToken cancellationToken = default)
    {
        await _emailService.SendWelcomeEmailAsync(@event.Email, cancellationToken);
    }
}

// Another handler for the same event
public class CreateUserProfileHandler : IEventHandler<UserCreatedEvent>
{
    private readonly IProfileService _profileService;

    public CreateUserProfileHandler(IProfileService profileService)
    {
        _profileService = profileService;
    }

    public async Task HandleAsync(UserCreatedEvent @event, CancellationToken cancellationToken = default)
    {
        await _profileService.CreateDefaultProfileAsync(@event.UserId, cancellationToken);
    }
}

Register Handlers

public class Startup : WebStartup
{
    public override void ConfigureServices(IServiceCollection services)
    {
        base.ConfigureServices(services);
        
        // Register event handlers
        services.AddScoped<IEventHandler<UserCreatedEvent>, SendWelcomeEmailHandler>();
        services.AddScoped<IEventHandler<UserCreatedEvent>, CreateUserProfileHandler>();
        services.AddScoped<IEventHandler<OrderPlacedEvent>, ProcessOrderHandler>();
    }
}

Publish Events

public class UserService
{
    private readonly IEventBus _eventBus;
    private readonly IUserRepository _userRepository;

    public UserService(IEventBus eventBus, IUserRepository userRepository)
    {
        _eventBus = eventBus;
        _userRepository = userRepository;
    }

    public async Task<User> CreateUserAsync(string email, string password)
    {
        var user = new User
        {
            Id = Guid.NewGuid(),
            Email = email,
            CreatedAt = DateTime.UtcNow
        };
        
        await _userRepository.AddAsync(user);
        
        // Publish event - all registered handlers will be invoked
        await _eventBus.PublishAsync(new UserCreatedEvent
        {
            UserId = user.Id,
            Email = user.Email,
            CreatedAt = user.CreatedAt
        });
        
        return user;
    }
}

Event Handler Patterns

Fire and Forget

public class AuditLogHandler : IEventHandler<UserCreatedEvent>
{
    private readonly IAuditService _auditService;

    public async Task HandleAsync(UserCreatedEvent @event, CancellationToken cancellationToken = default)
    {
        // Non-critical operation - log and continue
        try
        {
            await _auditService.LogUserCreationAsync(@event.UserId);
        }
        catch (Exception ex)
        {
            // Log error but don't throw
            Console.WriteLine($"Audit logging failed: {ex.Message}");
        }
    }
}

Critical Handler with Retry

public class SendOrderConfirmationHandler : IEventHandler<OrderPlacedEvent>
{
    private readonly IEmailService _emailService;
    private readonly ILogger<SendOrderConfirmationHandler> _logger;

    public async Task HandleAsync(OrderPlacedEvent @event, CancellationToken cancellationToken = default)
    {
        int maxRetries = 3;
        for (int attempt = 0; attempt < maxRetries; attempt++)
        {
            try
            {
                await _emailService.SendOrderConfirmationAsync(@event, cancellationToken);
                return; // Success
            }
            catch (Exception ex)
            {
                _logger.LogWarning($"Attempt {attempt + 1} failed: {ex.Message}");
                
                if (attempt == maxRetries - 1)
                    throw; // Final attempt failed
                
                await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, attempt)), cancellationToken);
            }
        }
    }
}

Conditional Handler

public class PremiumUserBonusHandler : IEventHandler<OrderPlacedEvent>
{
    private readonly IUserService _userService;
    private readonly IRewardService _rewardService;

    public async Task HandleAsync(OrderPlacedEvent @event, CancellationToken cancellationToken = default)
    {
        var user = await _userService.GetUserAsync(@event.UserId);
        
        // Only process for premium users
        if (user.IsPremium && @event.TotalAmount >= 100)
        {
            await _rewardService.AddBonusPointsAsync(@event.UserId, @event.TotalAmount * 0.1m);
        }
    }
}

Builder Pattern

EventBusBuilder

Fluent API for configuring event bus.

public class EventBusBuilder : IEventBusBuilder
{
    public IEventBusBuilder AddHandler<TEvent, THandler>() 
        where TEvent : IEvent 
        where THandler : IEventHandler<TEvent>;
    
    public IEventBusBuilder Configure(Action<EventBusOptions> options);
    
    public IEventBus Build();
}

Usage

var eventBus = new EventBusBuilder()
    .AddHandler<UserCreatedEvent, SendWelcomeEmailHandler>()
    .AddHandler<UserCreatedEvent, CreateUserProfileHandler>()
    .AddHandler<OrderPlacedEvent, ProcessOrderHandler>()
    .Configure(options =>
    {
        options.MaxConcurrentHandlers = 10;
        options.EnableRetry = true;
    })
    .Build();

Best Practices

1. Keep Events Immutable: Use read-only properties
2. Event Naming: Use past tense (UserCreated, OrderPlaced, PaymentProcessed)
3. Handler Independence: Handlers should not depend on execution order
4. Error Handling: Always handle exceptions in handlers
5. Idempotency: Design handlers to be idempotent
6. Single Responsibility: One handler, one responsibility
7. Event Versioning: Plan for event schema evolution

Event Patterns

Domain Events

public abstract class DomainEvent : IEvent
{
    public Guid EventId { get; } = Guid.NewGuid();
    public DateTime OccurredAt { get; } = DateTime.UtcNow;
    public string EventType => GetType().Name;
}

public class ProductStockChangedEvent : DomainEvent
{
    public Guid ProductId { get; set; }
    public int OldQuantity { get; set; }
    public int NewQuantity { get; set; }
    public string Reason { get; set; }
}

Integration Events

public abstract class IntegrationEvent : IEvent
{
    public Guid Id { get; set; } = Guid.NewGuid();
    public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
    public string CorrelationId { get; set; }
}

public class PaymentCompletedEvent : IntegrationEvent
{
    public Guid OrderId { get; set; }
    public decimal Amount { get; set; }
    public string TransactionId { get; set; }
}

Configuration Options

EventBusOptions

public class EventBusOptions
{
    public int MaxConcurrentHandlers { get; set; } = 10;
    public bool EnableRetry { get; set; } = true;
    public int MaxRetryAttempts { get; set; } = 3;
    public TimeSpan RetryDelay { get; set; } = TimeSpan.FromSeconds(1);
    public bool ThrowOnHandlerError { get; set; } = false;
}

Target Framework

• .NET 10

Dependencies

• Indiko.Blocks.Common.Abstractions

License

See LICENSE file in the repository root.

Related Packages

• Indiko.Blocks.EventBus.InMemory - In-memory event bus implementation
• Indiko.Blocks.EventBus.RabbitMQ - RabbitMQ-based event bus
• Indiko.Blocks.Mediation.Abstractions - CQRS and mediator pattern
• Indiko.Blocks.Common.Management - Block management system