Update C++ API for new daemon design

Cargo.lock

@@ -765,9 +765,9 @@ checksum = "b365fabc795046672053e29c954733ec3b05e4be654ab130fe8f1f94d7051f35"
 
 [[package]]
 name = "cxx"
-version = "1.0.73"
+version = "1.0.85"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "873c2e83af70859af2aaecd1f5d862f3790b747b1f4f50fb45a931d000ac0422"
+checksum = "5add3fc1717409d029b20c5b6903fc0c0b02fa6741d820054f4a2efa5e5816fd"
 dependencies = [
  "cc",
  "cxxbridge-flags",
@@ -792,15 +792,15 @@ dependencies = [
 
 [[package]]
 name = "cxxbridge-flags"
-version = "1.0.73"
+version = "1.0.85"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f46b787c15af80277db5c88c6ac6c502ae545e622f010e06f95e540d34931acf"
+checksum = "69a3e162fde4e594ed2b07d0f83c6c67b745e7f28ce58c6df5e6b6bef99dfb59"
 
 [[package]]
 name = "cxxbridge-macro"
-version = "1.0.73"
+version = "1.0.85"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ba3f3a7efa46626878fb5d324fabca4d19d2956b6ae97ce43044ef4515f5abc"
+checksum = "3e7e2adeb6a0d4a282e581096b06e1791532b7d576dcde5ccd9382acf55db8e6"
 dependencies = [
  "proc-macro2",
  "quote",

apis/c++/node/src/lib.rs

@@ -1,14 +1,27 @@
-use dora_node_api::{self, Input, Receiver};
+use dora_node_api::{
+    self,
+    daemon::{Event, EventStream},
+};
+use eyre::bail;
 
 #[cxx::bridge]
+#[allow(clippy::needless_lifetimes)]
 mod ffi {
     struct DoraNode {
-        inputs: Box<Inputs>,
+        events: Box<Events>,
         send_output: Box<OutputSender>,
     }
 
+    pub enum DoraEventType {
+        Stop,
+        Input,
+        InputClosed,
+        Error,
+        Unknown,
+        AllInputsClosed,
+    }
+
     struct DoraInput {
-        end_of_input: bool,
         id: String,
         data: Vec<u8>,
     }
@@ -18,12 +31,16 @@ mod ffi {
     }
 
     extern "Rust" {
-        type Inputs;
+        type Events;
         type OutputSender;
+        type DoraEvent<'a>;
 
         fn init_dora_node() -> Result<DoraNode>;
         fn free_dora_node(node: DoraNode);
-        fn next_input(inputs: &mut Box<Inputs>) -> DoraInput;
+
+        fn next_event(inputs: &mut Box<Events>) -> Box<DoraEvent<'_>>;
+        fn event_type(event: &Box<DoraEvent>) -> DoraEventType;
+        fn event_as_input(event: Box<DoraEvent>) -> Result<DoraInput>;
         fn send_output(
             output_sender: &mut Box<OutputSender>,
             id: String,
@@ -33,13 +50,12 @@ mod ffi {
 }
 
 fn init_dora_node() -> eyre::Result<ffi::DoraNode> {
-    let mut node = dora_node_api::DoraNode::init_from_env()?;
-    let input_stream = node.inputs()?;
-    let inputs = Inputs(input_stream);
+    let (node, events) = dora_node_api::DoraNode::init_from_env()?;
+    let inputs = Events(events);
     let send_output = OutputSender(node);
 
     Ok(ffi::DoraNode {
-        inputs: Box::new(inputs),
+        events: Box::new(inputs),
         send_output: Box::new(send_output),
     })
 }
@@ -48,33 +64,43 @@ fn free_dora_node(node: ffi::DoraNode) {
     let _ = node;
 }
 
-pub struct Inputs(Receiver<Input>);
-
-fn next_input(inputs: &mut Box<Inputs>) -> ffi::DoraInput {
-    match inputs.0.recv() {
-        Ok(input) => {
-            let id = input.id.clone().into();
-            let data = input.data();
-            ffi::DoraInput {
-                end_of_input: false,
-                id,
-                data: data.into_owned(),
-            }
-        }
-        Err(_) => ffi::DoraInput {
-            end_of_input: true,
-            id: String::new(),
-            data: Vec::new(),
+pub struct Events(EventStream);
+
+fn next_event(events: &mut Box<Events>) -> Box<DoraEvent> {
+    Box::new(DoraEvent(events.0.recv()))
+}
+
+pub struct DoraEvent<'a>(Option<Event<'a>>);
+
+fn event_type(event: &Box<DoraEvent>) -> ffi::DoraEventType {
+    match &event.0 {
+        Some(event) => match event {
+            Event::Stop => ffi::DoraEventType::Stop,
+            Event::Input { .. } => ffi::DoraEventType::Input,
+            Event::InputClosed { .. } => ffi::DoraEventType::InputClosed,
+            Event::Error(_) => ffi::DoraEventType::Error,
+            _ => ffi::DoraEventType::Unknown,
         },
+        None => ffi::DoraEventType::AllInputsClosed,
     }
 }
 
+fn event_as_input(event: Box<DoraEvent>) -> eyre::Result<ffi::DoraInput> {
+    let Some(Event::Input { id, metadata: _, data }) = event.0 else {
+        bail!("not an input event");
+    };
+    Ok(ffi::DoraInput {
+        id: id.into(),
+        data: data.map(|d| d.to_owned()).unwrap_or_default(),
+    })
+}
+
 pub struct OutputSender(dora_node_api::DoraNode);
 
 fn send_output(sender: &mut Box<OutputSender>, id: String, data: &[u8]) -> ffi::DoraResult {
     let result = sender
         .0
-        .send_output(&id.into(), Default::default(), data.len(), |out| {
+        .send_output(id.into(), Default::default(), data.len(), |out| {
             out.copy_from_slice(data)
         });
     let error = match result {