From 169ef7d86b608858c705f85dcc333da29d2d0f9a Mon Sep 17 00:00:00 2001 From: TekH Date: Thu, 23 Jul 2026 15:42:03 +0200 Subject: [PATCH] Remove RabbitMQ messaging functionality The application no longer uses RabbitMQ for command publishing and consumption. This commit removes all RabbitMQ-related code, including: - Removed RabbitMQ service registrations in `DependencyInjection.cs`. - Deleted `RabbitMqCommandConsumer.cs`, which implemented a background service for consuming commands. - Deleted `RabbitMqCommandPublisher.cs`, which implemented a publisher for RabbitMQ-based commands. - Removed RabbitMQ-specific properties (`ExchangeName`, `QueueName`, `RoutingKey`) from `RabbitMqConfiguration.cs`. These changes reflect a shift in the application's messaging strategy or architecture. --- .../DependencyInjection.cs | 6 - .../Messaging/RabbitMqCommandConsumer.cs | 216 ------------------ .../Messaging/RabbitMqCommandPublisher.cs | 158 ------------- .../Messaging/RabbitMqConfiguration.cs | 15 -- 4 files changed, 395 deletions(-) delete mode 100644 src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs delete mode 100644 src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandPublisher.cs diff --git a/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs b/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs index 42230a3..e724498 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs +++ b/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs @@ -37,12 +37,6 @@ public static class DependencyInjection services.Configure( configuration.GetSection(RabbitMqConfiguration.SectionName)); - // --- RabbitMQ Command Publisher --- - services.AddSingleton(); - - // --- RabbitMQ Command Consumer (Background Service) --- - services.AddHostedService(); - // --- Data Protection (for encryption) --- services.AddDataProtection() .PersistKeysToFileSystem(new DirectoryInfo(@"C:\ProgramData\EmailService\Keys")) diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs deleted file mode 100644 index 96e3d13..0000000 --- a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs +++ /dev/null @@ -1,216 +0,0 @@ -using System.Text; -using System.Text.Json; -using DigitalData.EmailProfiler.Infrastructure.Messaging; -using MediatR; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; -using RabbitMQ.Client; -using RabbitMQ.Client.Events; - -namespace DigitalData.EmailProfiler.Infrastructure.Messaging; - -/// -/// Background service that consumes commands from RabbitMQ and executes them via MediatR -/// -public class RabbitMqCommandConsumer( - IOptions config, - IServiceProvider ServiceProvider, - ILogger? Logger) : BackgroundService -{ - private readonly RabbitMqConfiguration Config = config.Value ?? throw new ArgumentNullException(nameof(config)); - private IConnection? Connection; - private IChannel? _channel; - - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - Logger?.LogInformation("RabbitMQ Command Consumer starting..."); - - try - { - // Create connection factory - var factory = new ConnectionFactory - { - HostName = Config.HostName, - Port = Config.Port, - UserName = Config.UserName, - Password = Config.Password, - VirtualHost = Config.VirtualHost, - AutomaticRecoveryEnabled = Config.AutomaticRecoveryEnabled, - NetworkRecoveryInterval = TimeSpan.FromSeconds(Config.NetworkRecoveryIntervalSeconds) - }; - - // Create connection and channel - Connection = await factory.CreateConnectionAsync(stoppingToken); - _channel = await Connection.CreateChannelAsync(cancellationToken: stoppingToken); - - // Declare exchange (Direct type for routing) - await _channel.ExchangeDeclareAsync( - exchange: Config.ExchangeName, - type: ExchangeType.Direct, - durable: true, - autoDelete: false, - cancellationToken: stoppingToken); - - // Declare queue (durable for persistence) - await _channel.QueueDeclareAsync( - queue: Config.QueueName, - durable: true, - exclusive: false, - autoDelete: false, - arguments: null, - cancellationToken: stoppingToken); - - // Bind queue to exchange with routing key - await _channel.QueueBindAsync( - queue: Config.QueueName, - exchange: Config.ExchangeName, - routingKey: Config.RoutingKey, - cancellationToken: stoppingToken); - - Logger?.LogInformation( - "RabbitMQ initialized: Exchange={Exchange}, Queue={Queue}, RoutingKey={RoutingKey}", - Config.ExchangeName, Config.QueueName, Config.RoutingKey); - - // Set prefetch count (process one message at a time) - await _channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false, cancellationToken: stoppingToken); - - // Create async consumer - var consumer = new AsyncEventingBasicConsumer(_channel); - consumer.ReceivedAsync += async (sender, ea) => - { - try - { - await ProcessMessageAsync(ea, stoppingToken); - await _channel.BasicAckAsync(ea.DeliveryTag, multiple: false, cancellationToken: stoppingToken); - } - catch (Exception ex) - { - Logger?.LogError(ex, "Error processing message {MessageId}", ea.BasicProperties?.MessageId); - - // Reject and requeue on error (be careful with infinite loops!) - await _channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: true, cancellationToken: stoppingToken); - } - }; - - // Start consuming - await _channel.BasicConsumeAsync( - queue: Config.QueueName, - autoAck: false, - consumer: consumer, - cancellationToken: stoppingToken); - - Logger?.LogInformation( - "RabbitMQ Command Consumer started. Listening on queue: {QueueName}", - Config.QueueName); - - // Keep running until cancellation requested - // Consumer will process messages in the background via event handlers - var tcs = new TaskCompletionSource(); - stoppingToken.Register(() => tcs.SetResult()); - await tcs.Task; - } - catch (OperationCanceledException) - { - Logger?.LogInformation("RabbitMQ Command Consumer is stopping due to cancellation"); - } - catch (Exception ex) - { - Logger?.LogError(ex, "Fatal error in RabbitMQ Command Consumer"); - throw; - } - } - - private async Task ProcessMessageAsync(BasicDeliverEventArgs ea, CancellationToken cancellationToken) - { - var messageId = ea.BasicProperties?.MessageId ?? "unknown"; - var commandType = ea.BasicProperties?.Type ?? "unknown"; - - Logger?.LogInformation( - "Processing command {CommandType} with MessageId {MessageId}", - commandType, messageId); - - // Deserialize message envelope - var json = Encoding.UTF8.GetString(ea.Body.ToArray()); - var envelope = JsonSerializer.Deserialize(json); - - if (envelope == null) - { - Logger?.LogWarning("Failed to deserialize command envelope for MessageId {MessageId}", messageId); - return; - } - - // Get command type from assembly - var type = Type.GetType(envelope.CommandType); - if (type == null) - { - Logger?.LogWarning( - "Command type {CommandType} not found in assembly for MessageId {MessageId}", - envelope.CommandType, messageId); - return; - } - - // Deserialize command payload - var command = JsonSerializer.Deserialize(envelope.Payload, type); - if (command == null) - { - Logger?.LogWarning( - "Failed to deserialize command payload for MessageId {MessageId}", - messageId); - return; - } - - // Create scope and execute command via MediatR - using var scope = ServiceProvider.CreateScope(); - var mediator = scope.ServiceProvider.GetRequiredService(); - - try - { - // Send command to MediatR (fire-and-forget) - await mediator.Send(command, cancellationToken); - - Logger?.LogInformation( - "Successfully processed command {CommandType} with MessageId {MessageId}", - commandType, messageId); - } - catch (Exception ex) - { - Logger?.LogError( - ex, - "Error executing command {CommandType} with MessageId {MessageId}", - commandType, messageId); - throw; - } - } - - public override async Task StopAsync(CancellationToken cancellationToken) - { - Logger?.LogInformation("RabbitMQ Command Consumer stopping..."); - - if (_channel != null) - { - await _channel.CloseAsync(cancellationToken); - _channel.Dispose(); - } - - if (Connection != null) - { - await Connection.CloseAsync(cancellationToken); - Connection.Dispose(); - } - - await base.StopAsync(cancellationToken); - } - - /// - /// Message envelope for command deserialization - /// - private class CommandEnvelope - { - public string CommandType { get; set; } = string.Empty; - public string Payload { get; set; } = string.Empty; - public DateTime PublishedAt { get; set; } - public string CorrelationId { get; set; } = string.Empty; - } -} diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandPublisher.cs b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandPublisher.cs deleted file mode 100644 index 11e67cc..0000000 --- a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandPublisher.cs +++ /dev/null @@ -1,158 +0,0 @@ -using System.Text; -using System.Text.Json; -using DigitalData.EmailProfiler.Application.Common.Interfaces; -using MediatR; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; -using RabbitMQ.Client; - -namespace DigitalData.EmailProfiler.Infrastructure.Messaging; - -/// -/// RabbitMQ implementation of ICommandPublisher for asynchronous command processing -/// -public class RabbitMqCommandPublisher : ICommandPublisher, IDisposable -{ - private readonly ILogger Logger; - - private readonly RabbitMqConfiguration Config; - - private readonly IConnection Connection; - - private readonly IChannel Channel; - - private bool Disposed { get; set; } - - public RabbitMqCommandPublisher( - IOptions config, - ILogger logger) - { - Config = config.Value ?? throw new ArgumentNullException(nameof(config)); - Logger = logger ?? throw new ArgumentNullException(nameof(logger)); - - try - { - // Create connection factory - var factory = new ConnectionFactory - { - HostName = Config.HostName, - Port = Config.Port, - UserName = Config.UserName, - Password = Config.Password, - VirtualHost = Config.VirtualHost, - AutomaticRecoveryEnabled = Config.AutomaticRecoveryEnabled, - NetworkRecoveryInterval = TimeSpan.FromSeconds(Config.NetworkRecoveryIntervalSeconds) - }; - - // Create connection and channel - Connection = factory.CreateConnectionAsync().GetAwaiter().GetResult(); - Channel = Connection.CreateChannelAsync().GetAwaiter().GetResult(); - - // Declare exchange (fanout for broadcasting commands) - Channel.ExchangeDeclareAsync( - exchange: Config.ExchangeName, - type: ExchangeType.Direct, - durable: true, - autoDelete: false).GetAwaiter().GetResult(); - - // Declare queue - Channel.QueueDeclareAsync( - queue: Config.QueueName, - durable: true, - exclusive: false, - autoDelete: false, - arguments: null).GetAwaiter().GetResult(); - - // Bind queue to exchange - Channel.QueueBindAsync( - queue: Config.QueueName, - exchange: Config.ExchangeName, - routingKey: Config.RoutingKey).GetAwaiter().GetResult(); - - Logger.LogInformation( - "RabbitMQ connection established: {HostName}:{Port}, Exchange: {Exchange}, Queue: {Queue}", - Config.HostName, Config.Port, Config.ExchangeName, Config.QueueName); - } - catch (Exception ex) - { - Logger.LogError(ex, "Failed to establish RabbitMQ connection"); - throw; - } - } - - /// - /// Publishes a command to RabbitMQ for asynchronous processing - /// - public async Task PublishAsync(TCommand command, CancellationToken cancellationToken = default) - where TCommand : IBaseRequest - { - ObjectDisposedException.ThrowIf(Disposed, typeof(RabbitMqCommandPublisher)); - - try - { - // Create message envelope with metadata - var envelope = new CommandEnvelope - { - CommandType = typeof(TCommand).AssemblyQualifiedName!, - Payload = JsonSerializer.Serialize(command), - PublishedAt = DateTime.Now, - CorrelationId = Guid.NewGuid().ToString() - }; - - // Serialize to JSON - var json = JsonSerializer.Serialize(envelope); - var body = Encoding.UTF8.GetBytes(json); - - // Set message properties - var properties = new BasicProperties - { - Persistent = true, - ContentType = "application/json", - MessageId = envelope.CorrelationId, - Timestamp = new AmqpTimestamp(DateTimeOffset.Now.ToUnixTimeSeconds()), - Type = typeof(TCommand).Name - }; - - // Publish to exchange - await Channel.BasicPublishAsync( - exchange: Config.ExchangeName, - routingKey: Config.RoutingKey, - mandatory: false, - basicProperties: properties, - body: body, - cancellationToken: cancellationToken); - - Logger.LogInformation( - "Published command {CommandType} with CorrelationId {CorrelationId} to RabbitMQ", - typeof(TCommand).Name, envelope.CorrelationId); - } - catch (Exception ex) - { - Logger.LogError(ex, "Failed to publish command {CommandType} to RabbitMQ", typeof(TCommand).Name); - throw; - } - } - - public void Dispose() - { - if (Disposed) - return; - - Channel?.Dispose(); - Connection?.Dispose(); - Disposed = true; - - Logger.LogInformation("RabbitMQ connection disposed"); - } - - /// - /// Message envelope for command serialization - /// - private class CommandEnvelope - { - public string CommandType { get; set; } = string.Empty; - public string Payload { get; set; } = string.Empty; - public DateTime PublishedAt { get; set; } - public string CorrelationId { get; set; } = string.Empty; - } -} diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqConfiguration.cs b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqConfiguration.cs index 974a7b5..52ac45e 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqConfiguration.cs +++ b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqConfiguration.cs @@ -35,21 +35,6 @@ public class RabbitMqConfiguration /// public string VirtualHost { get; set; } = "/"; - /// - /// Exchange name for commands - /// - public string ExchangeName { get; set; } = "emailprofiler.commands"; - - /// - /// Queue name for commands - /// - public string QueueName { get; set; } = "emailprofiler.command.queue"; - - /// - /// Routing key for commands - /// - public string RoutingKey { get; set; } = "command"; - /// /// Enable automatic recovery on connection failure ///