bft-crdt-experiment/side-node/src/websocket.rs

use async_trait::async_trait;
use bft_json_crdt::json_crdt::SignedOp;
use ezsockets::ClientConfig;
use tokio::sync::mpsc;

use crate::utils;

pub struct WebSocketClient {
    incoming_sender: mpsc::Sender<SignedOp>,
    handle: ezsockets::Client<WebSocketClient>,
}

impl WebSocketClient {
    /// Start the websocket client
    pub async fn new(
        incoming_sender: mpsc::Sender<SignedOp>,
    ) -> ezsockets::Client<WebSocketClient> {
        let config = ClientConfig::new("ws://localhost:8080/websocket");
        let (handle, future) = ezsockets::connect(
            |client| WebSocketClient {
                incoming_sender,
                handle: client,
            },
            config,
        )
        .await;
        tokio::spawn(async move {
            future.await.unwrap();
        });
        handle
    }
}

#[async_trait]
impl ezsockets::ClientExt for WebSocketClient {
    // 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<u8>) -> 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(())
    }
}