Merge branch 'fix-reduce' of github.com:numaproj/numaflow-rs into fix…

…-reduce

examples/simple-source/src/main.rs

@@ -13,7 +13,7 @@ pub(crate) mod simple_source {
         sync::RwLock,
     };
 
-    use numaflow::source::{Message, Offset, Sourcer, SourceReadRequest};
+    use numaflow::source::{Message, Offset, SourceReadRequest, Sourcer};
     use tokio::{sync::mpsc::Sender, time::Instant};
     use tonic::async_trait;
 

src/reduce.rs

@@ -350,8 +350,8 @@ where
 /// The `Task` struct represents a task in the reduce service.
 /// It is responsible for invoking the user's reducer and sending the response back to the client.
 struct Task {
-    tx: Option<Sender<ReduceRequest>>,
-    handle: Option<tokio::task::JoinHandle<()>>,
+    tx: Sender<ReduceRequest>,
+    handle: tokio::task::JoinHandle<()>,
 }
 
 impl Task {
@@ -360,13 +360,13 @@ impl Task {
     async fn new<R: Reducer + Send + Sync + 'static>(
         reducer: R,
         keys: Vec<String>,
-        md: Arc<Metadata>,
+        md: Metadata,
         response_tx: Sender<Result<proto::ReduceResponse, Status>>,
     ) -> Self {
         let (tx, rx) = channel::<ReduceRequest>(1);
 
         let handle = tokio::spawn(async move {
-            let messages = reducer.reduce(keys, rx, md.as_ref()).await;
+            let messages = reducer.reduce(keys, rx, &md).await;
             for message in messages {
                 response_tx
                     .send(Ok(proto::ReduceResponse {
@@ -388,25 +388,25 @@ impl Task {
         });
 
         Self {
-            tx: Some(tx),
-            handle: Some(handle),
+            tx: tx,
+            handle: handle,
         }
     }
 
     /// Sends a `ReduceRequest` to the task.
     async fn send(&self, rr: ReduceRequest) {
-        if let Some(ref tx) = self.tx {
-            tx.send(rr).await.expect("send failed, receiver dropped")
-        }
+        self.tx
+            .send(rr)
+            .await
+            .expect("send failed, receiver dropped")
     }
 
     /// Closes the task and waits for it to finish.
-    async fn close(&mut self) {
+    async fn close(self) {
         // drop the sender to close the task and wait for the task to finish
-        drop(self.tx.take());
-        if let Some(handle) = self.handle.take() {
-            handle.await.unwrap();
-        }
+        drop(self.tx);
+
+        self.handle.await.unwrap();
     }
 }
 
@@ -470,7 +470,7 @@ where
         self.window = Some(IntervalWindow::new(start_time, end_time));
 
         // Create Metadata with the extracted start and end time
-        let md = Arc::new(Metadata::new(IntervalWindow::new(start_time, end_time)));
+        let md = Metadata::new(IntervalWindow::new(start_time, end_time));
 
         // Create a new Task and add it to the TaskManager
         let task = Task::new(reducer, keys.clone(), md, self.response_stream.clone()).await;
@@ -530,7 +530,7 @@ where
     /// Closes all tasks in the task manager and sends an EOF message to the response stream.
     // Method to close all tasks
     async fn close_all_tasks(&mut self) {
-        for task in self.tasks.values_mut() {
+        for task in self.tasks..values_mut() {
             task.close().await;
         }