RSE/Backend/Handler/IpFilterHandler.cs

using System.Collections.Concurrent;
using Backend.Helper;
using Models.Model.Backend;

namespace Backend.Handler;

public class IpFilterHandler
{
    private readonly ConcurrentQueue<Discarded> _discardedQueue;
    private readonly ConcurrentQueue<UnfilteredQueueItem> _unfilteredQueue;
    private readonly ConcurrentQueue<Ip> _preFilteredQueue;
    private ThreadHandler _threadHandler;
    private bool _stop;
    private bool _done;
    private bool _stopAutoscaledThreads;
    private int _timeout;
    
    public IpFilterHandler(ConcurrentQueue<Discarded> discardedQueue,
        ConcurrentQueue<UnfilteredQueueItem> unfilteredQueue,
        ConcurrentQueue<Ip> preFilteredQueue, ThreadHandler threadHandler)
    {
        _discardedQueue = discardedQueue;
        _unfilteredQueue = unfilteredQueue;
        _preFilteredQueue = preFilteredQueue;
        _threadHandler = threadHandler;

        _timeout = 16;
    }
    
    public List<WaitHandle[]> Start(int threadCount)
    {
        WaitHandle[] waitHandle = new WaitHandle[64];
        
        int counter = 0;
        
        List<WaitHandle[]> waitHandles = [];
        
        for (int i = 0; i < threadCount; i++)
        {
            EventWaitHandle handle = new(false, EventResetMode.ManualReset);
            
            if (counter < 64)
            {
                waitHandle[counter] = handle;
                counter++;
                
                Thread f = new (Filter!);
                f.Start(handle);
                
                Console.WriteLine($"Filter thread ({i}) started");
                Thread.Sleep(16);
                
                continue;
            }
            
            counter = 0;
            
            waitHandles.Add(waitHandle);
            
            waitHandle = new WaitHandle[64];
        }
        
        return waitHandles;
    }
    
    private void Filter(object obj)
    {
        int counter = 0;
        while (!_stop)
        {
            if (_preFilteredQueue.IsEmpty)
            {
                if (counter == 30_000)
                {
                    _threadHandler.Stop();
                    _stop = true;
                }
                
                counter++;
                Thread.Sleep(128);
                
                continue;
            }
            
            _preFilteredQueue.TryDequeue(out Ip item);
            
            (int, int) ports = TcpClientHelper.CheckPort(item, 80, 443);
            
            if (ports is { Item1: 0, Item2: 0 })
            {
                _discardedQueue.Enqueue(CreateDiscardedQueueItem(item, item.ResponseCode));
                continue;
            }
            
            _unfilteredQueue.Enqueue(CreateUnfilteredQueueItem(item, ports));
        }
        
        ((EventWaitHandle) obj).Set();
    }
    
    private static Discarded CreateDiscardedQueueItem(Ip ip, int responseCode)
    {
        return new()
        {
            Ip = ip,
            ResponseCode = responseCode
        };
    }
    
    private static UnfilteredQueueItem CreateUnfilteredQueueItem(Ip ip, (int, int) ports)
    {
        Unfiltered unfiltered = new()
        {
            Ip = ip,
            Port1 = ports.Item1,
            Port2 = ports.Item2,
            Filtered = false
        };
        
        return new()
        {
            Unfiltered = unfiltered,
            Operations = Operations.Insert
        };
    }

    public void Stop()
    {
        _stop = true;
    }
}