chore: move message conversions to appropriate modules (#2306)

Signed-off-by: Yashash H L <[email protected]> Signed-off-by: Vigith Maurice <[email protected]> Co-authored-by: Vigith Maurice <[email protected]>
numaproj · Jan 2, 2025 · 0c86e82 · 0c86e82
1 parent bb4a0de
commit 0c86e82
Show file tree

Hide file tree

Showing 9 changed files with 330 additions and 368 deletions.
diff --git a/rust/numaflow-core/src/config/monovertex.rs b/rust/numaflow-core/src/config/monovertex.rs
@@ -47,7 +47,7 @@ impl Default for MonovertexConfig {
                 source_type: source::SourceType::Generator(GeneratorConfig::default()),
             },
             sink_config: SinkConfig {
-                sink_type: sink::SinkType::Log(sink::LogConfig::default()),
+                sink_type: SinkType::Log(sink::LogConfig::default()),
                 retry_config: None,
             },
             transformer_config: None,

diff --git a/rust/numaflow-core/src/mapper/map/user_defined.rs b/rust/numaflow-core/src/mapper/map/user_defined.rs
@@ -13,6 +13,7 @@ use tracing::error;
 use crate::config::get_vertex_name;
 use crate::error::{Error, Result};
 use crate::message::{Message, MessageID, Offset};
+use crate::shared::grpc::prost_timestamp_from_utc;
 
 type ResponseSenderMap =
     Arc<Mutex<HashMap<String, (ParentMessageInfo, oneshot::Sender<Result<Vec<Message>>>)>>>;
@@ -26,6 +27,23 @@ struct ParentMessageInfo {
     headers: HashMap<String, String>,
 }
 
+impl From<Message> for MapRequest {
+    fn from(message: Message) -> Self {
+        Self {
+            request: Some(map::map_request::Request {
+                keys: message.keys.to_vec(),
+                value: message.value.to_vec(),
+                event_time: prost_timestamp_from_utc(message.event_time),
+                watermark: None,
+                headers: message.headers,
+            }),
+            id: message.offset.unwrap().to_string(),
+            handshake: None,
+            status: None,
+        }
+    }
+}
+
 /// UserDefinedUnaryMap is a grpc client that sends unary requests to the map server
 /// and forwards the responses.
 pub(in crate::mapper) struct UserDefinedUnaryMap {