use crate::bcms_capnp::b_c_m_s as bcms;
use crate::data::IoTData;
use crate::devices::{DeviceInfo, WorkingMode};
use crate::{Error, Result};
use capnp_rpc::{rpc_twoparty_capnp, twoparty, RpcSystem};
use futures_util::io::AsyncReadExt;
use std::net::SocketAddr;
use std::time::Duration;
use tokio::net::TcpStream;
use tokio::time::timeout;
pub struct RpcClient {
server_address: SocketAddr,
}
impl RpcClient {
pub fn new(server_address: SocketAddr) -> Self {
Self { server_address }
}
async fn with_client<F, Fut, T>(&self, f: F) -> Result<T>
where
F: FnOnce(bcms::Client) -> Fut,
Fut: std::future::Future<Output = Result<T>>,
{
tracing::debug!("RPC client: connecting to server");
let stream = timeout(
Duration::from_secs(10),
TcpStream::connect(&self.server_address)
).await.map_err(|_| Error::General("RPC connection timeout".to_string()))??;
let (reader, writer) = tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split();
let rpc_network = Box::new(twoparty::VatNetwork::new(
reader,
writer,
rpc_twoparty_capnp::Side::Client,
Default::default(),
));
let mut rpc_system = RpcSystem::new(rpc_network, None);
let bcms_client: bcms::Client = rpc_system.bootstrap(rpc_twoparty_capnp::Side::Server);
tracing::debug!("RPC client: got bootstrap client, executing request");
let result = tokio::select! {
res = f(bcms_client) => {
tracing::debug!("RPC client: request completed");
res
}
res = rpc_system => {
tracing::error!("RPC system ended unexpectedly: {:?}", res);
Err(crate::Error::General("RPC system ended unexpectedly".to_string()))
}
};
tracing::debug!("RPC client: shutting down");
result
}
pub async fn list_devices(&self, only_approved: bool) -> Result<(Vec<DeviceInfo>, Vec<String>)> {
tracing::debug!("RPC client: requesting device list with only_approved={}", only_approved);
self.with_client(|client| async move {
let mut request = client.list_request();
request.get().set_only_approved(only_approved);
tracing::debug!("RPC client: sending list request");
let response = request.send().promise.await?;
tracing::debug!("RPC client: received response");
let results = response.get()?;
let devices_reader = results.get_devices()?;
let mut devices = Vec::new();
for device_reader in devices_reader.iter() {
let address = device_reader.get_address()?.to_string()?;
let name = device_reader.get_name()?.to_string()?;
let approved = device_reader.get_approved();
let paired = device_reader.get_paired();
let mut device = DeviceInfo::new(address, name);
device.approved = approved;
device.paired = paired;
devices.push(device);
}
let errors_reader = results.get_errors()?;
let mut errors = Vec::new();
for error_reader in errors_reader.iter() {
errors.push(error_reader?.to_string()?);
}
tracing::debug!("RPC client: parsed {} devices, {} errors", devices.len(), errors.len());
Ok((devices, errors))
}).await
}
pub async fn approve_device(&self, address: &str) -> Result<(bool, Vec<String>)> {
let address = address.to_string();
self.with_client(|client| async move {
let mut request = client.approve_request();
request.get().set_address(&address);
let response = request.send().promise.await?;
let results = response.get()?;
let status = results.get_status();
let errors_reader = results.get_errors()?;
let mut errors = Vec::new();
for error_reader in errors_reader.iter() {
errors.push(error_reader?.to_string()?);
}
Ok((status, errors))
}).await
}
pub async fn remove_device(&self, address: &str) -> Result<(bool, Vec<String>)> {
let address = address.to_string();
self.with_client(|client| async move {
let mut request = client.remove_request();
request.get().set_address(&address);
let response = request.send().promise.await?;
let results = response.get()?;
let status = results.get_status();
let errors_reader = results.get_errors()?;
let mut errors = Vec::new();
for error_reader in errors_reader.iter() {
errors.push(error_reader?.to_string()?);
}
Ok((status, errors))
}).await
}
pub async fn get_mode(&self) -> Result<WorkingMode> {
self.with_client(|client| async move {
let request = client.mode_request();
let response = request.send().promise.await?;
let results = response.get()?;
let mode_capnp = results.get_mode()?;
let mode = match mode_capnp {
crate::bcms_capnp::BCMSWorkingMode::Any => WorkingMode::Any,
crate::bcms_capnp::BCMSWorkingMode::Approved => WorkingMode::Approved,
};
Ok(mode)
}).await
}
pub async fn set_mode(&self, mode: WorkingMode) -> Result<(bool, Vec<String>)> {
self.with_client(|client| async move {
let mut request = client.set_mode_request();
let capnp_mode = match mode {
WorkingMode::Any => crate::bcms_capnp::BCMSWorkingMode::Any,
WorkingMode::Approved => crate::bcms_capnp::BCMSWorkingMode::Approved,
};
request.get().set_mode(capnp_mode);
let response = request.send().promise.await?;
let results = response.get()?;
let status = results.get_status();
let errors_reader = results.get_errors()?;
let mut errors = Vec::new();
for error_reader in errors_reader.iter() {
errors.push(error_reader?.to_string()?);
}
Ok((status, errors))
}).await
}
pub async fn pair_device(&self, address: &str) -> Result<(bool, Vec<String>)> {
let address = address.to_string();
self.with_client(|client| async move {
let mut request = client.pair_request();
request.get().set_address(&address);
let response = request.send().promise.await?;
let results = response.get()?;
let status = results.get_status();
let errors_reader = results.get_errors()?;
let mut errors = Vec::new();
for error_reader in errors_reader.iter() {
errors.push(error_reader?.to_string()?);
}
Ok((status, errors))
}).await
}
pub async fn unpair_device(&self, address: &str) -> Result<(bool, Vec<String>)> {
let address = address.to_string();
self.with_client(|client| async move {
let mut request = client.unpair_request();
request.get().set_address(&address);
let response = request.send().promise.await?;
let results = response.get()?;
let status = results.get_status();
let errors_reader = results.get_errors()?;
let mut errors = Vec::new();
for error_reader in errors_reader.iter() {
errors.push(error_reader?.to_string()?);
}
Ok((status, errors))
}).await
}
pub async fn is_device_paired(&self, address: &str) -> Result<(bool, Vec<String>)> {
let address = address.to_string();
self.with_client(|client| async move {
let mut request = client.is_paired_request();
request.get().set_address(&address);
let response = request.send().promise.await?;
let results = response.get()?;
let status = results.get_status();
let errors_reader = results.get_errors()?;
let mut errors = Vec::new();
for error_reader in errors_reader.iter() {
errors.push(error_reader?.to_string()?);
}
Ok((status, errors))
}).await
}
pub async fn submit_data(
&self,
device_address: &str,
device_name: &str,
connection_type: &str,
data: Vec<IoTData>
) -> Result<(bool, Vec<String>)> {
let device_address = device_address.to_string();
let device_name = device_name.to_string();
let connection_type = connection_type.to_string();
self.with_client(|client| async move {
let mut request = client.submit_data_request();
let mut payload_builder = request.get().init_payload();
{
let mut device_builder = payload_builder.reborrow().init_device();
device_builder.set_address(&device_address);
device_builder.set_name(&device_name);
device_builder.set_connection_type(&connection_type);
}
{
let mut data_builder = payload_builder.reborrow().init_data(data.len() as u32);
for (i, iot_data) in data.iter().enumerate() {
let json_data = serde_json::to_string(iot_data)?;
data_builder.set(i as u32, &json_data);
}
}
let response = request.send().promise.await?;
let results = response.get()?;
let status = results.get_status();
let errors_reader = results.get_errors()?;
let mut errors = Vec::new();
for error_reader in errors_reader.iter() {
errors.push(error_reader?.to_string()?);
}
Ok((status, errors))
}).await
}
pub async fn get_stats(&self) -> Result<(Vec<DeviceStats>, u32, Vec<String>)> {
self.with_client(|client| async move {
let request = client.stats_request();
let response = request.send().promise.await?;
let results = response.get()?;
let stats_result = results.get_result()?;
let devices_reader = stats_result.get_devices()?;
let mut device_stats = Vec::new();
for device_reader in devices_reader.iter() {
let address = device_reader.get_address()?.to_string()?;
let name = device_reader.get_name()?.to_string()?;
let approved = device_reader.get_approved();
let paired = device_reader.get_paired();
let connection_type = device_reader.get_connection_type()?.to_string()?;
let last_seen = device_reader.get_last_seen();
let backend_registered = device_reader.get_backend_registered();
let data_types_reader = device_reader.get_data_types()?;
let mut data_type_stats = Vec::new();
for data_type_reader in data_types_reader.iter() {
let data_type = data_type_reader.get_data_type()?.to_string()?;
let count = data_type_reader.get_count();
let last_timestamp = data_type_reader.get_last_timestamp();
data_type_stats.push(DataTypeStats {
data_type,
count,
last_timestamp,
});
}
device_stats.push(DeviceStats {
address,
name,
approved,
paired,
connection_type,
last_seen,
backend_registered,
data_types: data_type_stats,
});
}
let total_data_points = stats_result.get_total_data_points();
let errors_reader = stats_result.get_errors()?;
let mut errors = Vec::new();
for error_reader in errors_reader.iter() {
errors.push(error_reader?.to_string()?);
}
Ok((device_stats, total_data_points, errors))
}).await
}
}
#[derive(Debug, Clone)]
pub struct DeviceStats {
pub address: String,
pub name: String,
pub approved: bool,
pub paired: bool,
pub connection_type: String,
pub last_seen: i64,
pub backend_registered: bool,
pub data_types: Vec<DataTypeStats>,
}
#[derive(Debug, Clone)]
pub struct DataTypeStats {
pub data_type: String,
pub count: u32,
pub last_timestamp: i64,
}