EventTransit 2.0.1

EventTransit

EventTransit is a high-performance, production-ready .NET messaging library that provides reliable message delivery with support for multiple message brokers. Built with clean architecture principles, it offers transactional outbox/inbox patterns, delayed messages, and comprehensive monitoring capabilities.

🚀 Features

Core Capabilities

• ✅ Multi-Broker Support - RabbitMQ (with Kafka and Azure Service Bus planned)
• ✅ Transactional Outbox Pattern - Guaranteed message delivery with database transactions
• ✅ Inbox Pattern - Automatic duplicate detection and idempotency
• ✅ Delayed Messages - Schedule messages for future delivery using TTL + DLX pattern
• ✅ Dead Letter Handling - Automatic retry with exponential backoff and dead letter queues
• ✅ Direct Publishing - Fire-and-forget publishing for high-throughput scenarios
• ✅ Multiple Exchange Types - Direct, Topic, Fanout, and Headers exchanges
• ✅ Built-in Dashboard - Real-time monitoring UI for outbox, inbox, and dead letters
• ✅ Entity Framework Integration - Seamless integration with EF Core
• ✅ Repository Pattern Support - Use with or without Entity Framework

Production-Ready

• 🔒 Type-Safe - Strongly-typed message contracts
• 📊 Observable - Comprehensive logging and metrics
• 🔄 Resilient - Automatic reconnection and error recovery
• ⚡ High Performance - Optimized for throughput and low latency
• 🧪 Well Tested - Extensive test coverage
• 📖 Well Documented - Complete API documentation and examples

📦 Installation

dotnet add package EventTransit

🎯 Quick Start

1. Define Your Message

public class OrderCreatedEvent : IEventWithId
{
    public string Id { get; set; } = Guid.NewGuid().ToString();
    public string OrderId { get; set; }
    public decimal Amount { get; set; }
    public DateTime CreatedAt { get; set; }
}

2. Configure EventTransit

Option A: Using appsettings.json (Recommended)

{
  "EventTransit": {
    "Broker": {
      "BrokerType": "RabbitMQ",
      "ConnectionString": "amqp://guest:guest@localhost:5672",
      "ClientProvidedName": "my-service"
    },
    "DefaultExchange": {
      "Name": "my.events",
      "Type": "topic",
      "Durable": true
    },
    "Outbox": {
      "Enabled": true,
      "ProcessingIntervalSeconds": 1,
      "MaxRetryAttempts": 3,
      "DeadLetterAfterMaxRetries": true
    },
    "Inbox": {
      "Enabled": true,
      "MaxRetryAttempts": 3
    }
  }
}

// Program.cs
builder.Services.AddEventTransitWithEntityFramework<MyDbContext>(
    configureOptions: cfg => builder.Configuration.GetSection("EventTransit").Bind(cfg)
);

// Add Dashboard (optional but recommended)
builder.Services.AddEventTransitDashboardWithEntityFramework<MyDbContext>(builder.Configuration);

Option B: Programmatic Configuration

builder.Services.AddEventTransitWithEntityFramework<MyDbContext>(cfg =>
{
    cfg.Broker.BrokerType = BrokerType.RabbitMQ;
    cfg.Broker.ConnectionString = "amqp://guest:guest@localhost:5672";
    cfg.DefaultExchange.Name = "my.events";
    cfg.DefaultExchange.Type = ExchangeType.Topic;
    cfg.Outbox.Enabled = true;
    cfg.Inbox.Enabled = true;
});

3. Publish Messages

Using Outbox Publisher (Transactional - Recommended)

public class OrderService
{
    private readonly IOutboxPublisher _publisher;
    private readonly MyDbContext _dbContext;

    public OrderService(IOutboxPublisher publisher, MyDbContext dbContext)
    {
        _publisher = publisher;
        _dbContext = dbContext;
    }

    public async Task CreateOrderAsync(CreateOrderRequest request)
    {
        // Start transaction
        using var transaction = await _dbContext.Database.BeginTransactionAsync();
        
        try
        {
            // Save order to database
            var order = new Order { /* ... */ };
            _dbContext.Orders.Add(order);
            await _dbContext.SaveChangesAsync();
            
            // Publish event (saved to outbox table in same transaction)
            await _publisher.PublishAsync(new OrderCreatedEvent
            {
                OrderId = order.Id,
                Amount = order.Amount,
                CreatedAt = DateTime.UtcNow
            });
            
            // Commit transaction
            await transaction.CommitAsync();
        }
        catch
        {
            await transaction.RollbackAsync();
            throw;
        }
    }
}

Using Direct Publisher (Fire-and-Forget)

public class NotificationService
{
    private readonly IDirectPublisher _publisher;

    public NotificationService(IDirectPublisher publisher)
    {
        _publisher = publisher;
    }

    public async Task SendNotificationAsync(string userId, string message)
    {
        await _publisher.PublishAsync(new NotificationEvent
        {
            UserId = userId,
            Message = message,
            SentAt = DateTime.UtcNow
        });
    }
}

Publishing Delayed Messages

// Publish a message to be delivered after 5 minutes
await _publisher.PublishDelayedAsync(
    new ReminderEvent { UserId = "123", Message = "Don't forget!" },
    delay: TimeSpan.FromMinutes(5)
);

4. Consume Messages

[MessageBinding("my.events", "order-service-queue", RoutingKey = "order.created")]
public class OrderCreatedConsumer : ConsumerBase<OrderCreatedEvent>
{
    private readonly ILogger<OrderCreatedConsumer> _logger;
    private readonly IOrderProcessor _orderProcessor;

    public OrderCreatedConsumer(
        ILogger<OrderCreatedConsumer> logger,
        IOrderProcessor orderProcessor) : base(logger)
    {
        _logger = logger;
        _orderProcessor = orderProcessor;
    }

    protected override async Task HandleMessageAsync(OrderCreatedEvent message)
    {
        _logger.LogInformation("Processing order {OrderId}", message.OrderId);
        
        await _orderProcessor.ProcessOrderAsync(message.OrderId);
        
        _logger.LogInformation("Order {OrderId} processed successfully", message.OrderId);
    }
}

Consumer with Inbox (Automatic Duplicate Detection)

[MessageBinding("my.events", "payment-service-queue", RoutingKey = "order.created")]
public class PaymentConsumer : InboxConsumerBase<OrderCreatedEvent>
{
    private readonly IPaymentService _paymentService;

    public PaymentConsumer(
        ILogger<PaymentConsumer> logger,
        IInboxService inboxService,
        IPaymentService paymentService) : base(logger, inboxService)
    {
        _paymentService = paymentService;
    }

    protected override async Task HandleMessageAsync(OrderCreatedEvent message)
    {
        // This will only be called once per unique message ID
        // Duplicates are automatically detected and rejected
        await _paymentService.ProcessPaymentAsync(message.OrderId, message.Amount);
    }
}

5. Access the Dashboard

Navigate to: http://localhost:5000/eventtransit

The dashboard provides:

• 📊 Overview - Total messages, pending, completed, dead letters
• 📤 Outbox - View all outgoing messages with filtering and search
• 📥 Inbox - View all incoming messages with duplicate tracking
• ☠️ Dead Letters - View failed messages with retry/republish capabilities
• 🔍 Message Details - Click any message to see full payload, headers, and history

📚 Documentation

When to Use Which Publisher?

Recommendation: Use Outbox Publisher for critical business events where you cannot afford to lose messages. Use Direct Publisher for high-throughput scenarios where occasional message loss is acceptable.

Exchange Types

// Direct Exchange - Point-to-point messaging
[MessageBinding("orders.direct", "order-queue", RoutingKey = "order.created")]

// Topic Exchange - Pattern-based routing
[MessageBinding("events.topic", "payment-queue", RoutingKey = "order.*.created")]

// Fanout Exchange - Broadcast to all queues
[MessageBinding("notifications.fanout", "email-queue")]

// Headers Exchange - Route based on message headers
[MessageBinding("tasks.headers", "priority-queue", Headers = new { priority = "high" })]

Configuration Options

For complete configuration options, see EVENTTRANSIT-DOCUMENTATION.md

🏗️ Architecture

EventTransit follows clean architecture principles:

┌─────────────────────────────────────────────────────────────┐
│                        Your Application                      │
├─────────────────────────────────────────────────────────────┤
│  Publishers (IOutboxPublisher, IDirectPublisher)            │
│  Consumers (ConsumerBase<T>, InboxConsumerBase<T>)          │
├─────────────────────────────────────────────────────────────┤
│                      EventTransit Core                       │
│  - Message Routing                                           │
│  - Serialization                                             │
│  - Error Handling                                            │
├─────────────────────────────────────────────────────────────┤
│                   Outbox/Inbox Patterns                      │
│  - Transactional Outbox                                      │
│  - Duplicate Detection                                       │
│  - Retry Logic                                               │
├─────────────────────────────────────────────────────────────┤
│                    Broker Abstraction                        │
│  - RabbitMQ                                                  │
│  - Kafka (Coming Soon)                                       │
│  - Azure Service Bus (Coming Soon)                           │
└─────────────────────────────────────────────────────────────┘

🔧 Advanced Features

Custom Message Headers

await _publisher.PublishAsync(message, headers: new Dictionary<string, object>
{
    { "correlation-id", correlationId },
    { "user-id", userId },
    { "priority", 5 }
});

Batch Publishing

var messages = new List<OrderCreatedEvent> { /* ... */ };
await _publisher.PublishBatchAsync(messages);

Manual Retry from Dashboard

1. Navigate to Dead Letters page
2. Select failed messages
3. Click "Retry" or "Republish"
4. Messages are requeued automatically

🐛 Troubleshooting

"HandleMessageAsync doesn't exist" Error

If you encounter this error after installing EventTransit:

# Clear NuGet cache
dotnet nuget locals all --clear

# Restore without cache
dotnet restore --no-cache

# Rebuild
dotnet build

This ensures you're using the latest version of the library.

🤝 Contributing

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

📄 License

This project is licensed under the MIT License.

🙏 Acknowledgments

Built with ❤️ by Robert Orosume Eru