Create an example operator that receives inputs

Cargo.lock

@@ -662,6 +662,10 @@ version = "2.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77f3309417938f28bf8228fcff79a4a37103981e3e186d2ccd19c74b38f4eb71"
 
+[[package]]
+name = "example-operator"
+version = "0.1.0"
+
 [[package]]
 name = "eyre"
 version = "0.6.8"

Cargo.toml

@@ -6,7 +6,13 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [workspace]
-members = ["api", "coordinator", "common", "runtime"]
+members = [
+    "api",
+    "coordinator",
+    "common",
+    "runtime",
+    "coordinator/examples/example-operator",
+]
 
 [dependencies]
 eyre = "0.6.7"

coordinator/examples/example-operator/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "example-operator"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]

coordinator/examples/example-operator/src/lib.rs

@@ -0,0 +1,44 @@
+#![warn(unsafe_op_in_unsafe_fn)]
+
+use std::{ffi::c_void, slice};
+
+type OutputFn = unsafe extern "C" fn(
+    id_start: *const u8,
+    id_len: usize,
+    data_start: *const u8,
+    data_len: usize,
+    output_context: *const c_void,
+) -> isize;
+
+#[no_mangle]
+pub unsafe extern "C" fn dora_on_input(
+    id_start: *const u8,
+    id_len: usize,
+    data_start: *const u8,
+    data_len: usize,
+    output: OutputFn,
+    output_context: *const c_void,
+) -> isize {
+    let id = match std::str::from_utf8(unsafe { slice::from_raw_parts(id_start, id_len) }) {
+        Ok(id) => id,
+        Err(_) => return -1,
+    };
+    let data = unsafe { slice::from_raw_parts(data_start, data_len) };
+    match on_input(id, data, output, output_context) {
+        Ok(()) => 0,
+        Err(_) => -1,
+    }
+}
+
+pub fn on_input(
+    id: &str,
+    data: &[u8],
+    output: OutputFn,
+    output_context: *const c_void,
+) -> Result<(), ()> {
+    println!(
+        "operator got input `{id}` with value: {}",
+        String::from_utf8_lossy(data)
+    );
+    Ok(())
+}

coordinator/examples/mini-dataflow.yml

@@ -30,3 +30,11 @@ nodes:
       inputs:
         random: random/number
         time: timer/time
+
+  - id: runtime-node
+    operators:
+      - id: op-1
+        shared-library: ../target/debug/libexample_operator.so
+        inputs:
+          random: random/number
+          time: timer/time