From af9afc53827070cfe133ebefe9356ff9577ee283 Mon Sep 17 00:00:00 2001
From: Philipp Oppermann <dev@phil-opp.com>
Date: Wed, 25 May 2022 11:51:54 +0200
Subject: [PATCH] Create an example operator that receives inputs

---
 Cargo.lock                                    |  4 ++
 Cargo.toml                                    |  8 +++-
 .../examples/example-operator/Cargo.toml      | 11 +++++
 .../examples/example-operator/src/lib.rs      | 44 +++++++++++++++++++
 coordinator/examples/mini-dataflow.yml        |  8 ++++
 5 files changed, 74 insertions(+), 1 deletion(-)
 create mode 100644 coordinator/examples/example-operator/Cargo.toml
 create mode 100644 coordinator/examples/example-operator/src/lib.rs

diff --git a/Cargo.lock b/Cargo.lock
index 4ce5a5bf..c4502542 100644
--- a/Cargo.lock
+++ b/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"
diff --git a/Cargo.toml b/Cargo.toml
index bda89d42..0dd0520e 100644
--- a/Cargo.toml
+++ b/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"
diff --git a/coordinator/examples/example-operator/Cargo.toml b/coordinator/examples/example-operator/Cargo.toml
new file mode 100644
index 00000000..5d4ee761
--- /dev/null
+++ b/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]
diff --git a/coordinator/examples/example-operator/src/lib.rs b/coordinator/examples/example-operator/src/lib.rs
new file mode 100644
index 00000000..551ef4c9
--- /dev/null
+++ b/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(())
+}
diff --git a/coordinator/examples/mini-dataflow.yml b/coordinator/examples/mini-dataflow.yml
index 0c523d1a..e92e608c 100644
--- a/coordinator/examples/mini-dataflow.yml
+++ b/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