annotations/Azaion.CommonSecurity/ZeroMQExtensions.cs

using MessagePack;
using NetMQ;
using NetMQ.Sockets;

namespace Azaion.CommonSecurity;

public static class ZeroMqExtensions
{
    public static T? Get<T>(this DealerSocket dealer, Func<byte[], bool>? shouldInterceptFn = null, int retries = 24, int tryTimeoutSeconds = 5, CancellationToken ct = default) where T : class
    {
        var tryNum = 0;
        while (!ct.IsCancellationRequested && tryNum++ < retries)
        {
            if (!dealer.TryReceiveFrameBytes(TimeSpan.FromSeconds(tryTimeoutSeconds), out var bytes))
                continue;

            if (shouldInterceptFn != null && shouldInterceptFn(bytes))
                return null;

            return MessagePackSerializer.Deserialize<T>(bytes);
        }

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

        return null;
    }
}