# Node API Rust

> Rust Node API (dora-node-api)

- Skill: `zhanghandong/node-api-rust` (Agent Skill)
- Install (CLI): `npx skillmds@latest add zhanghandong/node-api-rust`
- Raw SKILL.md: https://api.skillmd.com/api/skills/zhanghandong/node-api-rust/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: ZhangHanDong (https://skillmd.com/u/zhanghandong)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/zhanghandong/node-api-rust

---


# Rust Node API (dora-node-api)

> Complete guide to building dora nodes in Rust

## Dependencies

```toml
[dependencies]
dora-node-api = "0.4"
eyre = "0.6"
```

## Basic Usage

```rust
use dora_node_api::{DoraNode, Event, MetadataParameters};
use dora_node_api::arrow::array::StringArray;

fn main() -> eyre::Result<()> {
    // Initialize node from environment (set by dora daemon)
    let (mut node, mut events) = DoraNode::init_from_env()?;

    // Process events
    while let Some(event) = events.recv() {
        match event {
            Event::Input { id, metadata, data } => {
                println!("Received input '{}' with {} bytes", id, data.len());

                // Send output
                let output_data = StringArray::from(vec!["hello"]);
                node.send_output(
                    "output_id".into(),
                    MetadataParameters::default(),
                    output_data,
                )?;
            }
            Event::InputClosed { id } => {
                println!("Input '{}' closed", id);
            }
            Event::Stop(cause) => {
                println!("Stopping: {:?}", cause);
                break;
            }
            Event::Error(e) => {
                eprintln!("Error: {}", e);
            }
            _ => {}
        }
    }

    Ok(())
}
```

## DoraNode Initialization

### Standard (Recommended)

```rust
// For nodes spawned by dora daemon
let (node, events) = DoraNode::init_from_env()?;
```

### Dynamic Nodes

```rust
use dora_node_api::dora_core::config::NodeId;

// For manually started nodes
let (node, events) = DoraNode::init_from_node_id(
    NodeId::from("my_node".to_string())
)?;
```

### Flexible

```rust
// Tries env first, falls back to dynamic
let (node, events) = DoraNode::init_flexible(
    NodeId::from("my_node".to_string())
)?;
```

### Interactive (Debugging)

```rust
// For terminal debugging
let (node, events) = DoraNode::init_interactive()?;
```

## Sending Outputs

### Arrow Array (Recommended)

```rust
use dora_node_api::arrow::array::*;

// Integer array
let data = Int32Array::from(vec![1, 2, 3]);
node.send_output("numbers".into(), MetadataParameters::default(), data)?;

// String array
let data = StringArray::from(vec!["hello", "world"]);
node.send_output("text".into(), MetadataParameters::default(), data)?;

// Float array
let data = Float64Array::from(vec![1.0, 2.5, 3.14]);
node.send_output("floats".into(), MetadataParameters::default(), data)?;
```

### Raw Bytes

```rust
let bytes = b"hello world";
node.send_output_bytes(
    "raw_data".into(),
    MetadataParameters::default(),
    bytes.len(),
    bytes,
)?;
```

### Zero-Copy (Large Data)

```rust
// For data > 4KB, uses shared memory automatically
node.send_output_raw(
    "large_data".into(),
    MetadataParameters::default(),
    data_len,
    |buffer| {
        buffer.copy_from_slice(&my_large_data);
    },
)?;
```

### Close Outputs Early

```rust
// Notify subscribers that output is finished
node.close_outputs(vec!["output_id".into()])?;
```

## Event Types

```rust
pub enum Event {
    // Input received from another node
    Input {
        id: DataId,           // Input ID as specified in YAML
        metadata: Metadata,   // Timestamp and type info
        data: ArrowData,      // Apache Arrow data
    },

    // Input was closed by sender
    InputClosed { id: DataId },

    // Node should stop
    Stop(StopCause),

    // Reload operator (runtime nodes only)
    Reload { operator_id: Option<OperatorId> },

    // Internal error
    Error(String),
}

pub enum StopCause {
    Manual,           // User issued stop command (dora stop / ctrl-c)
    AllInputsClosed,  // All inputs finished
}
```

## EventStream Methods

### Synchronous (Blocking)

```rust
// Block until next event
while let Some(event) = events.recv() { ... }

// With timeout
if let Some(event) = events.recv_timeout(Duration::from_secs(5)) { ... }
```

### Asynchronous

```rust
// Async receive
while let Some(event) = events.recv_async().await { ... }

// With timeout
let event = events.recv_async_timeout(Duration::from_secs(5)).await;
```

### Non-Blocking

```rust
match events.try_recv() {
    Ok(event) => { /* handle event */ }
    Err(TryRecvError::Empty) => { /* no event ready */ }
    Err(TryRecvError::Closed) => { /* stream closed */ }
}

// Drain all buffered events
if let Some(events) = events.drain() {
    for event in events { ... }
}
```

## Data Conversion

```rust
use dora_node_api::arrow::array::*;

// Convert to specific array type
if let Some(int_array) = data.as_any().downcast_ref::<Int32Array>() {
    for value in int_array.iter() {
        println!("{:?}", value);
    }
}

// For string data
if let Some(str_array) = data.as_any().downcast_ref::<StringArray>() {
    for s in str_array.iter() {
        println!("{}", s.unwrap_or(""));
    }
}

// For binary data
if let Some(bin_array) = data.as_any().downcast_ref::<BinaryArray>() {
    for bytes in bin_array.iter() {
        if let Some(b) = bytes {
            // Process bytes
        }
    }
}
```

## Node Information

```rust
let node_id = node.id();                    // Node ID
let dataflow_id = node.dataflow_id();       // Dataflow UUID
let config = node.node_config();            // Input/output config
let descriptor = node.dataflow_descriptor()?; // Full dataflow YAML
```

## Shared Memory

Data larger than `ZERO_COPY_THRESHOLD` (4096 bytes) automatically uses shared memory:

```rust
use dora_node_api::ZERO_COPY_THRESHOLD;

// Allocate data sample (auto uses shared memory for large data)
let mut sample = node.allocate_data_sample(large_data_len)?;
sample.copy_from_slice(&large_data);
```

## Complete Example: Image Processor

```rust
use dora_node_api::{DoraNode, Event, MetadataParameters};
use dora_node_api::arrow::array::{BinaryArray, StructArray};
use eyre::Result;

fn main() -> Result<()> {
    let (mut node, mut events) = DoraNode::init_from_env()?;

    while let Some(event) = events.recv() {
        match event {
            Event::Input { id, data, .. } => {
                if id.as_ref() == "image" {
                    // Process image data
                    if let Some(bin) = data.as_any().downcast_ref::<BinaryArray>() {
                        if let Some(image_bytes) = bin.value(0).into() {
                            // Process image...
                            let result = process_image(image_bytes);

                            // Send result
                            node.send_output(
                                "processed".into(),
                                MetadataParameters::default(),
                                result,
                            )?;
                        }
                    }
                }
            }
            Event::Stop(_) => break,
            _ => {}
        }
    }

    Ok(())
}
```

## Tracing (OpenTelemetry)

```rust
use dora_node_api::init_tracing;

fn main() -> eyre::Result<()> {
    let (node, events) = DoraNode::init_from_env()?;

    // Initialize tracing (requires tokio runtime)
    let _guard = init_tracing(node.id(), node.dataflow_id())?;

    // ... rest of node logic
}
```

## Related Skills

- **dataflow-config** - YAML configuration
- **operator-api** - Lightweight operators
- **integration-testing** - Testing nodes