annotations/Azaion.CommonSecurity/Services/InferenceClient.cs

using System.Diagnostics;
using System.Text;
using Azaion.CommonSecurity.DTO;
using Azaion.CommonSecurity.DTO.Commands;
using MessagePack;
using Microsoft.Extensions.Options;
using NetMQ;
using NetMQ.Sockets;

namespace Azaion.CommonSecurity.Services;

public interface IInferenceClient
{
    void Send(RemoteCommand create);
    T? Get<T>(int retries = 24, int tryTimeoutSeconds = 5, CancellationToken ct = default) where T : class;
    byte[]? GetBytes(int retries = 24, int tryTimeoutSeconds = 5, CancellationToken ct = default);
    void Stop();
}

public class InferenceClient : IInferenceClient
{
    private readonly DealerSocket _dealer = new();
    private readonly Guid _clientId = Guid.NewGuid();
    private readonly InferenceClientConfig _inferenceClientConfig;

    public InferenceClient(IOptions<InferenceClientConfig> config)
    {
        _inferenceClientConfig = config.Value;
        Start();
        _ = Task.Run(ProcessClientCommands);
    }

    private void Start()
    {
        try
        {
            using var process = new Process();
            process.StartInfo = new ProcessStartInfo
            {
                FileName = SecurityConstants.EXTERNAL_INFERENCE_PATH,
                Arguments = $"--port {_inferenceClientConfig.ZeroMqPort} --api {_inferenceClientConfig.ApiUrl}",
                //RedirectStandardOutput = true,
                //RedirectStandardError = true,
                //CreateNoWindow = true
            };

            process.OutputDataReceived += (_, e) => { if (e.Data != null) Console.WriteLine(e.Data); };
            process.ErrorDataReceived += (_, e) => { if (e.Data != null) Console.WriteLine(e.Data); };
            process.Start();
        }
        catch (Exception e)
        {
            Console.WriteLine(e);
            //throw;
        }

        _dealer.Options.Identity = Encoding.UTF8.GetBytes(_clientId.ToString("N"));
        _dealer.Connect($"tcp://{_inferenceClientConfig.ZeroMqHost}:{_inferenceClientConfig.ZeroMqPort}");
    }
    private async Task ProcessClientCommands()
    {
        //TODO: implement always on ready to client's requests. Utilize RemoteCommand

        await Task.CompletedTask;
    }

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

    public void Send(RemoteCommand command)
    {
        _dealer.SendFrame(MessagePackSerializer.Serialize(command));
    }

    public T? Get<T>(int retries = 24, int tryTimeoutSeconds = 5, CancellationToken ct = default) where T : class
    {
        var bytes = GetBytes(retries, tryTimeoutSeconds, ct);
        return bytes != null ? MessagePackSerializer.Deserialize<T>(bytes, cancellationToken: ct) : null;
    }

    public byte[]? GetBytes(int retries = 24, int tryTimeoutSeconds = 5, CancellationToken ct = default)
    {
        var tryNum = 0;
        while (!ct.IsCancellationRequested && tryNum < retries)
        {
            tryNum++;
            if (!_dealer.TryReceiveFrameBytes(TimeSpan.FromSeconds(tryTimeoutSeconds), out var bytes))
                continue;

            return bytes;
        }

        if (!ct.IsCancellationRequested)
            throw new Exception($"Unable to get bytes after {tryNum - 1} retries, {tryTimeoutSeconds} seconds each");

        return null;
    }
}