TAF.Infra.MassTransit 2.2.14

TAF.Infra.MassTransit

A comprehensive abstraction layer over MediatR and MassTransit that provides unified Command, Query, and Event handling patterns with support for both local (in-process) and distributed (message broker) execution.

🚀 Features

• Unified CQRS Pattern: Clean separation of Commands, Queries, and Events
• Dual Execution Modes:
  • Local execution via MediatR
  • Distributed execution via MassTransit (RabbitMQ, Azure Service Bus, Amazon SQS)
• Smart Queue Naming: Configurable naming conventions with automatic sanitization and duplicate suffix removal
• Context-Aware: Built-in context propagation for multi-tenant, multi-app scenarios with helper utilities
• Multiple Message Brokers: Support for RabbitMQ, Azure Service Bus, Amazon SQS, and In-Memory
• Automatic Registration: Auto-discovery and registration of handlers
• Retry Policies: Configurable retry mechanisms with exponential backoff
• Type Safety: Strongly-typed message contracts and handlers
• Environment Support: Multi-environment queue/topic naming with project and environment identifiers

📦 Installation

dotnet add package TAF.Infra.MassTransit

🗃️ Architecture Overview

The library provides three main abstractions:

• Commands: Actions that change system state (fire-and-forget or with response)
• Queries: Read-only operations that return data (always local via MediatR)
• Events: Notifications about things that have happened (distributed via message broker)

Execution Patterns

Queue/Topic Naming Convention

The library automatically generates structured names for queues and topics:

• Commands: {projectName}-{environment}-command-{commandName}
• Events: {projectName}-{environment}-event-{eventName}

Examples:

• CreateUserCommand → myproject-prod-command-createuser
• UserCreatedEvent → myproject-prod-event-usercreated

🚀 Quick Start

1. Configuration

appsettings.json:

{
  "MessageBus": {
    "Host": "localhost",
    "Port": 5672,
    "Username": "guest",
    "Password": "guest",
    "VirtualHost": "/",
    "ExchangeName": "default",
    "RetryLimit": 5,
    "UseMessageScheduler": true
  }
}

Program.cs:

using TAF.Infra.MassTransit;
using TAF.Infra.MassTransit.Enums;

var builder = WebApplication.CreateBuilder(args);

// Basic configuration from appsettings
builder.Services.AddMessageBus(builder.Configuration);

// With custom naming conventions
builder.Services.AddMessageBusWithNaming(
    builder.Configuration,
    projectName: "MyProject",
    environment: "production"
);

// Manual configuration with naming
builder.Services.AddMessageBusWithNaming(
    options =>
    {
        options.Provider = MessageBusConfigurationEnum.RabbitMQ;
        options.Settings["Host"] = "localhost";
        options.Settings["Username"] = "guest";
        options.Settings["Password"] = "guest";
    },
    projectName: "MyProject",
    environment: "dev",
    commandQueueTemplate: "{projectName}-{environment}-cmd-{commandName}",
    eventTopicTemplate: "{projectName}-{environment}-evt-{eventName}"
);

2. Create Messages

Commands:

using TAF.Infra.MassTransit.Base;

// Fire-and-forget command
public class CreateUserCommand : BaseCommand
{
    public string Name { get; set; }
    public string Email { get; set; }
}

// Command with response
public class GetUserCommand : BaseCommand<UserResponse>
{
    public Guid UserId { get; set; }
}

public class UserResponse
{
    public string Name { get; set; }
    public string Email { get; set; }
}

Queries:

using TAF.Infra.MassTransit.Base;

public class GetUsersQuery : BaseQuery<List<UserResponse>>
{
    public int PageSize { get; set; } = 10;
    public int PageNumber { get; set; } = 1;
}

Events:

using TAF.Infra.MassTransit.Base;

public class UserCreatedEvent : BaseEvent
{
    public Guid UserId { get; set; }
    public string Name { get; set; }
    public string Email { get; set; }
}

3. Create Handlers

Command Handler:

using TAF.Infra.MassTransit.Abstractions;
using TAF.Infra.MassTransit.Abstractions.Command;

public class CreateUserCommandHandler : ICommandHandler<CreateUserCommand>
{
    private readonly IUserRepository _userRepository;
    private readonly IEventBus _eventBus;

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

    public async Task HandleAsync(CreateUserCommand command, Context context, 
        CancellationToken cancellationToken = default)
    {
        var user = new User 
        { 
            Name = command.Name, 
            Email = command.Email 
        };
        
        await _userRepository.CreateAsync(user);

        // Publish event
        var userCreatedEvent = new UserCreatedEvent
        {
            UserId = user.Id,
            Name = user.Name,
            Email = user.Email
        };

        await _eventBus.PublishAsync(userCreatedEvent, context, cancellationToken);
    }
}

Query Handler:

using TAF.Infra.MassTransit.Abstractions.Query;

public class GetUsersQueryHandler : IQueryHandler<GetUsersQuery, List<UserResponse>>
{
    private readonly IUserRepository _userRepository;

    public GetUsersQueryHandler(IUserRepository userRepository)
    {
        _userRepository = userRepository;
    }

    public async Task<List<UserResponse>> HandleAsync(GetUsersQuery query, Context context, 
        CancellationToken cancellationToken = default)
    {
        var users = await _userRepository.GetPagedAsync(query.PageNumber, query.PageSize);
        return users.Select(u => new UserResponse { Name = u.Name, Email = u.Email }).ToList();
    }
}

Event Handler:

using TAF.Infra.MassTransit.Abstractions.Event;

public class UserCreatedEventHandler : IEventHandler<UserCreatedEvent>
{
    private readonly IEmailService _emailService;

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

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

4. Usage in Controllers/Services

using TAF.Infra.MassTransit.Helpers;

[ApiController]
[Route("api/[controller]")]
public class UsersController : ControllerBase
{
    private readonly ICommandBus _commandBus;
    private readonly IQueryBus _queryBus;

    public UsersController(ICommandBus commandBus, IQueryBus queryBus)
    {
        _commandBus = commandBus;
        _queryBus = queryBus;
    }

    [HttpPost]
    public async Task<IActionResult> CreateUser([FromBody] CreateUserRequest request)
    {
        // Create context from HTTP headers (extracts tenant, user info, etc.)
        var context = ContextHelper.CreateFromHeaders(Request.Headers);

        var command = new CreateUserCommand 
        { 
            Name = request.Name, 
            Email = request.Email 
        };

        // Execute locally via MediatR
        await _commandBus.SendAsync<CreateUserCommand, Unit>(command, context);
        
        // OR execute distributed via MassTransit
        // Queue: myproject-prod-command-createuser
        await _commandBus.PublishAsync(command, context);

        return Ok();
    }

    [HttpGet]
    public async Task<IActionResult> GetUsers([FromQuery] int page = 1, [FromQuery] int size = 10)
    {
        // Create context from parameters
        var context = ContextHelper.CreateFromParameters(
            userId: User.Identity.Name,
            tenantId: "tenant-123",
            appId: "web-api",
            environmentId: "production"
        );

        var query = new GetUsersQuery 
        { 
            PageNumber = page, 
            PageSize = size 
        };

        // Queries are always executed locally
        var result = await _queryBus.SendAsync<GetUsersQuery, List<UserResponse>>(query, context);

        return Ok(result);
    }
}

⚙️ Configuration Options

Message Broker Providers

services.AddMessageBus(options =>
{
    // RabbitMQ (default)
    options.Provider = MessageBusConfigurationEnum.RabbitMQ;
    
    // Azure Service Bus
    options.Provider = MessageBusConfigurationEnum.AzureServiceBus;
    
    // Amazon SQS
    options.Provider = MessageBusConfigurationEnum.AmazonSQS;
    
    // In-Memory (testing)
    options.Provider = MessageBusConfigurationEnum.InMemory;
});

Smart Naming Configuration

// Default templates: {projectName}-{environment}-command-{commandName}
services.AddMessageBusWithNaming(
    configuration,
    projectName: "MyProject",        // Default: assembly name
    environment: "production",       // Default: "dev"
    commandQueueTemplate: "{projectName}.{environment}.cmd.{commandName}",
    eventTopicTemplate: "{projectName}.{environment}.evt.{eventName}"
);

Naming Examples:

// Class: CreateUserCommand
// Result: myproject-prod-command-createuser

// Class: CreateUserCommandCommand (duplicate suffix)
// Result: myproject-prod-command-createusercommand (removes last "Command")

// Class: UserCreatedEvent  
// Result: myproject-prod-event-usercreated

// Class: UserCreatedEventEvent (duplicate suffix)
// Result: myproject-prod-event-usercreatedevent (removes last "Event")

Advanced Configuration

services.Configure<MessageBrokerSettings>(options =>
{
    options.Host = "rabbitmq.example.com";
    options.Port = 5672;
    options.Username = "myuser";
    options.Password = "mypassword";
    options.VirtualHost = "/myapp";
    options.UseSsl = true;
    options.PrefetchCount = 50;
    
    // Retry settings
    options.MassTransit.RetryPolicy.RetryLimit = 3;
    options.MassTransit.RetryPolicy.InitialIntervalMs = 1000;
    options.MassTransit.RetryPolicy.MaxIntervalMs = 30000;
});

🛠️ Context and Multi-Tenancy

The library provides built-in context support for cross-cutting concerns:

Context Helper Utilities

using TAF.Infra.MassTransit.Helpers;

// Create context from HTTP headers (in controllers/middleware)
var context = ContextHelper.CreateFromHeaders(Request.Headers);

// Create context from parameters (in services/background jobs)
var context = ContextHelper.CreateFromParameters(
    userId: "user-123",
    tenantId: "tenant-456",
    appId: "web-api",
    environmentId: "production",
    correlationId: Guid.NewGuid(),
    properties: new Dictionary<string, object>
    {
        { "TraceId", "trace-789" },
        { "ClientIP", "192.168.1.1" }
    }
);

// Manual context creation
var context = new Context
{
    CorrelationId = Guid.NewGuid(),
    TenantId = "tenant-123",
    AppId = "web-app",
    UserId = "user-456",
    EnvironmentId = "production"
};

// Add custom properties
context.SetProperty("TraceId", "trace-789");
context.SetProperty("ClientIP", "192.168.1.1");

Context in Middleware

public class ContextMiddleware
{
    public async Task InvokeAsync(HttpContext httpContext, RequestDelegate next)
    {
        // Automatically extract context from headers
        var messageContext = ContextHelper.CreateFromHeaders(httpContext.Request.Headers);
        
        // Store in some context provider for later use
        using var scope = new ContextScope(messageContext);
        
        await next(httpContext);
    }
}

📄 Execution Patterns

Local Execution (MediatR)

• Fast, in-process execution
• Suitable for simple operations
• Automatic transaction boundaries
• Commands and Queries

Distributed Execution (MassTransit)

• Resilient, scalable execution
• Cross-service communication
• Built-in retry and error handling
• Commands and Events only

🧪 Testing Support

// Use In-Memory provider for testing
services.AddMessageBusWithNaming(
    options =>
    {
        options.Provider = MessageBusConfigurationEnum.InMemory;
    },
    projectName: "TestProject",
    environment: "test"
);

🔧 Advanced Features

Custom Base Classes

public class AuditableCommand : BaseCommand
{
    public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
    public string CreatedBy { get; set; }
}

public class CreateOrderCommand : AuditableCommand
{
    public List<OrderItem> Items { get; set; }
    public decimal Total { get; set; }
}

Multiple Handlers

// Multiple event handlers for the same event
public class EmailNotificationHandler : IEventHandler<UserCreatedEvent> { ... }
public class AuditLogHandler : IEventHandler<UserCreatedEvent> { ... }
public class MetricsHandler : IEventHandler<UserCreatedEvent> { ... }

Queue Name Resolution

// The library automatically handles complex naming scenarios:

// Full class name: MyApp.Commands.User.CreateUserCommand
// Generated queue: myapp-prod-command-createuser

// Full class name: MyApp.Events.Order.OrderShippedEvent  
// Generated topic: myapp-prod-event-ordershipped

// Duplicate suffix handling:
// CreateUserCommandCommand → myapp-prod-command-createusercommand
// UserCreatedEventEvent → myapp-prod-event-usercreatedevent

📚 API Reference

Core Interfaces

• ICommandBus - Command execution (local/distributed)
• IQueryBus - Query execution (local only)
• IEventBus - Event publishing (distributed)
• ICommandHandler<T> - Command handler interface
• ICommandHandler<T, R> - Command handler with response interface
• IQueryHandler<T, R> - Query handler interface
• IEventHandler<T> - Event handler interface

Base Classes

• BaseCommand - Base class for commands
• BaseCommand<T> - Base class for commands with response
• BaseQuery<T> - Base class for queries
• BaseEvent - Base class for events

Helper Classes

• ContextHelper - Static helper for context creation
• MessageNamingConfiguration - Configuration for queue/topic naming
• ContextScope - Scoped context management

Configuration Classes

• MessageBusConfiguration - Main configuration
• MessageBrokerSettings - Broker-specific settings
• MessageNamingConfiguration - Naming convention settings

🤝 Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

📄 License

This project is licensed under the MIT License.

🐛 Issues

If you encounter any issues, please file them in the GitHub Issues section.

Built with ❤️ for distributed systems