Integration1/Client/RabbitMQClient.cs - lab4 - ArtSerg/IntegrationProject

Integration1/Client/RabbitMQClient.cs
129 строк5 KB
artS3rg
Добавлена интеграция RabbitMQ для асинхронного обмена сообщениями
14 дек 2025, 21:50
14 дек 2025, 21:50a02de12
using Integration1.Messaging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Json;

namespace Integration1.Client
{
    /// <summary>
    /// Клиент для отправки запросов в RabbitMQ и получения ответов
    /// </summary>
    public class RabbitMQClient : IDisposable
    {
        private readonly IConnection _connection;
        private readonly IModel _channel;
        private readonly string _requestQueue = "api.requests";
        private readonly string _responseQueue = "api.responses";
        private readonly Dictionary<string, TaskCompletionSource<ResponseMessage>> _pendingRequests = new();
        private readonly string _apiKey;

        public RabbitMQClient(IConfiguration config)
        {
            var factory = new ConnectionFactory
            {
                HostName = config["RabbitMQ:HostName"] ?? "localhost",
                Port = int.Parse(config["RabbitMQ:Port"] ?? "5672"),
                UserName = config["RabbitMQ:UserName"] ?? "guest",
                Password = config["RabbitMQ:Password"] ?? "guest"
            };

            _connection = factory.CreateConnection();
            _channel = _connection.CreateModel();

            // Создаем очередь для ответов (временная очередь для этого клиента)
            _channel.QueueDeclare(_responseQueue, durable: true, exclusive: false, autoDelete: false);

            // Настраиваем consumer для ответов
            var consumer = new EventingBasicConsumer(_channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                var response = JsonSerializer.Deserialize<ResponseMessage>(message);
                
                // Используем CorrelationId из properties, если он есть, иначе из сообщения
                var correlationId = ea.BasicProperties.CorrelationId ?? response?.CorrelationId;

                if (response != null && !string.IsNullOrEmpty(correlationId) && _pendingRequests.TryGetValue(correlationId, out var tcs))
                {
                    Console.WriteLine($"Received response with CorrelationId: {correlationId}, Status: {response.Status}");
                    tcs.SetResult(response);
                    _pendingRequests.Remove(correlationId);
                }
                else
                {
                    Console.WriteLine($"No pending request found for CorrelationId: {correlationId}");
                }
            };

            _channel.BasicConsume(_responseQueue, autoAck: true, consumer);

            _apiKey = config["Internal:ApiKey"] ?? "";
        }

        public async Task<ResponseMessage> SendRequestAsync(RequestMessage request, TimeSpan? timeout = null)
        {
            timeout ??= TimeSpan.FromSeconds(30);

            request.Auth = _apiKey;
            var correlationId = request.Id;
            
            // Проверяем, не ожидается ли уже ответ для этого correlation_id
            if (_pendingRequests.ContainsKey(correlationId))
            {
                // Если уже есть ожидающий запрос, ждем его завершения
                Console.WriteLine($"Request with CorrelationId {correlationId} already pending, waiting for existing request...");
                return await _pendingRequests[correlationId].Task;
            }
            
            var tcs = new TaskCompletionSource<ResponseMessage>();
            _pendingRequests[correlationId] = tcs;

            try
            {
                var props = _channel.CreateBasicProperties();
                props.MessageId = request.Id;
                props.CorrelationId = correlationId;
                props.ReplyTo = _responseQueue;

                var body = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(request));
                _channel.BasicPublish("", _requestQueue, props, body);
                Console.WriteLine($"Sent request with CorrelationId: {correlationId}, MessageId: {request.Id}");

                using var cts = new CancellationTokenSource(timeout.Value);
                cts.Token.Register(() => 
                {
                    if (_pendingRequests.TryGetValue(correlationId, out var pendingTcs) && pendingTcs == tcs)
                    {
                        _pendingRequests.Remove(correlationId);
                        tcs.TrySetCanceled();
                    }
                }, useSynchronizationContext: false);

                return await tcs.Task;
            }
            catch (OperationCanceledException)
            {
                if (_pendingRequests.TryGetValue(correlationId, out var pendingTcs) && pendingTcs == tcs)
                {
                    _pendingRequests.Remove(correlationId);
                }
                return new ResponseMessage
                {
                    CorrelationId = correlationId,
                    Status = "error",
                    Error = "Request timeout"
                };
            }
        }

        public void Dispose()
        {
            _channel?.Close();
            _connection?.Close();
            _channel?.Dispose();
            _connection?.Dispose();
        }
    }
}