Indiko.Blocks.EventBus.InMemory 2.7.8

Indiko.Blocks.EventBus.InMemory

In-memory event bus implementation for local, in-process event-driven communication.

Overview

This package provides a lightweight, in-memory implementation of the event bus abstractions, perfect for single-instance applications, testing, and development scenarios.

Features

• Zero Dependencies: No external message broker required
• In-Process Communication: Events published and consumed within the same process
• Concurrent Handling: Parallel execution of multiple event handlers
• Thread-Safe: Concurrent dictionary-based handler storage
• Fast: Minimal overhead for event delivery
• Testing-Friendly: Ideal for unit and integration tests
• Development: Perfect for local development without infrastructure

Installation

dotnet add package Indiko.Blocks.EventBus.InMemory

Quick Start

Configure Services

using Indiko.Blocks.EventBus.InMemory;

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

Define Events

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

public class OrderPlacedEvent : IEvent
{
    public Guid OrderId { get; set; }
    public Guid UserId { get; set; }
    public decimal TotalAmount { get; set; }
    public DateTime OrderDate { get; set; }
}

Implement Handlers

public class SendWelcomeEmailHandler : IEventHandler<UserCreatedEvent>
{
    private readonly IEmailService _emailService;
    private readonly ILogger<SendWelcomeEmailHandler> _logger;

    public SendWelcomeEmailHandler(IEmailService emailService, ILogger<SendWelcomeEmailHandler> logger)
    {
        _emailService = emailService;
        _logger = logger;
    }

    public async Task HandleAsync(UserCreatedEvent @event, CancellationToken cancellationToken = default)
    {
        _logger.LogInformation($"Sending welcome email to {event.Email}");
        await _emailService.SendWelcomeEmailAsync(@event.Email, cancellationToken);
    }
}

public class CreateUserProfileHandler : IEventHandler<UserCreatedEvent>
{
    private readonly IProfileService _profileService;

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

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 - handlers execute concurrently
        await _eventBus.PublishAsync(new UserCreatedEvent
        {
            UserId = user.Id,
            Email = user.Email,
            CreatedAt = user.CreatedAt
        });
        
        return user;
    }
}

How It Works

Concurrent Handler Execution

The in-memory event bus executes all registered handlers for an event type concurrently:

public async ValueTask PublishAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default)
{
    if (_eventHandlers.TryGetValue(typeof(TEvent), out var handlers))
    {
        // Execute all handlers concurrently
        var tasks = handlers.Cast<IEventHandler<TEvent>>()
                            .Select(handler => handler.HandleAsync(@event, cancellationToken));

        await Task.WhenAll(tasks); // Wait for all handlers to complete
    }
}

Thread-Safe Handler Registration

Handler registration is protected with locks to ensure thread safety:

public void RegisterEventHandler<TEvent>(IEventHandler<TEvent> eventHandler)
{
    lock (_lock)
    {
        _eventHandlers.AddOrUpdate(
            typeof(TEvent),
            [eventHandler],
            (key, existingHandlers) =>
            {
                existingHandlers.Add(eventHandler);
                return existingHandlers;
            }
        );
    }
}

Dynamic Handler Registration

Register at Runtime

public class EventHandlerRegistry
{
    private readonly IEventBus _eventBus;
    private readonly IServiceProvider _serviceProvider;

    public void RegisterHandlers()
    {
        // Get handler from DI container
        var emailHandler = _serviceProvider.GetRequiredService<IEventHandler<UserCreatedEvent>>();
        
        // Register dynamically
        _eventBus.RegisterEventHandler(emailHandler);
    }

    public void RegisterMultipleHandlers()
    {
        var handler1 = _serviceProvider.GetRequiredService<SendWelcomeEmailHandler>();
        var handler2 = _serviceProvider.GetRequiredService<CreateUserProfileHandler>();
        
        // Register multiple handlers at once
        _eventBus.RegisterEventHandlers<UserCreatedEvent>(handler1, handler2);
    }
}

Unregister Handlers

// Unregister all handlers for specific event type
_eventBus.UnRegisterEventHandler<UserCreatedEvent>();

// Unregister all handlers for all events
_eventBus.UnRegisterAllEventHandlers();

Testing

Unit Testing

[Fact]
public async Task UserCreation_Should_TriggerWelcomeEmail()
{
    // Arrange
    var emailServiceMock = new Mock<IEmailService>();
    var eventBus = new InMemoryEventBus();
    var handler = new SendWelcomeEmailHandler(emailServiceMock.Object, Mock.Of<ILogger>());
    
    eventBus.RegisterEventHandler(handler);
    
    var @event = new UserCreatedEvent
    {
        UserId = Guid.NewGuid(),
        Email = "test@example.com"
    };
    
    // Act
    await eventBus.PublishAsync(@event);
    
    // Assert
    emailServiceMock.Verify(
        x => x.SendWelcomeEmailAsync("test@example.com", It.IsAny<CancellationToken>()), 
        Times.Once
    );
}

Integration Testing

public class EventBusIntegrationTests : IClassFixture<WebApplicationFactory<Startup>>
{
    private readonly WebApplicationFactory<Startup> _factory;

    public EventBusIntegrationTests(WebApplicationFactory<Startup> factory)
    {
        _factory = factory;
    }

    [Fact]
    public async Task PublishEvent_Should_ExecuteAllHandlers()
    {
        // Arrange
        var serviceProvider = _factory.Services;
        var eventBus = serviceProvider.GetRequiredService<IEventBus>();
        
        var @event = new UserCreatedEvent
        {
            UserId = Guid.NewGuid(),
            Email = "test@example.com",
            CreatedAt = DateTime.UtcNow
        };
        
        // Act
        await eventBus.PublishAsync(@event);
        
        // Assert
        // Verify side effects of handlers
        // (e.g., database records created, emails sent, etc.)
    }
}

Performance Characteristics

Advantages

• Low Latency: No network overhead
• High Throughput: Direct in-memory method calls
• No Serialization: No message serialization/deserialization
• Simple Debugging: Easy to step through handler execution

Limitations

• Single Process: Events don't cross process boundaries
• No Persistence: Events are lost if process crashes
• No Retry: Failed handlers require manual retry logic
• Memory Bound: All handlers execute in the same memory space

Use Cases

1. Development Environment

public void ConfigureServices(IServiceCollection services)
{
    if (Environment.IsDevelopment())
    {
        // Use in-memory for local development
        services.AddInMemoryEventBus();
    }
    else
    {
        // Use RabbitMQ for production
        services.AddRabbitMQEventBus(Configuration);
    }
}

2. Unit Testing

Perfect for testing event-driven logic without external dependencies.

3. Monolithic Applications

For applications that don't require distributed event processing.

4. Domain Events

Internal domain events within a bounded context.

Error Handling

Handler Exceptions

public class RobustHandler : IEventHandler<UserCreatedEvent>
{
    public async Task HandleAsync(UserCreatedEvent @event, CancellationToken cancellationToken)
    {
        try
        {
            // Handler logic
            await DoWorkAsync(@event);
        }
        catch (Exception ex)
        {
            // Log error
            _logger.LogError(ex, $"Handler failed for user {event.UserId}");
            
            // Decide whether to throw or swallow
            // Throwing will cause PublishAsync to fail
            throw;
        }
    }
}

Publisher Error Handling

public async Task CreateUserWithErrorHandlingAsync(string email)
{
    var user = await CreateUserAsync(email);
    
    try
    {
        await _eventBus.PublishAsync(new UserCreatedEvent { ... });
    }
    catch (Exception ex)
    {
        // One or more handlers failed
        _logger.LogError(ex, "Event publishing failed");
        
        // Decide on compensation logic
        // (e.g., mark user for manual review, retry later, etc.)
    }
}

Comparison with Other Implementations

Migration Path

From InMemory to RabbitMQ

Your code remains the same when switching implementations:

// Development
services.AddInMemoryEventBus();

// Production
services.AddRabbitMQEventBus(options =>
{
    options.ConnectionString = Configuration.GetConnectionString("RabbitMQ");
});

All your events and handlers work without modification!

Target Framework

• .NET 10

Dependencies

• Indiko.Blocks.EventBus.Abstractions

License

See LICENSE file in the repository root.

Related Packages

• Indiko.Blocks.EventBus.Abstractions - Core event bus abstractions
• Indiko.Blocks.EventBus.RabbitMQ - Distributed event bus with RabbitMQ
• Indiko.Blocks.Mediation.Abstractions - CQRS and mediator pattern