use async_trait::async_trait; use bft_json_crdt::json_crdt::SignedOp; use ezsockets::ClientConfig; use tokio::sync::mpsc; use crate::utils; pub struct Client { incoming_sender: mpsc::Sender, handle: ezsockets::Client, } impl Client { /// Start the websocket client pub async fn new(incoming_sender: mpsc::Sender) -> ezsockets::Client { let config = ClientConfig::new("ws://localhost:8080/websocket"); let (handle, future) = ezsockets::connect( |client| Client { incoming_sender, handle: client, }, config, ) .await; tokio::spawn(async move { future.await.unwrap(); }); handle } } #[async_trait] impl ezsockets::ClientExt for Client { // Right now we're only using the Call type for sending signed ops // change this to an enum if we need to send other types of calls, and // match on it. type Call = String; /// When we receive a text message, apply the crdt operation contained in it to our /// local crdt. async fn on_text(&mut self, text: String) -> Result<(), ezsockets::Error> { let string_sha = utils::shassy(text.clone()); println!("received text, sha: {string_sha}"); let incoming: bft_json_crdt::json_crdt::SignedOp = serde_json::from_str(&text).unwrap(); let object_sha = utils::shappy(incoming.clone()); println!("deserialized: {}", object_sha); if string_sha != object_sha { panic!("sha mismatch: {string_sha} != {object_sha}, bft-crdt has failed"); } self.incoming_sender.send(incoming).await?; Ok(()) } /// When we receive a binary message, log the bytes. Currently unused. async fn on_binary(&mut self, bytes: Vec) -> Result<(), ezsockets::Error> { tracing::info!("received bytes: {bytes:?}"); Ok(()) } /// Call this with the `Call` type to send application data to the websocket client /// (and from there, to the server). async fn on_call(&mut self, call: Self::Call) -> Result<(), ezsockets::Error> { self.handle.text(call)?; Ok(()) } }