Replace AsyncInitWorker with ConsumerInitWorker
Replaced the `AsyncInitWorker` background service with the new `ConsumerInitWorker`, which introduces enhanced error handling, logging, and configurability for RabbitMQ consumer initialization. Updated `MessagingServiceDbContext` to remove the `UseCollation` configuration for the `Username` property in the `EmailAccount` entity, ensuring compatibility with the InMemory database. Added the `ContinueOnInitFailure` property to `RabbitMqConfiguration` to allow the application to continue running if RabbitMQ consumer initialization fails. Cleaned up unused `using` directives in `DependencyInjection.cs` and removed the obsolete `AsyncInitWorker` class.
This commit is contained in:
@@ -1,4 +1,3 @@
|
||||
using AutoMapper;
|
||||
using DigitalData.MessagingService.Application.Common.Interfaces;
|
||||
using DigitalData.MessagingService.Application.Common.Interfaces.Repositories;
|
||||
using DigitalData.MessagingService.Infrastructure.Mappings;
|
||||
@@ -9,7 +8,6 @@ using DigitalData.MessagingService.Infrastructure.Services;
|
||||
using DigitalData.MessagingService.Infrastructure.Services.Background;
|
||||
using DigitalData.MessagingService.Publisher;
|
||||
using DigitalData.MessagingService.RabbitMQ;
|
||||
using Microsoft.AspNetCore.DataProtection;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
@@ -61,7 +59,7 @@ public static class DependencyInjection
|
||||
services.AddRabbitMqConnectionFactory(configuration);
|
||||
|
||||
// Register Background Workers
|
||||
services.AddHostedService<AsyncInitWorker>();
|
||||
services.AddHostedService<ConsumerInitWorker>();
|
||||
services.AddHostedService<EmailSyncWorker>();
|
||||
|
||||
services.AddSingleton<IEmailSyncService>(p =>
|
||||
|
||||
@@ -18,8 +18,9 @@ public class MessagingServiceDbContext(DbContextOptions<MessagingServiceDbContex
|
||||
|
||||
modelBuilder.Entity<EmailAccount>(entity =>
|
||||
{
|
||||
entity.Property(e => e.Username)
|
||||
.UseCollation("SQL_Latin1_General_CP1_CI_AS");
|
||||
// UseCollation is only supported by relational databases (e.g. SQL Server).
|
||||
// Skipped here to remain compatible with the current InMemory database configuration.
|
||||
// entity.Property(e => e.Username).UseCollation("SQL_Latin1_General_CP1_CI_AS");
|
||||
});
|
||||
|
||||
modelBuilder.Entity<ReceivedEmail>(entity =>
|
||||
|
||||
@@ -1,19 +0,0 @@
|
||||
using DigitalData.MessagingService.Infrastructure.Queue;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
|
||||
namespace DigitalData.MessagingService.Infrastructure.Services.Background;
|
||||
|
||||
/// <summary>
|
||||
/// A hosted background service responsible for initializing the competing email consumer pool.
|
||||
/// Leverages a push-based, event-driven RabbitMQ consumer to eliminate polling overhead.
|
||||
/// Email account configuration is resolved exclusively from application settings; no database access is performed.
|
||||
/// </summary>
|
||||
public class AsyncInitWorker(SendingEmailConsumerPool ConsumerPool) : BackgroundService
|
||||
{
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await ConsumerPool.InitAsync();
|
||||
|
||||
await Task.Delay(Timeout.Infinite, stoppingToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
using DigitalData.MessagingService.Infrastructure.Queue;
|
||||
using DigitalData.MessagingService.RabbitMQ;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace DigitalData.MessagingService.Infrastructure.Services.Background;
|
||||
|
||||
/// <summary>
|
||||
/// A hosted background service responsible for initializing the competing email consumer pool.
|
||||
/// Leverages a push-based, event-driven RabbitMQ consumer to eliminate polling overhead.
|
||||
/// Email account configuration is resolved exclusively from application settings; no database access is performed.
|
||||
/// </summary>
|
||||
public class ConsumerInitWorker(SendingEmailConsumerPool ConsumerPool, ILogger<ConsumerInitWorker> Logger, IOptions<RabbitMqConfiguration> RabbitMqOptions) : BackgroundService
|
||||
{
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
await ConsumerPool.InitAsync();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Log and continue — a RabbitMQ connection failure must not bring down the entire host.
|
||||
// The API remains operational; only outgoing email delivery via the consumer pool is affected.
|
||||
Logger.LogError(ex, "Failed to initialize SendingEmailConsumerPool. Outgoing email consumers are not running.");
|
||||
|
||||
if (!RabbitMqOptions.Value.ContinueOnInitFailure)
|
||||
throw;
|
||||
}
|
||||
|
||||
await Task.Delay(Timeout.Infinite, stoppingToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
|
||||
}
|
||||
}
|
||||
@@ -81,5 +81,7 @@ namespace DigitalData.MessagingService.RabbitMQ
|
||||
/// Maps directly to RabbitMQ prefetchCount. Recommended: 3–5.
|
||||
/// </summary>
|
||||
public ushort ConsumerConcurrency { get; set; } = 5;
|
||||
|
||||
public bool ContinueOnInitFailure { get; set; } = false;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user