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

use crate::list_transaction_crdt::{self, CrdtList};
use bft_json_crdt::json_crdt::SignedOp;
use bft_json_crdt::json_crdt::{BaseCrdt, CrdtNode};
use bft_json_crdt::keypair::Ed25519KeyPair;
use tokio::time;
use websockets::WebSocket;

/// Starts a websocket and periodically sends a BFT-CRDT message to the websocket server
pub(crate) async fn start(
    keys: Ed25519KeyPair,
    bft_crdt: &mut BaseCrdt<CrdtList>,
) -> Result<(), websockets::WebSocketError> {
    println!("connecting to websocket at ws://127.0.0.1:8080/");
    let mut ws = WebSocket::connect("ws://127.0.0.1:8080/").await?;

    let mut interval = every_two_seconds();
    let mut count = 0;
    loop {
        let _ = list_transaction_crdt::send(count, bft_crdt, &mut ws, &keys).await;

        let msg = ws.receive().await?;
        println!("Received: {:?}", msg);

        // deserialize the received websocket Frame into a string
        let msg = msg.into_text().unwrap().0;

        // deserialize the message into a Transaction struct
        let incoming_operation: SignedOp = serde_json::from_str(&msg).unwrap();
        println!("Received a new network operation: {:?}", incoming_operation);

        bft_crdt.apply(incoming_operation);

        println!("New crdt state is: {}", bft_crdt.doc.view());
        count = count + 1;
        interval.tick().await;
    }
}

fn every_two_seconds() -> time::Interval {
    time::interval(time::Duration::from_secs(2))
}