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

use crate::list_transaction_crdt::{self, CrdtList};
use base64::{engine::general_purpose, Engine as _};
use bft_json_crdt::json_crdt::BaseCrdt;
use bft_json_crdt::json_crdt::SignedOp;
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_ten_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 from: {:?}",
            general_purpose::STANDARD.encode(&incoming_operation.author())
        );

        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))
// }

fn every_ten_seconds() -> time::Interval {
    time::interval(time::Duration::from_secs(10))
}