use read batch size as the channel size

Signed-off-by: Yashash H L <[email protected]>

src/sink.rs

@@ -19,7 +19,6 @@ const DEFAULT_FB_SERVER_INFO_FILE: &str = "/var/run/numaflow/fb-sinker-server-in
 const ENV_UD_CONTAINER_TYPE: &str = "NUMAFLOW_UD_CONTAINER_TYPE";
 const UD_CONTAINER_FB_SINK: &str = "fb-udsink";
 const DEFAULT_CHANNEL_SIZE: usize = 1000;
-
 /// Numaflow Sink Proto definitions.
 pub mod proto {
     tonic::include_proto!("sink.v1");
@@ -186,7 +185,7 @@ where
         let sink_handle = self.handler.clone();
         let cancellation_token = self.cancellation_token.clone();
         let shutdown_tx = self.shutdown_tx.clone();
-        // TODO: what should be the idle buffer size?
+        // FIXME: we should be using the batch size as the channel size
         let (tx, rx) = mpsc::channel::<SinkRequest>(DEFAULT_CHANNEL_SIZE);
 
         let reader_shutdown_tx = shutdown_tx.clone();

src/source.rs

@@ -15,7 +15,6 @@ use tonic::{async_trait, Request, Response, Status};
 const DEFAULT_MAX_MESSAGE_SIZE: usize = 64 * 1024 * 1024;
 const DEFAULT_SOCK_ADDR: &str = "/var/run/numaflow/source.sock";
 const DEFAULT_SERVER_INFO_FILE: &str = "/var/run/numaflow/sourcer-server-info";
-const DEFAULT_CHANNEL_SIZE: usize = 1000;
 
 /// Source Proto definitions.
 pub mod proto {
@@ -88,9 +87,10 @@ where
         let sr = request.into_inner().request.unwrap();
 
         // tx,rx pair for sending data over to user-defined source
-        let (stx, mut srx) = mpsc::channel::<Message>(DEFAULT_CHANNEL_SIZE);
+        let (stx, mut srx) = mpsc::channel::<Message>(sr.num_records as usize);
         // tx,rx pair for gRPC response
-        let (tx, rx) = mpsc::channel::<Result<proto::ReadResponse, Status>>(DEFAULT_CHANNEL_SIZE);
+        let (tx, rx) =
+            mpsc::channel::<Result<proto::ReadResponse, Status>>(sr.num_records as usize);
 
         // start the ud-source rx asynchronously and start populating the gRPC response, so it can be streamed to the gRPC client (numaflow).
         tokio::spawn(async move {