Messaging.RabbitMq.Core 1.3.0

Messaging.RabbitMq

Shared RabbitMQ configuration and publisher utilities for all services. REQUIRED for publishing notification events.

Installation

dotnet add package Messaging.RabbitMq

Configuration

Add to your appsettings.json:

{
  "RabbitMq": {
    "Host": "localhost",
    "Port": 5672,
    "VirtualHost": "/",
    "Username": "guest",
    "Password": "guest"
  }
}

Usage

Register in Program.cs

using Messaging.RabbitMq.Extensions;

var builder = WebApplication.CreateBuilder(args);

// Add RabbitMQ messaging (required for all services that publish notifications)
builder.Services.AddRabbitMqMessaging(builder.Configuration);

// For services that also consume events (e.g., NotificationService):
builder.Services.AddRabbitMqMessaging(builder.Configuration, x =>
{
    x.AddConsumer<NotificationEventConsumer>();
});

Custom Resilience Options

Override the default resilience settings per service:

using Messaging.RabbitMq.Configuration;

builder.Services.AddRabbitMqMessaging(builder.Configuration,
    configureConsumers: x => { x.AddConsumer<MyConsumer>(); },
    resilienceOptions: new ResilienceOptions
    {
        RetryIntervals = [TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(10)],
        CircuitBreakerTripThreshold = 3,
        CircuitBreakerActiveDuration = TimeSpan.FromSeconds(30),
    });

Publish Events

Inject INotificationEventPublisher and publish events:

using Messaging.RabbitMq.Publishers;
using NotificationService.Contracts.Events;

public class QuestionnaireService
{
    private readonly INotificationEventPublisher _notificationPublisher;

    public QuestionnaireService(INotificationEventPublisher notificationPublisher)
    {
        _notificationPublisher = notificationPublisher;
    }

    public async Task SubmitQuestionnaireAsync(QuestionnaireSubmission submission)
    {
        // ... save submission ...

        // Publish notification event
        await _notificationPublisher.PublishAsync(new QuestionnaireSubmittedEvent
        {
            TenantId = submission.TenantId,
            UserId = templateOwnerId,
            QuestionnaireId = submission.Id,
            TemplateId = submission.TemplateId,
            TemplateName = template.Name,
            RespondentName = submission.RespondentName
        });
    }
}

Graceful Degradation (Fire-and-Forget)

Use TryPublishAsync when notifications are non-critical and the primary operation must succeed even if RabbitMQ is down:

public async Task UpdateMenuAsync(Menu menu)
{
    // Primary operation: save menu (must succeed)
    await _repository.UpdateAsync(menu);

    // Non-critical: send notification (must NOT fail the menu update)
    await _notificationPublisher.TryPublishAsync(new MenuUpdatedEvent
    {
        TenantId = menu.TenantId,
        UserId = menu.UserId,
        MenuId = menu.Id,
        MenuName = menu.Name,
        UpdatedByUserName = currentUser.Name
    });
    // Returns false if publish failed -- error is logged, no exception thrown
}

Batch Publishing

For multiple notifications:

var events = users.Select(u => new TemplateUpdatedEvent
{
    TenantId = tenantId,
    UserId = u.Id,
    TemplateId = template.Id,
    TemplateName = template.Name,
    UpdatedByUserName = currentUser.Name
});

await _notificationPublisher.PublishBatchAsync(events);

// Or with graceful degradation:
int successCount = await _notificationPublisher.TryPublishBatchAsync(events);

Raw JSON interop (plain AMQP publishers, no MassTransit envelope) — since 1.3.0

Opt-in. Services that do not call these methods are unchanged.

Use it when the other side is not MassTransit (for example NestJS amqplib) and puts the bare JSON object on the wire, with no envelope and no MT-MessageType header.

services.AddRabbitMqMessaging(builder.Configuration, x =>
{
    // Consume a raw JSON body from a topic exchange + routing key into a named durable queue.
    x.AddRawJsonConsumer<VerificationCompletedConsumer>("aml.verification-completed",
        e => e.BindToExchange("proovid.verification", "verification.completed"));

    // Publish this message type as a raw JSON body (content-type application/json) to a fixed exchange + key.
    x.AddRawJsonPublisher<VerificationScreeningCompleted>(
        "proovid.verification", "verification.screening.completed");
},
new ResilienceOptions
{
    // Never retried: the message goes straight to aml.verification-completed_error.
    NonRetryableExceptions = [typeof(UnknownTenantException)]
});

Notes:

• JSON property names follow System.Text.Json; for a snake_case wire format put [JsonPropertyName("request_id")] on the contract.
• NonRetryableExceptions is bus-wide (MassTransit's retry filter is configured on the bus), so use exception types specific to the failure.
• Adding any raw JSON registration also registers the raw JSON serializer for application/json on the bus. Enveloped MassTransit traffic uses application/vnd.masstransit+json and is unaffected.
• Integration tests (tests/.../RawJson/RawJsonInteropTests.cs, trait Category=Integration) use Testcontainers and need Docker. Offline, set TESTCONTAINERS_RYUK_DISABLED=true if the Ryuk image cannot be pulled.

Resilience Features

• Automatic Retry: Messages are retried with exponential backoff (1s, 5s, 15s, 30s)
• Circuit Breaker: Trips after 5 failures in 30s, resets after 60s. Prevents cascading failures by pausing message consumption while the system recovers.
• In-Memory Outbox: Publishes made during a consumer are held until it completes successfully, so a consumer that throws does not announce work it did not do. ⚠️ This is NOT a transactional outbox — the pending messages live in process memory, are tied to no database transaction, and are lost on crash. For durability (message committed in the SAME transaction as the state change) you need MassTransit's AddEntityFrameworkOutbox + UseBusOutbox, which this package does not wire up, or a domain outbox table.
• Error Queues: MassTransit automatically creates _error and _skipped queues for each consumer endpoint. Messages that fail all retries are moved to _error (dead-letter).
• Graceful Degradation: TryPublishAsync / TryPublishBatchAsync swallow exceptions for fire-and-forget scenarios.
• Health Check Degradation: RabbitMQ health reports Degraded (not Unhealthy) so publisher-only services keep serving HTTP requests.
• Structured Logging: All publish operations are logged with correlation IDs

Default Resilience Settings

Dependencies

• MassTransit.RabbitMQ - Message broker abstraction
• NotificationService.Contracts - Event contracts

License

MIT