sequencing.hpp

Go to the documentation of this file.

#pragma once
 
 
#include <algorithm>
#include <cstdint>
#include <functional>
#include <optional>
#include <string>
#include <string_view>
#include <unordered_map>
 
namespace itch::transport {
 
class RetransmitRequester {
   public:
    RetransmitRequester() = default;
    RetransmitRequester(const RetransmitRequester&) = default;
    RetransmitRequester(RetransmitRequester&&) noexcept = default;
    auto operator=(const RetransmitRequester&) -> RetransmitRequester& = default;
    auto operator=(RetransmitRequester&&) noexcept -> RetransmitRequester& = default;
    virtual ~RetransmitRequester() = default;
 
    virtual auto request_retransmit(
        std::string_view session, std::uint64_t start_sequence, std::uint64_t count
    ) -> void = 0;
};
 
class SequenceTracker {
   public:
    using GapCallback = std::function<
        void(std::string_view session, std::uint64_t expected, std::uint64_t received)>;
 
    auto observe(std::string_view session, std::uint64_t first_sequence, std::uint64_t count)
        -> std::uint64_t {
        const std::string key {session};
        auto              iter = m_expected_by_session.find(key);
        if (iter == m_expected_by_session.end()) {
            // First time we see this session: anchor on its first sequence.
            m_expected_by_session.emplace(key, first_sequence + count);
            m_messages_seen += count;
            return 0;
        }
 
        std::uint64_t& expected = iter->second;
        std::uint64_t  gap      = 0;
        if (first_sequence > expected) {
            gap = first_sequence - expected;
            m_gap_count += gap;
            if (m_gap_callback) {
                m_gap_callback(session, expected, first_sequence);
            }
            if (m_requester != nullptr) {
                m_requester->request_retransmit(session, expected, gap);
            }
        }
 
        const std::uint64_t end_sequence = first_sequence + count;
        if (end_sequence > expected) {
            // Count only the genuinely new messages, not replayed duplicates.
            const std::uint64_t already_seen =
                expected > first_sequence ? expected - first_sequence : 0;
            m_messages_seen += count - std::min(count, already_seen);
            expected = end_sequence;
        }
        return gap;
    }
 
    auto set_gap_callback(GapCallback callback) -> void { m_gap_callback = std::move(callback); }
 
    auto set_retransmit_requester(RetransmitRequester* requester) noexcept -> void {
        m_requester = requester;
    }
 
    [[nodiscard]] auto expected_next(std::string_view session
    ) const -> std::optional<std::uint64_t> {
        const auto iter = m_expected_by_session.find(std::string {session});
        if (iter == m_expected_by_session.end()) {
            return std::nullopt;
        }
        return iter->second;
    }
 
    [[nodiscard]] auto gap_count() const noexcept -> std::uint64_t { return m_gap_count; }
 
    [[nodiscard]] auto messages_seen() const noexcept -> std::uint64_t { return m_messages_seen; }
 
    auto reset() -> void {
        m_expected_by_session.clear();
        m_gap_count     = 0;
        m_messages_seen = 0;
    }
 
   private:
    std::unordered_map<std::string, std::uint64_t> m_expected_by_session {};
    GapCallback                                    m_gap_callback {};
    RetransmitRequester*                           m_requester {nullptr};
    std::uint64_t                                  m_gap_count {0};
    std::uint64_t                                  m_messages_seen {0};
};
 
}  // namespace itch::transport