### Manage Dynamic Consumers with IDynamicConsumer

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Demonstrates a controller implementation for starting, stopping, and publishing to dynamic consumers using IDynamicConsumer. Includes examples for inline delegates, typed consumers, and event bus consumers.

```csharp
using Maomi.MQ;
using Maomi.MQ.Attributes;
using Microsoft.AspNetCore.Mvc;

[ApiController]
[Route("api/consumers")]
public class DynamicConsumerController : ControllerBase
{
    private readonly IDynamicConsumer _dynamicConsumer;
    private readonly IMessagePublisher _publisher;
    private readonly ILogger<DynamicConsumerController> _logger;

    public DynamicConsumerController(
        IDynamicConsumer dynamicConsumer,
        IMessagePublisher publisher,
        ILogger<DynamicConsumerController> logger)
    {
        _dynamicConsumer = dynamicConsumer;
        _publisher = publisher;
        _logger = logger;
    }

    // Start consumer with inline delegate handlers
    [HttpPost("start/{queueName}")]
    public async Task<IActionResult> StartConsumer(string queueName, [FromQuery] ushort qos = 10)
    {
        var options = new ConsumerAttribute(queueName) { Qos = qos };

        var consumerTag = await _dynamicConsumer.ConsumerAsync<DynamicMessage>(
            options,
            execute: async (header, message) =>
            {
                _logger.LogInformation(
                    "Dynamic consumer received. Queue={Queue}, MessageId={Id}",
                    queueName,
                    message.Id);
                await Task.CompletedTask;
            },
            faild: async (header, ex, retryCount, message) =>
            {
                _logger.LogWarning(ex, "Dynamic consumer failed. Retry={RetryCount}", retryCount);
                await Task.CompletedTask;
            },
            fallback: (header, message, ex) =>
            {
                _logger.LogError(ex, "Dynamic consumer fallback reached.");
                return Task.FromResult(ConsumerState.Ack);
            });

        return Ok(new { QueueName = queueName, ConsumerTag = consumerTag });
    }

    // Start consumer using existing IConsumer implementation
    [HttpPost("start-typed/{queueName}")]
    public async Task<IActionResult> StartTypedConsumer(string queueName)
    {
        var options = new ConsumerAttribute(queueName) { Qos = 5 };

        var consumerTag = await _dynamicConsumer.ConsumerAsync<MyTypedConsumer, DynamicMessage>(options);

        return Ok(new { QueueName = queueName, ConsumerTag = consumerTag });
    }

    // Start event bus consumer dynamically
    [HttpPost("start-eventbus/{queueName}")]
    public async Task<IActionResult> StartEventBusConsumer(string queueName)
    {
        var options = new ConsumerAttribute(queueName) { Qos = 5 };

        var consumerTag = await _dynamicConsumer.EventBusAsync<DynamicMessage>(options);

        return Ok(new { QueueName = queueName, ConsumerTag = consumerTag });
    }

    // Stop consumer by queue name
    [HttpDelete("stop/{queueName}")]
    public async Task<IActionResult> StopConsumer(string queueName)
    {
        await _dynamicConsumer.StopConsumerAsync(queueName);
        return Ok(new { QueueName = queueName, Stopped = true });
    }

    // Stop consumer by consumer tag
    [HttpDelete("stop-tag/{consumerTag}")]
    public async Task<IActionResult> StopConsumerByTag(string consumerTag)
    {
        await _dynamicConsumer.StopConsumerTagAsync(consumerTag);
        return Ok(new { ConsumerTag = consumerTag, Stopped = true });
    }

    // Publish to dynamic queue
    [HttpPost("publish/{queueName}")]
    public async Task<IActionResult> Publish(string queueName, [FromBody] string text)
    {
        var message = new DynamicMessage { Text = text };
        await _publisher.PublishAsync(string.Empty, queueName, message);
        return Ok(message);
    }
}

public class DynamicMessage
{
    public Guid Id { get; set; } = Guid.NewGuid();
    public string Text { get; set; } = string.Empty;
}

public class MyTypedConsumer : IConsumer<DynamicMessage>
{
    public Task ExecuteAsync(MessageHeader header, DynamicMessage message) => Task.CompletedTask;
    public Task FaildAsync(MessageHeader header, Exception ex, int retryCount, DynamicMessage message) => Task.CompletedTask;
    public Task<ConsumerState> FallbackAsync(MessageHeader header, DynamicMessage? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}
```

--------------------------------

### Start Event Bus Consumer Dynamically

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Dynamically starts a consumer for an event bus. This allows for flexible subscription to events without pre-configuration.

```APIDOC
## POST /api/consumers/start-eventbus/{queueName}

### Description
Dynamically starts an event bus consumer.

### Method
POST

### Endpoint
/api/consumers/start-eventbus/{queueName}

### Parameters
#### Path Parameters
- **queueName** (string) - Required - The name of the queue to consume from.

#### Query Parameters
(Not applicable for this endpoint)

#### Request Body
(Not applicable for this endpoint)

### Response
#### Success Response (200)
- **queueName** (string) - The name of the started consumer's queue.
- **consumerTag** (string) - The unique tag identifying the started consumer.

#### Response Example
```json
{
  "queueName": "your_queue_name",
  "consumerTag": "generated_consumer_tag"
}
```
```

--------------------------------

### Start Typed Consumer Using Existing IConsumer Implementation

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Starts a consumer dynamically by referencing an existing IConsumer implementation. This is useful for reusing consumer logic defined elsewhere in the application.

```APIDOC
## POST /api/consumers/start-typed/{queueName}

### Description
Starts a consumer dynamically using an existing IConsumer implementation.

### Method
POST

### Endpoint
/api/consumers/start-typed/{queueName}

### Parameters
#### Path Parameters
- **queueName** (string) - Required - The name of the queue to consume from.

#### Query Parameters
(Not applicable for this endpoint)

#### Request Body
(Not applicable for this endpoint)

### Response
#### Success Response (200)
- **queueName** (string) - The name of the started consumer's queue.
- **consumerTag** (string) - The unique tag identifying the started consumer.

#### Response Example
```json
{
  "queueName": "your_queue_name",
  "consumerTag": "generated_consumer_tag"
}
```
```

--------------------------------

### Start Consumer with Inline Delegate Handlers

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Starts a consumer dynamically with specified queue name and quality of service (QoS). It accepts inline delegate handlers for message processing, failure handling, and fallback scenarios.

```APIDOC
## POST /api/consumers/start/{queueName}

### Description
Starts a consumer dynamically with inline delegate handlers.

### Method
POST

### Endpoint
/api/consumers/start/{queueName}

### Parameters
#### Path Parameters
- **queueName** (string) - Required - The name of the queue to consume from.

#### Query Parameters
- **qos** (ushort) - Optional - The quality of service for the consumer. Defaults to 10.

### Request Body
(Not applicable for this endpoint, parameters are in path and query)

### Response
#### Success Response (200)
- **queueName** (string) - The name of the started consumer's queue.
- **consumerTag** (string) - The unique tag identifying the started consumer.

#### Response Example
```json
{
  "queueName": "your_queue_name",
  "consumerTag": "generated_consumer_tag"
}
```
```

--------------------------------

### Build All Samples with Dotnet CLI

Source: https://github.com/whuanle/maomi.mq/blob/main/examples/README.md

Build all sample projects for Maomi.MQ using the dotnet CLI. Ensure RabbitMQ is available and optionally set the RABBITMQ environment variable.

```bash
dotnet build examples/Maomi.MQ.Examples.sln
```

--------------------------------

### Run ScenarioHub API Sample

Source: https://github.com/whuanle/maomi.mq/blob/main/examples/README.md

Run the ScenarioHub API sample project using the dotnet CLI. This project is recommended for API-triggered end-to-end verification.

```bash
dotnet run --project examples/00-ScenarioHub/Maomi.MQ.Examples.ScenarioHub.Api
```

--------------------------------

### Configure Basic Consumer with Dead-Letter Handling

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Use this configuration for basic consumers that require dead-letter handling. It specifies the queue name, prefetch count, and whether to requeue messages on failure. The DeadExchange and DeadRoutingKey properties define where failed messages are routed.

```csharp
using Maomi.MQ;
using Maomi.MQ.Attributes;

// Basic consumer with dead-letter configuration
[Consumer(
    "payments.process",
    Qos = 5,                              // Prefetch count
    RetryFaildRequeue = false,            // Don't requeue on failure
    DeadExchange = "",                    // Default exchange for dead letters
    DeadRoutingKey = "payments.dead")]    // Dead-letter queue
public sealed class PaymentConsumer : IConsumer<PaymentMessage>
{
    public Task ExecuteAsync(MessageHeader header, PaymentMessage message)
        => Task.CompletedTask;
    public Task FaildAsync(MessageHeader header, Exception ex, int retryCount, PaymentMessage message)
        => Task.CompletedTask;
    public Task<ConsumerState> FallbackAsync(MessageHeader header, PaymentMessage? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}
```

--------------------------------

### Implement Basic Consumer with IConsumer<TMessage>

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Implement IConsumer<TMessage> to process messages from RabbitMQ. This pattern includes automatic deserialization, retry logic for transient failures, and fallback mechanisms for unrecoverable errors. Configure consumers with attributes like [Consumer] for routing and QoS settings.

```csharp
using Maomi.MQ;
using Maomi.MQ.Attributes;

[Consumer("orders.created", Qos = 10)]  // Queue name and prefetch count
public sealed class OrderCreatedConsumer : IConsumer<OrderCreatedMessage>
{
    private readonly ILogger<OrderCreatedConsumer> _logger;
    private readonly IOrderService _orderService;

    public OrderCreatedConsumer(
        ILogger<OrderCreatedConsumer> logger,
        IOrderService orderService)
    {
        _logger = logger;
        _orderService = orderService;
    }

    // Called when message is successfully deserialized
    public async Task ExecuteAsync(MessageHeader header, OrderCreatedMessage message)
    {
        _logger.LogInformation(
            "Processing order. HeaderId={HeaderId}, OrderId={OrderId}, Amount={Amount}",
            header.Id,
            message.OrderId,
            message.Amount);

        await _orderService.ProcessOrderAsync(message);
    }

    // Called on each retry attempt when ExecuteAsync throws
    public Task FaildAsync(
        MessageHeader header,
        Exception ex,
        int retryCount,
        OrderCreatedMessage message)
    {
        _logger.LogWarning(
            ex,
            "Order processing failed. OrderId={OrderId}, Retry={RetryCount}",
            message.OrderId,
            retryCount);
        return Task.CompletedTask;
    }

    // Called after all retries exhausted or unrecoverable error
    public Task<ConsumerState> FallbackAsync(
        MessageHeader header,
        OrderCreatedMessage? message,
        Exception? ex)
    {
        _logger.LogError(
            ex,
            "Order processing failed permanently. OrderId={OrderId}",
            message?.OrderId);

        // ConsumerState options:
        // - Ack: Acknowledge and remove from queue
        // - Nack: Negative ack with default requeue behavior
        // - NackAndRequeue: Nack and put back on queue
        // - NackAndNoRequeue: Nack and discard
        return Task.FromResult(ConsumerState.Ack);
    }
}
```

--------------------------------

### Implement Event Bus Pattern with Middleware and Handlers

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Defines an event message, a middleware for pipeline control, and multiple ordered handlers with compensation logic.

```csharp
using Maomi.MQ;
using Maomi.MQ.Attributes;
using Maomi.MQ.EventBus;

// Message definition
[RouterKey("orders.saga")]
public sealed class OrderSagaEvent
{
    public Guid OrderId { get; set; } = Guid.NewGuid();
    public string CustomerId { get; set; } = string.Empty;
    public decimal Amount { get; set; }
    public List<string> Items { get; set; } = new();
}

// Middleware: Entry point for event bus, controls the execution pipeline
[Consumer("orders.saga")]
public sealed class OrderSagaMiddleware : IEventMiddleware<OrderSagaEvent>
{
    private readonly ILogger<OrderSagaMiddleware> _logger;

    public OrderSagaMiddleware(ILogger<OrderSagaMiddleware> logger)
    {
        _logger = logger;
    }

    public async Task ExecuteAsync(
        MessageHeader header,
        OrderSagaEvent message,
        EventHandlerDelegate<OrderSagaEvent> next)
    {
        _logger.LogInformation("Starting order saga. OrderId={OrderId}", message.OrderId);

        // Execute the handler chain
        await next(header, message, CancellationToken.None);

        _logger.LogInformation("Order saga completed. OrderId={OrderId}", message.OrderId);
    }

    public Task FaildAsync(MessageHeader header, Exception ex, int retryCount, OrderSagaEvent? message)
    {
        _logger.LogWarning(ex, "Saga step failed. OrderId={OrderId}, Retry={RetryCount}",
            message?.OrderId, retryCount);
        return Task.CompletedTask;
    }

    public Task<ConsumerState> FallbackAsync(MessageHeader header, OrderSagaEvent? message, Exception? ex)
    {
        _logger.LogError(ex, "Saga failed completely. OrderId={OrderId}", message?.OrderId);
        return Task.FromResult(ConsumerState.Ack);
    }
}

// Handler 1: Reserve inventory (executes first)
[EventOrder(1)]
public sealed class ReserveInventoryHandler : IEventHandler<OrderSagaEvent>
{
    private readonly ILogger<ReserveInventoryHandler> _logger;
    private readonly IInventoryService _inventory;

    public ReserveInventoryHandler(ILogger<ReserveInventoryHandler> logger, IInventoryService inventory)
    {
        _logger = logger;
        _inventory = inventory;
    }

    public async Task ExecuteAsync(OrderSagaEvent message, CancellationToken cancellationToken)
    {
        _logger.LogInformation("[Step 1] Reserving inventory for OrderId={OrderId}", message.OrderId);
        await _inventory.ReserveAsync(message.OrderId, message.Items, cancellationToken);
    }

    // Called if a later handler fails - enables saga compensation
    public async Task CancelAsync(OrderSagaEvent message, CancellationToken cancellationToken)
    {
        _logger.LogInformation("[Step 1] Rolling back inventory for OrderId={OrderId}", message.OrderId);
        await _inventory.ReleaseReservationAsync(message.OrderId, cancellationToken);
    }
}

// Handler 2: Process payment (executes second)
[EventOrder(2)]
public sealed class ProcessPaymentHandler : IEventHandler<OrderSagaEvent>
{
    private readonly ILogger<ProcessPaymentHandler> _logger;
    private readonly IPaymentService _payment;

    public ProcessPaymentHandler(ILogger<ProcessPaymentHandler> logger, IPaymentService payment)
    {
        _logger = logger;
        _payment = payment;
    }

    public async Task ExecuteAsync(OrderSagaEvent message, CancellationToken cancellationToken)
    {
        _logger.LogInformation("[Step 2] Processing payment for OrderId={OrderId}, Amount={Amount}",
            message.OrderId, message.Amount);
        await _payment.ChargeAsync(message.CustomerId, message.Amount, cancellationToken);
    }

    public async Task CancelAsync(OrderSagaEvent message, CancellationToken cancellationToken)
    {
        _logger.LogInformation("[Step 2] Refunding payment for OrderId={OrderId}", message.OrderId);
        await _payment.RefundAsync(message.CustomerId, message.Amount, cancellationToken);
    }
}

// Handler 3: Send confirmation (executes third)
[EventOrder(3)]
public sealed class SendConfirmationHandler : IEventHandler<OrderSagaEvent>
{
    private readonly ILogger<SendConfirmationHandler> _logger;
    private readonly INotificationService _notifications;

    public SendConfirmationHandler(ILogger<SendConfirmationHandler> logger, INotificationService notifications)
    {
        _logger = logger;
        _notifications = notifications;
    }

    public async Task ExecuteAsync(OrderSagaEvent message, CancellationToken cancellationToken)
    {
        _logger.LogInformation("[Step 3] Sending confirmation for OrderId={OrderId}", message.OrderId);
        await _notifications.SendOrderConfirmationAsync(message.CustomerId, message.OrderId, cancellationToken);
    }

    public Task CancelAsync(OrderSagaEvent message, CancellationToken cancellationToken)
    {

```

--------------------------------

### Register Maomi.MQ Services

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Registers Maomi.MQ services with dependency injection. Configure RabbitMQ connection details and specify assemblies to scan for consumers. Ensure the Work ID is unique for distributed ID generation.

```csharp
using Maomi.MQ;
using Maomi.MQ.Models;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddMaomiMQ(
    (MqOptionsBuilder options) =>
    {
        options.WorkId = 1;                    // Worker ID (0-64) for distributed ID generation
        options.AppName = "my-service";        // Application name for tracing
        options.AutoQueueDeclare = true;       // Auto-create queues on startup
        options.Rabbit = rabbit =>
        {
            rabbit.Uri = new Uri("amqp://guest:guest@localhost:5672");
        };
        // Optional: Add custom serializers
        options.MessageSerializers = serializers =>
        {
            // serializers.Add(new ProtobufMessageSerializer());
        };
    },
    [typeof(Program).Assembly]);  // Assemblies to scan for consumers

var app = builder.Build();
app.Run();

```

--------------------------------

### Configure Consumer with Fanout Exchange Binding for Broadcast

Source: https://context7.com/whuanle/maomi.mq/llms.txt

This configuration is used for consumers that need to receive broadcast messages via a fanout exchange. It binds the consumer's queue to a specified exchange with a fanout type, enabling multiple consumers to share the same queue name and receive all published messages.

```csharp
// Consumer with exchange binding (fanout broadcast)
[Consumer(
    "notifications.broadcast",
    BindExchange = "notifications.exchange",
    ExchangeType = ExchangeType.Fanout,
    RoutingKey = "notifications.all",
    IsBroadcast = true,                   // Multiple consumers can share queue name
    Qos = 1)]
public sealed class NotificationConsumer : IConsumer<NotificationMessage>
{
    public Task ExecuteAsync(MessageHeader header, NotificationMessage message)
        => Task.CompletedTask;
    public Task FaildAsync(MessageHeader header, Exception ex, int retryCount, NotificationMessage message)
        => Task.CompletedTask;
    public Task<ConsumerState> FallbackAsync(MessageHeader header, NotificationMessage? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}
```

--------------------------------

### Implement Custom Message Serializer in C#

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Implement IMessageSerializer for custom serialization formats like Protobuf or MessagePack. Register custom serializers in Program.cs by clearing defaults and adding your implementation.

```csharp
using Maomi.MQ;
using System.Text.Json;

// Custom serializer implementation
public sealed class CustomJsonSerializer : IMessageSerializer
{
    private static readonly JsonSerializerOptions Options = new()
    {
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
        WriteIndented = false
    };

    public string ContentType => "application/json";

    public bool SerializerVerify<TObject>(TObject obj) => true;

    public bool SerializerVerify<TObject>() => true;

    public byte[] Serializer<TObject>(TObject obj)
    {
        return JsonSerializer.SerializeToUtf8Bytes(obj, Options);
    }

    public TObject? Deserialize<TObject>(ReadOnlySpan<byte> bytes)
    {
        return JsonSerializer.Deserialize<TObject>(bytes, Options);
    }
}
```

```csharp
// Registration in Program.cs
builder.Services.AddMaomiMQ(
    options =>
    {
        options.WorkId = 1;
        options.AppName = "custom-serializer-app";
        options.Rabbit = rabbit => rabbit.Uri = new Uri("amqp://localhost");

        // Add custom serializers
        options.MessageSerializers = serializers =>
        {
            serializers.Clear();  // Remove default JSON serializer
            serializers.Add(new CustomJsonSerializer());
            // serializers.Add(new ProtobufMessageSerializer());
            // serializers.Add(new MessagePackSerializer());
        };
    },
    [typeof(Program).Assembly]);
```

--------------------------------

### Implement Transactional Outbox Pattern

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Ensures reliable message delivery by registering messages within a database transaction. Requires ITransactionOutboxService and appropriate database transaction handling.

```csharp
using Maomi.MQ;
using Maomi.MQ.Transaction;
using Microsoft.AspNetCore.Mvc;
using MySqlConnector;

[ApiController]
[Route("api/orders")]
public class TransactionalOrdersController : ControllerBase
{
    private readonly ITransactionOutboxService _outboxService;
    private readonly string _connectionString;

    public TransactionalOrdersController(
        ITransactionOutboxService outboxService,
        IConfiguration configuration)
    {
        _outboxService = outboxService;
        _connectionString = configuration.GetConnectionString("DefaultConnection")!;
    }

    [HttpPost("create")]
    public async Task<IActionResult> CreateOrder([FromBody] CreateOrderRequest request)
    {
        var orderId = Guid.NewGuid();
        var message = new OrderCreatedMessage
        {
            OrderId = orderId,
            CustomerName = request.CustomerName,
            Amount = request.Amount,
            CreatedAt = DateTimeOffset.UtcNow
        };

        await using var connection = new MySqlConnection(_connectionString);
        await connection.OpenAsync();
        await using var transaction = await connection.BeginTransactionAsync();

        try
        {
            // Insert order into database
            const string insertSql = @"
                INSERT INTO orders (order_id, customer_name, amount, created_at)
                VALUES (@orderId, @customerName, @amount, @createdAt)";

            using var cmd = new MySqlCommand(insertSql, connection, transaction);
            cmd.Parameters.AddWithValue("@orderId", orderId.ToString("N"));
            cmd.Parameters.AddWithValue("@customerName", request.CustomerName);
            cmd.Parameters.AddWithValue("@amount", request.Amount);
            cmd.Parameters.AddWithValue("@createdAt", message.CreatedAt.UtcDateTime);
            await cmd.ExecuteNonQueryAsync();

            // Register message in outbox (same transaction)
            var outbox = await _outboxService.RegisterAsync(
                connection,
                transaction,
                exchange: string.Empty,
                routingKey: "orders.created",
                message: message);

            // Commit database transaction
            await transaction.CommitAsync();

            // Publish message after commit succeeds
            await outbox.PublishAsync();

            return Ok(new
            {
                OrderId = orderId,
                OutboxMessageId = outbox.MessageId,
                Status = "Created"
            });
        }
        catch
        {
            await transaction.RollbackAsync();
            throw;
        }
    }
}

// Setup for Transaction support in Program.cs
// builder.Services.AddMaomiMQTransaction(options =>
// {
//     options.ConnectionString = "your-connection-string";
// });

public class CreateOrderRequest
{
    public string CustomerName { get; set; } = string.Empty;
    public decimal Amount { get; set;
}
```

--------------------------------

### Main Consumer with Dead-Letter Forwarding

Source: https://context7.com/whuanle/maomi.mq/llms.txt

This consumer forwards messages that fail processing to a dead-letter queue. Configure DeadExchange and DeadRoutingKey to specify the dead-letter destination. Set RetryFaildRequeue to false to prevent requeuing failed messages back to the original queue.

```csharp
using Maomi.MQ;
using Maomi.MQ.Attributes;

// Main consumer that forwards failures to dead-letter queue
[Consumer(
    "orders.process",
    Qos = 1,
    RetryFaildRequeue = false,
    DeadExchange = "",
    DeadRoutingKey = "orders.dead")]
public sealed class OrderProcessConsumer : IConsumer<OrderMessage>
{
    private readonly ILogger<OrderProcessConsumer> _logger;
    private readonly IMessagePublisher _publisher;

    public OrderProcessConsumer(ILogger<OrderProcessConsumer> logger, IMessagePublisher publisher)
    {
        _logger = logger;
        _publisher = publisher;
    }

    public async Task ExecuteAsync(MessageHeader header, OrderMessage message)
    {
        _logger.LogInformation("Processing order. OrderId={OrderId}", message.OrderId);

        // Simulate failure condition
        if (message.Amount > 10000)
        {
            throw new InvalidOperationException("Amount exceeds limit, requires manual review.");
        }

        await ProcessOrderAsync(message);
    }

    public Task FaildAsync(MessageHeader header, Exception ex, int retryCount, OrderMessage message)
    {
        _logger.LogWarning(ex, "Order processing failed. OrderId={OrderId}, Retry={RetryCount}",
            message.OrderId, retryCount);
        return Task.CompletedTask;
    }

    public async Task<ConsumerState> FallbackAsync(MessageHeader header, OrderMessage? message, Exception? ex)
    {
        _logger.LogError(ex, "Order failed permanently. OrderId={OrderId}", message?.OrderId);

        // Manually publish to dead-letter queue with additional context
        if (message != null)
        {
            await _publisher.PublishAsync(
                exchange: string.Empty,
                routingKey: "orders.dead",
                message: new DeadOrderMessage
                {
                    OriginalOrderId = message.OrderId,
                    OriginalAmount = message.Amount,
                    FailureReason = ex?.Message ?? "Unknown error",
                    FailedAt = DateTimeOffset.UtcNow
                });
        }

        return ConsumerState.Ack;
    }

    private Task ProcessOrderAsync(OrderMessage message) => Task.CompletedTask;
}

public class OrderMessage
{
    public Guid OrderId { get; set; }
    public decimal Amount { get; set; }
}

public class DeadOrderMessage
{
    public Guid OriginalOrderId { get; set; }
    public decimal OriginalAmount { get; set; }
    public string FailureReason { get; set; } = string.Empty;
    public DateTimeOffset FailedAt { get; set; }
}

```

--------------------------------

### Define Message with RouterKeyAttribute

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Defines a message class using the `[RouterKey]` attribute for automatic routing to RabbitMQ exchanges and queues. This attribute specifies the queue or routing key for publishing.

```csharp
using Maomi.MQ.Attributes;

[RouterKey("orders.created")]  // Queue/routing key for auto-publishing
public sealed class OrderCreatedMessage
{
    public Guid OrderId { get; set; } = Guid.NewGuid();
    public string CustomerName { get; set; } = string.Empty;
    public decimal Amount { get; set; }
    public DateTimeOffset CreatedAt { get; set; } = DateTimeOffset.UtcNow;
}

// With exchange binding
[RouterKey("orders.exchange", "orders.created")]
public sealed class OrderWithExchangeMessage
{
    public Guid OrderId { get; set; } = Guid.NewGuid();
    public decimal Amount { get; set; }
}

```

--------------------------------

### Configure Consumer with Message Expiration

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Use this configuration for consumers that process time-sensitive messages. The Expiration property sets a Time-To-Live (TTL) for messages in milliseconds, and AutoQueueDeclare.Enable ensures the queue is declared automatically if it doesn't exist.

```csharp
// Consumer with message expiration
[Consumer(
    "cache.invalidate",
    Qos = 20,
    Expiration = 30000,                   // Message TTL in milliseconds
    AutoQueueDeclare = AutoQueueDeclare.Enable)]
public sealed class CacheInvalidateConsumer : IConsumer<CacheInvalidateMessage>
{
    public Task ExecuteAsync(MessageHeader header, CacheInvalidateMessage message)
        => Task.CompletedTask;
    public Task FaildAsync(MessageHeader header, Exception ex, int retryCount, CacheInvalidateMessage message)
        => Task.CompletedTask;
    public Task<ConsumerState> FallbackAsync(MessageHeader header, CacheInvalidateMessage? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}
```

--------------------------------

### Publish Messages with IMessagePublisher

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Use IMessagePublisher to send messages to RabbitMQ. Supports automatic routing based on message attributes, explicit exchange and routing key specification, and custom message properties like TTL and priority.

```csharp
using Maomi.MQ;
using Microsoft.AspNetCore.Mvc;
using RabbitMQ.Client;

[ApiController]
[Route("api/orders")]
public class OrdersController : ControllerBase
{
    private readonly IMessagePublisher _publisher;

    public OrdersController(IMessagePublisher publisher)
    {
        _publisher = publisher;
    }

    // Auto-publish using [RouterKey] attribute on message class
    [HttpPost("auto")]
    public async Task<IActionResult> PublishAuto()
    {
        var message = new OrderCreatedMessage
        {
            CustomerName = "John Doe",
            Amount = 99.99m
        };

        await _publisher.AutoPublishAsync(message);
        return Ok(new { message.OrderId });
    }

    // Explicit exchange and routing key
    [HttpPost("explicit")]
    public async Task<IActionResult> PublishExplicit()
    {
        var message = new OrderCreatedMessage
        {
            CustomerName = "Jane Doe",
            Amount = 149.99m
        };

        await _publisher.PublishAsync(
            exchange: "orders.exchange",
            routingKey: "orders.created",
            message: message);

        return Ok(new { message.OrderId });
    }

    // With custom properties (TTL, priority, etc.)
    [HttpPost("with-properties")]
    public async Task<IActionResult> PublishWithProperties()
    {
        var message = new OrderCreatedMessage
        {
            CustomerName = "Bob Smith",
            Amount = 299.99m
        };

        await _publisher.PublishAsync(
            exchange: "",
            routingKey: "orders.priority",
            message: message,
            properties: props =>
            {
                props.Expiration = "60000";  // TTL in milliseconds
                props.Priority = 5;
                props.Headers = new Dictionary<string, object?>
                {
                    ["x-custom-header"] = "custom-value"
                };
            });

        return Ok(new { message.OrderId });
    }
}
```

--------------------------------

### Publish messages using ITransactionPublisher

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Uses RabbitMQ native transactions to ensure multiple messages are published atomically. Requires an ITransactionPublisher instance created within a service scope.

```csharp
using Maomi.MQ;
using Microsoft.AspNetCore.Mvc;

[ApiController]
[Route("api/batch")]
public class BatchController : ControllerBase
{
    private readonly IServiceProvider _serviceProvider;

    public BatchController(IServiceProvider serviceProvider)
    {
        _serviceProvider = serviceProvider;
    }

    [HttpPost("publish-transactional")]
    public async Task<IActionResult> PublishTransactional([FromBody] List<string> items)
    {
        // Create transaction publisher (implements IDisposable)
        using var scope = _serviceProvider.CreateScope();
        var transactionPublisher = scope.ServiceProvider.GetRequiredService<ITransactionPublisher>();

        try
        {
            // Start transaction
            await transactionPublisher.TxSelectAsync();

            // Publish multiple messages in transaction
            foreach (var item in items)
            {
                var message = new BatchItemMessage { Item = item };
                await transactionPublisher.PublishAsync(
                    exchange: string.Empty,
                    routingKey: "batch.items",
                    message: message);
            }

            // Commit all messages atomically
            await transactionPublisher.TxCommitAsync();

            return Ok(new { PublishedCount = items.Count, Status = "Committed" });
        }
        catch (Exception ex)
        {
            // Rollback on any failure
            await transactionPublisher.TxRollbackAsync();
            return StatusCode(500, new { Error = ex.Message, Status = "RolledBack" });
        }
    }
}

public class BatchItemMessage
{
    public Guid Id { get; set; } = Guid.NewGuid();
    public string Item { get; set; } = string.Empty;
}
```

--------------------------------

### Publish Message to Dynamic Queue

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Publishes a message to a specified dynamic queue. This endpoint is used to send data to consumers that are dynamically managed.

```APIDOC
## POST /api/consumers/publish/{queueName}

### Description
Publishes a message to a dynamic queue.

### Method
POST

### Endpoint
/api/consumers/publish/{queueName}

### Parameters
#### Path Parameters
- **queueName** (string) - Required - The name of the queue to publish the message to.

#### Query Parameters
(Not applicable for this endpoint)

#### Request Body
- **text** (string) - Required - The content of the message to be published.

### Response
#### Success Response (200)
- **id** (string) - The unique identifier of the published message.
- **text** (string) - The content of the published message.

#### Response Example
```json
{
  "id": "a1b2c3d4-e5f6-7890-1234-567890abcdef",
  "text": "Hello, dynamic world!"
}
```
```

--------------------------------

### Stop Consumer by Consumer Tag

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Stops a running consumer identified by its unique consumer tag. This provides a precise way to manage individual consumer instances.

```APIDOC
## DELETE /api/consumers/stop-tag/{consumerTag}

### Description
Stops a consumer by its consumer tag.

### Method
DELETE

### Endpoint
/api/consumers/stop-tag/{consumerTag}

### Parameters
#### Path Parameters
- **consumerTag** (string) - Required - The unique tag of the consumer to stop.

#### Query Parameters
(Not applicable for this endpoint)

#### Request Body
(Not applicable for this endpoint)

### Response
#### Success Response (200)
- **consumerTag** (string) - The tag of the consumer that was stopped.
- **stopped** (boolean) - Indicates if the consumer was successfully stopped.

#### Response Example
```json
{
  "consumerTag": "generated_consumer_tag",
  "stopped": true
}
```
```

--------------------------------

### Stop Consumer by Queue Name

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Stops a running consumer identified by its queue name. This is a convenient way to halt consumption from a specific queue.

```APIDOC
## DELETE /api/consumers/stop/{queueName}

### Description
Stops a consumer by its queue name.

### Method
DELETE

### Endpoint
/api/consumers/stop/{queueName}

### Parameters
#### Path Parameters
- **queueName** (string) - Required - The name of the queue whose consumer should be stopped.

#### Query Parameters
(Not applicable for this endpoint)

#### Request Body
(Not applicable for this endpoint)

### Response
#### Success Response (200)
- **queueName** (string) - The name of the queue whose consumer was stopped.
- **stopped** (boolean) - Indicates if the consumer was successfully stopped.

#### Response Example
```json
{
  "queueName": "your_queue_name",
  "stopped": true
}
```
```

--------------------------------

### Dead-Letter Consumer

Source: https://context7.com/whuanle/maomi.mq/llms.txt

This consumer processes messages that have been moved to the dead-letter queue. It logs the failure details for further investigation or manual intervention.

```csharp
// Dead-letter consumer for failed messages
[Consumer("orders.dead", Qos = 1)]
public sealed class DeadOrderConsumer : IConsumer<DeadOrderMessage>
{
    private readonly ILogger<DeadOrderConsumer> _logger;

    public DeadOrderConsumer(ILogger<DeadOrderConsumer> logger)
    {
        _logger = logger;
    }

    public Task ExecuteAsync(MessageHeader header, DeadOrderMessage message)
    {
        _logger.LogWarning(
            "Dead-letter received. OriginalOrderId={OrderId}, Reason={Reason}, FailedAt={FailedAt}",
            message.OriginalOrderId,
            message.FailureReason,
            message.FailedAt);

        // Store for manual review, send alerts, etc.
        return Task.CompletedTask;
    }

    public Task FaildAsync(MessageHeader header, Exception ex, int retryCount, DeadOrderMessage message)
        => Task.CompletedTask;

    public Task<ConsumerState> FallbackAsync(MessageHeader header, DeadOrderMessage? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}

```

--------------------------------

### Access MessageHeader Properties in C# Consumer

Source: https://context7.com/whuanle/maomi.mq/llms.txt

Access metadata about received messages using the MessageHeader struct within a consumer. This includes message ID, timestamp, content type, and routing information for tracing and debugging.

```csharp
using Maomi.MQ;
using Maomi.MQ.Attributes;

[Consumer("audit.events")]
public sealed class AuditConsumer : IConsumer<AuditEvent>
{
    private readonly ILogger<AuditConsumer> _logger;

    public AuditConsumer(ILogger<AuditConsumer> logger)
    {
        _logger = logger;
    }

    public Task ExecuteAsync(MessageHeader header, AuditEvent message)
    {
        // MessageHeader properties
        _logger.LogInformation(
            "Message received: " +
            "Id={Id}, " +
            "Timestamp={Timestamp}, " +
            "ContentType={ContentType}, " +
            "Type={Type}, " +
            "AppId={AppId}, " +
            "Exchange={Exchange}, " +
            "RoutingKey={RoutingKey}",
            header.Id,           // Unique message ID
            header.Timestamp,    // When message was created
            header.ContentType,  // MIME type (e.g., "application/json")
            header.Type,         // Object type name
            header.AppId,        // Source application name
            header.Exchange,     // Original exchange
            header.RoutingKey);  // Original routing key

        // Access raw RabbitMQ properties if needed
        // var basicProperties = (IReadOnlyBasicProperties)header.Properties;

        return Task.CompletedTask;
    }

    public Task FaildAsync(MessageHeader header, Exception ex, int retryCount, AuditEvent message)
        => Task.CompletedTask;

    public Task<ConsumerState> FallbackAsync(MessageHeader header, AuditEvent? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}

public class AuditEvent
{
    public string Action { get; set; } = string.Empty;
    public string UserId { get; set; } = string.Empty;
}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.