ekko/lws_client
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1. 修改配置的解析方式，使用建立rpc链接的时候使用一个get_config的请求同步，这样不用维护windows和linux两个配置文件

Browse Source 
2. 修改命令行参数使其更加符合CLI交互，使用clap库实现。

遗留问题：
权限全是0777。需要改一下
This commit is contained in:
Ekko.bao 2024-08-06 08:38:35 +08:00
parent 80c7a14ff2
commit 1d49f19301
5 changed files with 180 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
Cargo.lock generated
View File

@ -190,6 +190,30 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
[[package]]
name = "clap"
version = "3.2.25"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4ea181bf566f71cb9a5d17a59e1871af638180a18fb0035c92ae62b705207123"
dependencies = [
"atty",
"bitflags 1.3.2",
"clap_lex",
"indexmap 1.9.3",
"strsim",
"termcolor",
"textwrap",
]
[[package]]
name = "clap_lex"
version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2850f2f5a82cbf437dd5af4d49848fbdfc27c157c3d010345776f952765261c5"
dependencies = [
"os_str_bytes",
]
[[package]]
name = "either"
version = "1.13.0"
@ -518,6 +542,7 @@ checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"
name = "lws_client"
version = "0.1.0"
dependencies = [
"clap",
"env_logger",
"fuse_mt",
"hashbrown 0.9.1",
@ -600,6 +625,12 @@ version = "1.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92"
[[package]]
name = "os_str_bytes"
version = "6.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e2355d85b9a3786f481747ced0e0ff2ba35213a1f9bd406ed906554d7af805a1"
[[package]]
name = "page_size"
version = "0.5.0"
@ -897,6 +928,12 @@ dependencies = [
"windows-sys 0.52.0",
]
[[package]]
name = "strsim"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
[[package]]
name = "syn"
version = "2.0.71"
@ -935,6 +972,12 @@ dependencies = [
"winapi-util",
]
[[package]]
name = "textwrap"
version = "0.16.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "23d434d3f8967a09480fb04132ebe0a3e088c173e6d0ee7897abbdf4eab0f8b9"
[[package]]
name = "threadpool"
version = "1.8.1"

1
Cargo.toml
View File

@ -18,6 +18,7 @@ libc = "0.2"
hashbrown = "0.9.0"
log = "0.4"
env_logger = "0.8"
clap = "3.0"
[build-dependencies]
tonic-build = "0.11"

5
proto/lws.proto
View File

@ -19,6 +19,7 @@ package lws_vfs;
service LwsVfs {
// Sends a greeting
rpc SayHello(HelloRequest) returns (HelloReply) {}
rpc GetConfig(get_config) returns (get_config) {}
rpc fgetattr(getattr) returns (getattr) {}
rpc fsetxattr(setxattr) returns (setxattr) {}
@ -48,6 +49,10 @@ message HelloRequest { string name = 1; }
// The response message containing the greetings
message HelloReply { string message = 1; }
message get_config {
string config = 1;
}
message file_info {
uint32 flags = 1;
uint32 fh_old = 2;

93
src/client.rs
View File

@ -1,11 +1,100 @@
use lws_client::LwsVfsIns;
use std::thread;
use clap::{Arg, App};
extern crate log;
use std::env;
const DEFAULT_PORT: u16 = 33444;
// use std::process::{Command,Stdio};
// fn get_ssh_clinet() -> String {
// // Run `who` command
// let who = Command::new("who")
// .stdout(Stdio::piped())
// .spawn().unwrap();
// // Run `grep $(whoami)` command
// let whoami = Command::new("whoami")
// .stdout(Stdio::piped())
// .output().unwrap();
// let user = String::from_utf8_lossy(&whoami.stdout).trim().to_string();
// let grep = Command::new("grep")
// .arg(&user)
// .stdin(Stdio::from(who.stdout.unwrap()))
// .stdout(Stdio::piped())
// .spawn().unwrap();
// // Run `awk '{print $5}'` command
// let awk = Command::new("awk")
// .arg("{print $5}")
// .stdin(Stdio::from(grep.stdout.unwrap()))
// .stdout(Stdio::piped())
// .spawn().unwrap();
// // Collect the output
// let output = awk.wait_with_output().unwrap();
// let output = String::from_utf8_lossy(&output.stdout).to_string();
// println!("output: {}",output);
// let mut iter = output.split("(");
// iter.next();
// let output = iter.next().unwrap().split(")").collect::<Vec<&str>>()[0].to_string();
// println!("output: {}",output);
// output
// }
fn get_ssh_clinet() -> String {
// 获取特定环境变量的值
let client = env::var("SSH_CLIENT").expect("only support auto get ssh connection");
let client = client.split(' ').next().unwrap();
client.to_string()
}
fn param_parser() -> (String, String) {
let matches = App::new("lws_client")
.version("1.0")
.author("Ekko.bao")
.about("linux&windows shared filesystem")
.arg(Arg::with_name("s")
.short('s')
.long("server")
.takes_value(true)
.value_name("server")
.help(format!("server addr: [ip[:port]], default is [sshclient:{}]", DEFAULT_PORT).as_str()))
.arg(Arg::with_name("m")
.short('m')
.long("mount")
.takes_value(true)
.value_name("mount point")
.required(true)
.help("mount point, eg: ~/mnt"))
.get_matches();
let server = match matches.value_of("s") {
Some(server) => {
let split = server.split(":").collect::<Vec<&str>>();
if split.len() == 2 {
server.to_string()
} else {
format!("{}:{}", split[0], DEFAULT_PORT)
}
},
None => {
let ip = get_ssh_clinet();
format!("{}:{}", ip, DEFAULT_PORT)
}
};
log::info!("args server: [{}]", server);
let mount_point = matches.value_of("m").unwrap().to_string();
log::info!("args mount_point: [{}]", mount_point);
(server, mount_point)
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
env_logger::init();
let lws_ins = match LwsVfsIns::new("config.json").await {
let (server, mount_point) = param_parser();
let lws_ins = match LwsVfsIns::new(&server).await {
Ok(ins) => ins,
Err(e) => {
log::error!("Error creating lws server instance: {:?}", e);
@ -22,7 +111,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
}
log::info!("start mount process");
let handle = thread::spawn(move ||{
match LwsVfsIns::mount(lws_ins) {
match LwsVfsIns::mount(&mount_point, lws_ins) {
Ok(_) => {
Ok::<i32, String>(0)
},

81
src/lib.rs
View File

@ -1,16 +1,16 @@
use lws_client::lws_vfs_client::LwsVfsClient;
use lws_client::{
Access, Create, FileInfo, Flush, Fstat, Getattr, HelloRequest, Mkdir, Open, Opendir, Read,
Readdir, Release, Releasedir, Rename, Rmdir, Timespec, Truncate, Unlink, Utimens, Write,
Readdir, Release, Releasedir, Rename, Rmdir, Timespec, Truncate, Unlink, Utimens, Write, GetConfig,
};
use serde_json::{self, Value};
use hashbrown::HashMap;
use std::error::Error;
use std::ffi::{OsStr, OsString};
use std::ffi::{OsStr};
use std::fs::File;
use std::{env, path, vec};
use std::io::{Error as IoError, ErrorKind, Read as IoRead};
use tonic::transport::{channel, Channel as RpcChannel};
use tonic::transport::{Channel as RpcChannel};
use tokio::runtime::{Builder,Runtime};
use std::thread::{self, JoinHandle};
extern crate log;
@ -21,28 +21,12 @@ pub mod lws_client {
#[derive(Debug, Default)]
pub struct Config {
port: u16,
addr: String,
mount_map: HashMap<String, String>,
}
impl Config {
pub fn new(json: &str) -> Result<Config, Box<dyn Error>> {
let mut file = File::open(json)?;
let mut buffer = String::new();
file.read_to_string(&mut buffer)?;
let json: Value = serde_json::from_str(buffer.as_ref())?;
let port: u16 = match json.get("port") {
Some(port) => port.as_u64().expect("expect port is a number but its not") as u16,
None => 5001,
};
let addr: String = match json.get("addr") {
Some(addr) => addr
.as_str()
.expect("expect addr is a String but its not")
.to_string(),
None => "localhost".to_string(),
};
let mounts = match json.get("mount") {
let json: Value = serde_json::from_str(json)?;
let mounts = match json.get("mount_map") {
Some(mounts) => mounts.as_object().unwrap(),
None => {
return Err(Box::new(std::io::Error::new(
@ -61,19 +45,10 @@ impl Config {
})
.collect();
Ok(Config {
port,
addr,
mount_map,
})
}
pub fn get_port(&self) -> u16 {
self.port
}
pub fn get_addr(&self) -> &String {
&self.addr
}
pub fn get_mount(&self) -> Vec<&String> {
let mut ret = vec![];
for (_path, mount) in &self.mount_map {
@ -101,7 +76,7 @@ impl RpcPool {
async fn new(size: usize, uri: String) -> Result<Self, Box<dyn std::error::Error>> {
let mut clients = Vec::with_capacity(size);
let uri = uri.to_ascii_lowercase();
log::info!("create rpc pool {}", uri);
log::info!("create rpc pool size: {}", size);
let channel = RpcChannel::from_shared(uri)?.connect().await?;
for _ in 0..size {
let client = LwsVfsClient::new(channel.clone());
@ -198,7 +173,7 @@ impl VirFs {
VirFs {
_name: name.to_string(),
stat: lws_client::Fstat {
fst_mode: 0o0755 | {
fst_mode: 0o0555 | {
if is_dir {
libc::S_IFDIR
} else {
@ -284,7 +259,7 @@ impl VirFs {
for name in self.sub.keys() {
ret.push(name.to_string());
}
log::info!("readdir: {:?}", ret);
log::trace!("readdir: {:?}", ret);
ret
}
}
@ -1535,10 +1510,11 @@ impl FilesystemMT for LwsVfsIns {
impl LwsVfsIns {
pub async fn new(json: &str) -> Result<LwsVfsIns, Box<dyn Error>> {
let config = Config::new(json)?;
let addr = format!("http://{}:{}", config.get_addr(), config.get_port());
pub async fn new(server: &String) -> Result<LwsVfsIns, Box<dyn Error>> {
let addr = format!("http://{}", server);
log::info!("lws server is {}", addr);
let rpcs = RpcPool::new(10, addr).await?;
let config = LwsVfsIns::fetch_config(&rpcs).await?;
let mut vir_root = VirFs::new();
for mount in config.get_mount() {
vir_root.add(mount);
@ -1550,6 +1526,31 @@ impl LwsVfsIns {
cache:FileAttrCacheManager::new(),
})
}
async fn fetch_config(rpcs: &RpcPool) -> Result<Config, Box<dyn Error>> {
let (rpc, resp) = {
let request = tonic::Request::new(GetConfig {
..GetConfig::default()
});
let client = rpcs.get().unwrap();
let mut rpc = client.write().unwrap();
let rpc = &mut (rpc.client);
let resp = rpc.get_config(request).await;
(client.clone(), resp)
};
rpcs.put(rpc);
let resp = match resp {
Ok(resp) => resp.into_inner(),
Err(e) => {
log::error!("get resp err: {:?}", e);
return Err(Box::new(std::io::Error::new(std::io::ErrorKind::Other, "get config err")));
}
};
let config = Config::new(&resp.config).expect("parse config err");
log::info!("config: {:?}", config);
Ok(config)
}
pub fn init(&self) -> Result<(), Box<dyn Error>> {
let request = tonic::Request::new(HelloRequest {
name: "Ekko lws hello".into(),
@ -1582,16 +1583,14 @@ impl LwsVfsIns {
Ok(())
}
pub fn mount<F>(file_system: F) -> Result<(), Box<dyn Error>>
pub fn mount<F>(mount_point:&str, file_system: F) -> Result<(), Box<dyn Error>>
where
F: FilesystemMT + Sync + Send + 'static,
{
let args: Vec<OsString> = env::args_os().collect();
log::info!("args is {:?}", args);
let fuse_args = [OsStr::new("-o"), OsStr::new("fsname=passthrufs")];
let fuse_args = [OsStr::new("-o"), OsStr::new("fsname=lws_vfs")];
fuse_mt::mount(
fuse_mt::FuseMT::new(file_system, 10),
&args[2],
&mount_point,
&fuse_args[..],
)?;
Ok(())