IronAlpine.EventBus 3.0.0

IronAlpine.EventBus

Kafka-based event publishing with outbox pattern, SSL/SASL security, and dead-letter handling.

• Target Frameworks: net9.0, net10.0
• Dependencies: Confluent.Kafka
• Package Size: ~80 KB
• Replaces: 2 v2 packages (EventBus.Contracts, EventBus.Kafka)

What It Is

IronAlpine.EventBus provides event-driven infrastructure:

• Event Publishing — async fire-and-forget to Kafka
• Event Subscription — handlers consume topics
• Outbox Pattern — durability (write to DB, Kafka worker picks up)
• SSL/SASL — parameterized security (certificates, passwords)
• Dead-Letter Queue — failed messages captured for replay
• Correlation Tracking — trace requests across services

Installation

dotnet add package IronAlpine.EventBus

Quick Setup

services.AddIronAlpineEventBusKafka(
    configuration,
    kafkaOptions =>
    {
        kafkaOptions.ConsumerGroupId = "TimeOffService-Group";
    },
    topology =>
    {
        topology.PublishEvent<TimeOffRequestedEvent>("timeoff.requested");
        topology.PublishEvent<TimeOffApprovedEvent>("timeoff.approved");
    });

Configuration

Basic

{
  "IronAlpine": {
    "EventBus": {
      "Kafka": {
        "Enabled": true,
        "BootstrapServers": "kafka:9092",
        "ConsumerGroupId": "TimeOffService-Group",
        "ClientId": "TimeOffService"
      }
    }
  }
}

With SSL/SASL

{
  "IronAlpine": {
    "EventBus": {
      "Kafka": {
        "Enabled": true,
        "BootstrapServers": "kafka:9093",
        "ConsumerGroupId": "TimeOffService-Group",
        "ClientId": "TimeOffService",
        
        "SecurityProtocol": "SaslSsl",
        "SaslMechanism": "Plain",
        "SaslUsername": "username",
        "SaslPassword": "password",
        
        "SslCaLocation": "/app/certs/ca-cert",
        "SslCertLocation": "/app/certs/client-cert",
        "SslKeyLocation": "/app/certs/client-key",
        "SslKeyPassword": "key-password"
      }
    }
  }
}

Publish Events

From Domain

// 1. Define domain event
public class TimeOffRequestedEvent : IDomainEvent
{
    public Guid TimeOffId { get; init; }
    public Guid UserId { get; init; }
    public DateTime StartDate { get; init; }
}

// 2. Publish from aggregate
public class TimeOff : AggregateRoot
{
    public void Request(UserId userId, DateRange dates)
    {
        // ... business logic
        AddDomainEvent(new TimeOffRequestedEvent
        {
            TimeOffId = Id.Value,
            UserId = userId.Value,
            StartDate = dates.Start
        });
    }
}

// 3. SaveChangesAsync dispatches automatically
var repo = _unitOfWork.Repository<TimeOff>();
await repo.AddAsync(timeOff);
await _unitOfWork.SaveChangesAsync(ct);  // Event → Kafka

Explicitly from Handler

public class RequestTimeOffCommandHandler : IRequestHandler<RequestTimeOffCommand, Unit>
{
    private readonly IUnitOfWork _unitOfWork;
    private readonly IEventPublisher _eventBus;

    public async Task<Unit> Handle(RequestTimeOffCommand request, CancellationToken ct)
    {
        var timeOff = TimeOff.Create(request.UserId, request.StartDate, request.EndDate);
        
        var repo = _unitOfWork.Repository<TimeOff>();
        await repo.AddAsync(timeOff, ct);
        
        // Explicit publication (if not using aggregate events)
        await _eventBus.PublishAsync(
            new TimeOffRequestedEvent
            {
                TimeOffId = timeOff.Id.Value,
                UserId = request.UserId
            },
            ct);
        
        await _unitOfWork.SaveChangesAsync(ct);
        return Unit.Value;
    }
}

Subscribe to Events

Define Handler

public class TimeOffRequestedEventHandler : IEventHandler<TimeOffRequestedEvent>
{
    private readonly INotificationService _notificationService;
    private readonly IRepository<Manager> _managerRepository;

    public async Task Handle(TimeOffRequestedEvent @event, CancellationToken ct)
    {
        // Get event details
        var manager = await _managerRepository.GetByEmployeeAsync(@event.UserId, ct);
        
        // Take action
        await _notificationService.NotifyAsync(
            manager.Id,
            $"Time off request from employee {@event.UserId}",
            ct);
    }
}

Register Handler

services.AddIronAlpineEventBusKafka(
    configuration,
    null,
    topology =>
    {
        // Define what events are published
        topology.PublishEvent<TimeOffRequestedEvent>("timeoff.requested");
        topology.PublishEvent<TimeOffApprovedEvent>("timeoff.approved");
        
        // Register handlers for subscribed events
        topology
            .PublishEvent<TimeOffRequestedEvent>("timeoff.requested")
            .SubscribeWith<TimeOffRequestedEventHandler>();
        
        topology
            .PublishEvent<TimeOffApprovedEvent>("timeoff.approved")
            .SubscribeWith<NotifyApprovedEventHandler>();
    });

Event Flow

Command Handler
  → Calls aggregate method
  → Aggregate publishes domain event
  → SaveChangesAsync called
  ↓
Database Transaction
  → Writes aggregate state
  → Writes OutboxMessage to DB
  ↓
Kafka Worker (Background Service)
  → Polls OutboxMessage table
  → Publishes to Kafka topic
  → Deletes from OutboxMessage
  ↓
Consumer Services
  → Receive message from topic
  → EventHandler processes it
  → Side effects (notifications, etc.)

Topology Configuration

Define all topics, publishers, and subscribers in one place:

services.AddIronAlpineEventBusKafka(
    configuration,
    null,
    topology =>
    {
        // TimeOff events
        topology
            .PublishEvent<TimeOffRequestedEvent>("timeoff.requested")
            .SubscribeWith<SendApprovalNotificationHandler>()
            .SubscribeWith<LogTimeOffRequestHandler>();
        
        topology
            .PublishEvent<TimeOffApprovedEvent>("timeoff.approved")
            .SubscribeWith<NotifyEmployeeApprovedHandler>()
            .SubscribeWith<PublishPayrollEventHandler>();
        
        // Assignment events
        topology
            .PublishEvent<AssignmentCreatedEvent>("assignment.created")
            .SubscribeWith<CreateAuditLogHandler>();
    });

Security

SSL Certificates

{
  "IronAlpine": {
    "EventBus": {
      "Kafka": {
        "SecurityProtocol": "SaslSsl",
        "SslCaLocation": "/app/certs/ca.crt",
        "SslCertLocation": "/app/certs/client.crt",
        "SslKeyLocation": "/app/certs/client.key",
        "SslKeyPassword": "password"
      }
    }
  }
}

SASL Authentication

{
  "IronAlpine": {
    "EventBus": {
      "Kafka": {
        "SecurityProtocol": "SaslSsl",
        "SaslMechanism": "Plain",
        "SaslUsername": "serviceuser",
        "SaslPassword": "servicepassword"
      }
    }
  }
}

Environment Variables (Docker)

FROM mcr.microsoft.com/dotnet/aspnet:10.0

ENV IronAlpine__EventBus__Kafka__BootstrapServers=kafka:9093
ENV IronAlpine__EventBus__Kafka__SecurityProtocol=SaslSsl
ENV IronAlpine__EventBus__Kafka__SaslUsername=${KAFKA_USER}
ENV IronAlpine__EventBus__Kafka__SaslPassword=${KAFKA_PASSWORD}
ENV IronAlpine__EventBus__Kafka__SslCaLocation=/app/certs/ca.crt

Correlation Tracking

Trace requests across microservices:

// Web request arrives with trace headers
// EventMetadataContext captures: TraceId, CorrelationId, CausationId, TenantId

// When event published, metadata automatically included
await _eventPublisher.PublishAsync(
    new TimeOffRequestedEvent { ... },
    ct);
// Kafka message includes: TraceId, CorrelationId, etc.

// Subscriber receives event with full context
public class EventHandler : IEventHandler<TimeOffRequestedEvent>
{
    private readonly IEventMetadataAccessor _metadataAccessor;

    public async Task Handle(TimeOffRequestedEvent @event, CancellationToken ct)
    {
        var metadata = _metadataAccessor.GetCurrent();
        
        // Log with full trace context
        _logger.LogInformation(
            "Processing event {@Event} with TraceId={TraceId} CorrelationId={CorrelationId}",
            @event,
            metadata.TraceId,
            metadata.CorrelationId);
    }
}

Dead-Letter Handling

Failed messages captured for analysis and replay:

// If EventHandler throws exception
public class EventHandler : IEventHandler<TimeOffRequestedEvent>
{
    public async Task Handle(TimeOffRequestedEvent @event, CancellationToken ct)
    {
        // If exception here → message goes to DLQ
        await _repository.SaveAsync(/* fail */);
    }
}

// Query dead-letter queue
public class GetDeadLetterMessagesQuery : IRequest<List<DeadLetterMessageDto>>
{
}

public class GetDeadLetterMessagesHandler
    : IRequestHandler<GetDeadLetterMessagesQuery, List<DeadLetterMessageDto>>
{
    private readonly IDeadLetterQueue _dlq;

    public async Task<List<DeadLetterMessageDto>> Handle(
        GetDeadLetterMessagesQuery request,
        CancellationToken ct)
    {
        var messages = await _dlq.ListAsync(ct);
        return messages.Select(Map).ToList();
    }
}

Best Practices

✅ DO

// 1. Publish from aggregates (domain events)
public void Approve(UserId approver, string comment)
{
    Status = TimeOffStatus.Approved;
    AddDomainEvent(new TimeOffApprovedEvent(...));  // Published automatically
}

// 2. Name topics clearly
topology.PublishEvent<TimeOffApprovedEvent>("timeoff.approved");

// 3. Handle idempotency
public class EventHandler : IEventHandler<TimeOffApprovedEvent>
{
    public async Task Handle(TimeOffApprovedEvent @event, CancellationToken ct)
    {
        // Check if already processed
        if (await _repository.ExistsAsync(@event.TimeOffId, ct))
            return;  // Idempotent
    }
}

// 4. Use correlation headers
var metadata = _metadataAccessor.GetCurrent();
_logger.LogInformation("TraceId: {TraceId}", metadata.TraceId);

❌ DON'T

// 1. Publish infrastructure events
// Don't: AddDomainEvent(new ServiceStartedEvent(...))

// 2. Ignore error handling
public async Task Handle(TimeOffApprovedEvent @event, CancellationToken ct)
{
    // Missing try-catch → unhandled exception → DLQ
    await _service.ProcessAsync(@event.TimeOffId, ct);
}

// 3. Block in event handlers
public async Task Handle(TimeOffApprovedEvent @event, CancellationToken ct)
{
    // Missing await or async/await
    _service.ProcessAsync(@event.TimeOffId, ct);  // Fire and forget
}

// 4. Use synchronous publishing
// Don't call Publish(...).Result

Troubleshooting

Q: Events not appearing in Kafka?
A: Check (1) Kafka is running, (2) topic exists or auto-creation enabled, (3) OutboxMessage table has records.

Q: Consumer group not receiving messages?
A: Ensure (1) handlers registered via topology, (2) topic name matches, (3) consumer group ID unique.

Q: SSL certificate errors?
A: Verify (1) cert paths are correct, (2) certificate is valid, (3) permissions on cert files.

Examples

See IRONALPINE_V3_DOCUMENTATION.md for detailed examples.

License

MIT