AppStd/DigitalData.MessagingService
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(infrastructure): inject IServiceScopeFactory into consumer pool for scoped IMAP access and register new services (OAuth2, POP3)

Browse Source
This commit is contained in:
tekh
2026-08-17 10:15:52 +02:00
parent 1d7f269fe1
commit 7782e86db3
3 changed files with 21 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs
View File

@@ -29,13 +29,17 @@ public static class DependencyInjection
IConfiguration configuration)
{
// --- External Services ---
// Email Service - SMTP outbound (Limilabs Mail.dll)
services.AddSingleton<IEmailService, LimilabsEmailService>();
// Email Service - SMTP outbound (Limilabs Mail.dll) - OAuth2 aware
services.AddSingleton<IOAuth2TokenService, MicrosoftOAuth2TokenService>();
services.AddSingleton<LimilabsEmailService>();
services.AddSingleton<IEmailService>(sp => sp.GetRequiredService<LimilabsEmailService>());
// Email Service - IMAP inbound (Limilabs Mail.dll)
// Fresh connection per call — stateless and thread-safe.
services.AddScoped<IImapEmailService, LimilabsImapEmailService>();
// Email Service - POP3 inbound (Limilabs Mail.dll)
services.AddScoped<IPop3EmailService, LimilabsPop3EmailService>();
// PDF Processing Service (using DevExpress.Pdf)
services.AddScoped<IPdfProcessingService, DevExpressPdfProcessingService>();

15
src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumer.cs
View File

@@ -2,6 +2,7 @@ using System.Text;
using System.Text.Json;
using DigitalData.MessagingService.Application.Common.Interfaces;
using DigitalData.MessagingService.RabbitMQ;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
@@ -31,7 +32,7 @@ public sealed class SendingEmailConsumer : IAsyncDisposable
/// </summary>
public Guid RuntimeId { get; } = Guid.NewGuid();
public SendingEmailConsumer(string queueName, IEmailService emailService, RabbitMqConnectionFactory cnnFactory, ILogger<SendingEmailConsumer>? logger = null)
public SendingEmailConsumer(string queueName, IEmailService emailService, IServiceScopeFactory scopeFactory, RabbitMqConnectionFactory cnnFactory, ILogger<SendingEmailConsumer>? logger = null)
{
_logger = logger;
_queueName = queueName;
@@ -41,8 +42,6 @@ public sealed class SendingEmailConsumer : IAsyncDisposable
{
var channel = await _lazyChannel.Value;
// prefetchCount=1 ensures this consumer processes one message at a time before acking.
// Parallelism comes from running multiple consumer instances, not from within a single channel.
await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false);
var consumer = new AsyncEventingBasicConsumer(channel);
@@ -57,8 +56,14 @@ public sealed class SendingEmailConsumer : IAsyncDisposable
if (oMailEvent is not null)
{
// Send email via SMTP (SMTP config is injected in IEmailService via IOptions)
await emailService.SendEmailAsync(oMailEvent.Mail, args.CancellationToken);
if (oMailEvent.UseImapAppend)
{
await using var scope = scopeFactory.CreateAsyncScope();
var imapService = scope.ServiceProvider.GetRequiredService<IImapEmailService>();
await imapService.SendAndAppendAsync(oMailEvent.Mail, oMailEvent.SentFolder, args.CancellationToken);
}
else
await emailService.SendEmailAsync(oMailEvent.Mail, args.CancellationToken);
// Acknowledge message after successful processing
await channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken);

6
src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumerPool.cs
View File

@@ -1,5 +1,6 @@
using DigitalData.MessagingService.Application.Common.Interfaces;
using DigitalData.MessagingService.RabbitMQ;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
@@ -8,8 +9,6 @@ namespace DigitalData.MessagingService.Infrastructure.Queue;
/// <summary>
/// Manages a pool of <see cref="SendingEmailConsumer"/> instances that compete for messages
/// on the same RabbitMQ queue (competing consumers pattern).
/// Each consumer owns a dedicated channel, so they process messages fully in parallel
/// without any shared locking or synchronization primitives.
/// </summary>
public sealed class SendingEmailConsumerPool : IAsyncDisposable
{
@@ -20,6 +19,7 @@ public sealed class SendingEmailConsumerPool : IAsyncDisposable
public SendingEmailConsumerPool(
IOptions<RabbitMqConfiguration> config,
IEmailService emailService,
IServiceScopeFactory scopeFactory,
RabbitMqConnectionFactory cnnFactory,
ILogger<SendingEmailConsumerPool>? logger = null,
ILogger<SendingEmailConsumer>? consumerLogger = null)
@@ -29,7 +29,7 @@ public sealed class SendingEmailConsumerPool : IAsyncDisposable
_consumers = [.. Enumerable
.Range(0, _concurrency)
.Select(_ => new SendingEmailConsumer(config.Value.QueueName, emailService, cnnFactory, consumerLogger))];
.Select(_ => new SendingEmailConsumer(config.Value.QueueName, emailService, scopeFactory, cnnFactory, consumerLogger))];
}
/// <summary>