Mq Consumer

PARR.BLL/Services/Implementations/MqService.cs

@@ -3,6 +3,7 @@ using PARR.BLL.Contracts.Interfaces;
 using PARR.BLL.Domain.Mq;
 using PARR.BLL.Services.Interfaces;
 using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
 using System.Text;
 
 namespace PARR.BLL.Services.Implementations
@@ -11,6 +12,8 @@ namespace PARR.BLL.Services.Implementations
     {
         private readonly ILogger<MqService> logger;
 
+        public event ReceivedHandler? Received;
+
         public MqService(ILogger<MqService> logger)
         {
             this.logger = logger;
@@ -83,5 +86,61 @@ namespace PARR.BLL.Services.Implementations
                 return new MqSendResult { IsSuccess = false, NotSendMessages = notSendMsg.ToArray() };
             }
         }
+
+
+        //TODO: rename ListenQueue
+        public void InitConsumer(IMqSettings mqSettings)
+        {
+            logger.LogInformation($"Устанавливаю соединение с RabbitMQ: {mqSettings.HostName}, {mqSettings.QueueName}");
+
+            var facory = new ConnectionFactory
+            {
+                HostName = mqSettings.HostName,
+                UserName = mqSettings.User,
+                Password = mqSettings.Password,
+                AutomaticRecoveryEnabled = true,
+                DispatchConsumersAsync = true
+            };
+
+            try
+            {
+                var connection = facory.CreateConnection();
+                var channel = connection.CreateModel();
+
+                var args = new Dictionary<string, object> { { "x-queue-mode", "lazy" } };
+
+                channel.QueueDeclare(
+                        queue: mqSettings.QueueName,
+                        durable: true,
+                        exclusive: false,
+                        autoDelete: false,
+                        arguments: args
+                    );
+
+                logger.LogInformation($"Соединение с RabbitMQ установлено: {mqSettings.HostName}, {mqSettings.QueueName}");
+
+
+                var consumer = new AsyncEventingBasicConsumer(channel);
+                consumer.Received += async (ch, ea) =>
+                {
+                    var content = Encoding.UTF8.GetString(ea.Body.ToArray());
+                    logger.LogDebug($"Получено сообщение: {content}");
+
+                    //https://metanit.com/sharp/tutorial/3.14.php
+                    Received?.Invoke(content);
+
+                    channel.BasicAck(ea.DeliveryTag, false);
+                    await Task.Yield();
+                };
+
+                channel.BasicConsume(mqSettings.QueueName, false, consumer);
+
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(ex, $"Ошибка при получении сообщений из очереди: {mqSettings.HostName}, {mqSettings.QueueName}");
+            }
+        }
+
     }
 }