IndexMatchQueue.h

#pragma once
 
#include <cstdint>
#include <type_traits>
#include <utility>
#include <limits>
#include <functional>
 
#include <NDEVR/QueueBuffer.h>
 
namespace NDEVR
{
    template <class t_a, class t_b, class t_index = uint04>
    class IndexMatchQueue
    {
    public:
        using A = t_a;      
        using B = t_b;      
        using Index = t_index;  
        using UIndex = std::make_unsigned_t<Index>; 
 
        using OnMatchFn = std::function<void(const A&, const B&)>;      
        using OnNearMatchFn = std::function<void(const A&, const B&)>;  

        IndexMatchQueue() = default;

        IndexMatchQueue(OnMatchFn on_match, OnNearMatchFn on_near_match = {})
            : m_on_match(std::move(on_match))
            , m_on_near_match(std::move(on_near_match))
        {}

        void setOnMatch(OnMatchFn on_match)
        {
            m_on_match = std::move(on_match);
        }

        void setOnNearMatch(OnNearMatchFn on_near_match)
        {
            m_on_near_match = std::move(on_near_match);
        }

        void clear()
        {
            m_a.clear();
            m_b.clear();
        }

        void pushA(Index idx, A a)
        {
            if (m_a.size() > 0 && m_a.last().index == idx)
            {
                m_a.last().value = a;
                return;
            }
            m_a.push(NodeA{ idx, std::move(a) });
            drain();
            if (m_a.size() > HalfRange())
                m_a.pop();
        }

        void pushB(Index idx, B b)
        {
            if (m_b.size() > 0 && m_b.last().index == idx)
            {
                m_b.last().value = b;
                return;
            }
            m_b.push(NodeB{ idx, std::move(b) });
            drain();
            if (m_b.size() > HalfRange())
                m_b.pop();
        }
 
    protected:
        struct NodeA
        {
            Index   index;  
            A       value;  
        };

        struct NodeB
        {
            Index   index;  
            B       value;  
        };

        static constexpr UIndex HalfRange() noexcept
        {
            return (UIndex(~UIndex(0)) >> 1) + 1;
        }

        static constexpr bool seqLess(Index a, Index b) noexcept
        {
            const UIndex ua = static_cast<UIndex>(a);
            const UIndex ub = static_cast<UIndex>(b);
            const UIndex d = ub - ua;
            return d != 0 && d < HalfRange();
        }

        static constexpr UIndex circularDistance(Index a, Index b) noexcept
        {
            const UIndex ua = static_cast<UIndex>(a);
            const UIndex ub = static_cast<UIndex>(b);
            const UIndex d1 = ub - ua;
            const UIndex d2 = ua - ub;
            return (d1 < d2) ? d1 : d2;
        }

        void drain()
        {
            static_assert(std::is_integral_v<Index>, "Index must be integral");
 
            while (!m_a.empty() && !m_b.empty())
            {
                const Index ia = m_a.front().index;
                const Index ib = m_b.front().index;
 
                if (ia == ib)
                {
                    if (m_on_match)
                        m_on_match(m_a.front().value, m_b.front().value);
 
                    m_a.pop();
                    m_b.pop();
                    continue;
                }
                if (seqLess(ia, ib))
                {
                    m_a.pop();
                    while (m_a.size() > 0 && seqLess(m_a.front().index, ib))
                        m_a.pop();
                }
                else
                {
                    const UIndex dist = circularDistance(ia, ib);
                    //lib_assert(dist <= distance_tolerance, "unmatched record");
                    if (dist <= distance_tolerance)
                    {
                        if (m_on_near_match)
                            m_on_near_match(m_a.front().value, m_b.front().value);
                        m_b.pop();
                    }
                    else
                    {
                        while (m_b.size() > 0 && !seqLess(ia, m_b.front().index))
                            m_b.pop();
                    }
                    //while (m_b.size() > 0 && !seqLess(ia, m_b.front().index))
                        //m_b.pop();
                }
            }
 
        }
 
    private:
        uint04 distance_tolerance = 50U;    
        QueueBuffer<NodeA> m_a;             
        QueueBuffer<NodeB> m_b;             
 
        OnMatchFn m_on_match;               
        OnNearMatchFn m_on_near_match;      
    };
}