tyler
/
proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

src/server/command_socket.rs: correctly handle/spawn handle parallel connections

This commit is contained in:
Dietmar Maurer 2019-04-10 11:05:00 +02:00
parent 9b002cbc5f
commit cfb2d3c17c
1 changed files with 27 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
src/server/command_socket.rs
View File

@ -18,7 +18,6 @@ pub fn create_control_socket<P, F>(path: P, f: F) -> Result<impl Future<Item=(),
F: Send + Sync +'static + Fn(Value) -> Result<Value, Error>, F: Send + Sync +'static + Fn(Value) -> Result<Value, Error>,
{ {
let path: PathBuf = path.into(); let path: PathBuf = path.into();
let path1: PathBuf = path.clone();
let socket = UnixListener::bind(&path)?; let socket = UnixListener::bind(&path)?;
@ -35,29 +34,34 @@ pub fn create_control_socket<P, F>(path: P, f: F) -> Result<impl Future<Item=(),
let path = path3.clone(); let path = path3.clone();
let path2 = path3.clone(); let path2 = path3.clone();
tokio::io::lines(std::io::BufReader::new(rx)) let abort_future = super::last_worker_future().map_err(|_| {});
.map_err(move |err| { eprintln!("control socket {:?} read error: {}", path, err); })
.and_then(move |cmd| { tokio::spawn(
let res = try_block!({ tokio::io::lines(std::io::BufReader::new(rx))
let param = match cmd.parse::<Value>() { .map_err(move |err| { eprintln!("control socket {:?} read error: {}", path, err); })
Ok(p) => p, .and_then(move |cmd| {
Err(err) => bail!("unable to parse json value - {}", err), let res = try_block!({
let param = match cmd.parse::<Value>() {
Ok(p) => p,
Err(err) => bail!("unable to parse json value - {}", err),
};
f1(param)
});
let resp = match res {
Ok(v) => format!("OK: {}\n", v),
Err(err) => format!("ERROR: {}\n", err),
}; };
Ok(resp)
f1(param) })
}); .for_each(move |resp| {
tx.write_all(resp.as_bytes())
let resp = match res { .map_err(|err| { eprintln!("control socket {:?} write response error: {}", path2, err); })
Ok(v) => format!("OK: {}\n", v), })
Err(err) => format!("ERROR: {}\n", err), .select(abort_future)
}; .then(move |_| { Ok(()) })
Ok(resp) )
})
.for_each(move |resp| {
tx.write_all(resp.as_bytes())
.map_err(|err| { eprintln!("control socket {:?} write response error: {}", path2, err); })
})
}); });
let abort_future = super::last_worker_future().map_err(|_| {}); let abort_future = super::last_worker_future().map_err(|_| {});