Lab4/Client/TaskerMQClient.cs - master - ivanstrike/tasker

Lab4/Client/TaskerMQClient.cs
511 строк16 KB
ivanstrike
add DLQ simulation
23 дек 2025, 14:22
23 дек 2025, 14:22e6b7124
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Collections.Concurrent;
using System.Text;
using TaskerMQ.Client.Models;

namespace TaskerMQ.Client;

/// <summary>
/// Клиент для асинхронного взаимодействия с TaskerMQ через RabbitMQ
/// </summary>
public class TaskerMQClient : IDisposable
{
    #region Fields

    private readonly string _host;
    private readonly int _port;
    private readonly string _username;
    private readonly string _password;
    private readonly string _requestQueue;
    private readonly string _responseQueue;
    private readonly string _apiKey;
    private readonly int _defaultTimeout;

    private IConnection? _connection;
    private IChannel? _channel;
    private readonly ConcurrentDictionary<string, TaskCompletionSource<ResponseMessage>> _pendingRequests = new();
    private readonly SemaphoreSlim _connectionLock = new(1, 1);
    private bool _isDisposed;

    #endregion

    #region Constructor

    /// <summary>
    /// Создаёт новый экземпляр клиента TaskerMQ
    /// </summary>
    public TaskerMQClient(
        string host = "localhost",
        int port = 5672,
        string username = "guest",
        string password = "guest",
        string requestQueue = "api.requests",
        string responseQueue = "api.responses",
        string apiKey = "dev-api-key-12345",
        int defaultTimeoutSeconds = 30)
    {
        _host = host ?? throw new ArgumentNullException(nameof(host));
        _port = port;
        _username = username ?? throw new ArgumentNullException(nameof(username));
        _password = password ?? throw new ArgumentNullException(nameof(password));
        _requestQueue = requestQueue ?? throw new ArgumentNullException(nameof(requestQueue));
        _responseQueue = responseQueue ?? throw new ArgumentNullException(nameof(responseQueue));
        _apiKey = apiKey ?? throw new ArgumentNullException(nameof(apiKey));
        _defaultTimeout = defaultTimeoutSeconds;
    }

    #endregion

    #region Connection Management

    /// <summary>
    /// Подключается к RabbitMQ и начинает слушать ответы
    /// </summary>
    public async Task ConnectAsync()
    {
        if (_isDisposed)
            throw new ObjectDisposedException(nameof(TaskerMQClient));

        await _connectionLock.WaitAsync();
        try
        {
            // Проверяем существующее соединение
            if (_connection != null && _connection.IsOpen && _channel != null && _channel.IsOpen)
            {
                return;
            }

            // Закрываем старое соединение если есть
            await DisconnectInternalAsync();

            // Создаём новое соединение
            var factory = new ConnectionFactory
            {
                HostName = _host,
                Port = _port,
                UserName = _username,
                Password = _password,
                AutomaticRecoveryEnabled = true,
                NetworkRecoveryInterval = TimeSpan.FromSeconds(10)
            };

            _connection = await factory.CreateConnectionAsync();
            _channel = await _connection.CreateChannelAsync();

            // Проверяем существование очередей
            await _channel.QueueDeclarePassiveAsync(_requestQueue);
            await _channel.QueueDeclarePassiveAsync(_responseQueue);

            // Начинаем слушать очередь ответов
            var consumer = new AsyncEventingBasicConsumer(_channel);
            consumer.ReceivedAsync += OnResponseReceivedAsync;

            await _channel.BasicConsumeAsync(
                queue: _responseQueue,
                autoAck: true,
                consumer: consumer);

            Console.WriteLine($"✓ Connected to RabbitMQ at {_host}:{_port}");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"✗ Failed to connect to RabbitMQ: {ex.Message}");
            throw;
        }
        finally
        {
            _connectionLock.Release();
        }
    }

    /// <summary>
    /// Отключается от RabbitMQ
    /// </summary>
    public async Task DisconnectAsync()
    {
        await _connectionLock.WaitAsync();
        try
        {
            await DisconnectInternalAsync();
        }
        finally
        {
            _connectionLock.Release();
        }
    }

    private async Task DisconnectInternalAsync()
    {
        if (_channel != null)
        {
            try
            {
                await _channel.CloseAsync();
                _channel.Dispose();
            }
            catch { }
            _channel = null;
        }

        if (_connection != null)
        {
            try
            {
                await _connection.CloseAsync();
                _connection.Dispose();
            }
            catch { }
            _connection = null;
        }

        // Отменяем все ожидающие запросы
        foreach (var (requestId, tcs) in _pendingRequests)
        {
            tcs.TrySetCanceled();
        }
        _pendingRequests.Clear();
    }

    /// <summary>
    /// Проверяет состояние подключения
    /// </summary>
    public bool IsConnected => _connection?.IsOpen == true && _channel?.IsOpen == true;

    #endregion

    #region Message Handling

    /// <summary>
    /// Обработчик входящих ответов из очереди
    /// </summary>
    private Task OnResponseReceivedAsync(object sender, BasicDeliverEventArgs ea)
    {
        try
        {
            var body = ea.Body.ToArray();
            var messageJson = Encoding.UTF8.GetString(body);

            var response = JsonConvert.DeserializeObject<ResponseMessage>(messageJson);

            if (response != null && !string.IsNullOrEmpty(response.CorrelationId))
            {
                if (_pendingRequests.TryRemove(response.CorrelationId, out var tcs))
                {
                    tcs.SetResult(response);
                }
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"✗ Error processing response: {ex.Message}");
        }

        return Task.CompletedTask;
    }

    /// <summary>
    /// Отправляет запрос и ждёт ответ
    /// </summary>
    public async Task<ResponseMessage> SendRequestAsync(
        string action,
        object? data = null,
        string version = "v1",
        int? timeoutSeconds = null)
    {
        if (_isDisposed)
            throw new ObjectDisposedException(nameof(TaskerMQClient));

        if (string.IsNullOrEmpty(action))
            throw new ArgumentException("Action cannot be null or empty", nameof(action));

        // Проверяем/переподключаемся при необходимости
        if (!IsConnected)
        {
            await ConnectAsync();
        }

        var timeout = timeoutSeconds ?? _defaultTimeout;
        var requestId = Guid.NewGuid().ToString();
        var tcs = new TaskCompletionSource<ResponseMessage>();

        // Создаём запрос
        var request = new RequestMessage
        {
            Id = requestId,
            Version = version,
            Action = action,
            Data = data,
            Auth = _apiKey
        };

        // Регистрируем ожидание ответа
        _pendingRequests[requestId] = tcs;

        try
        {
            // Сериализуем и отправляем
            var messageJson = JsonConvert.SerializeObject(request, new JsonSerializerSettings
            {
                NullValueHandling = NullValueHandling.Ignore
            });
            var messageBytes = Encoding.UTF8.GetBytes(messageJson);

            await _channel!.BasicPublishAsync(
                exchange: "",
                routingKey: _requestQueue,
                body: messageBytes);

            Console.WriteLine($"→ Sent: {action} (ID: {requestId[..8]}...)");

            // Ждём ответ с таймаутом
            using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(timeout));
            var completedTask = await Task.WhenAny(tcs.Task, Task.Delay(Timeout.Infinite, cts.Token));

            if (completedTask == tcs.Task)
            {
                var response = await tcs.Task;
                Console.WriteLine($"← Received: {action} - {response.Status}");
                return response;
            }
            else
            {
                throw new TimeoutException($"Request '{action}' (ID: {requestId}) timed out after {timeout} seconds");
            }
        }
        catch (Exception ex) when (ex is not TimeoutException)
        {
            Console.WriteLine($"✗ Error sending request '{action}': {ex.Message}");
            throw;
        }
        finally
        {
            _pendingRequests.TryRemove(requestId, out _);
        }
    }

    #endregion

    #region User Operations

    /// <summary>
    /// Создаёт нового пользователя
    /// </summary>
    public async Task<ResponseMessage> CreateUserAsync(string username, string email, string password)
    {
        return await SendRequestAsync("create_user", new
        {
            username,
            email,
            password
        });
    }

    /// <summary>
    /// Получает пользователя по ID
    /// </summary>
    public async Task<ResponseMessage> GetUserAsync(int id)
    {
        return await SendRequestAsync("get_user", new { id });
    }

    /// <summary>
    /// Обновляет данные пользователя
    /// </summary>
    public async Task<ResponseMessage> UpdateUserAsync(int id, string? username = null, string? email = null, string? password = null)
    {
        var data = new Dictionary<string, object> { { "id", id } };
        if (username != null) data["username"] = username;
        if (email != null) data["email"] = email;
        if (password != null) data["password"] = password;

        return await SendRequestAsync("update_user", data);
    }

    /// <summary>
    /// Удаляет пользователя
    /// </summary>
    public async Task<ResponseMessage> DeleteUserAsync(int id)
    {
        return await SendRequestAsync("delete_user", new { id });
    }

    /// <summary>
    /// Получает список всех пользователей
    /// </summary>
    public async Task<ResponseMessage> ListUsersAsync()
    {
        return await SendRequestAsync("list_users");
    }

    #endregion

    #region Project Operations

    /// <summary>
    /// Создаёт новый проект
    /// </summary>
    public async Task<ResponseMessage> CreateProjectAsync(string name, string? description = null, int? userId = null)
    {
        var data = new Dictionary<string, object> { { "name", name } };
        if (description != null) data["description"] = description;
        if (userId.HasValue) data["user_id"] = userId.Value;

        return await SendRequestAsync("create_project", data);
    }

    /// <summary>
    /// Получает проект по ID
    /// </summary>
    public async Task<ResponseMessage> GetProjectAsync(int id)
    {
        return await SendRequestAsync("get_project", new { id });
    }

    /// <summary>
    /// Обновляет проект
    /// </summary>
    public async Task<ResponseMessage> UpdateProjectAsync(int id, string? name = null, string? description = null)
    {
        var data = new Dictionary<string, object> { { "id", id } };
        if (name != null) data["name"] = name;
        if (description != null) data["description"] = description;

        return await SendRequestAsync("update_project", data);
    }

    /// <summary>
    /// Удаляет проект
    /// </summary>
    public async Task<ResponseMessage> DeleteProjectAsync(int id)
    {
        return await SendRequestAsync("delete_project", new { id });
    }

    /// <summary>
    /// Получает список проектов (опционально по пользователю)
    /// </summary>
    public async Task<ResponseMessage> ListProjectsAsync(int? userId = null)
    {
        var data = userId.HasValue ? new { user_id = userId.Value } : null;
        return await SendRequestAsync("list_projects", data);
    }

    #endregion

    #region Task Operations

    /// <summary>
    /// Создаёт новую задачу
    /// </summary>
    public async Task<ResponseMessage> CreateTaskAsync(
        string title,
        string? description = null,
        int? projectId = null,
        int? userId = null,
        DateTime? dueDate = null)
    {
        var data = new Dictionary<string, object> { { "title", title } };
        if (description != null) data["description"] = description;
        if (projectId.HasValue) data["project_id"] = projectId.Value;
        if (userId.HasValue) data["user_id"] = userId.Value;
        if (dueDate.HasValue) data["due_date"] = dueDate.Value;

        return await SendRequestAsync("create_task", data);
    }

    /// <summary>
    /// Получает задачу по ID
    /// </summary>
    public async Task<ResponseMessage> GetTaskAsync(int id)
    {
        return await SendRequestAsync("get_task", new { id });
    }

    /// <summary>
    /// Обновляет задачу
    /// </summary>
    public async Task<ResponseMessage> UpdateTaskAsync(
        int id,
        string? title = null,
        string? description = null,
        string? status = null,
        DateTime? dueDate = null)
    {
        var data = new Dictionary<string, object> { { "id", id } };
        if (title != null) data["title"] = title;
        if (description != null) data["description"] = description;
        if (status != null) data["status"] = status;
        if (dueDate.HasValue) data["due_date"] = dueDate.Value;

        return await SendRequestAsync("update_task", data);
    }

    /// <summary>
    /// Удаляет задачу
    /// </summary>
    public async Task<ResponseMessage> DeleteTaskAsync(int id)
    {
        return await SendRequestAsync("delete_task", new { id });
    }

    /// <summary>
    /// Получает список задач с фильтрацией
    /// </summary>
    public async Task<ResponseMessage> ListTasksAsync(int? projectId = null, int? userId = null, string? status = null)
    {
        var data = new Dictionary<string, object>();
        if (projectId.HasValue) data["project_id"] = projectId.Value;
        if (userId.HasValue) data["user_id"] = userId.Value;
        if (status != null) data["status"] = status;

        return await SendRequestAsync("list_tasks", data.Count > 0 ? data : null);
    }

    #endregion

    #region DLQ

    /// <summary>
    /// Демонстрирует работу Dead Letter Queue (DLQ)
    /// </summary>
    public async Task<ResponseMessage> DemonstrateDLQAsync()
    {
        return await SendRequestAsync("demonstrate_dlq", null);
    }

    /// <summary>
    /// Отправляет запрос с симуляцией ошибки для демонстрации попадания в DLQ
    /// </summary>
    public async Task<ResponseMessage> SimulateFailureAsync()
    {
        return await SendRequestAsync("simulate_failure", null);
    }

    #endregion

    #region IDisposable

    /// <summary>
    /// Освобождает ресурсы
    /// </summary>
    public void Dispose()
    {
        if (_isDisposed) return;

        _isDisposed = true;

        try
        {
            DisconnectInternalAsync().GetAwaiter().GetResult();
        }
        catch { }

        _connectionLock?.Dispose();

        Console.WriteLine("✓ Client disposed");
    }

    #endregion
}