@ -17,6 +17,7 @@ use serde_json;
|
|||||||
pub struct AccountantSkel {
|
pub struct AccountantSkel {
|
||||||
pub acc: Accountant,
|
pub acc: Accountant,
|
||||||
pub last_id: Hash,
|
pub last_id: Hash,
|
||||||
|
pub ledger: Vec<Entry>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Serialize, Deserialize, Debug)]
|
#[derive(Serialize, Deserialize, Debug)]
|
||||||
@ -37,13 +38,18 @@ pub enum Response {
|
|||||||
impl AccountantSkel {
|
impl AccountantSkel {
|
||||||
pub fn new(acc: Accountant) -> Self {
|
pub fn new(acc: Accountant) -> Self {
|
||||||
let last_id = acc.first_id;
|
let last_id = acc.first_id;
|
||||||
AccountantSkel { acc, last_id }
|
AccountantSkel {
|
||||||
|
acc,
|
||||||
|
last_id,
|
||||||
|
ledger: vec![],
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn sync(self: &mut Self) -> Hash {
|
pub fn sync(self: &mut Self) -> Hash {
|
||||||
while let Ok(entry) = self.acc.historian.receiver.try_recv() {
|
while let Ok(entry) = self.acc.historian.receiver.try_recv() {
|
||||||
self.last_id = entry.id;
|
self.last_id = entry.id;
|
||||||
println!("{}", serde_json::to_string(&entry).unwrap());
|
println!("{}", serde_json::to_string(&entry).unwrap());
|
||||||
|
self.ledger.push(entry);
|
||||||
}
|
}
|
||||||
self.last_id
|
self.last_id
|
||||||
}
|
}
|
||||||
@ -60,11 +66,20 @@ impl AccountantSkel {
|
|||||||
let val = self.acc.get_balance(&key);
|
let val = self.acc.get_balance(&key);
|
||||||
Some(Response::Balance { key, val })
|
Some(Response::Balance { key, val })
|
||||||
}
|
}
|
||||||
Request::GetEntries { .. } => Some(Response::Entries { entries: vec![] }),
|
Request::GetEntries { last_id } => {
|
||||||
|
self.sync();
|
||||||
|
let entries = self.ledger
|
||||||
|
.iter()
|
||||||
|
.skip_while(|x| x.id != last_id) // log(n) way to find Entry with id == last_id.
|
||||||
|
.skip(1) // Skip the entry with last_id.
|
||||||
|
.take(256) // TODO: Take while the serialized entries fit into a 64k UDP packet.
|
||||||
|
.cloned()
|
||||||
|
.collect();
|
||||||
|
Some(Response::Entries { entries })
|
||||||
|
}
|
||||||
Request::GetId { is_last } => Some(Response::Id {
|
Request::GetId { is_last } => Some(Response::Id {
|
||||||
id: if is_last {
|
id: if is_last {
|
||||||
self.sync();
|
self.sync()
|
||||||
self.last_id
|
|
||||||
} else {
|
} else {
|
||||||
self.acc.first_id
|
self.acc.first_id
|
||||||
},
|
},
|
||||||
|
@ -14,7 +14,6 @@ use accountant_skel::{Request, Response};
|
|||||||
pub struct AccountantStub {
|
pub struct AccountantStub {
|
||||||
pub addr: String,
|
pub addr: String,
|
||||||
pub socket: UdpSocket,
|
pub socket: UdpSocket,
|
||||||
pub last_id: Option<Hash>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl AccountantStub {
|
impl AccountantStub {
|
||||||
@ -22,7 +21,6 @@ impl AccountantStub {
|
|||||||
AccountantStub {
|
AccountantStub {
|
||||||
addr: addr.to_string(),
|
addr: addr.to_string(),
|
||||||
socket,
|
socket,
|
||||||
last_id: None,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -75,16 +73,8 @@ impl AccountantStub {
|
|||||||
self.get_id(true)
|
self.get_id(true)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn wait_on_signature(&mut self, wait_sig: &Signature) -> io::Result<()> {
|
pub fn wait_on_signature(&mut self, wait_sig: &Signature, last_id: &Hash) -> io::Result<Hash> {
|
||||||
let last_id = match self.last_id {
|
let mut last_id = *last_id;
|
||||||
None => {
|
|
||||||
let first_id = self.get_id(false)?;
|
|
||||||
self.last_id = Some(first_id);
|
|
||||||
first_id
|
|
||||||
}
|
|
||||||
Some(last_id) => last_id,
|
|
||||||
};
|
|
||||||
|
|
||||||
let req = Request::GetEntries { last_id };
|
let req = Request::GetEntries { last_id };
|
||||||
let data = serialize(&req).unwrap();
|
let data = serialize(&req).unwrap();
|
||||||
self.socket.send_to(&data, &self.addr).map(|_| ())?;
|
self.socket.send_to(&data, &self.addr).map(|_| ())?;
|
||||||
@ -94,11 +84,11 @@ impl AccountantStub {
|
|||||||
let resp = deserialize(&buf).expect("deserialize signature");
|
let resp = deserialize(&buf).expect("deserialize signature");
|
||||||
if let Response::Entries { entries } = resp {
|
if let Response::Entries { entries } = resp {
|
||||||
for Entry { id, events, .. } in entries {
|
for Entry { id, events, .. } in entries {
|
||||||
self.last_id = Some(id);
|
last_id = id;
|
||||||
for event in events {
|
for event in events {
|
||||||
if let Some(sig) = event.get_signature() {
|
if let Some(sig) = event.get_signature() {
|
||||||
if sig == *wait_sig {
|
if sig == *wait_sig {
|
||||||
return Ok(());
|
return Ok(last_id);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -106,7 +96,7 @@ impl AccountantStub {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TODO: Loop until we found it.
|
// TODO: Loop until we found it.
|
||||||
Ok(())
|
Ok(last_id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -138,7 +128,7 @@ mod tests {
|
|||||||
let last_id = acc.get_last_id().unwrap();
|
let last_id = acc.get_last_id().unwrap();
|
||||||
let sig = acc.transfer(500, &alice.keypair(), bob_pubkey, &last_id)
|
let sig = acc.transfer(500, &alice.keypair(), bob_pubkey, &last_id)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
acc.wait_on_signature(&sig).unwrap();
|
acc.wait_on_signature(&sig, &last_id).unwrap();
|
||||||
assert_eq!(acc.get_balance(&bob_pubkey).unwrap().unwrap(), 500);
|
assert_eq!(acc.get_balance(&bob_pubkey).unwrap().unwrap(), 500);
|
||||||
*exit.lock().unwrap() = true;
|
*exit.lock().unwrap() = true;
|
||||||
for t in threads.iter() {
|
for t in threads.iter() {
|
||||||
|
@ -65,7 +65,7 @@ fn main() {
|
|||||||
acc.transfer_signed(tr).unwrap();
|
acc.transfer_signed(tr).unwrap();
|
||||||
}
|
}
|
||||||
println!("Waiting for last transaction to be confirmed...",);
|
println!("Waiting for last transaction to be confirmed...",);
|
||||||
acc.wait_on_signature(&sig).unwrap();
|
acc.wait_on_signature(&sig, &last_id).unwrap();
|
||||||
|
|
||||||
let duration = now.elapsed();
|
let duration = now.elapsed();
|
||||||
let ns = duration.as_secs() * 1_000_000_000 + duration.subsec_nanos() as u64;
|
let ns = duration.as_secs() * 1_000_000_000 + duration.subsec_nanos() as u64;
|
||||||
|
Reference in New Issue
Block a user