Struct DoraNode 

pub struct DoraNode { /* private fields */ }

Allows sending outputs and retrieving node information.

The main purpose of this struct is to send outputs via Dora. There are also functions available for retrieving the node configuration.

Implementations

impl DoraNode

pub fn init_from_env() -> NodeResult<(Self, EventStream)>

Initiate a node from environment variables set by the Dora daemon or fall back to interactive mode.

This is the recommended initialization function for Dora nodes, which are spawned by Dora daemon instances. The daemon will set a DORA_NODE_CONFIG environment variable to configure the node.

When the node is started manually without the DORA_NODE_CONFIG environment variable set, the initialization will fall back to init_interactive if stdin is a terminal (detected through isatty).

If the DORA_NODE_CONFIG environment variable is not set and DORA_TEST_WITH_INPUTS is set, the node will be initialized in integration test mode. See the integration testing module for details.

This function will also initialize the node in integration test mode when the setup_integration_testing function was called before. This takes precedence over all environment variables.

use dora_node_api::DoraNode;

let (mut node, mut events) = DoraNode::init_from_env().expect("Could not init node.");

pub fn init_from_env_force() -> NodeResult<(Self, EventStream)>

Initialize the node from environment variables set by the Dora daemon; error if not set.

This function behaves the same as init_from_env, but it does not fall back to init_interactive. Instead, an error is returned when the DORA_NODE_CONFIG environment variable is missing.

pub fn builder() -> DoraNodeBuilder

Create a builder for configuring a node connection.

Setting a node_id selects the dynamic-node path; without one, build() falls back to init_from_env. Use this builder when you need a custom daemon port — the other init functions cover the common cases. Source-compatible with upstream dora 0.5.x: .dynamic() is accepted (no-op) so code written against upstream still compiles.

use dora_node_api::DoraNode;
use dora_node_api::dora_core::config::NodeId;

let (mut node, mut events) = DoraNode::builder()
    .node_id(NodeId::from("plot".to_string()))
    .daemon_port(6789)
    .build()
    .expect("Could not init node");

pub fn init_from_node_id(node_id: NodeId) -> NodeResult<(Self, EventStream)>

Initiate a node from a dataflow id and a node id.

This initialization function should be used for dynamic nodes.

use dora_node_api::DoraNode;
use dora_node_api::dora_core::config::NodeId;

let (mut node, mut events) = DoraNode::init_from_node_id(NodeId::from("plot".to_string())).expect("Could not init node plot");

pub fn init_flexible(node_id: NodeId) -> NodeResult<(Self, EventStream)>

Dynamic initialization function for nodes that are sometimes used as dynamic nodes.

This function first tries initializing the traditional way through init_from_env. If this fails, it falls back to init_from_node_id.

pub fn init_interactive() -> NodeResult<(Self, EventStream)>

Initialize the node in a standalone mode that prompts for inputs on the terminal.

Instead of connecting to a dora daemon, this interactive mode prompts for node inputs on the terminal. In this mode, the node is completely isolated from the dora daemon and other nodes, so it cannot be part of a dataflow.

Note that this function will hang indefinitely if no input is supplied to the interactive prompt. So it should be only used through a terminal.

Because of the above limitations, it is not recommended to use this function directly. Use init_from_env instead, which supports both normal daemon connections and manual interactive runs.

Example

Run any node that uses init_interactive or init_from_env directly from a terminal. The node will then start in “interactive mode” and prompt you for the next input:

> cargo build -p rust-dataflow-example-node
> target/debug/rust-dataflow-example-node
hello
Starting node in interactive mode as DORA_NODE_CONFIG env variable is not set
Node asks for next input
? Input ID
[empty input ID to stop]

The rust-dataflow-example-node expects a tick input, so let’s set the input ID to tick. Tick messages don’t have any data, so we leave the “Data” empty when prompted:

Node asks for next input
> Input ID tick
> Data
tick 0, sending 0x943ed1be20c711a4
node sends output random with data: PrimitiveArray<UInt64>
[
  10682205980693303716,
]
Node asks for next input
? Input ID
[empty input ID to stop]

We see that both the stdout output of the node and also the output messages that it sends are printed to the terminal. Then we get another prompt for the next input.

If you want to send an input with data, you can either send it as text (for string data) or as a JSON object (for struct, string, or array data). Other data types are not supported currently.

Empty input IDs are interpreted as stop instructions:

> Input ID
given input ID is empty -> stopping
Received stop
Node asks for next input
event channel was stopped -> returning empty event list
node reports EventStreamDropped
node reports closed outputs []
node reports OutputsDone

In addition to the node output, we see log messages for the different events that the node reports. After OutputsDone, the node should exit.

JSON data

In addition to text input, the Data prompt also supports JSON objects, which will be converted to Apache Arrow struct arrays:

Node asks for next input
> Input ID some_input
> Data { "field_1": 42, "field_2": { "inner": "foo" } }

This JSON data is converted to the following Arrow array:

StructArray
-- validity: [valid, ]
[
  -- child 0: "field_1" (Int64)
     PrimitiveArray<Int64>
     [42,]
  -- child 1: "field_2" (Struct([Field { name: "inner", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]))
     StructArray
     -- validity: [valid,]
     [
       -- child 0: "inner" (Utf8)
       StringArray
       ["foo",]
     ]
]

pub fn init_testing( input: TestingInput, output: TestingOutput, options: TestingOptions, ) -> NodeResult<(Self, EventStream)>

Initializes a node in integration test mode.

No connection to a dora daemon is made in this mode. Instead, inputs are read from the specified TestingInput, and outputs are written to the specified TestingOutput. Additional options for the testing mode can be specified through TestingOptions.

It is recommended to use this function only within test functions.

pub fn validate_output(&mut self, output_id: &DataId) -> bool

Check whether output_id is declared as an output of this node.

Returns true if the output is declared (or this node is interactive, which has no static output declaration); false and emits a one-time warning if the output is unknown. Public so callers building higher-level send helpers (e.g. the Python send_output_raw zero-copy path) can validate before allocating a buffer.

pub fn send_output_raw<F>( &mut self, output_id: DataId, parameters: MetadataParameters, data_len: usize, data: F, ) -> NodeResult<()>

where F: FnOnce(&mut [u8]),

Send raw data from the node to the other nodes.

We take a closure as an input to enable zero copy on send.

use dora_node_api::{DoraNode, MetadataParameters};
use dora_core::config::DataId;

let (mut node, mut events) = DoraNode::init_from_env().expect("Could not init node.");

let output = DataId::from("output_id".to_owned());

let data: &[u8] = &[0, 1, 2, 3];
let parameters = MetadataParameters::default();

node.send_output_raw(
   output,
   parameters,
   data.len(),
   |out| {
        out.copy_from_slice(data);
    }).expect("Could not send output");

Ignores the output if the given output_id is not specified as node output in the dataflow configuration file.

pub fn send_output( &mut self, output_id: DataId, parameters: MetadataParameters, data: impl Array, ) -> NodeResult<()>

Sends the give Arrow array as an output message.

Uses shared memory for efficient data transfer if suitable.

This method might copy the message once to move it to shared memory.

Ignores the output if the given output_id is not specified as node output in the dataflow configuration file.

pub fn send_output_bytes( &mut self, output_id: DataId, parameters: MetadataParameters, data_len: usize, data: &[u8], ) -> NodeResult<()>

Send the given raw byte data as output.

Might copy the data once to move it into shared memory.

Ignores the output if the given output_id is not specified as node output in the dataflow configuration file.

pub fn send_output_sample( &mut self, output_id: DataId, parameters: MetadataParameters, sample: Option<DataSample>, ) -> NodeResult<()>

Sends the given DataSample as output.

The sample must already be a self-describing Arrow IPC stream (the FRAMING_ARROW_IPC parameter should be set). It is recommended to use a function like send_output instead, which handles the encoding.

Ignores the output if the given output_id is not specified as node output in the dataflow configuration file.

pub fn close_outputs(&mut self, outputs_ids: Vec<DataId>) -> NodeResult<()>

Report the given outputs IDs as closed.

The node is not allowed to send more outputs with the closed IDs.

Closing outputs early can be helpful to receivers.

pub fn id(&self) -> &NodeId

Returns the ID of the node as specified in the dataflow configuration file.

pub fn dataflow_id(&self) -> &DataflowId

Returns the unique identifier for the running dataflow instance.

Dora assigns each dataflow instance a random identifier when started.

pub fn node_config(&self) -> &NodeRunConfig

Returns the input and output configuration of this node.

pub fn zero_copy_threshold(&self) -> usize

Returns the zero-copy SHM threshold in bytes.

Outputs whose raw payload is at least this many bytes are published via zenoh shared memory (zero-copy for local subscribers); smaller outputs are published via zenoh with a heap-buffered put. Configured via the DORA_ZERO_COPY_THRESHOLD env var, defaulting to ZERO_COPY_THRESHOLD.

pub fn is_restart(&self) -> bool

Returns true if this node was restarted after a previous exit or failure.

Nodes can use this to decide whether to restore saved state or start fresh.

pub fn restart_count(&self) -> u32

Returns how many times this node has been restarted.

Returns 0 on the first run, 1 after the first restart, etc.

pub fn timestamp(&self) -> Timestamp

Returns the current timestamp from the node’s Hybrid Logical Clock.

This generates a new HLC timestamp, which combines the physical wall-clock time with a logical counter to ensure uniqueness and monotonicity even across nodes. The HLC is the same clock dora stamps every outgoing message with, so this is the right value to subtract from an input event’s metadata.timestamp when measuring per-event processing latency — using std::time::SystemTime::now() instead would mix two unrelated clocks and give meaningless results across daemons.

pub fn log(&self, level: &str, message: &str, target: Option<&str>)

Send a structured log message.

Outputs a JSONL line to stdout that the daemon parses automatically. Works with min_log_level filtering and send_logs_as routing.

level should be one of: "error", "warn", "info", "debug", "trace". Unknown levels default to "info".

pub fn log_with_fields( &self, level: &str, message: &str, target: Option<&str>, fields: Option<&BTreeMap<String, String>>, )

Send a structured log message with optional key-value fields.

Like log, but accepts additional structured fields that are included in the JSON payload and preserved through send_logs_as.

pub fn log_error(&self, message: &str)

Log an error message.

pub fn log_warn(&self, message: &str)

Log a warning message.

pub fn log_info(&self, message: &str)

Log an info message.

pub fn log_debug(&self, message: &str)

Log a debug message.

pub fn log_trace(&self, message: &str)

Log a trace message.

pub fn new_request_id() -> String

Generate a new unique request/goal ID (UUID v7, time-ordered).

Uses a per-thread monotonic counter context to guarantee uniqueness even when multiple IDs are generated within the same clock tick.

pub fn new_goal_id() -> String

Generate a new unique goal ID (UUID v7, time-ordered).

This is an alias for new_request_id that reads more naturally in action (goal/feedback/result) contexts.

pub fn send_service_request( &mut self, output_id: DataId, parameters: MetadataParameters, data: impl Array, ) -> NodeResult<String>

Send a service request, automatically injecting a request_id into the metadata parameters. Returns the generated request ID.

Any existing request_id key in parameters is replaced.

pub fn send_service_response( &mut self, output_id: DataId, parameters: MetadataParameters, data: impl Array, ) -> NodeResult<()>

Send a service response. This is a semantic alias for send_output.

The caller is expected to pass through the request_id parameter from the incoming request’s metadata.

pub fn send_stream_chunk( &mut self, output_id: DataId, segment: &mut StreamSegment, fin: bool, data: impl Array, ) -> NodeResult<()>

Send a streaming segment chunk. Convenience wrapper around send_output that builds metadata from the StreamSegment builder.

pub fn allocate_data_sample( &mut self, data_len: usize, ) -> NodeResult<DataSample>

Allocates a DataSample of the specified size.

For payloads at or above the zero-copy threshold the buffer is allocated directly from the zenoh SHM provider (when available), so the producer writes straight into shared memory and publishing moves the buffer into zenoh’s put without a further copy. Smaller payloads — or the case where no SHM provider exists (interactive/testing mode) — use a heap-allocated, 128-byte-aligned buffer; the SHM provider is page-aligned, so dedicating a full page to a small message is pure waste.

pub fn dataflow_descriptor(&self) -> NodeResult<&Descriptor>

Returns the full dataflow descriptor that this node is part of.

This method returns the parsed dataflow YAML file.

pub fn register_pinned_memory( &mut self, shared_memory_id: String, metadata: Metadata, ) -> Result<(), Error>

Register a pinned memory pool with the daemon for lifecycle tracking.

Send the memory pool metadata to the daemon so it can track the pool and provide it to other nodes for zero-copy access.

pub fn read_pinned_memory( &mut self, shared_memory_id: String, free: bool, ) -> Result<Metadata, Error>

Read pinned memory metadata from the daemon.

When free is true, the daemon also frees the pool after reading.

pub fn free_pinned_memory( &mut self, shared_memory_id: String, ) -> Result<(), Error>

Free a pinned memory pool via the daemon.

Trait Implementations

impl Drop for DoraNode

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.