proto: get_latest_transactions

Author: tharvik

Cargo.toml

@@ -10,6 +10,7 @@ resolver = "2"
 drop = { git = "https://github.com/Distributed-EPFL/drop" }
 sieve = { git = "https://github.com/Distributed-EPFL/sieve" }
 bincode = "1.3.3"
+chrono = { version = "0.4", features = ["serde"] }
 http = "0.2"
 prost = { version = "0.8", default-features = false }
 serde = { version = "1", features = ["derive"] }

src/at2.proto

@@ -5,15 +5,14 @@ service AT2 {
 	rpc SendAsset (SendAssetRequest) returns (SendAssetReply);
 	rpc GetBalance (GetBalanceRequest) returns (GetBalanceReply);
 	rpc GetLastSequence (GetLastSequenceRequest) returns (GetLastSequenceReply);
+	rpc GetLatestTransactions (GetLatestTransactionsRequest) returns (GetLatestTransactionsReply);
 }
 
 message SendAssetRequest {
 	bytes sender = 1;
 	uint32 sequence = 2;
-
-	bytes receiver = 3;
+	bytes recipient = 3;
 	uint64 amount = 4;
-
 	bytes signature = 5;
 }
 message SendAssetReply {}
@@ -31,3 +30,15 @@ message GetLastSequenceRequest {
 message GetLastSequenceReply {
 	uint32 sequence = 1;
 }
+
+message ProcessedTransaction {
+	string timestamp = 1;
+	bytes sender = 2;
+	bytes recipient = 3;
+	uint64 amount = 4;
+}
+
+message GetLatestTransactionsRequest {}
+message GetLatestTransactionsReply {
+	repeated ProcessedTransaction transactions = 1;
+}

src/bin/client/main.rs

@@ -24,6 +24,7 @@ enum Commands {
     },
     GetBalance,
     GetLastSequence,
+    GetLatestTransactions,
 }
 
 #[derive(Debug, StructOpt)]
@@ -52,6 +53,8 @@ enum CommandsError {
     GetBalance { source: CommandError },
     #[snafu(display("get last sequence: {}", source))]
     GetLastSequence { source: CommandError },
+    #[snafu(display("get latest transactions: {}", source))]
+    GetLatestTransactions { source: CommandError },
 }
 
 fn config(cmd: CommandsConfig) -> Result<(), config::Error> {
@@ -120,6 +123,25 @@ async fn get_last_sequence() -> Result<(), CommandError> {
     Ok(())
 }
 
+async fn get_latest_transactions() -> Result<(), CommandError> {
+    let config = config::from_reader(stdin()).context(ReadConfig)?;
+
+    Client::new(config.rpc_address)
+        .context(ClientError)?
+        .get_latest_transactions()
+        .await
+        .context(ClientError)?
+        .iter()
+        .for_each(|tx| {
+            println!(
+                "{}: {} send {}¤ to {}",
+                tx.timestamp, tx.sender, tx.amount, tx.recipient,
+            )
+        });
+
+    Ok(())
+}
+
 #[tokio::main(flavor = "multi_thread")]
 async fn main() {
     let ret = match Commands::from_args() {
@@ -133,6 +155,9 @@ async fn main() {
             .context(SendAsset),
         Commands::GetBalance => get_balance().await.context(GetBalance),
         Commands::GetLastSequence => get_last_sequence().await.context(GetLastSequence),
+        Commands::GetLatestTransactions => get_latest_transactions()
+            .await
+            .context(GetLatestTransactions),
     };
 
     if let Err(err) = ret {

src/bin/server/main.rs

@@ -11,6 +11,7 @@ use tracing_fmt::FmtSubscriber;
 mod accounts;
 mod config;
 mod rpc;
+mod recent_transactions;
 
 #[derive(Debug, StructOpt)]
 enum Commands {

src/bin/server/recent_transactions.rs

@@ -0,0 +1,113 @@
+use std::collections::VecDeque;
+
+use at2_node::{FullTransaction, ThinTransaction};
+use drop::crypto::sign;
+use tokio::sync::{mpsc, oneshot};
+
+const LATEST_TRANSACTIONS_MAX_SIZE: usize = 10;
+
+#[derive(snafu::Snafu, Debug)]
+pub enum Error {
+    #[snafu(display("gone on send"))]
+    GoneOnSend,
+    #[snafu(display("gone on recv"))]
+    GoneOnRecv,
+}
+
+#[derive(Debug)]
+enum Commands {
+    Put {
+        thin: ThinTransaction,
+        sender: Box<sign::PublicKey>,
+        resp: oneshot::Sender<()>,
+    },
+    GetAll {
+        resp: oneshot::Sender<Vec<FullTransaction>>,
+    },
+}
+
+#[derive(Clone)]
+pub struct RecentTransactions {
+    agent: mpsc::Sender<Commands>,
+}
+
+/// Own the recent transactions
+struct RecentTransactionsHandler(VecDeque<FullTransaction>);
+
+impl RecentTransactions {
+    pub fn new() -> Self {
+        Self {
+            agent: RecentTransactionsHandler::new().spawn(),
+        }
+    }
+
+    /// Add a new transaction
+    pub async fn put(
+        &self,
+        sender: Box<sign::PublicKey>,
+        thin: ThinTransaction,
+    ) -> Result<(), Error> {
+        let (tx, rx) = oneshot::channel();
+
+        self.agent
+            .send(Commands::Put {
+                sender,
+                thin,
+                resp: tx,
+            })
+            .await
+            .map_err(|_| Error::GoneOnSend)?;
+
+        rx.await.map_err(|_| Error::GoneOnRecv)
+    }
+
+    /// Return the recently seen transactions
+    pub async fn get_all(&self) -> Result<Vec<FullTransaction>, Error> {
+        let (tx, rx) = oneshot::channel();
+
+        self.agent
+            .send(Commands::GetAll { resp: tx })
+            .await
+            .map_err(|_| Error::GoneOnSend)?;
+
+        rx.await.map_err(|_| Error::GoneOnRecv)
+    }
+}
+
+impl RecentTransactionsHandler {
+    fn new() -> Self {
+        Self(VecDeque::with_capacity(LATEST_TRANSACTIONS_MAX_SIZE))
+    }
+
+    fn spawn(mut self) -> mpsc::Sender<Commands> {
+        let (tx, mut rx) = mpsc::channel(32);
+
+        tokio::spawn(async move {
+            while let Some(cmd) = rx.recv().await {
+                match cmd {
+                    Commands::Put { sender, thin, resp } => {
+                        self.put(*sender, thin);
+                        let _ = resp.send(());
+                    }
+                    Commands::GetAll { resp } => {
+                        let _ = resp.send(self.get_all());
+                    }
+                }
+            }
+        });
+
+        tx
+    }
+
+    fn put(&mut self, sender: sign::PublicKey, thin: ThinTransaction) {
+        if self.0.len() == LATEST_TRANSACTIONS_MAX_SIZE {
+            self.0.pop_front();
+        }
+        let full = FullTransaction::with_thin(sender, thin);
+        self.0.push_back(full);
+    }
+
+    fn get_all(&self) -> Vec<FullTransaction> {
+        self.0.clone().into()
+    }
+}