Fixed transaction structure. Remote transactions not apply()ing yet.

This commit is contained in:
Dave Hrycyszyn
2024-06-05 18:07:59 +01:00
parent c866774612
commit e85683e865
3 changed files with 48 additions and 34 deletions

View File

@@ -109,7 +109,7 @@ pub struct BaseCrdt<T: CrdtNode> {
/// An [`Op<Value>`] with a few bits of extra metadata /// An [`Op<Value>`] with a few bits of extra metadata
#[serde_as] #[serde_as]
#[derive(Clone, Serialize, Deserialize, Debug)] #[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
pub struct SignedOp { pub struct SignedOp {
// Note that this can be different from the author of the inner op as the inner op could have been created // Note that this can be different from the author of the inner op as the inner op could have been created
// by a different person // by a different person

View File

@@ -81,7 +81,7 @@ pub fn parse_field(path: Vec<PathSegment>) -> Option<String> {
} }
/// Represents a single node in a CRDT /// Represents a single node in a CRDT
#[derive(Clone, Serialize, Deserialize, Debug)] #[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
pub struct Op<T> pub struct Op<T>
where where
T: CrdtNode, T: CrdtNode,

View File

@@ -1,14 +1,17 @@
use std::io::{self, BufRead};
use bft_crdt_derive::add_crdt_fields; use bft_crdt_derive::add_crdt_fields;
use bft_json_crdt::json_crdt::SignedOp; use bft_json_crdt::json_crdt::SignedOp;
use bft_json_crdt::keypair::KeyPair; use bft_json_crdt::keypair::{Ed25519KeyPair, KeyPair};
use bft_json_crdt::{ use bft_json_crdt::{
json_crdt::{BaseCrdt, CrdtNode, IntoCrdtNode}, json_crdt::{BaseCrdt, CrdtNode, IntoCrdtNode},
keypair::{make_keypair, ED25519_PUBLIC_KEY_LENGTH}, keypair::make_keypair,
list_crdt::ListCrdt, list_crdt::ListCrdt,
op::ROOT_ID, op::ROOT_ID,
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use tokio::time::{self}; use serde_json::{json, Value};
// use tokio::time::{self};
use websockets::WebSocket; use websockets::WebSocket;
/// Starts a websocket and periodically sends a BFT-CRDT message to the websocket server /// Starts a websocket and periodically sends a BFT-CRDT message to the websocket server
@@ -21,21 +24,16 @@ pub(crate) async fn start() -> Result<(), websockets::WebSocketError> {
let mut bft_crdt = BaseCrdt::<ListExample>::new(&keys); let mut bft_crdt = BaseCrdt::<ListExample>::new(&keys);
println!("Generated a new CRDT with public key: {}", keys.public()); println!("Generated a new CRDT with public key: {}", keys.public());
// generate a placeholder transaction // let mut interval = every_two_seconds();
let transaction = generate_transaction();
let json = convert_to_json(&transaction).unwrap();
// next job is to keep adding to this guy
let signed_op = bft_crdt.doc.list.insert(ROOT_ID, json.clone()).sign(&keys);
println!("SignedOp before send is: {:?}", signed_op);
let mut interval = every_two_seconds();
loop { loop {
interval.tick().await; let stdin = io::stdin();
println!("Sending: {:?}", signed_op); let mut handle = stdin.lock();
ws.send_text(serde_json::to_string(&signed_op).unwrap())
.await?; let mut line = String::new();
handle.read_line(&mut line).unwrap();
println!("You entered: {}", line);
let _ = send_a_transaction(line, &mut bft_crdt, &mut ws, &keys).await;
let msg = ws.receive().await?; let msg = ws.receive().await?;
println!("Received: {:?}", msg); println!("Received: {:?}", msg);
@@ -46,15 +44,15 @@ pub(crate) async fn start() -> Result<(), websockets::WebSocketError> {
// deserialize the message into a Transaction struct // deserialize the message into a Transaction struct
let operation: SignedOp = serde_json::from_str(&msg).unwrap(); let operation: SignedOp = serde_json::from_str(&msg).unwrap();
// TODO: bft_crdt.apply() changes in here when we receive socket input from other nodes
bft_crdt.apply(operation); bft_crdt.apply(operation);
println!("New crdt state is: {}", bft_crdt.doc.view()) println!("New crdt state is: {}", bft_crdt.doc.view())
} }
} }
fn every_two_seconds() -> time::Interval { // fn every_two_seconds() -> time::Interval {
time::interval(time::Duration::from_secs(2)) // time::interval(time::Duration::from_secs(2))
} // }
#[add_crdt_fields] #[add_crdt_fields]
#[derive(Clone, CrdtNode, Serialize, Deserialize)] #[derive(Clone, CrdtNode, Serialize, Deserialize)]
@@ -71,17 +69,33 @@ struct Transaction {
amount: f64, amount: f64,
} }
fn generate_transaction() -> Transaction { fn generate_transaction(from: String) -> Value {
Transaction { json!({
from: "Alice".to_string(), "from": from,
to: "Bob".to_string(), "to": "Bob",
amount: 100.0, "amount": 100.0
path: vec![], })
id: [0; ED25519_PUBLIC_KEY_LENGTH],
}
} }
fn convert_to_json(transaction: &Transaction) -> serde_json::Result<String> { async fn send_a_transaction(
let json = serde_json::to_string(&transaction).unwrap(); input: String,
Ok(json) bft_crdt: &mut BaseCrdt<ListExample>,
ws: &mut WebSocket,
keys: &Ed25519KeyPair,
) -> Result<(), websockets::WebSocketError> {
// generate a placeholder transaction
let transaction = generate_transaction(input);
// next job is to keep adding to this guy
let signed_op = bft_crdt
.doc
.list
.insert(ROOT_ID, transaction.clone())
.sign(&keys);
println!("SignedOp being sent is: {:?}", signed_op);
println!("Sending: {:?}", signed_op);
Ok(ws
.send_text(serde_json::to_string(&signed_op).unwrap())
.await?)
} }