C++ API Reference

Dora provides C++ bindings for both standalone nodes and in-process operators via CXX (Rust-C++ interop). The CXX bridge generates type-safe C++ headers from Rust definitions -- no raw FFI or manual extern "C" declarations are needed.

Two crates provide the C++ surface:

Crate	Library	Use case
dora-node-api-cxx	libdora_node_api_cxx.a	Standalone node executable
dora-operator-api-cxx	libdora_operator_api_cxx.a	Shared-library operator loaded by the runtime

Generated headers: dora-node-api.h and dora-operator-api.h.

---

Node API (dora-node-api-cxx)

Initialization

#include "dora-node-api.h"

// Initialize a node from environment variables set by the Dora daemon.
// Returns an DoraNode struct containing the event stream and output sender.
// Throws on failure.
DoraNode init_dora_node();

DoraNode

Returned by init_dora_node(). Owns the event stream and the output sender for the lifetime of the node.

struct DoraNode {
    rust::Box<Events>        events;       // event stream (blocking receiver)
    rust::Box<OutputSender>  send_output;  // output sender
};

Events

Opaque Rust type exposed to C++. Provides blocking iteration over the node's incoming events.

// Member function -- call on the boxed object directly.
rust::Box<DoraEvent> Events::next();

// Free function form -- equivalent to events->next().
rust::Box<DoraEvent> next_event(rust::Box<Events>& events);

Both forms block until the next event arrives and return an owned DoraEvent.

Non-blocking and timed receive

For nodes that need to do work between events, or react to deadlines, the Events stream also offers non-blocking and timed variants of next_event():

// Block up to `timeout_ms` milliseconds for the next event. Returns
// an event with `event_type() == Timeout` if the deadline elapses
// before one arrives, or `AllInputsClosed` if the stream closed.
rust::Box<DoraEvent> next_event_timeout(rust::Box<Events>& events, uint64_t timeout_ms);

// Non-blocking poll. Returns `event_type() == Empty` if no event is
// immediately available, or `AllInputsClosed` if the stream is closed.
rust::Box<DoraEvent> try_next_event(rust::Box<Events>& events);

// Hint: true when the event queue is currently empty. Treat as
// advisory only -- the daemon can produce a new event between the
// check and a subsequent receive.
bool events_is_empty(const rust::Box<Events>& events);

For draining buffered events into a snapshot (useful when you want to process a batch all at once without racing the daemon):

// Take a snapshot of all currently-buffered events. Subsequent
// `next_event` / `try_next_event` calls only see events that arrive
// after this point.
rust::Box<DrainedEvents> drain_events(rust::Box<Events>& events);

size_t drained_events_len(const rust::Box<DrainedEvents>& drained);

// Pop the next event from a drained snapshot. Returns
// `event_type() == Empty` once the snapshot is exhausted.
rust::Box<DoraEvent> drained_events_next(rust::Box<DrainedEvents>& drained);

DoraEvent

Opaque Rust type. Inspect its kind with event_type(), then downcast with one of the variant extractors below.

// Determine the event kind.
DoraEventType event_type(const rust::Box<DoraEvent>& event);

// Downcast to a raw-byte input. Throws if the event is not Input.
DoraInput event_as_input(rust::Box<DoraEvent> event);

// Downcast to an Arrow FFI input (writes Arrow C Data Interface structs).
// out_array and out_schema must point to valid ArrowArray / ArrowSchema structs.
// Returns DoraResult with empty error on success.
DoraResult event_as_arrow_input(
    rust::Box<DoraEvent> event,
    uint8_t* out_array,
    uint8_t* out_schema);

// Same as above, but also returns the input ID and metadata.
ArrowInputInfo event_as_arrow_input_with_info(
    rust::Box<DoraEvent> event,
    uint8_t* out_array,
    uint8_t* out_schema);

// Downcast to a NodeFailed payload. Throws if the event is not NodeFailed.
DoraNodeFailed event_as_node_failed(rust::Box<DoraEvent> event);

DoraEventType

enum class DoraEventType : uint8_t {
    Stop,             // graceful shutdown requested
    Input,            // new data arrived on an input
    InputClosed,      // a single input was closed
    Error,            // an error occurred
    Unknown,          // unrecognized event variant
    AllInputsClosed,  // all inputs closed (stream ended)
    Empty,            // try_next_event / drained_events_next found no event ready
    Timeout,          // next_event_timeout deadline elapsed without an event
    NodeFailed,       // an upstream node failed (use event_as_node_failed to inspect)
    Reload,           // hot-reload notification for an operator
};

Empty is distinct from Timeout: Empty means "no event available right now" (the caller did not request a timeout); Timeout means "the caller-supplied deadline elapsed before an event arrived". A non-blocking poll never returns Timeout; a timed receive returns either Timeout or AllInputsClosed if no event arrived in time.

DoraInput

Returned by event_as_input(). Contains raw bytes.

struct DoraInput {
    rust::String     id;    // input identifier (e.g. "tick", "image")
    rust::Vec<uint8_t> data;  // raw payload bytes
};

DoraNodeFailed

Returned by event_as_node_failed(). Carries the failure information from an upstream node that exited unexpectedly. Use it to log the cause, flush downstream state, or trigger graceful shutdown.

struct DoraNodeFailed {
    rust::Vec<rust::String> affected_input_ids;  // inputs on this node that will stop receiving data
    rust::String            error;               // human-readable error message from the failed node
    rust::String            source_node_id;      // id of the node that failed
};

Example:

auto event = next_event(dora_node.events);
if (event_type(event) == DoraEventType::NodeFailed) {
    auto failed = event_as_node_failed(std::move(event));
    std::cerr << "Upstream node `" << std::string(failed.source_node_id)
              << "` failed: " << std::string(failed.error) << std::endl;
}

ArrowInputInfo

Returned by event_as_arrow_input_with_info(). Contains the input ID, metadata, and an error string.

struct ArrowInputInfo {
    rust::String       id;        // input identifier
    rust::Box<Metadata> metadata; // attached metadata
    rust::String       error;     // empty on success
};

DoraResult

Returned by output-sending functions. Check the error field -- empty means success.

struct DoraResult {
    rust::String error;  // empty string on success
};

OutputSender

Opaque Rust type. All methods take rust::Box<OutputSender>& as the first argument (the sender from DoraNode::send_output).

send_output

Send raw bytes on a named output.

DoraResult send_output(
    rust::Box<OutputSender>& sender,
    rust::String id,
    rust::Slice<const uint8_t> data);

send_output_with_metadata

Send raw bytes with attached metadata.

DoraResult send_output_with_metadata(
    rust::Box<OutputSender>& sender,
    rust::String id,
    rust::Slice<const uint8_t> data,
    rust::Box<Metadata> metadata);

send_arrow_output

Send an Arrow array via the C Data Interface. The pointers must reference valid ArrowArray and ArrowSchema structs. Ownership of the Arrow data transfers to Rust on success.

DoraResult send_arrow_output(
    rust::Box<OutputSender>& sender,
    rust::String id,
    uint8_t* array_ptr,
    uint8_t* schema_ptr);

// Overload with metadata (same C++ name via cxx_name attribute).
DoraResult send_arrow_output(
    rust::Box<OutputSender>& sender,
    rust::String id,
    uint8_t* array_ptr,
    uint8_t* schema_ptr,
    rust::Box<Metadata> metadata);

close_outputs

Selectively close one or more of this node's outputs without shutting the whole node down. Downstream subscribers see the corresponding InputClosed event for each closed output.

DoraResult close_outputs(
    rust::Box<OutputSender>& sender,
    rust::Vec<rust::String> output_ids);

output_ids are validated as DataIds; an invalid id (e.g. one containing characters not allowed by the data-id grammar) returns a DoraResult whose error names the offending id and the underlying validation message, instead of aborting the process.

node_config_json

Return this node's NodeRunConfig (inputs, outputs, type annotations, framing overrides, etc. — the per-node block from the dataflow descriptor) serialized as JSON. Lets a C++ node reason about its own declared interface at runtime without re-parsing the dataflow yaml.

rust::String node_config_json(const rust::Box<OutputSender>& sender);

Throws rust::Error if serialization fails (rare — NodeRunConfig has no fields that can produce serde errors in practice).

dataflow_descriptor_json

Return the full parsed Descriptor (the entire dataflow yaml) serialized as JSON. Useful for introspecting peer nodes, listing all topics, etc.

rust::String dataflow_descriptor_json(const rust::Box<OutputSender>& sender);

Throws rust::Error if the daemon hasn't delivered the descriptor yet, or on serialization failure.

Caution: the returned JSON includes any env blocks declared on nodes in the dataflow yaml. Inline literals (API_KEY: "...") and host-environment substitutions (API_KEY: "${HOST_VAR}", expanded at descriptor parse time) both appear as plain strings in the output. Don't pipe the result into shared logs or telemetry without sanitizing if your dataflow can carry secrets in env vars.

log_message

Send a log message through the Dora logging system.

DoraResult log_message(
    const rust::Box<OutputSender>& sender,
    rust::String level,    // e.g. "info", "warn", "error"
    rust::String message);

Metadata

Opaque Rust type for attaching typed key-value pairs to outputs.

Construction

rust::Box<Metadata> new_metadata();

Reading

uint64_t     Metadata::timestamp() const;

bool         Metadata::get_bool(const rust::Str key) const;        // throws on missing/wrong type
int64_t      Metadata::get_int(const rust::Str key) const;
double       Metadata::get_float(const rust::Str key) const;
rust::String Metadata::get_str(const rust::Str key) const;

rust::Vec<int64_t>      Metadata::get_list_int(const rust::Str key) const;
rust::Vec<double>       Metadata::get_list_float(const rust::Str key) const;
rust::Vec<rust::String> Metadata::get_list_string(const rust::Str key) const;

int64_t      Metadata::get_timestamp(const rust::Str key) const;   // nanoseconds since epoch
rust::String Metadata::get_json(const rust::Str key) const;        // single value as JSON string

Writing

All setters throw on failure.

void Metadata::set_bool(const rust::Str key, bool value);
void Metadata::set_int(const rust::Str key, int64_t value);
void Metadata::set_float(const rust::Str key, double value);
void Metadata::set_string(const rust::Str key, rust::String value);

void Metadata::set_list_int(const rust::Str key, rust::Vec<int64_t> value);
void Metadata::set_list_float(const rust::Str key, rust::Vec<double> value);
void Metadata::set_list_string(const rust::Str key, rust::Vec<rust::String> value);

void Metadata::set_timestamp(const rust::Str key, int64_t nanos);  // nanoseconds since epoch

Introspection

MetadataValueType Metadata::type(const rust::Str key) const;  // throws if key missing
rust::String      Metadata::to_json() const;                   // full metadata as JSON
rust::Vec<rust::String> Metadata::list_keys() const;

MetadataValueType

enum class MetadataValueType : uint8_t {
    Bool,
    Integer,
    Float,
    String,
    ListInt,
    ListFloat,
    ListString,
    Timestamp,
};

Service, Action, and Streaming Patterns

C++ nodes can implement communication patterns using the metadata API. The well-known metadata keys are:

Key	Description
"request_id"	Service request/response correlation (UUID v7)
"goal_id"	Action goal identification (UUID v7)
"goal_status"	Action result status: "succeeded", "aborted", or "canceled"
"session_id"	Streaming session identifier
"segment_id"	Streaming segment within a session (integer)
"seq"	Streaming chunk sequence number (integer)
"fin"	Last chunk of a streaming segment (bool)
"flush"	Discard older queued messages on input (bool)

// Service server: pass through request_id from input metadata
auto input_metadata = event_as_arrow_input_with_info(event);
send_output_with_metadata(sender, "response", result, std::move(input_metadata.metadata));

// Action server: set goal_id and goal_status on result
auto meta = new_metadata();
meta->set_string("goal_id", goal_id);
meta->set_string("goal_status", "succeeded");
send_output_with_metadata(sender, "result", result_data, std::move(meta));

CombinedEvents (ROS2 integration)

When using the optional ros2-bridge feature, node events and ROS2 subscription events can be merged into a single stream.

// Convert Dora events into a combined stream.
CombinedEvents dora_events_into_combined(rust::Box<Events> events);

// Create an empty combined stream (for ROS2-only nodes).
CombinedEvents empty_combined_events();

CombinedEvents struct

struct CombinedEvents {
    rust::Box<MergedEvents> events;

    CombinedEvent next();  // blocking -- returns the next merged event
};

CombinedEvent struct

struct CombinedEvent {
    rust::Box<MergedDoraEvent> event;

    bool is_dora() const;  // true if this is a standard Dora event
};

// Downcast a combined event back to an DoraEvent. Throws if not an Dora event.
rust::Box<DoraEvent> downcast_dora(CombinedEvent event);

ROS2 subscriptions add their own events to the merged stream. Use subscription->matches(event) and subscription->downcast(event) to handle ROS2-specific events (see the ROS2 Bridge docs).

---

Operator API (dora-operator-api-cxx)

Operators are shared libraries loaded by the Dora runtime. The C++ side implements five functions that the CXX bridge calls into.

Breaking change vs. earlier dora releases: the operator API used to require only new_operator + on_input; Event::InputClosed and Event::Stop were silently dropped on the C++ side. As of dora #1849 the bridge calls on_input_closed and on_stop instead. As of dora #1879 the bridge also calls on_input_parse_error. Existing operators must add a no-op stub for each new function — return { rust::String(), false }; is sufficient.

Required C++ interface

You must provide a header operator.h and an implementation file. The header declares an Operator class and five free functions:

// operator.h
#pragma once
#include <memory>
#include "dora-operator-api.h"

class Operator {
public:
    Operator();
    // Add any state your operator needs.
};

std::unique_ptr<Operator> new_operator();

DoraOnInputResult on_input(
    Operator& op,
    rust::Str id,
    rust::Slice<const uint8_t> data,
    OutputSender& output_sender);

DoraOnInputResult on_input_closed(Operator& op, rust::Str id, OutputSender& output_sender);
DoraOnInputResult on_stop(Operator& op, OutputSender& output_sender);
DoraOnInputResult on_input_parse_error(Operator& op, rust::Str id, rust::Str error, OutputSender& output_sender);

• new_operator() -- called once at startup; returns the operator instance.
• on_input() -- called for every input event; process data and optionally send outputs.
• on_input_closed() -- called when an upstream input stream closes (the daemon delivers Event::InputClosed { id }). Operator can log, flush per-input state, or send_output(output_sender, ...) to emit a final/status message in response. Set result.stop = true to request shutdown.
• on_stop() -- called on graceful shutdown (the daemon delivers Event::Stop). Operator can drain output queues, persist final state, or send_output(output_sender, ...) to flush buffered data before returning.
• on_input_parse_error() -- called when an input's Arrow data fails to deserialize (Event::InputParseError { id, error }). id identifies the affected input; error is the deserialization error string. Operator can log the error, surface it as health state, or request shutdown. Previously these events were silently dropped.

Default implementations that simply log + return success are sufficient for operators that don't need to react to these events. See examples/c++-dataflow/operator-rust-api/operator.cc for a minimal reference.

OutputSender (operator)

Available inside all operator callbacks. Sends data on a named output.

DoraSendOutputResult send_output(
    OutputSender& sender,
    rust::Str id,
    rust::Slice<const uint8_t> data);

Result types

struct DoraOnInputResult {
    rust::String error;  // empty on success
    bool         stop;   // true to request graceful shutdown
};

error and stop are mutually exclusive. If error is non-empty the runtime treats the result as a fatal failure regardless of the stop field. Use stop = true only when error is empty.

struct DoraSendOutputResult {
    rust::String error;  // empty on success
};

---

Quick Start: Node Example

A minimal node that receives timer ticks and sends a counter.

#include "dora-node-api.h"
#include <iostream>
#include <vector>

int main() {
    auto dora_node = init_dora_node();
    unsigned char counter = 0;

    for (;;) {
        auto event = next_event(dora_node.events);
        auto ty = event_type(event);

        if (ty == DoraEventType::AllInputsClosed) {
            break;
        }
        if (ty == DoraEventType::Stop) {
            break;
        }
        if (ty == DoraEventType::Input) {
            auto input = event_as_input(std::move(event));
            counter += 1;

            std::cout << "Input: " << std::string(input.id)
                      << " counter=" << (int)counter << std::endl;

            std::vector<unsigned char> out{counter};
            rust::Slice<const uint8_t> slice{out.data(), out.size()};
            auto result = send_output(dora_node.send_output, "counter", slice);
            if (!result.error.empty()) {
                std::cerr << "Send error: " << std::string(result.error) << std::endl;
                return 1;
            }
        }
    }
    return 0;
}

Dataflow YAML:

nodes:
  - id: cxx-node
    path: build/my_node
    inputs:
      tick: dora/timer/millis/300
    outputs:
      - counter

---

Quick Start: Arrow Node Example

A node that receives and sends Arrow arrays via the C Data Interface, with metadata.

#include "dora-node-api.h"
#include <arrow/api.h>
#include <arrow/c/bridge.h>
#include <iostream>

int main() {
    auto dora_node = init_dora_node();

    for (int i = 0; i < 10; i++) {
        auto event = dora_node.events->next();
        auto ty = event_type(event);

        if (ty == DoraEventType::AllInputsClosed || ty == DoraEventType::Stop) {
            break;
        }
        if (ty == DoraEventType::Input) {
            // Receive Arrow input with metadata
            struct ArrowArray c_array;
            struct ArrowSchema c_schema;
            auto info = event_as_arrow_input_with_info(
                std::move(event),
                reinterpret_cast<uint8_t*>(&c_array),
                reinterpret_cast<uint8_t*>(&c_schema));

            if (!info.error.empty()) {
                std::cerr << std::string(info.error) << std::endl;
                continue;
            }

            std::cout << "Input: " << std::string(info.id)
                      << " ts=" << info.metadata->timestamp() << std::endl;

            auto imported = arrow::ImportArray(&c_array, &c_schema);
            auto array = imported.ValueOrDie();
            std::cout << "Arrow: " << array->ToString() << std::endl;

            // Build an output Arrow array
            arrow::Int32Builder builder;
            builder.Append(i * 10);
            std::shared_ptr<arrow::Array> out_array;
            builder.Finish(&out_array);

            // Export and send with metadata
            struct ArrowArray out_c_array;
            struct ArrowSchema out_c_schema;
            arrow::ExportArray(*out_array, &out_c_array, &out_c_schema);

            auto meta = new_metadata();
            meta->set_string("source", "cpp-arrow-node");
            meta->set_int("iteration", i);

            auto result = send_arrow_output(
                dora_node.send_output, "counter",
                reinterpret_cast<uint8_t*>(&out_c_array),
                reinterpret_cast<uint8_t*>(&out_c_schema),
                std::move(meta));

            if (!result.error.empty()) {
                std::cerr << "Send error: " << std::string(result.error) << std::endl;
            }
        }
    }
    return 0;
}

---

Quick Start: Operator Example

A minimal operator shared library.

// operator.cc
#include "operator.h"
#include <iostream>
#include <vector>

Operator::Operator() {}

std::unique_ptr<Operator> new_operator() {
    return std::make_unique<Operator>();
}

DoraOnInputResult on_input(
    Operator& op,
    rust::Str id,
    rust::Slice<const uint8_t> data,
    OutputSender& output_sender)
{
    op.counter += 1;

    std::vector<unsigned char> out{op.counter};
    rust::Slice<const uint8_t> slice{out.data(), out.size()};
    auto send_result = send_output(output_sender, rust::Str("status"), slice);

    return DoraOnInputResult{send_result.error, false};
}

Dataflow YAML:

nodes:
  - id: runtime-node
    operators:
      - id: my-operator
        shared-library: build/my_operator
        inputs:
          data: some-node/output
        outputs:
          - status

---

Build Integration (CMake)

The recommended build approach uses CMake with the DoraTargets.cmake helper (see examples/cmake-dataflow/).

Project structure

my-project/
  CMakeLists.txt
  DoraTargets.cmake       # copied from examples/cmake-dataflow/
  node/main.cc
  operator/operator.h
  operator/operator.cc
  dataflow.yml

CMakeLists.txt

cmake_minimum_required(VERSION 3.21)
project(my-dataflow LANGUAGES C CXX)

set(CMAKE_CXX_STANDARD 20)
set(CMAKE_CXX_FLAGS "-fPIC")

include(DoraTargets.cmake)
link_directories(${dora_link_dirs})

# Standalone node (executable)
add_executable(my_node node/main.cc ${node_bridge})
add_dependencies(my_node Dora_cxx)
target_include_directories(my_node PRIVATE ${dora_cxx_include_dir})
target_link_libraries(my_node dora_node_api_cxx)

# Operator (shared library)
add_library(my_operator SHARED
    operator/operator.cc ${operator_bridge})
add_dependencies(my_operator Dora_cxx)
target_include_directories(my_operator PRIVATE
    ${dora_cxx_include_dir} ${dora_c_include_dir}
    ${CMAKE_CURRENT_SOURCE_DIR}/operator)
target_link_libraries(my_operator dora_operator_api_cxx)

install(TARGETS my_node DESTINATION ${CMAKE_CURRENT_SOURCE_DIR}/bin)
install(TARGETS my_operator DESTINATION ${CMAKE_CURRENT_SOURCE_DIR}/lib)

What DoraTargets.cmake provides

Variable	Description
dora_cxx_include_dir	Path to generated CXX headers (dora-node-api.h, dora-operator-api.h)
dora_c_include_dir	Path to C API headers (for mixed C/C++ projects)
dora_link_dirs	Library search path for libdora_node_api_cxx.a / libdora_operator_api_cxx.a
node_bridge	Generated CXX bridge source file for nodes (node_bridge.cc)
operator_bridge	Generated CXX bridge source file for operators (operator_bridge.cc)
Dora_cxx	CMake target dependency that builds the CXX crates

Build steps

# Option A: Build against local Dora source
mkdir build && cd build
cmake .. -DDORA_ROOT_DIR=/path/to/dora
cmake --build .

# Option B: Build against Dora from GitHub (cloned automatically)
mkdir build && cd build
cmake ..
cmake --build .

Requirements

• C++20 compiler
• Rust toolchain (for building the Dora static libraries via Cargo)
• CMake 3.21+
• For Arrow integration: Apache Arrow C++ library

---

CXX Bridge Notes

• All Rust opaque types (Events, OutputSender, DoraEvent, Metadata, MergedEvents, MergedDoraEvent) are accessed through rust::Box<T>.
• rust::String, rust::Vec<T>, and rust::Slice<const T> are CXX bridge types that interoperate with their C++ standard library counterparts. See the CXX type reference.
• Functions that return Result<T> in Rust throw C++ exceptions on the error path.
• Arrow FFI functions (event_as_arrow_input, send_arrow_output) are unsafe on the Rust side. The caller must pass valid pointers to ArrowArray / ArrowSchema structs cast to uint8_t*.
• The node library is a static archive (staticlib). Link it into your executable with -ldora_node_api_cxx.
• The operator library is also a static archive. Link it into your shared library with -ldora_operator_api_cxx.