Send metadata through TCP for now

apis/rust/node/src/daemon.rs

@@ -52,11 +52,16 @@ impl ControlChannel {
     pub fn prepare_message(
         &mut self,
         output_id: DataId,
-        len: usize,
+        metadata: dora_message::Metadata<'static>,
+        data_len: usize,
     ) -> eyre::Result<MessageSample> {
         tcp_send(
             &mut self.0,
-            &ControlRequest::PrepareOutputMessage { output_id, len },
+            &ControlRequest::PrepareOutputMessage {
+                output_id,
+                metadata,
+                data_len,
+            },
         )
         .wrap_err("failed to send PrepareOutputMessage request to dora-daemon")?;
         match tcp_receive(&mut self.0)

apis/rust/node/src/lib.rs

@@ -67,15 +67,11 @@ impl DoraNode {
         if !self.node_config.outputs.contains(&output_id) {
             eyre::bail!("unknown output");
         }
-        let metadata = Metadata::from_parameters(self.hlc.new_timestamp(), parameters);
-        let serialized_metadata = metadata
-            .serialize()
-            .with_context(|| format!("failed to serialize `{}` message", output_id))?;
-        let full_len = serialized_metadata.len() + data_len;
+        let metadata = Metadata::from_parameters(self.hlc.new_timestamp(), parameters.into_owned());
 
         let sample = self
             .control_channel
-            .prepare_message(output_id.clone(), full_len)
+            .prepare_message(output_id.clone(), metadata, data_len)
             .wrap_err("failed to prepare sample for output message")?;
 
         // map shared memory and fill in data
@@ -86,8 +82,7 @@ impl DoraNode {
                 .wrap_err("failed to open shared memory sample")?;
 
             let raw = unsafe { shared_memory.as_slice_mut() };
-            raw[..serialized_metadata.len()].copy_from_slice(&serialized_metadata);
-            data(&mut raw[serialized_metadata.len()..]);
+            data(raw);
         }
 
         self.control_channel

binaries/daemon/src/listener.rs

@@ -68,9 +68,15 @@ pub async fn handle_connection(mut connection: TcpStream, events_tx: mpsc::Sende
                 }
             }
             daemon_messages::ControlRequest::Stopped => DaemonNodeEvent::Stopped,
-            daemon_messages::ControlRequest::PrepareOutputMessage { output_id, len } => {
-                DaemonNodeEvent::PrepareOutputMessage { output_id, len }
-            }
+            daemon_messages::ControlRequest::PrepareOutputMessage {
+                output_id,
+                metadata,
+                data_len,
+            } => DaemonNodeEvent::PrepareOutputMessage {
+                output_id,
+                metadata,
+                data_len,
+            },
             daemon_messages::ControlRequest::SendOutMessage { id } => {
                 DaemonNodeEvent::SendOutMessage { id }
             }

binaries/daemon/src/main.rs

@@ -3,7 +3,6 @@ use dora_core::{
     daemon_messages::{self, ControlReply, DaemonCoordinatorEvent, DataflowId, SpawnDataflowNodes},
     topics::DORA_COORDINATOR_PORT_DEFAULT,
 };
-use dora_message::{uhlc, Metadata};
 use eyre::{bail, eyre, Context, ContextCompat};
 use futures_concurrency::stream::Merge;
 use shared_memory::{Shmem, ShmemConf};
@@ -44,8 +43,7 @@ async fn run() -> eyre::Result<()> {
 
 struct Daemon {
     port: u16,
-    hlc: uhlc::HLC,
-    uninit_shared_memory: HashMap<String, (DataId, Shmem)>,
+    uninit_shared_memory: HashMap<String, (DataId, dora_message::Metadata<'static>, Shmem)>,
     sent_out_shared_memory: HashMap<String, Shmem>,
 
     running: HashMap<DataflowId, RunningDataflow>,
@@ -74,7 +72,6 @@ impl Daemon {
 
         let daemon = Self {
             port,
-            hlc: uhlc::HLC::default(),
             uninit_shared_memory: Default::default(),
             sent_out_shared_memory: Default::default(),
             running: HashMap::new(),
@@ -172,14 +169,18 @@ impl Daemon {
                 };
                 let _ = reply_sender.send(ControlReply::Result(result));
             }
-            DaemonNodeEvent::PrepareOutputMessage { output_id, len } => {
+            DaemonNodeEvent::PrepareOutputMessage {
+                output_id,
+                metadata,
+                data_len,
+            } => {
                 let memory = ShmemConf::new()
-                    .size(len)
+                    .size(data_len)
                     .create()
                     .wrap_err("failed to allocate shared memory")?;
                 let id = memory.get_os_id().to_owned();
                 self.uninit_shared_memory
-                    .insert(id.clone(), (output_id, memory));
+                    .insert(id.clone(), (output_id, metadata, memory));
 
                 let reply = ControlReply::PreparedMessage {
                     shared_memory_id: id.clone(),
@@ -190,7 +191,7 @@ impl Daemon {
                 }
             }
             DaemonNodeEvent::SendOutMessage { id } => {
-                let (output_id, memory) = self
+                let (output_id, metadata, memory) = self
                     .uninit_shared_memory
                     .remove(&id)
                     .ok_or_else(|| eyre!("invalid shared memory id"))?;
@@ -214,7 +215,7 @@ impl Daemon {
                         if channel
                             .send_async(daemon_messages::NodeEvent::Input {
                                 id: input_id.clone(),
-                                metadata: Metadata::new(self.hlc.new_timestamp()), // TODO
+                                metadata: metadata.clone(),
                                 data: unsafe { daemon_messages::InputData::new(id.clone()) },
                             })
                             .await
@@ -272,7 +273,8 @@ pub enum Event {
 pub enum DaemonNodeEvent {
     PrepareOutputMessage {
         output_id: DataId,
-        len: usize,
+        metadata: dora_message::Metadata<'static>,
+        data_len: usize,
     },
     SendOutMessage {
         id: MessageId,

libraries/core/src/daemon_messages.rs

@@ -29,7 +29,8 @@ pub enum ControlRequest {
     },
     PrepareOutputMessage {
         output_id: DataId,
-        len: usize,
+        metadata: Metadata<'static>,
+        data_len: usize,
     },
     SendOutMessage {
         id: SharedMemoryId,

libraries/message/src/lib.rs

@@ -22,7 +22,7 @@ pub struct MetadataParameters<'a> {
 }
 
 impl MetadataParameters<'_> {
-    fn into_owned(self) -> MetadataParameters<'static> {
+    pub fn into_owned(self) -> MetadataParameters<'static> {
         MetadataParameters {
             open_telemetry_context: self.open_telemetry_context.into_owned().into(),
             ..self