annotations/Azaion.Common/Services/LoaderClient.cs

using System.Diagnostics;
using System.IO;
using System.Text;
using Azaion.Common.DTO;
using MessagePack;
using NetMQ;
using NetMQ.Sockets;
using Serilog;
using Exception = System.Exception;

namespace Azaion.Common.Services;

public class LoaderClient : IDisposable
{
    private readonly LoaderClientConfig _config;
    private readonly ILogger _logger;
    private readonly CancellationToken _ct;
    private readonly IProcessLauncher _processLauncher;
    private readonly DealerSocket _dealer = new();
    private readonly Guid _clientId = Guid.NewGuid();

    public LoaderClient(LoaderClientConfig config, ILogger logger, IProcessLauncher processLauncher, CancellationToken ct = default)
    {
        _config = config;
        _logger = logger;
        _processLauncher = processLauncher;
        _ct = ct;
    }

    public void StartClient()
    {
        try
        {
            _processLauncher.Launch(
                Constants.EXTERNAL_LOADER_PATH,
                $"--port {_config.ZeroMqPort} --api {_config.ApiUrl}");
        }
        catch (Exception e)
        {
            _logger.Error(e, e.Message);
            throw;
        }
    }

    public void Connect()
    {
        _dealer.Options.Identity = Encoding.UTF8.GetBytes(_clientId.ToString("N"));
        _dealer.Connect($"tcp://{_config.ZeroMqHost}:{_config.ZeroMqPort}");
    }

    public void Login(ApiCredentials credentials)
    {
        var result = SendCommand(RemoteCommand.Create(CommandType.Login, credentials));
        if (result.CommandType != CommandType.Ok)
            throw new Exception(result.Message);
    }

    public MemoryStream LoadFile(string filename, string folder)
    {
        var result = SendCommand(RemoteCommand.Create(CommandType.Load, new LoadFileData(filename, folder)));
        if (result.Data?.Length == 0)
            throw new Exception($"Can't load {filename}. Returns 0 bytes");
        return new MemoryStream(result.Data!);
    }

    private RemoteCommand SendCommand(RemoteCommand command, int retryCount = 50, int retryDelayMs = 800)
    {
        try
        {
            _dealer.SendFrame(MessagePackSerializer.Serialize(command));

            var tryNum = 0;
            while (!_ct.IsCancellationRequested && tryNum++ < retryCount)
            {
                if (!_dealer.TryReceiveFrameBytes(TimeSpan.FromMilliseconds(retryDelayMs), out var bytes))
                    continue;
                var res = MessagePackSerializer.Deserialize<RemoteCommand>(bytes, cancellationToken: _ct);
                if (res.CommandType == CommandType.Error)
                    throw new Exception(res.Message);
                return res;
            }

            throw new Exception($"Sent {command} {retryCount} times, with wait time {retryDelayMs}ms for each call. No response from client.");
        }
        catch (Exception e)
        {
            _logger.Error(e, e.Message);
            throw;
        }
    }

    public void Stop()
    {
        _dealer.SendFrame(MessagePackSerializer.Serialize(new RemoteCommand(CommandType.Exit)));
    }

    public void Dispose()
    {
        _dealer.Dispose();
    }
}