example_support.hpp

1#pragma once2#include <TwoBufQueue.hpp>3#include <cstdio>4#include <cstring>5#include <stdexcept>6#include <string_view>78namespace example {9inline void require(bool ok, const char* message) {10    if (!ok) throw std::runtime_error(message);11}1213// A frame contains multiple [uint16 little-endian length][payload] records.14// Payloads may be unaligned: use memcpy, never reinterpret_cast<Record*>.15template<class TVisitor>16bool messages(const TwoBufQueue::DataBatch& batch, TVisitor&& visit) noexcept {17    for (const auto& frame : batch.frames) {18        const auto* p = frame.data;19        auto remaining = frame.size;20        while (remaining) {21            if (remaining < 2) return false;22            const auto length = std::to_integer<unsigned>(p[0]) |23                                (std::to_integer<unsigned>(p[1]) << 8);24            p += 2;25            remaining -= 2;26            if (length > remaining || !visit(frame.thread_id, p, length)) return false;27            p += length;28            remaining -= length;29        }30    }31    return true;32}3334template<class TVisitor>35bool messages(const TwoBufQueue::SpscWriteBlock* block, const std::byte* p,36              std::size_t remaining, TVisitor&& visit) noexcept {37    while (remaining) {38        if (remaining < 2) return false;39        const auto length = std::to_integer<unsigned>(p[0]) | (std::to_integer<unsigned>(p[1]) << 8);40        p += 2; remaining -= 2;41        if (length > remaining || !visit(block->source_id(), p, length)) return false;42        p += length; remaining -= length;43    }44    return true;45}4647struct Record { std::uint64_t producer; std::uint64_t sequence; };4849inline TwoBufQueue::WriterBackpressureConfig blocking() {50    TwoBufQueue::WriterBackpressureConfig config;51    config.mode = TwoBufQueue::BackpressureMode::Block;52    return config; // bounded wait; write still returns false on timeout53}54}