feat: improve connection protocol

cli/src/actors/client.rs

@@ -3,8 +3,9 @@ use std::time::Duration;
 use bytes::Bytes;
 use handle_macro::Handle;
 use remux_core::{
-    communication,
+    comm,
     events::{CliEvent, DaemonEvent},
+    states::DaemonState,
 };
 use tokio::{io::AsyncReadExt, net::UnixStream, sync::mpsc, time::interval};
 use tracing::{Instrument, debug};
@@ -13,7 +14,6 @@ use crate::{
     actors::ui::{UI, UIHandle},
     input_parser::{Action, InputParser, ParsedEvent},
     prelude::*,
-    states::daemon_state::DaemonState,
     utils::DisplayableVec,
 };
 
@@ -43,12 +43,12 @@ pub struct Client {
 }
 impl Client {
     #[instrument(skip(stream))]
-    pub fn spawn(stream: UnixStream) -> Result<CliTask> {
-        Client::new(stream)?.run()
+    pub fn spawn(stream: UnixStream, daemon_state: DaemonState) -> Result<CliTask> {
+        Client::new(stream, daemon_state)?.run()
     }
 
     #[instrument(skip(stream))]
-    fn new(stream: UnixStream) -> Result<Self> {
+    fn new(stream: UnixStream, daemon_state: DaemonState) -> Result<Self> {
         let (tx, rx) = mpsc::channel(100);
         let (ui_stdin_tx, ui_stdin_rx) = mpsc::channel(100);
         let handle = ClientHandle { tx };
@@ -59,8 +59,8 @@ impl Client {
             rx,
             ui_stdin_tx,
             ui_handle,
-            daemon_state: DaemonState::default(),
-            sync_daemon_state: false,
+            daemon_state,
+            sync_daemon_state: true,
             input_parser: InputParser::new(),
             client_state: ClientState::Normal,
         })
@@ -88,7 +88,7 @@ impl Client {
                                             if let Some(index) = index {
                                                 let selected_session = self.daemon_state.session_ids[index];
                                                 debug!("sending session selection: {selected_session}");
-                                                communication::send_event(&mut self.stream, CliEvent::SwitchSession(selected_session)).await.unwrap();
+                                                comm::send_event(&mut self.stream, CliEvent::SwitchSession(selected_session)).await.unwrap();
                                             }
                                             debug!("Returning to normal state");
                                             self.client_state = ClientState::Normal;
@@ -97,7 +97,7 @@ impl Client {
                                 }
                             }
                         },
-                        res = communication::recv_daemon_event(&mut self.stream) => {
+                        res = comm::recv_daemon_event(&mut self.stream) => {
                             match res {
                                 Ok(event) => {
                                     match event {
@@ -145,7 +145,7 @@ impl Client {
                                             for event in self.input_parser.process(&stdin_buf[..n]) {
                                                 match event {
                                                     ParsedEvent::DaemonAction(cli_event) => {
-                                                        communication::send_event(&mut self.stream, cli_event).await?;
+                                                        comm::send_event(&mut self.stream, cli_event).await?;
                                                     },
                                                     ParsedEvent::LocalAction(local_action) => {
                                                         match local_action {

cli/src/actors/lua.rs

@@ -8,13 +8,10 @@ use std::{
 };
 
 use mlua::Lua as MLua;
+use remux_core::states::DaemonState;
 use tokio::runtime::Handle;
 
-use crate::{
-    actors::ui::UIHandle,
-    prelude::*,
-    states::{daemon_state::DaemonState, status_line_state::StatusLineState},
-};
+use crate::{actors::ui::UIHandle, prelude::*, states::status_line_state::StatusLineState};
 
 pub enum LuaEvent {
     Kill,

cli/src/actors/ui.rs

@@ -11,6 +11,7 @@ use crossterm::{
 };
 use handle_macro::Handle;
 use ratatui::{Terminal, prelude::CrosstermBackend};
+use remux_core::states::DaemonState;
 use tokio::{
     sync::{broadcast, mpsc},
     time::interval,
@@ -25,7 +26,7 @@ use crate::{
         lua::{Lua, LuaHandle},
     },
     prelude::*,
-    states::{daemon_state::DaemonState, status_line_state::StatusLineState},
+    states::status_line_state::StatusLineState,
     utils::DisplayableVec,
     widgets::{BasicSelector, FuzzySelector, Selector},
 };

cli/src/args.rs

@@ -1,5 +1,5 @@
 use clap::{Parser, Subcommand};
-use remux_core::messages::{RequestBody, RequestMessage};
+use remux_core::messages::{CliRequestMessage, RequestBody, RequestBuilder, request};
 
 #[derive(Parser, Debug)]
 pub struct Args {
@@ -24,34 +24,54 @@ pub enum SessionCommands {
     List,
 }
 
-#[allow(clippy::from_over_into)]
-impl Into<RequestBody> for Commands {
-    fn into(self) -> RequestBody {
+impl Commands {
+    pub fn into_request(self) -> CliRequestMessage<impl RequestBody> {
         match self {
-            Self::Attach { session_id } => RequestBody::Attach { session_id },
-            Self::Session { action } => action.into(),
+            Self::Attach { session_id } => RequestBuilder::default()
+                .body(request::Attach {
+                    session_id,
+                    create: true,
+                })
+                .build(),
+            Self::Session { .. } => todo!(),
         }
     }
 }
-#[allow(clippy::from_over_into)]
-impl Into<RequestBody> for SessionCommands {
-    fn into(self) -> RequestBody {
-        match self {
-            SessionCommands::List => RequestBody::SessionsList,
-        }
-    }
-}
-#[allow(clippy::from_over_into)]
-impl Into<RequestMessage> for Commands {
-    fn into(self) -> RequestMessage {
-        let body: RequestBody = self.into();
-        RequestMessage::body(body)
-    }
-}
-#[allow(clippy::from_over_into)]
-impl Into<RequestMessage> for SessionCommands {
-    fn into(self) -> RequestMessage {
-        let body: RequestBody = self.into();
-        RequestMessage::body(body)
-    }
-}
+
+// #[allow(clippy::from_over_into)]
+// impl<T: RequestBody> Into<RequestMessage<T>> for Commands {
+//     fn into(self) -> RequestMessage<T> {
+//         match self {
+//             Self::Attach { session_id } => RequestBuilder::default()
+//                 .body(request::Attach {
+//                     session_id,
+//                     create: true,
+//                 })
+//                 .build(),
+//             Self::Session { action } => action.into(),
+//         }
+//     }
+// }
+
+// #[allow(clippy::from_over_into)]
+// impl Into<RequestBody> for SessionCommands {
+//     fn into(self) -> RequestBody {
+//         match self {
+//             SessionCommands::List => RequestBody::SessionsList,
+//         }
+//     }
+// }
+// #[allow(clippy::from_over_into)]
+// impl Into<RequestBody> for Commands {
+//     fn into(self) -> RequestBody {
+//         let body: RequestBody = self.into();
+//         RequestBuilder::default().body(body).build()
+//     }
+// }
+// #[allow(clippy::from_over_into)]
+// impl Into<RequestBody> for SessionCommands {
+//     fn into(self) -> RequestBody {
+//         let body: RequestBody = self.into();
+//         RequestBuilder::default().body(body).build()
+//     }
+// }

cli/src/error.rs

@@ -1,5 +1,4 @@
 use bytes::Bytes;
-use remux_core::messages::RequestMessage;
 use thiserror::Error;
 use tokio::sync::mpsc::error::SendError;
 
@@ -30,9 +29,9 @@ pub enum Error {
         source: std::io::Error,
     },
 
-    #[error("Error sending message {message}: {source}")]
+    #[error("Error sending message {message:?}: {source}")]
     SendRequestMessage {
-        message: RequestMessage,
+        message: String,
         source: remux_core::error::Error,
     },
 

cli/src/main.rs

@@ -10,15 +10,18 @@ mod widgets;
 use clap::Parser;
 use ratatui::crossterm::terminal::{disable_raw_mode, enable_raw_mode};
 use remux_core::{
-    communication,
+    comm,
     daemon_utils::get_sock_path,
-    messages::{RequestMessage, ResponseBody, ResponseMessage},
+    messages::{
+        CliRequestMessage, RequestBuilder,
+        request::{self, Attach},
+    },
 };
 use tokio::net::UnixStream;
 
 use crate::{
     actors::Client,
-    args::{Args, Commands, SessionCommands},
+    args::{Args, Commands},
     error::{Error, Result},
     prelude::*,
 };
@@ -71,39 +74,52 @@ async fn connect() -> Result<UnixStream> {
         })
 }
 
-#[instrument(skip(stream))]
-async fn handle_session_command(mut stream: UnixStream, command: SessionCommands) -> Result<()> {
-    let req: RequestMessage = command.into();
-    let res: ResponseMessage = communication::send_and_recv(&mut stream, &req).await?;
-    match res.body {
-        ResponseBody::SessionsList { sessions } => {
-            println!("{sessions:?}");
-        }
-    }
-    Ok(())
-}
+// #[instrument(skip(stream))]
+// async fn handle_session_command(mut stream: UnixStream, command: SessionCommands) -> Result<()> {
+//     let req: Request = command.into();
+//     let res: Response = comm::send_and_recv(&mut stream, &req).await?;
+//     assert_eq!(res.status, ResponseStatus::Ok);
+//     match res.body {
+//         ResponseBody::SessionsList { sessions } => {
+//             println!("{sessions:?}");
+//         }
+//         _ => {}
+//     }
+//     Ok(())
+// }
 
+#[instrument]
 async fn run(command: Commands) -> Result<()> {
     let stream = connect().await?;
+    debug!("Running command: {:?}", command);
     match command {
-        a @ Commands::Attach { .. } => attach(stream, a.into()).await,
-        Commands::Session { action } => handle_session_command(stream, action).await,
+        Commands::Attach { session_id } => {
+            attach(
+                stream,
+                RequestBuilder::default()
+                    .body(request::Attach {
+                        session_id,
+                        create: true,
+                    })
+                    .build(),
+            )
+            .await
+        }
+        // Commands::Session { action } => handle_session_command(stream, action).await,
+        _ => todo!(),
     }
 }
 
-#[instrument(skip(stream, attach_message))]
-async fn attach(mut stream: UnixStream, attach_message: RequestMessage) -> Result<()> {
-    debug!("Sending attach request");
-    communication::write_message(&mut stream, &attach_message)
-        .await
-        .map_err(|source| Error::SendRequestMessage {
-            message: attach_message,
-            source,
-        })?;
-    debug!("Sent attach request successfully");
+#[instrument(skip(stream, attach_request))]
+async fn attach(mut stream: UnixStream, attach_request: CliRequestMessage<Attach>) -> Result<()> {
+    debug!("Sending attach request: {:?}", attach_request);
+    let res = comm::send_and_recv_message(&mut stream, &attach_request).await?;
+    debug!("Recieved attach response: {:?}", res);
+    debug!("Recieved initial daemon state: {:?}", res.initial_daemon_state);
+
     enable_raw_mode()?;
-    debug!("raw mode enabled");
-    if let Ok(task) = Client::spawn(stream) {
+    debug!("Enabled raw mode");
+    if let Ok(task) = Client::spawn(stream, res.initial_daemon_state) {
         match task.await {
             Ok(Err(e)) => {
                 error!("Error joining client task: {e}");

cli/src/states/mod.rs

@@ -1,2 +1 @@
-pub mod daemon_state;
 pub mod status_line_state;

core/Cargo.toml

@@ -14,3 +14,4 @@ tokio.workspace = true
 
 bincode = "2.0.1"
 fs2 = "0.4.3"
+rand = "0.9.2"