xEventService/Providers/XRabbitMQConsumerServiceBase.cs

using System;
using System.Text;
using RabbitMQ.Client;
using System.Threading;
using xCommons.Extensions;
using xEventService.Models;
using xExceptions.Constants;
using System.Threading.Tasks;
using RabbitMQ.Client.Events;
using xEventService.Constants;
using xEventService.Extensions;
using xEventService.Configuration;
using Microsoft.Extensions.Logging;

namespace xEventService.Providers
{
    /// <summary>
    /// an Abstract Kafka Consumer Service ...
    /// </summary>
    public abstract class XRabbitMQConsumerServiceBase : XBaseEventConsumerBackgroundService
    {
        //
        protected IModel channel;
        protected IConnection connection;
        protected AsyncEventingBasicConsumer consumer;

        protected XRabbitMQConsumerServiceBase(
            string[] allowedActions,
            string[] allowedSenders,
            IServiceProvider serviceProvider,
            XEventServiceConfiguration configuration,
            ILogger<XRabbitMQConsumerServiceBase> logger,
            string topic = XEventServiceConstants.XDefaultTopic
        ) : base(
            topic: topic,
            logger: logger,
            configuration: configuration,
            allowedActions: allowedActions,
            allowedSenders: allowedSenders,
            serviceProvider: serviceProvider
        )
        {
            //
            // Validating Configuration ...
            if (!configuration.RabbitMQ.IsValid() ||
                configuration.Broker != XEventBroker.XRabbitMQ
            )
            {
                XException.InvalidConfiguration.Throw();
            }

            //
            SetupConnection();
            DeclareTopology(topic);
            consumer = new AsyncEventingBasicConsumer(channel);
        }

        /// <summary>
        /// Default Action Execution for Background Services ...
        /// </summary>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        protected override async Task ExecuteAsync(
            CancellationToken cancellationToken = default
        )
        {
            //
            // For Fix Blocking Synchronous ...
            await Task.Yield();
            using var cts = new CancellationTokenSource();
            logger.LogInformation($"RabbitMQ Cosume Topic: {topic} ...");

            //
            // Subscribe to Topic ...
            consumer.Received += async (model, ea) =>
            {
                //
                try
                {
                    //
                    var body = ea.Body.ToArray();
                    var json = Encoding.UTF8.GetString(body);
                    var message = json.FromJSON<XEventMessage>();
                    if (!message.IsNullOrDefault() &&
                        message.IsValid(
                        allowedSenders: allowedSenders,
                        allowedActions: allowedActions)
                    )
                    {
                        //
                        await DoWorkAsync(
                            payload: message,
                            cancellationToken: cancellationToken
                        );
                    }

                    //
                    if (configuration.RabbitMQ.ManualAck)
                    {
                        //
                        channel.BasicAck(
                            multiple: false,
                            deliveryTag: ea.DeliveryTag
                        );
                    }
                }
                catch (Exception ex)
                {
                    //
                    logger.LogError($"RabbitMQ Cosume Error: {ex.Message} ...");
                    if (configuration.RabbitMQ.ManualAck)
                    {
                        //
                        channel.BasicNack(
                            requeue: true,
                            multiple: false,
                            deliveryTag: ea.DeliveryTag
                        );
                    }
                }
            };

            //
            consumer.Shutdown += (model, ea) =>
            {
                return Task.CompletedTask;
            };

            //
            channel.BasicQos(
                prefetchSize: 0,
                global: false,
                prefetchCount: configuration.RabbitMQ.PrefetchCount
            );

            //
            channel.BasicConsume(
                queue: GetQueueName(),
                consumer: consumer,
                autoAck: !configuration.RabbitMQ.ManualAck
            );

            //
            while (!cancellationToken.IsCancellationRequested)
            {
                await Task.Delay(1000, cancellationToken);
            }
        }

        /// <summary>
        /// Overriding Stop Action ...
        /// </summary>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public override async Task StopAsync(CancellationToken cancellationToken)
        {
            //
            if (channel != null)
            {
                //
                channel.Close();
                channel.Dispose();
            }

            //
            if (connection != null)
            {
                //
                connection.Close();
                connection.Dispose();
            }

            //
            await base.StopAsync(cancellationToken);
        }

        /// <summary>
        /// Dispose Implementation ...
        /// </summary>
        public override void Dispose()
        {
            //
            if (channel != null)
            {
                channel.Dispose();
            }

            //
            if (connection != null)
            {
                connection.Dispose();
            }

            //
            GC.SuppressFinalize(this);
            base.Dispose();
        }

        //
        #region Private ...
        private string GetQueueName()
        {
            //
            var result = string.Empty;
            if (!configuration.RabbitMQ.Queue.IsNullOrEmpty())
            {
                result = configuration.RabbitMQ.Queue;
            }

            //
            if (result.IsNullOrEmpty())
            {
                //
                var exchangeName = topic ?? configuration.RabbitMQ.Exchange;
                var digits = Guid.NewGuid().ToString().GetDigits();
                result = $"{exchangeName}[{digits}]";
            }

            //
            return result;
        }

        private void SetupConnection()
        {
            var factory = new ConnectionFactory
            {
                DispatchConsumersAsync = true,
                Port = configuration.GetRabbitMQPort(),
                HostName = configuration.GetRabbitMQHost(),
                UserName = configuration.RabbitMQ.Username,
                Password = configuration.RabbitMQ.Password,
                NetworkRecoveryInterval = TimeSpan.FromSeconds(
                    configuration.RabbitMQ.NetworkRecoveryIntervalSeconds
                ),
                VirtualHost = configuration.RabbitMQ.VirtualHost,
                AutomaticRecoveryEnabled = configuration.RabbitMQ.AutomaticRecoveryEnabled,
            };

            //
            connection = factory.CreateConnection();
            channel = connection.CreateModel();
        }

        private void DeclareTopology(string topic)
        {
            //
            channel.ExchangeDeclare(
                arguments: null,
                exchange: topic ?? configuration.RabbitMQ.Exchange,
                durable: configuration.RabbitMQ.ExchangeDurable,
                autoDelete: configuration.RabbitMQ.ExchangeAutoDelete,
                type: configuration.RabbitMQ.ExchangeType.GetStringValue()
            );

            //
            var queueName = GetQueueName();

            //
            channel.QueueDeclare(
                arguments: null,
                queue: queueName,
                durable: configuration.RabbitMQ.QueueDurable,
                exclusive: configuration.RabbitMQ.QueueExclusive,
                autoDelete: configuration.RabbitMQ.QueueAutoDelete
            );

            //
            channel.QueueBind(
                arguments: null,
                queue: queueName,
                exchange: topic ?? configuration.RabbitMQ.Exchange,
                routingKey: configuration.RabbitMQ.RoutingKey
            );
        }
        #endregion
    }
}