remove todo, avoid clone

Signed-off-by: Yashash H L <[email protected]>
numaproj · Jan 15, 2025 · fd44a6b · fd44a6b
1 parent 9431405
commit fd44a6b
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 9 deletions.
diff --git a/rust/numaflow-core/src/metrics.rs b/rust/numaflow-core/src/metrics.rs
@@ -718,7 +718,6 @@ struct TimestampedPending {
 #[derive(Clone)]
 pub(crate) enum LagReader {
     Source(Source),
-    // TODO: Arc<[T]>
     ISB(Vec<JetstreamReader>), // multiple partitions
 }
 

diff --git a/rust/serving/src/app/callback.rs b/rust/serving/src/app/callback.rs
@@ -14,22 +14,21 @@ pub(crate) mod store;
 
 /// As message passes through each component (map, transformer, sink, etc.). it emits a beacon via callback
 /// to inform that message has been processed by this component.
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize)]
 pub(crate) struct Callback {
     pub(crate) id: String,
     pub(crate) vertex: String,
     pub(crate) cb_time: u64,
     pub(crate) from_vertex: String,
     /// Due to flat-map operation, we can have 0 or more responses.
-    // TODO: Arc<[T]>
     pub(crate) responses: Vec<Response>,
 }
 
 /// It contains details about the `To` vertex via tags (conditional forwarding).
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize)]
 pub(crate) struct Response {
     /// If tags is None, the message is forwarded to all vertices, if len(Vec) == 0, it means that
-    /// the message has been dropped. 
+    /// the message has been dropped.
     pub(crate) tags: Option<Vec<String>>,
 }
 

diff --git a/rust/serving/src/app/jetstream_proxy.rs b/rust/serving/src/app/jetstream_proxy.rs
@@ -415,10 +415,10 @@ mod tests {
         let app = jetstream_proxy(app_state).await.unwrap();
 
         tokio::spawn(async move {
-            let cbs = create_default_callbacks(ID_VALUE);
             let mut retries = 0;
             loop {
-                match callback_state.insert_callback_requests(cbs.clone()).await {
+                let cbs = create_default_callbacks(ID_VALUE);
+                match callback_state.insert_callback_requests(cbs).await {
                     Ok(_) => break,
                     Err(e) => {
                         retries += 1;
@@ -487,14 +487,14 @@ mod tests {
         let app = jetstream_proxy(app_state).await.unwrap();
 
         // pipeline is in -> cat -> out, so we will have 3 callback requests
-        let cbs = create_default_callbacks(ID_VALUE);
 
         // spawn a tokio task which will insert the callback requests to the callback state
         // if it fails, sleep for 10ms and retry
         tokio::spawn(async move {
             let mut retries = 0;
             loop {
-                match callback_state.insert_callback_requests(cbs.clone()).await {
+                let cbs = create_default_callbacks(ID_VALUE);
+                match callback_state.insert_callback_requests(cbs).await {
                     Ok(_) => {
                         // save a test message, we should get this message when serve is invoked
                         // with foobar id