04_layer1_callback.cpp

1// Layer-1 callback: read the writer's source blocks directly, skipping layer-2 copying and per-record dispatch.2// Implement the six methods of ReadWriterDataCallback; the reader thread invokes them.34#include "../example_support.hpp"5#include <cstring>6using namespace TwoBufQueue;78// Consumer: implement the six layer-1 callbacks to process source-block bytes directly.9struct Collector {10    std::size_t count = 0;11    long long total = 0;1213    // These six fixed methods form the ReadService interface; see read_writer_data_callback.hpp.14    void begin_read_complete(std::int64_t) noexcept {}     // Before a group of completed blocks.15    bool read_complete(SpscWriteBlock* block) noexcept {   // One full or finished block.16        const auto bytes = block->readable();              // The complete block's data range.17        return parse(block, bytes.data(), bytes.size());18    }19    void end_read_complete() noexcept {}20    void begin_read_current(std::int64_t) noexcept {}      // Before reading the current block still being written.21    bool read_current(SpscWriteBlock* block, const std::byte* ptr, std::size_t size) noexcept {22        return parse(block, ptr, size);                    // Newly available data in the current block.23    }24    void end_read() noexcept {}2526private:27    // Parse [ptr, ptr + size) as [length prefix][payload] records, using the same format as example_support.28    bool parse(SpscWriteBlock* block, const std::byte* ptr, std::size_t size) noexcept {29        return example::messages(block, ptr, size,30            [&](std::int32_t, const std::byte* data, std::size_t n) noexcept {31                if (n != sizeof(int)) return false;32                int value = 0;33                std::memcpy(&value, data, sizeof(value));34                total += value;35                ++count;36                return true;37            });38    }39};4041int main() try {42    //---------------1. Initialize one writer and one layer-1 reader---------43    auto writer = make_writer<Writer<>>();                 // Default Writer includes length prefixes.44    Collector collector;45    ReadWorker reader({}, collector);                      // {} selects the default ReadConfig.46    if (!reader.add_writer(&writer)) throw std::runtime_error("add_writer failed");47    if (!reader.start()) throw std::runtime_error("start failed");4849    //---------------2. Produce 100 int records------------------------------50    bool written = true;51    for (int i = 1; i <= 100; ++i)52        written &= writer.write(i);5354    //---------------3. Finish the producer, then drain and stop the reader--55    writer.finish();                                       // End production and hand off the final block.56    reader.stop();                                         // Wait for the remaining data before stopping.5758    //---------------4. Verify----------------------------------------------59    example::require(written && reader.error_code() == ReadError::none, "write/read failed");60    example::require(collector.count == 100 && collector.total == 5050, "delivery mismatch");61    std::puts("PASS: 100 records read directly at layer 1 (no layer-2 copy)");62} catch (const std::exception& error) { std::fprintf(stderr, "%s\n", error.what()); return 1; }