liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

checkpoint

Browse Source
This commit is contained in:
John Smith
2023-06-08 14:07:09 -04:00
parent 59c14f3b22
commit 419bfcd8ce
19 changed files with 563 additions and 602 deletions
Download Patch File Download Diff File Expand all files Collapse all files

509
veilid-cli/src/client_api_connection.rs
View File

@@ -1,12 +1,12 @@
use crate::command_processor::*;
use crate::tools::*;
use serde::de::DeserializeOwned;
use std::cell::RefCell;
use core::str::FromStr;
use futures::stream::FuturesUnordered;
use futures::StreamExt;
use std::net::SocketAddr;
use std::rc::Rc;
use stop_token::{future::FutureExt as _, StopSource, StopToken};
use std::time::SystemTime;
use stop_token::{future::FutureExt as _, StopSource};
use veilid_tools::*;
cfg_if! {
if #[cfg(feature="rt-async-std")] {
use async_std::io::prelude::BufReadExt;
@@ -19,80 +19,41 @@ cfg_if! {
}
}
// fn map_to_internal_error<T: ToString>(e: T) -> VeilidAPIError {
// VeilidAPIError::Internal {
// message: e.to_string(),
// }
// }
// fn decode_api_result<T: DeserializeOwned + fmt::Debug>(
// reader: &api_result::Reader,
// ) -> VeilidAPIResult<T> {
// match reader.which().map_err(map_to_internal_error)? {
// api_result::Which::Ok(v) => {
// let ok_val = v.map_err(map_to_internal_error)?;
// let res: T = veilid_core::deserialize_json(ok_val).map_err(map_to_internal_error)?;
// Ok(res)
// }
// api_result::Which::Err(e) => {
// let err_val = e.map_err(map_to_internal_error)?;
// let res: VeilidAPIError =
// veilid_core::deserialize_json(err_val).map_err(map_to_internal_error)?;
// Err(res)
// }
// }
// }
// struct VeilidClientImpl {
// comproc: CommandProcessor,
// }
// impl VeilidClientImpl {
// pub fn new(comproc: CommandProcessor) -> Self {
// Self { comproc }
// }
// }
// }
struct ClientApiConnectionInner {
comproc: CommandProcessor,
connect_addr: Option<SocketAddr>,
server: Option<flume::Sender<String>>,
request_sender: Option<flume::Sender<String>>,
server_settings: Option<String>,
disconnector: Option<StopSource>,
disconnect_requested: bool,
cancel_eventual: Eventual,
reply_channels: HashMap<u32, flume::Sender<json::JsonValue>>,
next_req_id: u32,
}
type Handle<T> = Rc<RefCell<T>>;
#[derive(Clone)]
pub struct ClientApiConnection {
inner: Handle<ClientApiConnectionInner>,
inner: Arc<Mutex<ClientApiConnectionInner>>,
}
impl ClientApiConnection {
pub fn new(comproc: CommandProcessor) -> Self {
Self {
inner: Rc::new(RefCell::new(ClientApiConnectionInner {
inner: Arc::new(Mutex::new(ClientApiConnectionInner {
comproc,
connect_addr: None,
server: None,
request_sender: None,
server_settings: None,
disconnector: None,
disconnect_requested: false,
cancel_eventual: Eventual::new(),
reply_channels: HashMap::new(),
next_req_id: 0,
})),
}
}
pub fn cancel(&self) {
let eventual = {
let inner = self.inner.borrow();
inner.cancel_eventual.clone()
};
eventual.resolve(); // don't need to await this
pub fn cancel_all(&self) {
let mut inner = self.inner.lock();
inner.reply_channels.clear();
}
// async fn process_veilid_state<'a>(
@@ -106,8 +67,33 @@ impl ClientApiConnection {
// Ok(())
// }
async fn process_response(&self, response: json::JsonValue) {
// find the operation id and send the response to the channel for it
let Some(id_str) = response["id"].as_str() else {
error!("missing id: {}", response);
return;
};
let Ok(id) = u32::from_str(id_str) else {
error!("invalid id: {}", response);
return;
};
let reply_channel = {
let mut inner = self.inner.lock();
inner.reply_channels.remove(&id)
};
let Some(reply_channel) = reply_channel else {
warn!("received cancelled reply: {}", response);
return;
};
if let Err(e) = reply_channel.send_async(response).await {
error!("failed to process reply: {}", e);
return;
}
}
async fn process_update(&self, update: json::JsonValue) {
let comproc = self.inner.borrow().comproc.clone();
let comproc = self.inner.lock().comproc.clone();
let Some(kind) = update["kind"].as_str() else {
comproc.log_message(format!("missing update kind: {}", update));
return;
@@ -145,7 +131,7 @@ impl ClientApiConnection {
}
// async fn spawn_rpc_system(
// &mut self,
// &self,
// connect_addr: SocketAddr,
// mut rpc_system: RpcSystem<rpc_twoparty_capnp::Side>,
// ) -> Result<(), String> {
@@ -235,18 +221,9 @@ impl ClientApiConnection {
// res.map_err(|e| format!("client RPC system error: {}", e))
// }
async fn handle_connection(&mut self, connect_addr: SocketAddr) -> Result<(), String> {
async fn handle_connection(&self, connect_addr: SocketAddr) -> Result<(), String> {
trace!("ClientApiConnection::handle_connection");
let stop_token = {
let stop_source = StopSource::new();
let token = stop_source.token();
let mut inner = self.inner.borrow_mut();
inner.connect_addr = Some(connect_addr);
inner.disconnector = Some(stop_source);
token
};
// Connect the TCP socket
let stream = TcpStream::connect(connect_addr)
.await
@@ -255,283 +232,245 @@ impl ClientApiConnection {
// If it succeed, disable nagle algorithm
stream.set_nodelay(true).map_err(map_to_string)?;
// State we connected
let comproc = self.inner.lock().comproc.clone();
comproc.set_connection_state(ConnectionState::Connected(connect_addr, SystemTime::now()));
// Split the stream
cfg_if! {
if #[cfg(feature="rt-async-std")] {
use futures::AsyncReadExt;
let (reader, writer) = stream.split();
let (reader, mut writer) = stream.split();
let mut reader = BufReader::new(reader);
} else if #[cfg(feature="rt-tokio")] {
let (reader, writer) = stream.into_split();
let (reader, mut writer) = stream.into_split();
let mut reader = BufReader::new(reader);
}
}
// Requests to send
let (requests_tx, requests_rx) = flume::unbounded();
let stop_token = {
let stop_source = StopSource::new();
let token = stop_source.token();
let mut inner = self.inner.lock();
inner.connect_addr = Some(connect_addr);
inner.disconnector = Some(stop_source);
inner.request_sender = Some(requests_tx);
token
};
// Futures to process unordered
let mut unord = FuturesUnordered::new();
// Process lines
let mut line = String::new();
while let Ok(r) = reader
.read_line(&mut line)
.timeout_at(stop_token.clone())
.await
{
match r {
Ok(size) => {
// Exit on EOF
if size == 0 {
// Disconnected
return Err("Connection closed".to_owned());
}
}
Err(e) => {
let this = self.clone();
let recv_messages_future = async move {
let mut line = String::new();
while let Ok(size) = reader.read_line(&mut line).await {
// Exit on EOF
if size == 0 {
// Disconnected
return Err("Connection lost".to_owned());
break;
}
// Unmarshal json
let j = match json::parse(line.trim()) {
Ok(v) => v,
Err(e) => {
error!("failed to parse server response: {}", e);
continue;
}
};
if j["type"] == "Update" {
this.process_update(j).await;
} else if j["type"] == "Response" {
this.process_response(j).await;
}
}
//
let mut inner = this.inner.lock();
inner.request_sender = None;
};
unord.push(system_boxed(recv_messages_future));
// Unmarshal json
let j = match json::parse(line.trim()) {
Ok(v) => v,
Err(e) => {
error!("failed to parse server response: {}", e);
continue;
// Requests send processor
let send_requests_future = async move {
while let Ok(req) = requests_rx.recv_async().await {
if let Err(e) = writer.write_all(req.as_bytes()).await {
error!("failed to write request: {}", e)
}
};
if j["type"] == "Update" {
self.process_update(j).await;
}
}
};
unord.push(system_boxed(send_requests_future));
// Connection finished
Ok(())
// let rpc_network = Box::new(twoparty::VatNetwork::new(
// reader,
// writer,
// rpc_twoparty_capnp::Side::Client,
// Default::default(),
// ));
// // Create the rpc system
// let rpc_system = RpcSystem::new(rpc_network, None);
// // Process the rpc system until we decide we're done
// match self.spawn_rpc_system(connect_addr, rpc_system).await {
// Ok(()) => {}
// Err(e) => {
// error!("Failed to spawn client RPC system: {}", e);
// }
// }
// Send and receive until we're done or a stop is requested
while let Ok(Some(())) = unord.next().timeout_at(stop_token.clone()).await {}
// // Drop the server and disconnector too (if we still have it)
// let mut inner = self.inner.borrow_mut();
// let disconnect_requested = inner.disconnect_requested;
// inner.server_settings = None;
// inner.server = None;
// inner.disconnector = None;
// inner.disconnect_requested = false;
// inner.connect_addr = None;
let mut inner = self.inner.lock();
let disconnect_requested = inner.disconnect_requested;
inner.server_settings = None;
inner.request_sender = None;
inner.disconnector = None;
inner.disconnect_requested = false;
inner.connect_addr = None;
// Connection finished
if disconnect_requested {
Ok(())
} else {
Err("Connection lost".to_owned())
}
}
// pub fn cancellable<T>(&mut self, p: Promise<T, capnp::Error>) -> Promise<T, capnp::Error>
// where
// T: 'static,
// {
// let (mut cancel_instance, cancel_eventual) = {
// let inner = self.inner.borrow();
// (
// inner.cancel_eventual.instance_empty().fuse(),
// inner.cancel_eventual.clone(),
// )
// };
// let mut p = p.fuse();
async fn perform_request(&self, mut req: json::JsonValue) -> Option<json::JsonValue> {
let (sender, reply_rx) = {
let mut inner = self.inner.lock();
// Promise::from_future(async move {
// let out = select! {
// a = p => {
// a
// },
// _ = cancel_instance => {
// Err(capnp::Error::failed("cancelled".into()))
// }
// };
// drop(cancel_instance);
// cancel_eventual.reset();
// out
// })
// }
// Get the request sender
let Some(sender) = inner.request_sender.clone() else {
error!("dropping request, not connected");
return None;
};
pub async fn server_attach(&mut self) -> Result<(), String> {
// Get next id
let id = inner.next_req_id;
inner.next_req_id += 1;
// Add the id
req["id"] = id.into();
// Make a reply receiver
let (reply_tx, reply_rx) = flume::bounded(1);
inner.reply_channels.insert(id, reply_tx);
(sender, reply_rx)
};
// Send the request
let req_ndjson = req.dump() + "\n";
if let Err(e) = sender.send_async(req_ndjson).await {
error!("failed to send request: {}", e);
return None;
}
// Wait for the reply
let Ok(r) = reply_rx.recv_async().await else {
// Cancelled
return None;
};
Some(r)
}
pub async fn server_attach(&self) -> Result<(), String> {
trace!("ClientApiConnection::server_attach");
let server = {
let inner = self.inner.borrow();
inner
.server
.as_ref()
.ok_or_else(|| "Not connected, ignoring attach request".to_owned())?
.clone()
let mut req = json::JsonValue::new_object();
req["op"] = "Attach".into();
let Some(resp) = self.perform_request(req).await else {
return Err("Cancelled".to_owned());
};
let request = server.borrow().attach_request();
let response = self
.cancellable(request.send().promise)
.await
.map_err(map_to_string)?;
let reader = response
.get()
.map_err(map_to_string)?
.get_result()
.map_err(map_to_string)?;
let res: VeilidAPIResult<()> = decode_api_result(&reader);
res.map_err(map_to_string)
if resp.has_key("error") {
return Err(resp["error"].to_string());
}
Ok(())
}
pub async fn server_detach(&mut self) -> Result<(), String> {
pub async fn server_detach(&self) -> Result<(), String> {
trace!("ClientApiConnection::server_detach");
let server = {
let inner = self.inner.borrow();
inner
.server
.as_ref()
.ok_or_else(|| "Not connected, ignoring detach request".to_owned())?
.clone()
let mut req = json::JsonValue::new_object();
req["op"] = "Detach".into();
let Some(resp) = self.perform_request(req).await else {
return Err("Cancelled".to_owned());
};
let request = server.borrow().detach_request();
let response = self
.cancellable(request.send().promise)
.await
.map_err(map_to_string)?;
let reader = response
.get()
.map_err(map_to_string)?
.get_result()
.map_err(map_to_string)?;
let res: VeilidAPIResult<()> = decode_api_result(&reader);
res.map_err(map_to_string)
if resp.has_key("error") {
return Err(resp["error"].to_string());
}
Ok(())
}
pub async fn server_shutdown(&mut self) -> Result<(), String> {
pub async fn server_shutdown(&self) -> Result<(), String> {
trace!("ClientApiConnection::server_shutdown");
let server = {
let inner = self.inner.borrow();
inner
.server
.as_ref()
.ok_or_else(|| "Not connected, ignoring attach request".to_owned())?
.clone()
let mut req = json::JsonValue::new_object();
req["op"] = "Control".into();
req["args"] = json::JsonValue::new_array();
req["args"].push("Shutdown").unwrap();
let Some(resp) = self.perform_request(req).await else {
return Err("Cancelled".to_owned());
};
let request = server.borrow().shutdown_request();
let response = self
.cancellable(request.send().promise)
.await
.map_err(map_to_string)?;
response.get().map(drop).map_err(map_to_string)
if resp.has_key("error") {
return Err(resp["error"].to_string());
}
Ok(())
}
pub async fn server_debug(&mut self, what: String) -> Result<String, String> {
pub async fn server_debug(&self, what: String) -> Result<String, String> {
trace!("ClientApiConnection::server_debug");
let server = {
let inner = self.inner.borrow();
inner
.server
.as_ref()
.ok_or_else(|| "Not connected, ignoring debug request".to_owned())?
.clone()
let mut req = json::JsonValue::new_object();
req["op"] = "Debug".into();
req["command"] = what.into();
let Some(resp) = self.perform_request(req).await else {
return Err("Cancelled".to_owned());
};
let mut request = server.borrow().debug_request();
request.get().set_command(&what);
let response = self
.cancellable(request.send().promise)
.await
.map_err(map_to_string)?;
let reader = response
.get()
.map_err(map_to_string)?
.get_result()
.map_err(map_to_string)?;
let res: VeilidAPIResult<String> = decode_api_result(&reader);
res.map_err(map_to_string)
if resp.has_key("error") {
return Err(resp["error"].to_string());
}
Ok(resp["value"].to_string())
}
pub async fn server_change_log_level(
&mut self,
&self,
layer: String,
log_level: VeilidConfigLogLevel,
log_level: String,
) -> Result<(), String> {
trace!("ClientApiConnection::change_log_level");
let server = {
let inner = self.inner.borrow();
inner
.server
.as_ref()
.ok_or_else(|| "Not connected, ignoring change_log_level request".to_owned())?
.clone()
let mut req = json::JsonValue::new_object();
req["op"] = "Control".into();
req["args"] = json::JsonValue::new_array();
req["args"].push("ChangeLogLevel").unwrap();
req["args"].push(layer).unwrap();
req["args"].push(log_level).unwrap();
let Some(resp) = self.perform_request(req).await else {
return Err("Cancelled".to_owned());
};
let mut request = server.borrow().change_log_level_request();
request.get().set_layer(&layer);
let log_level_json = veilid_core::serialize_json(&log_level);
request.get().set_log_level(&log_level_json);
let response = self
.cancellable(request.send().promise)
.await
.map_err(map_to_string)?;
let reader = response
.get()
.map_err(map_to_string)?
.get_result()
.map_err(map_to_string)?;
let res: VeilidAPIResult<()> = decode_api_result(&reader);
res.map_err(map_to_string)
if resp.has_key("error") {
return Err(resp["error"].to_string());
}
Ok(())
}
pub async fn server_appcall_reply(
&mut self,
id: OperationId,
msg: Vec<u8>,
) -> Result<(), String> {
pub async fn server_appcall_reply(&self, id: u64, msg: Vec<u8>) -> Result<(), String> {
trace!("ClientApiConnection::appcall_reply");
let server = {
let inner = self.inner.borrow();
inner
.server
.as_ref()
.ok_or_else(|| "Not connected, ignoring change_log_level request".to_owned())?
.clone()
let mut req = json::JsonValue::new_object();
req["op"] = "AppCallReply".into();
req["call_id"] = id.to_string().into();
req["message"] = data_encoding::BASE64URL_NOPAD.encode(&msg).into();
let Some(resp) = self.perform_request(req).await else {
return Err("Cancelled".to_owned());
};
let mut request = server.borrow().app_call_reply_request();
request.get().set_id(id.as_u64());
request.get().set_message(&msg);
let response = self
.cancellable(request.send().promise)
.await
.map_err(map_to_string)?;
let reader = response
.get()
.map_err(map_to_string)?
.get_result()
.map_err(map_to_string)?;
let res: VeilidAPIResult<()> = decode_api_result(&reader);
res.map_err(map_to_string)
if resp.has_key("error") {
return Err(resp["error"].to_string());
}
Ok(())
}
// Start Client API connection
pub async fn connect(&mut self, connect_addr: SocketAddr) -> Result<(), String> {
pub async fn connect(&self, connect_addr: SocketAddr) -> Result<(), String> {
trace!("ClientApiConnection::connect");
// Save the address to connect to
self.handle_connection(connect_addr).await
}
// End Client API connection
pub async fn disconnect(&mut self) {
pub async fn disconnect(&self) {
trace!("ClientApiConnection::disconnect");
let disconnector = self.inner.borrow_mut().disconnector.take();
match disconnector {
Some(d) => {
self.inner.borrow_mut().disconnect_requested = true;
d.await.unwrap();
}
None => {
debug!("disconnector doesn't exist");
}
let mut inner = self.inner.lock();
if inner.disconnector.is_some() {
inner.disconnector = None;
inner.disconnect_requested = true;
}
}
}