In-process pub/sub (L4 graph)

The smallest complete libtracer node: a single-process graph with no transport and no wire bytes. A publisher writes /sensor/temp; three subscribers each receive the value a different way:

  1. a direct in-process callback — the subscribe(src, callback) sugar;

  2. a spec-faithful target vertex — subscribe(src, target) re-dispatches to a handler-backed sink vertex;

  3. a thread blocking in await() — the single-shot readiness primitive.

Delivery to (1) and (2) is a refcount-bump clone of the same rope value — no byte copy. The example finishes by reading the last-known-value back and field-writing a QoS setting, then discovering it via the :schema control read.

What to notice

  • Handles, not strings, on the hot path — the path is encoded once via the parse-once path_t("…") constructor; register_vertex returns a vertex_handle_t reused for every op.

  • await is join-safe — the publisher join()s the waiter after the write, so every delivery is complete and visible when the checks run.

  • It self-checks — each delivery path is asserted, so the ctest smoke test guards behavior, not just a clean exit.

Source

  1/*
  2 * SPDX-License-Identifier: Apache-2.0
  3 * SPDX-FileCopyrightText: Copyright 2026 avatarsd LLC
  4 */
  5
  6/**
  7 * @file
  8 * @brief In-process publish/subscribe over the L4 graph — the M3 P0 node, end to end.
  9 *
 10 * No transport, no wire bytes leave the process. A publisher writes
 11 * `/sensor/temp`; three subscribers receive the value three different ways:
 12 *   1. a direct in-process callback  — `subscribe(src, callback)` sugar;
 13 *   2. a spec-faithful target vertex — `subscribe(src, target)` → a handler sink;
 14 *   3. a thread blocking in `await()` — the single-shot primitive.
 15 * Delivery to (1) and (2) is a refcount-bump clone of the same rope value — no
 16 * byte copy.
 17 *
 18 * Runs under ctest as `example_in_process_pubsub`: it @ref check "checks" every
 19 * delivered value and returns non-zero on any mismatch, so the smoke test guards
 20 * behavior (each subscriber sees the write) rather than merely exiting cleanly.
 21 */
 22
 23#include <chrono>
 24#include <cstddef>
 25#include <cstdint>
 26#include <cstdio>
 27#include <thread>
 28
 29#include "libtracer/tracer.hpp"
 30
 31namespace {
 32
 33using namespace std::chrono_literals;
 34using tr::graph::path_t;
 35using tr::graph::role_t;
 36
 37/** @brief A 4-byte little-endian VALUE view over @p v (one heap segment). */
 38tr::view::view_t value_u32(std::uint32_t v) {
 39    tr::view::segment_ptr_t seg = tr::view::heap_alloc(4);
 40    for (int i = 0; i < 4; ++i)
 41        seg->bytes[static_cast<std::size_t>(i)] = static_cast<std::byte>((v >> (8 * i)) & 0xFF);
 42    return tr::view::view_t::over(std::move(seg));
 43}
 44
 45/** @brief Decode a little-endian u32 from @p view's first (≤4) bytes. */
 46std::uint32_t as_u32(const tr::view::view_t& view) {
 47    const auto b = view.bytes();
 48    std::uint32_t v = 0;
 49    for (std::size_t i = 0; i < b.size() && i < 4; ++i)
 50        v |= static_cast<std::uint32_t>(std::to_integer<std::uint8_t>(b[i])) << (8 * i);
 51    return v;
 52}
 53
 54/** @brief Record a failed expectation on @p ok and report it; leaves @p ok false on failure. */
 55void check(bool& ok, bool cond, const char* what) {
 56    if (!cond) {
 57        std::printf("  [FAIL] %s\n", what);
 58        ok = false;
 59    }
 60}
 61
 62}  // namespace
 63
 64int main() {
 65    constexpr std::uint32_t kSent = 23;
 66    tr::graph::graph_t g;
 67
 68    const tr::graph::vertex_handle_t temp =
 69        g.register_vertex(path_t("/sensor/temp"), role_t::STORED_VALUE);
 70
 71    // Values captured from each delivery path, verified at the end.
 72    std::uint32_t cb_got = 0;     // subscriber 1 (callback)
 73    std::uint32_t sink_got = 0;   // subscriber 2 (target vertex handler)
 74    std::uint32_t await_got = 0;  // subscriber 3 (await)
 75
 76    // A "sink" vertex backed by a callback handler — the target of a spec-faithful
 77    // SUBSCRIBER (subscriber 2 below re-dispatches to it).
 78    tr::graph::handlers_t sink;
 79    sink.on_write = [&sink_got](const tr::view::rope_t& in) -> tr::graph::result_t<void> {
 80        sink_got = as_u32(in.only());
 81        std::printf("  [sink vertex /log/temp] received %u\n", sink_got);
 82        return {};
 83    };
 84    (void)g.register_vertex(path_t("/log/temp"), role_t::HANDLER, std::move(sink));
 85
 86    // subscriber_t 1 — direct in-process callback.
 87    (void)g.subscribe(path_t("/sensor/temp"), [&cb_got](const tr::view::rope_t& v) {
 88        cb_got = as_u32(v.only());
 89        std::printf("  [callback sub] received %u\n", cb_got);
 90    });
 91    // subscriber_t 2 — spec-faithful target-path subscription -> /log/temp.
 92    (void)g.subscribe(path_t("/sensor/temp"), path_t("/log/temp"));
 93
 94    // subscriber_t 3 — a thread blocking in await().
 95    std::thread waiter([&] {
 96        auto r = g.await(temp, 2s);
 97        if (r) {
 98            await_got = as_u32(r->only());
 99            std::printf("  [await sub] received %u\n", await_got);
100        }
101    });
102    std::this_thread::sleep_for(50ms);  // let the waiter park in await()
103
104    std::printf("publisher: write /sensor/temp = %u\n", kSent);
105    (void)g.write(temp, value_u32(kSent));
106    waiter.join();  // joins-after-write: every delivery is complete + visible here
107
108    // Read back the last-known-value (a clone — keeps the segment alive for us).
109    auto rb = g.read(temp);
110    const std::uint32_t rb_got = rb ? as_u32(rb->only()) : 0u;
111    std::printf("read-back /sensor/temp = %u\n", rb_got);
112
113    // Field-write a QoS setting, then discover it via :schema.
114    (void)g.write(path_t("/sensor/temp:settings.deadline_ns"), value_u32(5000));
115    auto schema = g.read(path_t("/sensor/temp:schema"));
116    std::size_t schema_children = 0;
117    if (schema) {
118        if (auto point = tr::wire::view_as_tlv(schema->only()))
119            schema_children = point->children.size();
120    }
121    std::printf(":schema is a POINT with %zu children\n", schema_children);
122
123    // Guard the semantics, not just a clean exit: every path must see the write.
124    bool ok = true;
125    check(ok, cb_got == kSent, "callback subscriber received the written value");
126    check(ok, sink_got == kSent, "target-vertex subscriber received the written value");
127    check(ok, await_got == kSent, "await subscriber received the written value");
128    check(ok, rb_got == kSent, "read-back returns the last-known-value");
129    check(ok, schema_children == 2, ":schema resolves to a 2-child POINT");
130    return ok ? 0 : 1;
131}

See also: graph module · views · path.