feat: use postcard encoding

Author: Mivik

binaries/cli/src/command/build/distributed.rs

@@ -1,10 +1,10 @@
-use communication_layer_request_reply::{
-    Transport, encoding::JsonEncoding, transport::FramedTransport,
-};
+use communication_layer_request_reply::{Transport, transport::FramedTransport};
 use dora_core::descriptor::Descriptor;
 use dora_message::{
     BuildId,
-    cli_to_coordinator::{CliToCoordinatorClient, CliToCoordinatorRequest},
+    cli_to_coordinator::{
+        CliToCoordinatorClient, CliToCoordinatorEncoding, CliToCoordinatorRequest,
+    },
     common::{GitSource, LogMessage},
     id::NodeId,
 };
@@ -45,7 +45,7 @@ pub fn wait_until_dataflow_built(
     let mut log_session = FramedTransport::new(
         TcpStream::connect(coordinator_socket).wrap_err("failed to connect to dora coordinator")?,
     )
-    .with_encoding::<_, CliToCoordinatorRequest, LogMessage>(JsonEncoding);
+    .with_encoding::<_, CliToCoordinatorRequest, LogMessage>(CliToCoordinatorEncoding);
     log_session
         .send(&CliToCoordinatorRequest::BuildLogSubscribe {
             build_id,

binaries/cli/src/command/logs.rs

@@ -9,12 +9,12 @@ use crate::{
     output::print_log_message,
 };
 use clap::Args;
-use communication_layer_request_reply::{
-    Transport, encoding::JsonEncoding, transport::FramedTransport,
-};
+use communication_layer_request_reply::{Transport, transport::FramedTransport};
 use dora_core::topics::{DORA_COORDINATOR_PORT_CONTROL_DEFAULT, LOCALHOST};
 use dora_message::{
-    cli_to_coordinator::{CliToCoordinatorClient, CliToCoordinatorRequest},
+    cli_to_coordinator::{
+        CliToCoordinatorClient, CliToCoordinatorEncoding, CliToCoordinatorRequest,
+    },
     common::LogMessage,
     id::NodeId,
 };
@@ -90,7 +90,7 @@ pub fn logs(
     let mut log_session = FramedTransport::new(
         TcpStream::connect(coordinator_addr).wrap_err("failed to connect to dora coordinator")?,
     )
-    .with_encoding::<_, CliToCoordinatorRequest, LogMessage>(JsonEncoding);
+    .with_encoding::<_, CliToCoordinatorRequest, LogMessage>(CliToCoordinatorEncoding);
     log_session.send(&CliToCoordinatorRequest::LogSubscribe {
         dataflow_id: uuid,
         level: log_level,

binaries/cli/src/command/start/attach.rs

@@ -1,9 +1,8 @@
-use communication_layer_request_reply::{
-    Transport, encoding::JsonEncoding, transport::FramedTransport,
-};
+use communication_layer_request_reply::{Transport, transport::FramedTransport};
 use dora_core::descriptor::{CoreNodeKind, Descriptor, DescriptorExt, resolve_path};
 use dora_message::cli_to_coordinator::{
-    CheckResp, CliToCoordinatorClient, CliToCoordinatorRequest, DataflowStopped,
+    CheckResp, CliToCoordinatorClient, CliToCoordinatorEncoding, CliToCoordinatorRequest,
+    DataflowStopped,
 };
 use dora_message::common::LogMessage;
 use dora_message::id::{NodeId, OperatorId};
@@ -142,7 +141,7 @@ pub fn attach_dataflow(
     let mut log_session = FramedTransport::new(
         TcpStream::connect(coordinator_socket).wrap_err("failed to connect to dora coordinator")?,
     )
-    .with_encoding::<_, CliToCoordinatorRequest, LogMessage>(JsonEncoding);
+    .with_encoding::<_, CliToCoordinatorRequest, LogMessage>(CliToCoordinatorEncoding);
     log_session
         .send(&CliToCoordinatorRequest::LogSubscribe {
             dataflow_id,

binaries/cli/src/command/start/mod.rs

@@ -9,15 +9,15 @@ use crate::{
     output::print_log_message,
     session::DataflowSession,
 };
-use communication_layer_request_reply::{
-    Transport, encoding::JsonEncoding, transport::FramedTransport,
-};
+use communication_layer_request_reply::{Transport, transport::FramedTransport};
 use dora_core::{
     descriptor::{Descriptor, DescriptorExt},
     topics::{DORA_COORDINATOR_PORT_CONTROL_DEFAULT, LOCALHOST},
 };
 use dora_message::{
-    cli_to_coordinator::{CliToCoordinatorClient, CliToCoordinatorRequest, StartReq},
+    cli_to_coordinator::{
+        CliToCoordinatorClient, CliToCoordinatorEncoding, CliToCoordinatorRequest, StartReq,
+    },
     common::LogMessage,
 };
 use eyre::Context;
@@ -148,7 +148,7 @@ fn wait_until_dataflow_started(
     let mut log_session = FramedTransport::new(
         TcpStream::connect(coordinator_addr).wrap_err("failed to connect to dora coordinator")?,
     )
-    .with_encoding::<_, CliToCoordinatorRequest, LogMessage>(JsonEncoding);
+    .with_encoding::<_, CliToCoordinatorRequest, LogMessage>(CliToCoordinatorEncoding);
     log_session
         .send(&CliToCoordinatorRequest::LogSubscribe {
             dataflow_id,

binaries/coordinator/src/control.rs

@@ -1,11 +1,9 @@
 use crate::{
     Coordinator, handler::CliRequestHandler, log_subscriber::LogSubscriber, send_log_message,
 };
-use communication_layer_request_reply::{
-    AsyncTransport, encoding::JsonEncoding, transport::FramedTransport,
-};
+use communication_layer_request_reply::{AsyncTransport, transport::FramedTransport};
 use dora_message::cli_to_coordinator::{
-    CliToCoordinator, CliToCoordinatorRequest, CliToCoordinatorResponse,
+    CliToCoordinator, CliToCoordinatorEncoding, CliToCoordinatorRequest, CliToCoordinatorResponse,
 };
 use eyre::{Context, eyre};
 use futures::{
@@ -84,7 +82,9 @@ pub(crate) async fn listen(
 async fn handle_requests(state: Arc<ListenState>, connection: TcpStream) {
     let peer_addr = connection.peer_addr().ok();
     let mut transport = FramedTransport::new(connection)
-        .with_encoding::<_, CliToCoordinatorResponse, CliToCoordinatorRequest>(JsonEncoding);
+        .with_encoding::<_, CliToCoordinatorResponse, CliToCoordinatorRequest>(
+            CliToCoordinatorEncoding,
+        );
     loop {
         let next_request = transport.receive().map(Either::Left);
         let coordinator_stopped = state.cancel_token.cancelled().map(Either::Right);
@@ -147,7 +147,7 @@ async fn handle_requests(state: Arc<ListenState>, connection: TcpStream) {
 
             log_subscribers.push(LogSubscriber::new(
                 level,
-                transport.into_inner().into_inner(),
+                transport.into_inner(),
             ));
             let buffered = std::mem::take(buffered_log_messages);
             for message in buffered {

binaries/coordinator/src/log_subscriber.rs

@@ -1,18 +1,22 @@
-use dora_message::coordinator_to_cli::LogMessage;
+use communication_layer_request_reply::{
+    AsyncTransport, EncodedTransport, transport::FramedTransport,
+};
+use dora_message::{cli_to_coordinator::CliToCoordinatorEncoding, coordinator_to_cli::LogMessage};
 use eyre::{Context, ContextCompat};
-
-use crate::tcp_utils::tcp_send;
+use tokio::net::TcpStream;
 
 pub struct LogSubscriber {
     pub level: log::LevelFilter,
-    connection: Option<tokio::net::TcpStream>,
+    transport: Option<
+        EncodedTransport<FramedTransport<TcpStream>, CliToCoordinatorEncoding, LogMessage, ()>,
+    >,
 }
 
 impl LogSubscriber {
-    pub fn new(level: log::LevelFilter, connection: tokio::net::TcpStream) -> Self {
+    pub fn new(level: log::LevelFilter, transport: FramedTransport<TcpStream>) -> Self {
         Self {
             level,
-            connection: Some(connection),
+            transport: Some(transport.with_encoding(CliToCoordinatorEncoding)),
         }
     }
 
@@ -26,19 +30,19 @@ impl LogSubscriber {
             dora_core::build::LogLevelOrStdout::Stdout => {}
         }
 
-        let message = serde_json::to_vec(&message)?;
-        let connection = self.connection.as_mut().context("connection is closed")?;
-        tcp_send(connection, &message)
+        let connection = self.transport.as_mut().context("connection is closed")?;
+        connection
+            .send(message)
             .await
             .context("failed to send message")?;
         Ok(())
     }
 
     pub fn is_closed(&self) -> bool {
-        self.connection.is_none()
+        self.transport.is_none()
     }
 
     pub fn close(&mut self) {
-        self.connection = None;
+        self.transport = None;
     }
 }

libraries/message/src/cli_to_coordinator.rs

@@ -15,6 +15,8 @@ use crate::{
     id::{NodeId, OperatorId},
 };
 
+pub use communication_layer_request_reply::encoding::JsonEncoding as CliToCoordinatorEncoding;
+
 #[dora_schema_macro::dora_schema]
 pub trait CliToCoordinator {
     /// Builds a dataflow by distributing build tasks to appropriate daemons.
@@ -48,10 +50,10 @@ pub trait CliToCoordinator {
     /// spawning of nodes across connected daemons and returns a [`UUID`] for tracking
     /// the dataflow execution.
     ///
-    /// # Response  
+    /// # Response
     /// A [`UUID`] that identifies this specific dataflow execution
     ///
-    /// # Error Handling  
+    /// # Error Handling
     ///
     /// Returns `ControlRequestReply::Error` if:
     /// - Dataflow name is already in use