Add an example C++ operator based on the C API

examples/c++-dataflow/dataflow.yml

@@ -27,3 +27,9 @@ nodes:
           counter_2: cxx-node-rust-api/counter
         outputs:
           - status
+      - id: operator-c-api
+        shared-library: build/operator_c_api.so
+        inputs:
+          op_status: runtime-node/operator-rust-api/status
+        outputs:
+          - half-status

examples/c++-dataflow/operator-c-api/operator.cc

@@ -0,0 +1,71 @@
+extern "C"
+{
+#include "../../../apis/c/operator/operator_api.h"
+}
+
+#include <memory>
+#include <iostream>
+#include <vector>
+
+class Operator
+{
+public:
+    Operator();
+};
+
+Operator::Operator() {}
+
+extern "C" int dora_init_operator(void **operator_context)
+{
+    Operator *op = std::make_unique<Operator>().release();
+    *operator_context = (void *)op;
+
+    return 0;
+}
+
+extern "C" void dora_drop_operator(void *operator_context)
+{
+    delete (Operator *)operator_context;
+}
+
+extern "C" int dora_on_input(
+    const char *id_start,
+    size_t id_len,
+    const char *data_start,
+    size_t data_len,
+    const int (*output_fn_raw)(const char *id_start,
+                               size_t id_len,
+                               const char *data_start,
+                               size_t data_len,
+                               const void *output_context),
+    void *output_context,
+    const void *operator_context)
+{
+
+    std::string id(id_start, id_len);
+
+    std::vector<unsigned char> data;
+    for (size_t i = 0; i < data_len; i++)
+    {
+        data.push_back(*(data_start + i));
+    }
+
+    std::cout
+        << "C++ Operator (C-API) received input `" << id << "` with data: [";
+    for (unsigned char &v : data)
+    {
+        std::cout << (unsigned int)v << ", ";
+    }
+    std::cout << "]" << std::endl;
+
+    char out = data[0] / 2;
+    std::string out_id = "half-status";
+    int result = output_fn_raw(&out_id[0], out_id.length(), &out, 1, output_context);
+    if (result != 0)
+    {
+        std::cerr << "failed to send output" << std::endl;
+        return 1;
+    }
+
+    return 0;
+}

examples/c++-dataflow/run.rs

@@ -7,7 +7,7 @@ async fn main() -> eyre::Result<()> {
     std::env::set_current_dir(root.join(file!()).parent().unwrap())
         .wrap_err("failed to set working dir")?;
 
-    tokio::fs::create_dir_all(root.join("build")).await?;
+    tokio::fs::create_dir_all("build").await?;
 
     build_package("cxx-dataflow-example-node-rust-api").await?;
     build_package("cxx-dataflow-example-operator-rust-api").await?;
@@ -19,6 +19,14 @@ async fn main() -> eyre::Result<()> {
         "node_c_api",
     )
     .await?;
+    build_cxx_operator(
+        &Path::new("operator-c-api")
+            .join("operator.cc")
+            .canonicalize()?,
+        "operator_c_api",
+    )
+    .await?;
+
     build_package("dora-runtime").await?;
 
     dora_coordinator::run(dora_coordinator::Command::Run {
@@ -58,3 +66,31 @@ async fn build_cxx_node(root: &Path, path: &Path, out_name: &str) -> eyre::Resul
     };
     Ok(())
 }
+
+async fn build_cxx_operator(path: &Path, out_name: &str) -> eyre::Result<()> {
+    let object_file_path = Path::new("../build").join(out_name).with_extension("o");
+
+    let mut compile = tokio::process::Command::new("clang++");
+    compile.arg("-c").arg(path);
+    compile.arg("-o").arg(&object_file_path);
+    compile.arg("-fPIC");
+    if let Some(parent) = path.parent() {
+        compile.current_dir(parent);
+    }
+    if !compile.status().await?.success() {
+        bail!("failed to compile cxx operator");
+    };
+
+    let mut link = tokio::process::Command::new("clang++");
+    link.arg("-shared").arg(&object_file_path);
+    link.arg("-o")
+        .arg(Path::new("../build").join(out_name).with_extension("so"));
+    if let Some(parent) = path.parent() {
+        link.current_dir(parent);
+    }
+    if !link.status().await?.success() {
+        bail!("failed to create shared library from cxx operator (c api)");
+    };
+
+    Ok(())
+}