07_quick_write_muti_msg.cpp

1#include "../example_support.hpp"2#include <array>3#include <chrono>4#include <cstring>5using namespace TwoBufQueue;67int main() try {8    //---------------1. Initialization--------------------------------------9    using HotWriter = Writer<UINT32_MAX>; // Publish at block handoff and keep length prefixes; Session needs no additional template argument.10    auto writer = make_writer<HotWriter>({.block_size = 4096, .pool_min = 8, .pool_max = 8});11    std::size_t count = 0;12    auto rcvCallback = [&](std::int32_t thread_id, const std::byte* data, std::size_t size) noexcept {13        (void)thread_id;14        std::array<std::uint64_t, 4> value{};15        if (size != sizeof(value)) std::terminate();16        std::memcpy(value.data(), data, size);17        if (value != std::array<std::uint64_t,4>{count,count+1,count+2,count+3}) std::terminate();18        ++count;19    };20    ::TwoBufQueue::TwoBufQueue queue(rcvCallback, writer);2122    //---------------2. Producer writes-------------------------------------23    std::uint64_t sent = 0;24    auto session = writer.get_current_session(std::chrono::seconds{1}); // Handle acquisition failure and return.25    example::require(static_cast<bool>(session), "cannot acquire session within timeout");26    while (sent != 10000) { // Outer loop: the caller explicitly selects the next block.27        while (sent != 10000) { // Inner loop: write only into this block; leave the loop when it lacks space.28            const auto result = session.write(std::array<std::uint64_t,4>{sent,sent+1,sent+2,sent+3});29            if (result == SessionWriteResult::insufficient_space) break;30            example::require(result == SessionWriteResult::success, "record exceeds block capacity or write error");31            ++sent; // Advance the message position only after success.32        }33        if (sent == 10000) break; // All messages are written; do not acquire another block.34        session = {}; // End the old Session: synchronize the cursor only; neither publish nor switch blocks.35        session = writer.get_next_session(std::chrono::seconds{1});36        example::require(static_cast<bool>(session), "cannot acquire session within timeout");37    }38    session.flush(); // Optional explicit flush: publish the position without switching blocks or ending the Session.39    session = {}; // End the Session before calling Writer.finish; scope-based destruction works too.4041    //---------------3. Finish----------------------------------------------42    // With current scanning disabled, flush does not hand off the tail; finish hands off the final block.43    writer.finish();44    queue.stop();45    //---------------4. Check the result------------------------------------46    example::require(count == 10000, "message count mismatch");47    std::puts("PASS: 10000 messages with current/next single-block sessions");48} catch (const std::exception& error) { std::fprintf(stderr, "%s\n", error.what()); return 1; }