Added header values

proto/map.proto

@@ -21,6 +21,8 @@ message MapRequest {
   bytes value = 2;
   google.protobuf.Timestamp event_time = 3;
   google.protobuf.Timestamp watermark = 4;
+  map<string, string> headers = 5;
+
 }
 
 /**

proto/reduce.proto

@@ -22,6 +22,7 @@ message ReduceRequest {
   bytes value = 2;
   google.protobuf.Timestamp event_time = 3;
   google.protobuf.Timestamp watermark = 4;
+  map<string, string> headers = 5;
 }
 
 /**

proto/sourcetransform.proto

@@ -23,6 +23,7 @@ message SourceTransformRequest {
   bytes value = 2;
   google.protobuf.Timestamp event_time = 3;
   google.protobuf.Timestamp watermark = 4;
+  map<string, string> headers = 5;
 }
 
 /**

src/map.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::future::Future;
 use std::path::PathBuf;
 
@@ -87,6 +88,7 @@ pub struct Message {
     pub value: Vec<u8>,
     /// Tags are used for [conditional forwarding](https://numaflow.numaproj.io/user-guide/reference/conditional-forwarding/).
     pub tags: Vec<String>,
+
 }
 
 impl From<Message> for proto::map_response::Result {
@@ -109,6 +111,9 @@ pub struct MapRequest {
     pub watermark: DateTime<Utc>,
     /// Time of the element as seen at source or aligned after a reduce operation.
     pub eventtime: DateTime<Utc>,
+    /// Headers for the message.
+    pub headers: HashMap<String, String>,
+
 }
 
 impl From<proto::MapRequest> for MapRequest {
@@ -118,6 +123,7 @@ impl From<proto::MapRequest> for MapRequest {
             value: value.value,
             watermark: shared::utc_from_timestamp(value.watermark),
             eventtime: shared::utc_from_timestamp(value.event_time),
+            headers: value.headers
         }
     }
 }
@@ -268,6 +274,7 @@ mod tests {
             value: "hello".into(),
             watermark: Some(prost_types::Timestamp::default()),
             event_time: Some(prost_types::Timestamp::default()),
+            headers:Default::default()
         });
 
         let resp = client.map_fn(request).await?;

src/reduce.rs

@@ -195,6 +195,8 @@ pub struct ReduceRequest {
     pub watermark: DateTime<Utc>,
     /// Time of the element as seen at source or aligned after a reduce operation.
     pub eventtime: DateTime<Utc>,
+    /// Headers for the message.
+    pub headers: HashMap<String, String>,
 }
 
 impl From<proto::ReduceRequest> for ReduceRequest {
@@ -204,6 +206,7 @@ impl From<proto::ReduceRequest> for ReduceRequest {
             value: mr.value,
             watermark: shared::utc_from_timestamp(mr.watermark),
             eventtime: shared::utc_from_timestamp(mr.event_time),
+            headers: mr.headers,
         }
     }
 }

src/sourcetransform.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::future::Future;
 use std::path::PathBuf;
 
@@ -88,6 +89,8 @@ pub struct SourceTransformRequest {
     pub watermark: DateTime<Utc>,
     /// event_time is the time of the element as seen at source or aligned after a reduce operation.
     pub eventtime: DateTime<Utc>,
+    /// Headers for the message.
+    pub headers: HashMap<String, String>,
 }
 
 impl From<Message> for proto::source_transform_response::Result {
@@ -108,6 +111,7 @@ impl From<proto::SourceTransformRequest> for SourceTransformRequest {
             value: value.value,
             watermark: shared::utc_from_timestamp(value.watermark),
             eventtime: shared::utc_from_timestamp(value.event_time),
+            headers:value.headers
         }
     }
 }
@@ -289,6 +293,7 @@ mod tests {
             value: "hello".into(),
             watermark: Some(prost_types::Timestamp::default()),
             event_time: Some(prost_types::Timestamp::default()),
+            headers:Default::default()
         });
 
         let resp = client.source_transform_fn(request).await?;