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

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

use crate::utils;

pub(crate) struct WebSocketClient {
    incoming_sender: mpsc::Sender<SignedOp>,
}

impl WebSocketClient {
    /// Start the websocket client
    pub(crate) async fn start(
        incoming_sender: mpsc::Sender<SignedOp>,
        // stdin_sender: mpsc::Sender<String>,
    ) -> tokio::task::JoinHandle<()> {
        tracing_subscriber::fmt::init();
        let config = ClientConfig::new("ws://localhost:8080/websocket");
        let (handle, future) =
            ezsockets::connect(|_client| WebSocketClient { incoming_sender }, config).await;
        tokio::spawn(async move {
            future.await.unwrap();
        });
        tokio::spawn(async move {
            let stdin = std::io::stdin();
            let lines = stdin.lock().lines();
            for line in lines {
                println!("We don't get here until we send a message");
                let line = line.unwrap();
                let signed_op = if let "exit" = line.as_str() {
                    break;
                    // } else if let "trace" = line.as_str() {
                    // node.trace_crdt();
                    // continue;
                } else {
                    let fake = utils::fake_transaction("foo123".to_string());
                    // stdin_sender.send(fake).await.unwrap();
                    fake
                };
                tracing::info!("sending {:?}", signed_op);
                let json = serde_json::to_string(&signed_op).unwrap();
                handle.text(json).unwrap();
            }
        })
    }
}

#[async_trait]
impl ezsockets::ClientExt for WebSocketClient {
    type Call = ();

    async fn on_text(&mut self, text: String) -> Result<(), ezsockets::Error> {
        tracing::info!("received message: {text}");
        let incoming: bft_json_crdt::json_crdt::SignedOp = serde_json::from_str(&text).unwrap();
        tracing::info!("received signed op: {incoming:?}");
        self.incoming_sender.send(incoming).await.unwrap();
        Ok(())
    }

    async fn on_binary(&mut self, bytes: Vec<u8>) -> Result<(), ezsockets::Error> {
        tracing::info!("received bytes: {bytes:?}");
        Ok(())
    }

    async fn on_call(&mut self, call: Self::Call) -> Result<(), ezsockets::Error> {
        let () = call;
        Ok(())
    }
}