Skip to main content

Messaging

Overview​

Package contains extensions for MassTransit initialization and configuration with RabbitMQ message broker.

  • Event-driven approach support
  • AMQP messaging infrastructure.

Install​

  • ASOL.Core.Messaging - contains extensions for standard MassTransit initialization
  • ASOL.Core.Messaging.RabbitMq - contains initialization configuration for MassTransit with RabbitMq
  • MassTransit.AspNetCore - hosting API project references this NuGet with compatible version
    • For ASOL.Core.Messaging.0.0.1.14347-dev use MassTransit.AspNetCore.6.3.2
    • For ASOL.Core.Messaging.0.0.1.19625-dev use MassTransit.AspNetCore.7.3.1
    • For ASOL.Core.Messaging.0.0.1.32021-dev usage of MassTransit.AspNetCore package reference is deprecated

Usage of third-party libraries (NuGets)​

Changelog​

  • 0.0.1.47855-dev

    • BREAKING CHANGE: Dependency on ASOL.Core.Identity.Abstractions of the version 0.0.1.47819-dev

  • 0.0.1.45790-dev

    • Support for Application Insights has been removed; OpenTelemetry is now used instead. For proper functioning, the minimum required version of ASOL.Core.Telemetry.Client is 0.0.1.45792-dev.

  • 0.0.1.44942-dev

    • BREAKING CHANGE: Support for older .NET frameworks has been removed; only .NET 8 is now supported.

  • 0.0.1.32021-dev

    • Breaking changes
      • Change RabbitMQ registration from config.AddBus(CreateBus) to config.UsingRabbitMq(UseRabbitMq)
      • Change method signature from IBusControl CreateBus(IBusRegistrationContext regContext) to void UseRabbitMq(IBusRegistrationContext regContext, IRabbitMqBusFactoryConfigurator cfg) and remove return Bus.Factory.CreateUsingRabbitMq method call and leave only the body of this factory method
      • Remove package reference MassTransit.AspNetCore (7.3.1) from your project
      • Remove MassTransitHostedService services.AddMassTransitHostedService(); registration
      • All changes for MassTransit 8 upgrade: MassTransit upgrade
      • Configuration of a (de)serialization is setup only by the IBusFactoryConfigurator extension method ConfigureJsonSerializer of the IBusFactoryConfigurator interface. Therefore, the extension method ConfigureJsonDeserializer was dropped.
      • Required min. version of others ASOL NuGets:
      • ASOL.IdentityManager.AspNetCore.Extensions 0.0.1.32191-dev
      • ASOL.Scheduler.AspNetCore 0.0.1.32072-dev
      • ASOL.Core.Multitenancy.Messaging 0.0.1.32021-dev

Dependency Injection​

Mandatory dependency injection for messaging:

/// <summary>
/// Configuration path to set up the message broker (RabbitMQ).
/// </summary>
public const string RabbitMqConfigurationSectionName = "Messaging:RabbitMq";
/// <summary>
/// Add messaging and message broker connection hosted service.
/// </summary>
public static IServiceCollection AddMessagingAndHostedService(this IServiceCollection services, IConfiguration configuration)
{
    services.Configure<RabbitMqConnectionOptions>(configuration.GetSection(RabbitMqConfigurationSectionName));
    services.AddMultitenancyMessaging();
    services.AddMessaging(config =>
    {
        config.AddConsumers(typeof(Startup).Assembly); //  Add all consumers in the API assembly
        config.AddConsumers(typeof(EventPublisherBase).Assembly); //  Add all consumers in the infrastructure assembly
        config.UsingRabbitMq(UseRabbitMq); 
    });
    return services;
}

.NET 8.0 or greater​

/// <summary>
/// Create RabbitMQ connection and configuration for messaging.
/// </summary>
private static void UseRabbitMq(IBusRegistrationContext regContext, IRabbitMqBusFactoryConfigurator cfg)
{
    var configuration = regContext.GetRequiredService<IConfiguration>();
    var serviceName = configuration?["ServiceName"] ?? throw new ArgumentNullException(nameof(regContext), "ServiceName is not defined in AppSettings.");

    cfg.Host(regContext.GetRequiredService<IConfiguration>(), RabbitMqConfigurationSectionName);
    
    cfg.ConfigureJsonSerializer(jsonConfiguration =>
    {
        new JsonSerializationConfiguration().Configure(jsonConfiguration);
        return jsonConfiguration;
    });
    
    cfg.ReceiveEndpoint(serviceName, ep =>
    {
        ep.UseServiceScope(regContext);
        ep.UseIdentity();
        ep.UseMultitenancy();
        
        // Here you can register consumers for events
        // Example:
        ep.Consumer<XEntityNameChangedEventConsumer>(regContext);
    });
}

RabbitMQ - Quorum Queues​

To ensure better availability, it is recommended to use quorum queues.

If you want to migrate from classic queues to quorum queues, use the EnsureEligibilityForQuorumQueue method and pass the name of the queue you want to migrate as a parameter. The method EnsureEligibilityForQuorumQueue(queueName) must be called for each queue that you want to set to a quorum type.

Quorum queue implementation steps​

  1. AddRabbitMqSupportedFeatures(): Register RabbitMQ supported features inside the AddMessaging method.
  2. EnsureEligibilityForQuorumQueue(queueName): Ensure that the queue is a quorum type. If not, it will be migrated.
  3. UseQuorumQueue(serviceProvider): MassTransit configuration for quorum queue.

Supported migration scenarios​

The EnsureEligibilityForQuorumQueue(queueName) method supports the following migration scenarios from classic to quorum queue:

  • The queue is a classic type and has no messages: The queue will be deleted, and a new quorum queue will be created.

  • The queue is a classic type and contains messages: The messages will be backed up, the queue will be recreated as a quorum type, and messages will be restored to the new quorum queue.

  • The queue is a quorum type: No action will be performed.

  • it is needed to call EnsureEligibilityForQuorumQueue(queueName) method for each queue / endpoint that you want to migrate to a quorum type. If the method is not called, the queue will not be migrated, and it will remain a classic type causing application to shutdown on error.

Quorum queues - Examples​

services.AddMessaging(config =>
{
    // ...
    config.AddRabbitMqSupportedFeatures(); // <-- namespace: ASOL.Core.Messaging.RabbitMq.Extensions
    config.UsingRabbitMq(UseRabbitMq);
});
private static void UseRabbitMq(IBusRegistrationContext regContext, IRabbitMqBusFactoryConfigurator cfg)
{
  // ...
  regContext.EnsureEligibilityForQuorumQueue(queueName); <-- namespace: ASOL.Core.Messaging.RabbitMq.Extensions
  cfg.ReceiveEndpoint(queueName, ep =>
  {
    // ..
    ep.UseQuorumQueue(regContext); // <-- namespace: ASOL.Core.Messaging.RabbitMq.Extensions
    // ..
  });
  // ...
}

Development against hosted RabbitMQ​

todo local queues

If you are using a hosted RabbitMQ instance without RabbitMQ API access, you can use the following configuration:

private static void UseRabbitMq(IBusRegistrationContext regContext, IRabbitMqBusFactoryConfigurator cfg)
{
  // ... 
  var env = regContext.GetRequiredService<IHostEnvironment>();
  if (!env.IsDevelopment());
  {
      regContext.EnsureEligibilityForQuorumQueue(queueName); <-- namespace: ASOL.Core.Messaging.RabbitMq.Extensions
  } 
  cfg.ReceiveEndpoint(queueName, ep =>
  {
    // ..
    if (!env.IsDevelopment());
    {
      ep.UseQuorumQueue(regContext); // <-- namespace: ASOL.Core.Messaging.RabbitMq.Extensions
    }
    // ..
  });
  // ...
}

If the destination queue has not yet been migrated to the quorum type, you first need to deploy changes with quorum queues configuration to the environment used for local development.