moves code into the api layer
This commit is contained in:
@@ -1,273 +1,71 @@
|
||||
use anyhow::anyhow;
|
||||
use api::client::error::MessageError;
|
||||
use api::client::Client;
|
||||
use api::data_type::DataType;
|
||||
use api::data_value::DataValue;
|
||||
use api::request::{
|
||||
RequestMessage, RequestMessagePayload, TelemetryDefinitionRequest, TelemetryEntry,
|
||||
};
|
||||
use api::response::{ResponseMessage, ResponseMessagePayload, TelemetryDefinitionResponse};
|
||||
use api::messages::telemetry_definition::TelemetryDefinitionRequest;
|
||||
use api::messages::telemetry_entry::TelemetryEntry;
|
||||
use chrono::{DateTime, TimeDelta, Utc};
|
||||
use futures_util::future::join_all;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use num_traits::FloatConst;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::net::TcpStream;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::sync::mpsc::channel;
|
||||
use tokio::time::{sleep, sleep_until, Instant};
|
||||
use tokio::{pin, select};
|
||||
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
|
||||
use tokio::time::{sleep_until, Instant};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use uuid::Uuid;
|
||||
|
||||
struct Telemetry {
|
||||
is_connected: Arc<AtomicBool>,
|
||||
tx: mpsc::Sender<RequestMessage>,
|
||||
callback_new: mpsc::Sender<(Uuid, mpsc::Sender<ResponseMessagePayload>)>,
|
||||
callback_delete: mpsc::Sender<Uuid>,
|
||||
cancel: CancellationToken,
|
||||
client: Arc<Client>,
|
||||
}
|
||||
|
||||
struct TelemetryItemHandle<'a> {
|
||||
struct TelemetryItemHandle {
|
||||
uuid: Uuid,
|
||||
tlm: &'a Telemetry,
|
||||
client: Arc<Client>,
|
||||
}
|
||||
|
||||
impl Telemetry {
|
||||
async fn connect<R>(request: R) -> anyhow::Result<WebSocketStream<MaybeTlsStream<TcpStream>>>
|
||||
where
|
||||
R: IntoClientRequest + Unpin,
|
||||
{
|
||||
let (client, _) = match connect_async(request).await {
|
||||
Ok(o) => o,
|
||||
Err(e) => {
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
return Err(e.into());
|
||||
}
|
||||
};
|
||||
Ok(client)
|
||||
}
|
||||
|
||||
pub async fn new<R>(request: R) -> anyhow::Result<Self>
|
||||
where
|
||||
R: IntoClientRequest + Unpin + Clone + Send + 'static,
|
||||
{
|
||||
let cancel = CancellationToken::new();
|
||||
let cancel_stored = cancel.clone();
|
||||
let (tx_tx, mut tx_rx) = channel(128);
|
||||
let (callback_new_tx, mut callback_new_rx) = channel(16);
|
||||
let (callback_delete_tx, mut callback_delete_rx) = channel(16);
|
||||
|
||||
let is_connected = Arc::new(AtomicBool::new(false));
|
||||
let connected = is_connected.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
while !cancel.is_cancelled() {
|
||||
let client = Self::connect(request.clone());
|
||||
pin!(client);
|
||||
let mut client = match loop {
|
||||
break select! {
|
||||
c = &mut client => {
|
||||
c
|
||||
},
|
||||
Some(_) = tx_rx.recv() => {continue;},
|
||||
Some(_) = callback_new_rx.recv() => {continue;},
|
||||
Some(_) = callback_delete_rx.recv() => {continue;},
|
||||
};
|
||||
} {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
println!("Connect Error: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let mut close_connection = true;
|
||||
let mut callbacks = HashMap::<Uuid, mpsc::Sender<ResponseMessagePayload>>::new();
|
||||
connected.store(true, Ordering::SeqCst);
|
||||
|
||||
loop {
|
||||
select! {
|
||||
biased;
|
||||
_ = cancel.cancelled() => { break; },
|
||||
Some(msg) = callback_new_rx.recv() => {
|
||||
let (uuid, callback) = msg;
|
||||
callbacks.insert(uuid, callback);
|
||||
},
|
||||
Some(msg) = callback_delete_rx.recv() => {
|
||||
callbacks.remove(&msg);
|
||||
},
|
||||
Some(msg) = client.next() => {
|
||||
match msg {
|
||||
Ok(msg) => {
|
||||
match msg {
|
||||
Message::Text(msg) => {
|
||||
let msg: ResponseMessage = match serde_json::from_str(&msg) {
|
||||
Ok(m) => m,
|
||||
Err(e) => {
|
||||
println!("Failed to deserialize {e}");
|
||||
break;
|
||||
}
|
||||
};
|
||||
if let Some(cb) = callbacks.get_mut(&msg.uuid) {
|
||||
if let Err(e) = cb.send(msg.payload).await {
|
||||
println!("Failed to call callback {e}");
|
||||
callbacks.remove(&msg.uuid);
|
||||
}
|
||||
} else {
|
||||
unimplemented!("Unexpected Message: {msg:?}");
|
||||
}
|
||||
}
|
||||
Message::Binary(_) => unimplemented!("Binary Unsupported"),
|
||||
Message::Ping(data) => {
|
||||
if let Err(e) = client.send(Message::Pong(data)).await {
|
||||
println!("Failed to send Pong {e}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
Message::Pong(_) => {
|
||||
// Intentionally Left Empty
|
||||
}
|
||||
Message::Close(_) => {
|
||||
println!("Websocket Closed");
|
||||
close_connection = false;
|
||||
break;
|
||||
}
|
||||
Message::Frame(_) => unreachable!("Not Possible"),
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
println!("Receive Error {e}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
},
|
||||
Some(msg) = tx_rx.recv() => {
|
||||
let msg = match serde_json::to_string(&msg) {
|
||||
Ok(m) => m,
|
||||
Err(e) => {
|
||||
println!("Encode Error {e}");
|
||||
break;
|
||||
}
|
||||
};
|
||||
if let Err(e) = client.send(Message::Text(msg.into())).await {
|
||||
println!("Send Error {e}");
|
||||
break;
|
||||
}
|
||||
},
|
||||
else => { break; },
|
||||
}
|
||||
}
|
||||
connected.store(false, Ordering::SeqCst);
|
||||
if close_connection {
|
||||
if let Err(e) = client.close(None).await {
|
||||
println!("Close Error {e}");
|
||||
}
|
||||
}
|
||||
|
||||
// Callbacks gets dropped here - closing all listeners automatically
|
||||
}
|
||||
});
|
||||
|
||||
Ok(Self {
|
||||
tx: tx_tx,
|
||||
callback_new: callback_new_tx,
|
||||
callback_delete: callback_delete_tx,
|
||||
cancel: cancel_stored,
|
||||
is_connected,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn wait_connected(&self) {
|
||||
while !self.is_connected.load(Ordering::Relaxed) {
|
||||
sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a message and don't expect a response
|
||||
pub async fn send_message<P: Into<RequestMessagePayload>>(
|
||||
&self,
|
||||
payload: P,
|
||||
) -> anyhow::Result<()> {
|
||||
self.tx
|
||||
.send(RequestMessage {
|
||||
uuid: Uuid::new_v4(),
|
||||
payload: payload.into(),
|
||||
})
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn send_request<
|
||||
P: Into<RequestMessagePayload>,
|
||||
R: TryFrom<ResponseMessagePayload, Error = impl std::error::Error + Send + Sync + 'static>,
|
||||
>(
|
||||
&self,
|
||||
payload: P,
|
||||
) -> anyhow::Result<R> {
|
||||
let uuid = Uuid::new_v4();
|
||||
let (tx, mut rx) = channel(1);
|
||||
|
||||
self.wait_connected().await;
|
||||
self.callback_new.send((uuid, tx)).await?;
|
||||
self.tx
|
||||
.send(RequestMessage {
|
||||
uuid,
|
||||
payload: payload.into(),
|
||||
})
|
||||
.await?;
|
||||
|
||||
let response = rx.recv().await.ok_or(anyhow!("No Response Received"))?;
|
||||
|
||||
let response = R::try_from(response)?;
|
||||
|
||||
self.callback_delete.send(uuid).await?;
|
||||
|
||||
Ok(response)
|
||||
pub fn new(client: Arc<Client>) -> Self {
|
||||
Self { client }
|
||||
}
|
||||
|
||||
pub async fn register(
|
||||
&self,
|
||||
name: String,
|
||||
data_type: DataType,
|
||||
) -> anyhow::Result<TelemetryItemHandle<'_>> {
|
||||
let response: TelemetryDefinitionResponse = self
|
||||
) -> anyhow::Result<TelemetryItemHandle> {
|
||||
let response = self
|
||||
.client
|
||||
.send_request(TelemetryDefinitionRequest { name, data_type })
|
||||
.await?;
|
||||
|
||||
Ok(TelemetryItemHandle {
|
||||
uuid: response.uuid,
|
||||
tlm: self,
|
||||
client: self.client.clone(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Telemetry {
|
||||
fn drop(&mut self) {
|
||||
self.cancel.cancel();
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a> TelemetryItemHandle<'a> {
|
||||
impl TelemetryItemHandle {
|
||||
pub async fn publish(&self, value: DataValue, timestamp: DateTime<Utc>) -> anyhow::Result<()> {
|
||||
self.tlm
|
||||
.send_message(TelemetryEntry {
|
||||
self.client
|
||||
.send_message_if_connected(TelemetryEntry {
|
||||
uuid: self.uuid,
|
||||
value,
|
||||
timestamp,
|
||||
})
|
||||
.await?;
|
||||
.await
|
||||
.or_else(|e| match e {
|
||||
MessageError::TokioLockError(_) => Ok(()),
|
||||
e => Err(e),
|
||||
})?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
let tlm = Telemetry::new("ws://[::1]:8080/backend").await?;
|
||||
let client = Arc::new(Client::connect("ws://[::1]:8080/backend")?);
|
||||
let tlm = Telemetry::new(client);
|
||||
|
||||
{
|
||||
let time_offset = tlm
|
||||
|
||||
Reference in New Issue
Block a user