Syed.BuildingBlocks 1.2.1

Syed.Messaging

A transport-agnostic .NET messaging framework with built-in retry, DLQ, middleware, outbox, inbox, sagas, and observability.

Why Syed.Messaging?

Most .NET messaging libraries force you into a specific broker. Syed.Messaging gives you one API across RabbitMQ, Kafka, and Azure Service Bus — with production patterns built in, not bolted on.

// Register everything in one fluent chain
services.AddMessaging(builder =>
{
    builder.UseRabbitMq(o => o.ConnectionString = "amqp://localhost");
    builder.AddMiddleware<TenantContextMiddleware>();
    builder.AddConsumer<OrderCreated, OrderCreatedHandler>(o =>
    {
        o.Destination = "orders.created";
        o.SubscriptionName = "orders.consumer";
        o.RetryPolicy = new RetryPolicy { MaxRetries = 3, Backoff = RetryBackoff.Exponential };
        o.MaxConcurrency = 4;
    });
});

✨ Features

📦 Packages

Installation

Packages are published to GitHub Packages:

dotnet add package Syed.Messaging.Core --version 1.2.0
dotnet add package Syed.Messaging.RabbitMq --version 1.2.0

🚀 Quick Start

1. Define a message and handler

[MessageType("orders.created")]
public record OrderCreated(Guid OrderId, string CustomerId);

public class OrderCreatedHandler : IMessageHandler<OrderCreated>
{
    public async Task HandleAsync(OrderCreated message, MessageContext ctx, CancellationToken ct)
    {
        Console.WriteLine($"Processing order {message.OrderId} for {message.CustomerId}");
    }
}

2. Wire up in Program.cs

var builder = Host.CreateDefaultBuilder(args);
builder.ConfigureServices(services =>
{
    services.AddMessaging(m =>
    {
        m.UseRabbitMq(o =>
        {
            o.ConnectionString = "amqp://guest:guest@localhost:5672";
            o.MainExchangeName = "myapp.events";
        });

        m.AddConsumer<OrderCreated, OrderCreatedHandler>(o =>
        {
            o.Destination = "orders.created";
            o.SubscriptionName = "orders.consumer";
        });
    });
});

await builder.Build().RunAsync();

3. Publish messages

var bus = serviceProvider.GetRequiredService<IMessageBus>();
await bus.PublishAsync("orders.created", new OrderCreated(Guid.NewGuid(), "cust-123"));

🔌 Middleware Pipeline

Middleware runs before every handler — ideal for tenant context, logging, auth propagation:

public class TenantContextMiddleware : IMessageMiddleware
{
    public async Task InvokeAsync(IMessageEnvelope envelope, IServiceProvider sp, Func<Task> next)
    {
        if (envelope.Headers.TryGetValue("tenant-id", out var tenantId))
        {
            var ctx = sp.GetRequiredService<ITenantContext>();
            ctx.SetTenant(tenantId);
        }
        await next(); // handler runs with tenant context set
    }
}

// Register
services.AddMessaging(m => m.AddMiddleware<TenantContextMiddleware>());

Middlewares execute in registration order (first registered = outermost wrapper).

🐇 RabbitMQ Transport

Per-destination queue routing (v1.2.0):

Publisher ──routing key──► Main Exchange (Direct)
                              │
                    ┌─────────┼──────────┐
                    ▼         ▼          ▼
            orders.queue  billing.queue  notifications.queue
                    │
                    ▼ (on failure)
              Retry Exchange ──► Retry Queue (TTL) ──DLX──► Main Exchange
                                                            (preserves routing key)

• Each AddConsumer<T>() auto-declares its own queue bound by destination
• Retry queue preserves original routing key on DLX — messages return to the correct queue
• DLQ captures poison messages with diagnostic headers (x-poison-*)

🧵 Kafka Partition Strategy

Kafka message ordering is partition-scoped, not topic-scoped. To preserve ordering for a business entity, publish a stable partition-key.

await bus.PublishRawAsync(
    "orders.created",
    JsonSerializer.SerializeToUtf8Bytes(new OrderCreated(orderId, customerId)),
    "OrderCreated",
    new Dictionary<string, string> { ["partition-key"] = customerId });

Practical guidance

• Use aggregate IDs (CustomerId, OrderId, TenantId) as partition-key.
• Same key means same partition, which gives deterministic in-order handling for that key.
• Different keys can run in parallel with:

services.AddMessaging(m =>
{
    m.UseKafka(k =>
    {
        k.Consumer.MaxConcurrentPartitions = 4;
        k.Consumer.PartitionAssignmentStrategy = KafkaPartitionAssignmentStrategy.CooperativeSticky;
    });
});

This gives you the common production shape: strict ordering per entity, concurrency across entities.

📤 Outbox Pattern

Guarantee at-least-once delivery with transactional outbox:

// Save to DB + queue atomically
var outbox = scope.ServiceProvider.GetRequiredService<IOutboxStore>();
await outbox.SaveAsync(new OutboxMessage
{
    Destination = "orders.created",
    MessageType = "OrderCreated",
    Payload = JsonSerializer.SerializeToUtf8Bytes(order)
});
await dbContext.SaveChangesAsync(); // single transaction

// OutboxPublisherService polls and publishes in background
services.AddHostedService<OutboxPublisherService>();

Supports raw mode for anonymous payloads (no CLR type resolution needed).

♻ Saga Orchestration

Long-running workflows with state management and distributed locking:

public class OrderSaga : ISagaHandler<OrderSagaState, OrderCreated>
{
    public async Task HandleAsync(OrderSagaState state, OrderCreated msg, ISagaContext ctx)
    {
        state.OrderId = msg.OrderId;
        await ctx.SendAsync("inventory.reserve", new ReserveInventory(msg.OrderId));
        ctx.SetTimeout(TimeSpan.FromMinutes(5), "InventoryTimeout");
    }
}

Persistence: EF Core or in-memory. Locking: Redis, in-memory, or no-op.

📊 Observability

// OpenTelemetry integration
services.AddOpenTelemetry()
    .WithTracing(b => b.AddSource("Syed.Messaging"));

// Built-in metrics (System.Diagnostics.Metrics)
// - messages_published, messages_consumed, messages_retried
// - messages_dead_lettered, handler_duration, handler_errors
// - active_consumers

🧪 Testing

76 tests across 5 test projects:

dotnet test --configuration Release

📋 Changelog

v1.2.0 — Per-Destination Queue Routing

• SubscribeAsync auto-declares per-destination queues bound by routing key
• Each consumer only receives its own messages — fixes shared-queue poison issue
• Retry/DLQ routing preserves original destination
• MainQueueName deprecated

v1.1.0 — Middleware Pipeline

• IMessageMiddleware interface for pre-handler cross-cutting concerns
• Integrated into GenericMessageConsumer and RpcMessageConsumer
• MessagingBuilder.AddMiddleware<T>() for fluent registration

🤝 Contributing

PRs, issues, and design discussions are welcome. This is a platform-style library — architectural feedback is especially valuable.

See ROADMAP.md for planned features and current status.

📝 License

MIT