Documentation for CD4 RabbitMQ SDK

This is the CD4 standard way to communicate with RabbitMQ and should replace all earlier individual implementations of RabbitMQ communication implementations. Any bugs in this standard should be fixed promptly.

If this implementation has any features missing or missing features noticed during migration of existing CD4 services then, this implementation should be improved to include any missing features.

Version History

• Version: 0.1.0.0 - Initial Release
• Version: 0.15.2.0 (2025-09-20) - No API changes. Bug fixes and improvements.
• Version: 0.16.0.0 (2025-10-14) - Added configurable assembly scanning for EventTypeCache. No breaking changes.

Required Configuration

1. appsettings.json Configuration

This SDK requires appsettings.json to have a configuration section named MessageBusConfiguration. An example of the section given below:

{
  "MessageBusConfiguration": {
    "RabbitMqConnectionString": "amqp://username:password@IPAddressOrHostname:5672",
    "RabbitMqMaxConnections": 10,
    "RabbitMqListener": {
      "QueueName": "CD4.ReportDispatcher.Incoming.SMS",
      "ExchangeName": "cd4.topic",
      "ExchangeType": "topic",
      "Durable": true,
      "Exclusive": false,
      "AutoDelete": false,
      "Arguments": null
    },
    "RabbitMqSender": {
      "ExchangeName": "cd4.topic",
      "ExchangeType": "topic",
      "Durable": true,
      "Exclusive": true,
      "AutoDelete": false,
      "Arguments": null
    }
  }
}

2. Event Type Cache Configuration (Optional)

By default, the SDK scans the CD4.EventDispatchingLibrary assembly for event types. If you have events in additional assemblies, configure them during application startup:

// In Program.cs or Startup.cs - BEFORE builder.Build()

// Option 1: Add additional assemblies to scan (recommended)
EventTypeCache.ConfigureAssemblies([
    "YourApp.Events",
    "YourApp.Orders.Events",
    "YourApp.Payments.Events"
]);

// Option 2: Replace the entire assembly list
EventTypeCache.SetAssemblies([
    "CD4.EventDispatchingLibrary",
    "YourApp.Events",
    "YourApp.Orders.Events"
]);

// Optional: For plugin scenarios where assemblies may not be present
EventTypeCache.ConfigureAssemblies([
    "YourApp.OptionalPlugin.Events"
], allowMissingAssemblies: true);

// Optional: Verify cache initialization
Console.WriteLine($"Event cache initialized with {EventTypeCache.CachedEventCount} event types");

Important Notes:

• Call ConfigureAssemblies() or SetAssemblies() during application startup before any message processing
• By default, the cache will fail fast if a specified assembly is not found (crash on startup with clear error message)
• Use allowMissingAssemblies: true only for optional/plugin assemblies
• Thread-safe and can be called multiple times if needed

Features

Initial Release (0.1.0.0)

• Gets config from appsettings.json
• Subscribes for the queue
• Dynamically deserializes the application events json messages to concrete types without having to recompile every time a new event is added
• Raises events for application event messages received with concrete types

Version 0.15.2.0

• Bug fixes and improvements
• Enhanced stability

Version 0.16.0.0

• Configurable assembly scanning for event types
• Support for multiple assemblies containing ApplicationEvents
• Fail-fast validation on startup (prevents runtime errors from missing assemblies)
• Optional graceful handling for plugin scenarios
• Thread-safe cache initialization
• Diagnostic properties for troubleshooting

Important Interfaces - For Users of This Library

Service Registration

namespace CD4.RabbitMq.Sdk;

public static class Cd4Messaging
{
  public static void AddCd4Rabbit(this IServiceCollection services, IConfiguration configuration)
  {
    services.Configure<MessageBusConfiguration>(configuration.GetRequiredSection("MessageBusConfiguration"));
    services.AddSingleton<IRabbitMqConnectionPool, RabbitMqConnectionPool>();
    services.AddSingleton<IRabbitMqMessaging, RabbitMqMessaging>();
    services.AddSingleton<IRabbitMqMessageParser, RabbitMqMessageParser>();
  }
}

Connection Pool Interface

namespace CD4.RabbitMq.Sdk;

public interface IRabbitMqConnectionPool
{
  void Dispose();
  Task<IConnection> GetConnectionAsync();
  void ReturnConnection(IConnection? connection);
}

Messaging Interface

namespace CD4.RabbitMq.Sdk;

/// <summary>
/// Enhanced interface for RabbitMQ messaging with automatic consumer recovery.
/// </summary>
public interface IRabbitMqMessaging : IDisposable
{
  /// <summary>
  /// Event fired when a message is received from any queue.
  /// </summary>
  event EventHandler<MessageReceivedEventArgs> MessageReceived;

  /// <summary>
  /// Sends a message to the specified queue.
  /// </summary>
  /// <param name="queueName">The name of the queue to send the message to.</param>
  /// <param name="message">The message to send.</param>
  /// <param name="senderConfig">Optional sender configuration. If null, uses configuration from settings.</param>
  /// <returns>A task representing the asynchronous operation.</returns>
  Task SendMessageAsync(string queueName, string message, RabbitMqSenderConfiguration? senderConfig = null);

  /// <summary>
  /// Starts listening to the specified queue for incoming messages with automatic recovery.
  /// </summary>
  /// <param name="queueName">The name of the queue to listen to.</param>
  /// <returns>A task representing the asynchronous operation.</returns>
  Task StartListeningAsync(string queueName);
}

EventTypeCache Interface

namespace CD4.RabbitMq.Sdk.Parsing;

public static class EventTypeCache
{
  /// <summary>
  /// Configure additional assemblies to scan for event types.
  /// Call this before any message processing occurs (e.g., in Startup/Program.cs).
  /// </summary>
  public static void ConfigureAssemblies(string[] assemblyNames, bool allowMissingAssemblies = false);

  /// <summary>
  /// Configure assemblies and replace the default list entirely.
  /// </summary>
  public static void SetAssemblies(string[] assemblyNames, bool allowMissingAssemblies = false);

  /// <summary>
  /// Get the count of cached event types.
  /// </summary>
  public static int CachedEventCount { get; }

  /// <summary>
  /// Check if the cache has been initialized.
  /// </summary>
  public static bool IsInitialized { get; }

  /// <summary>
  /// Get all cached event type names.
  /// </summary>
  public static IEnumerable<string> GetCachedEventTypeNames();
}

Usage Example

Complete Setup in Program.cs

using CD4.RabbitMq.Sdk;
using CD4.RabbitMq.Sdk.Parsing;

var builder = WebApplication.CreateBuilder(args);

// Step 1: Configure EventTypeCache (if you have events in multiple assemblies)
EventTypeCache.ConfigureAssemblies([
    "YourApp.Events",
    "YourApp.Orders.Events",
    "YourApp.Payments.Events"
]);

// Step 2: Register CD4 RabbitMQ services
builder.Services.AddCd4Rabbit(builder.Configuration);

// Step 3: Register your event handlers
builder.Services.AddHostedService<YourMessageHandlerService>();

var app = builder.Build();

// ... rest of your app configuration

app.Run();

Message Handler Example

public class YourMessageHandlerService : BackgroundService
{
  private readonly IRabbitMqMessaging _messaging;
  private readonly ILogger<YourMessageHandlerService> _logger;

  public YourMessageHandlerService(
      IRabbitMqMessaging messaging,
      ILogger<YourMessageHandlerService> logger)
  {
    _messaging = messaging;
    _logger = logger;
  }

  protected override async Task ExecuteAsync(CancellationToken stoppingToken)
  {
    // Subscribe to message received event
    _messaging.MessageReceived += OnMessageReceived;

    // Start listening to the queue
    await _messaging.StartListeningAsync("your.queue.name");

    // Keep the service running
    await Task.Delay(Timeout.Infinite, stoppingToken);
  }

  private void OnMessageReceived(object? sender, MessageReceivedEventArgs e)
  {
    // Deserialize to strongly-typed event
    var result = e.DeserializeMessage(_logger);

    if (result.IsSuccess && result.Event is YourSpecificEvent specificEvent)
    {
      // Handle your event
      _logger.LogInformation("Received event: {EventType}", specificEvent.EventType);
      // ... your business logic
    }
    else
    {
      _logger.LogError("Failed to deserialize message: {Error}", result.ErrorMessage);
    }
  }

  public override void Dispose()
  {
    _messaging.MessageReceived -= OnMessageReceived;
    base.Dispose();
  }
}

Troubleshooting

Assembly Not Found Error

If you get an error like:

InvalidOperationException: Assembly 'YourApp.Events' not found. Ensure the assembly is referenced and deployed with your application.

Solutions:

1. Ensure the assembly is referenced in your project
2. Verify the assembly name is correct (check in Solution Explorer)
3. Make sure the assembly is copied to the output directory
4. Check that the assembly contains classes that inherit from ApplicationEvents

Event Type Not Found in Cache

If you see warnings about event types not found:

Event type 'YourEventType' not found in cache.

Solutions:

1. Verify the assembly containing that event is configured in EventTypeCache.ConfigureAssemblies()
2. Check that the event class inherits from ApplicationEvents
3. Use EventTypeCache.GetCachedEventTypeNames() to see all cached types
4. Verify the EventType property value matches the class name

Debugging Cache Initialization

// Add after EventTypeCache.ConfigureAssemblies()
Console.WriteLine($"Cached event count: {EventTypeCache.CachedEventCount}");
Console.WriteLine($"Is initialized: {EventTypeCache.IsInitialized}");
Console.WriteLine($"Cached events: {string.Join(", ", EventTypeCache.GetCachedEventTypeNames())}");

Migration Guide

If you're migrating from an existing RabbitMQ implementation:

1. Add the NuGet package reference to CD4.RabbitMq.Sdk
2. Update appsettings.json with the MessageBusConfiguration section
3. Configure EventTypeCache if you have events in multiple assemblies
4. Replace service registration with services.AddCd4Rabbit(configuration)
5. Update message handlers to use IRabbitMqMessaging interface
6. Remove old RabbitMQ implementation code

Support

For issues, bugs, or feature requests, please contact the CD4 development team or create an issue in the repository.