us changes

noxis-cli/src/cli.rs

@@ -2,6 +2,12 @@ use clap::{Parser, Subcommand};
 
 #[derive(Debug, Parser, serde::Serialize, serde::Deserialize)]
 pub struct Cli {
+    #[arg(
+        short,
+        default_value="noxis-rs.sock",
+        help="explicit specify of NOXIS Socket file"
+    )]
+    pub socket : String,
     #[command(
         subcommand,
         help = "to manage Noxis work",

noxis-cli/src/cli_error.rs

@@ -1,14 +1,15 @@
 use thiserror::Error;
-use super::cli_net::NOXIS_RS_CREDS;
 
 #[derive(Debug, Error)]
 pub enum NoxisCliError {
-    #[error("Can't send any data to {:?}. Noxis-rs daemon is disabled or can't be accessed", NOXIS_RS_CREDS)]
-    NoxisDaemonMissing,
-    #[error("Noxis CLI can't write any data to the Noxis-rs port. Check daemon and it's web-functionality")]
+    #[error("Can't find socket `{0}`. Noxis-rs daemon is disabled or can't be accessed using Unix-Socket")]
+    NoxisDaemonMissing(String),
+    #[error("Noxis CLI can't write any data to the Noxis-rs port. Check daemon and it's runtime!")]
     PortIsNotWritable,
     #[error("Can't send Cli-prompt to the Noxis-rs. Check it's state")]
     CliPromptCanNotBeSent,
     #[error("Can't parse CLI struct and send as byte stream")]
     ToStringCliParsingParsing,
+    #[error("Can't read Noxis response")]
+    CliResponseReadError
 }

noxis-cli/src/cli_net.rs

@@ -1,32 +1,30 @@
-use tokio::net::TcpStream;
-use tokio::io::AsyncWriteExt;
+use tokio::net::UnixStream;
+use tokio::io::{AsyncWriteExt, AsyncReadExt};
 use tokio::time::{Duration, sleep};
 use anyhow::Result;
 use super::Cli;
 use super::cli_error::NoxisCliError;
 
-pub const NOXIS_RS_CREDS: &str = "127.0.0.1:7753";
-
-
-pub async fn create_tcp_stream() -> Result<TcpStream> {
-    Ok(TcpStream::connect(NOXIS_RS_CREDS).await.map_err(|_| NoxisCliError::NoxisDaemonMissing)?)
+async fn create_us_stream(cli: &Cli) -> Result<UnixStream> {
+    Ok(UnixStream::connect(&cli.socket).await.map_err(|_| NoxisCliError::NoxisDaemonMissing((&cli.socket).to_string()))?)
 }
 
-pub async fn try_send(stream: Result<TcpStream>, params: Cli) -> Result<()> {
-    use serde_json::to_string;
-    let mut stream = stream.map_err(|_| NoxisCliError::NoxisDaemonMissing)?;
-    loop {
-        if stream.writable().await.is_err() {
-            sleep(Duration::from_millis(100)).await;
-            continue;
-        }
-        // let msg: Cli = from_str(&format!("{:?}", params))?;
-        let msg= to_string(&params).map_err(|_| NoxisCliError::ToStringCliParsingParsing)?;
-        // let msg = r"HTTP/1.1 POST\r\nContent-Length: 14\r\nContent-Type: text/plain\r\n\r\nHello, World!@";
+pub async fn try_send(cli: Cli) -> Result<()> {
+    // let stream = create_us_stream(&cli).await;
+    let mut stream = create_us_stream(&cli).await?;
 
-        stream.write_all(msg.as_bytes()).await.map_err(|_| NoxisCliError::CliPromptCanNotBeSent)?;
-        // ...
-        break;
-    }
+    let msg = serde_json::to_vec(&cli)
+        .map_err(|_| NoxisCliError::ToStringCliParsingParsing)?;
+
+    stream.write_all(&msg).await
+        .map_err(|_| NoxisCliError::CliPromptCanNotBeSent)?;
+
+    stream.shutdown().await?;
+
+    let mut response = Vec::new();
+    stream.read_to_end(&mut response).await
+        .map_err(|_| NoxisCliError::CliResponseReadError)?;
+
+    println!("Received response: {}", String::from_utf8_lossy(&response));
     Ok(())
 }

noxis-cli/src/main.rs

@@ -4,12 +4,12 @@ mod cli_error;
 
 use clap::Parser;
 use cli::Cli;
-use cli_net::{create_tcp_stream, try_send};
+use cli_net::try_send;
 use anyhow::Result;
 
 #[tokio::main]
 async fn main() -> Result<()>{
     let cli = Cli::parse();
-    try_send(create_tcp_stream().await, cli).await?;
+    try_send(cli).await?;
     Ok(())
 }

noxis-rs/src/main.rs

@@ -1,7 +1,6 @@
 mod options;
 mod utils;
 
-use anyhow::Error;
 use clap::Parser;
 use log::{error, info};
 use options::config::*;

noxis-rs/src/options/cli_pipeline.rs

@@ -1,13 +1,11 @@
-use log::{error, info, warn};
-use tokio::net::{TcpListener, TcpStream, UnixStream};
-use anyhow::{Result as DynResult, Error};
+use log::{error, info};
+use tokio::net::{ UnixStream, UnixListener };
+use anyhow::Result as DynResult;
 use tokio::time::{sleep, Duration};
-use std::{borrow::BorrowMut, fs, net::{IpAddr, Ipv4Addr}};
+use std::fs;
 // use std::io::BufReader;
-use tokio::io::{BufReader, AsyncWriteExt, AsyncBufReadExt};
-use tokio::{io::AsyncReadExt, net::UnixListener};
+use tokio::io::{ AsyncWriteExt, AsyncReadExt};
 use noxis_cli::Cli;
-use serde_json::from_str;
 
 /// # Fn `init_cli_pipeline`
 /// ## for catching all input requests from CLI
@@ -25,12 +23,19 @@ use serde_json::from_str;
 pub async fn init_cli_pipeline() -> DynResult<()> {
     match init_listener().await {
         Ok(list) => {
-            info!("Successfully opened UnixListener for CLI");
+            // TODO: remove `unwrap`s
+            info!("Listening on {}", &list.local_addr()?.as_pathname().unwrap().display());
             loop {
-                if let Ok((socket, _)) = list.accept().await {
+                
+                if let Ok((socket, a)) = list.accept().await {
+                    info!("CLI connection from {}", a.as_pathname().unwrap().display());
                     process_connection(socket).await;
+                } else {
+                    error!("Cannot poll connection to CLI");
                 }
-                sleep(Duration::from_millis(500)).await;
+                dbg!(1);
+                sleep(Duration::from_millis(300)).await;
+                
             }
             // Ok(())
         },
@@ -65,7 +70,7 @@ async fn init_listener() -> anyhow::Result<UnixListener> {
     //         None
     //     } 
     // }
-    let socket_path = "noxis-rs";
+    let socket_path = "noxis-rs.sock";
     let _ = fs::remove_file(socket_path);
     Ok(UnixListener::bind(socket_path)?)
 }
@@ -84,25 +89,28 @@ async fn init_listener() -> anyhow::Result<UnixListener> {
 /// *depends on* :  `tokio::net::TcpStream`
 /// 
 async fn process_connection(mut stream: UnixStream) {
-    let buf_reader = BufReader::new(stream.borrow_mut());
-    let mut rqst = buf_reader.lines();
+    info!("Processing new connection");
     
-    while let Ok(Some(line)) = rqst.next_line().await {
-        if line.is_empty() {
-            break
+    let mut buf = Vec::new();
+    match stream.read_to_end(&mut buf).await {
+        Ok(_) => {
+            match serde_json::from_slice::<Cli>(&buf) {
+                Ok(cli) => {
+                    info!("Received CLI request: {:?}", cli);
+                    // Обработка запроса
+                    let response = "OK";
+                    if let Err(e) = stream.write_all(response.as_bytes()).await {
+                        error!("Failed to send response: {}", e);
+                    }
+                }
+                Err(e) => {
+                    error!("Failed to parse CLI request: {}", e);
+                }
+            }
         }
-        match from_str::<Cli>(&line) {
-            Ok(req) => {
-                // TODO: func wrapper
-                dbg!(req);
-            },
-            Err(_) => {
-                break
-            },
+        Err(e) => {
+            error!("Failed to read from socket: {}", e);
         }
-        println!("{}", line);
     }
-
-    let response = "OK";
-    stream.write_all(response.as_bytes()).await.unwrap();
+    let _ = stream.shutdown().await;
 }