Ack/Nack for routing in transform_processor

rust/otap-dataflow/crates/engine/src/control.rs

@@ -20,7 +20,7 @@ use std::time::{Duration, Instant};
 /// A 8-byte context value. Supports conversion to and from plain data
 /// using bytemuck.
 #[repr(transparent)]
-#[derive(Clone, Copy, Debug)]
+#[derive(Clone, Copy, Debug, PartialEq)]
 pub struct Context8u8([u8; 8]);
 
 impl<T: Pod> From<T> for Context8u8 {

rust/otap-dataflow/crates/otap/src/accessory/slots.rs

@@ -81,6 +81,12 @@ impl<UData> State<UData> {
         Some((key, ures))
     }
 
+    /// Get immutable reference to user data in a slot (if key is valid).
+    #[must_use]
+    pub fn get(&self, key: Key) -> Option<&UData> {
+        self.slots.get(key)
+    }
+
     /// Modify user data in a slot (if key is valid).
     #[must_use]
     pub fn get_mut(&mut self, key: Key) -> Option<&mut UData> {

rust/otap-dataflow/crates/otap/src/pdata.rs

@@ -23,7 +23,7 @@ use otap_df_engine::{
 use otap_df_pdata::OtapPayload;
 
 /// Context for OTAP requests
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug, Default, PartialEq)]
 pub struct Context {
     stack: Vec<Frame>,
 }
@@ -120,7 +120,7 @@ impl Context {
 }
 
 /// Per-node interests, context, and identity.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct Frame {
     /// Declares the set of interests this node has (Acks, Nacks, ...)
     pub interests: Interests,