apt-swarm 0.6.0

🥸 Experimental p2p gossip network for OpenPGP signature transparency 🥸
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

use super::{Database, DatabaseClient};
use crate::db;
use crate::errors::*;
use crate::signed::Signed;
use crate::sync;
use async_trait::async_trait;
use sequoia_openpgp::Fingerprint;
use tokio::sync::mpsc;

pub enum Query {
    AddRelease(Fingerprint, Signed, mpsc::Sender<String>),
    IndexFromScan(sync::TreeQuery, mpsc::Sender<(String, usize)>),
    Spill(Vec<u8>, mpsc::Sender<Vec<(db::Key, db::Value)>>),
    GetValue(Vec<u8>, mpsc::Sender<db::Value>),
    // Delete(Vec<u8>, mpsc::Sender<()>),
    Count(Vec<u8>, mpsc::Sender<u64>),
}

#[derive(Debug)]
pub struct DatabaseServer {
    db: Database,
    rx: mpsc::Receiver<Query>,
}

impl DatabaseServer {
    pub fn new(db: Database) -> (DatabaseServer, DatabaseServerClient) {
        let (tx, rx) = mpsc::channel(32);

        let server = DatabaseServer { db, rx };
        let client = DatabaseServerClient { tx };

        (server, client)
    }

    pub async fn run(&mut self) -> Result<()> {
        while let Some(msg) = self.rx.recv().await {
            match msg {
                Query::AddRelease(fp, signed, tx) => {
                    let hash = self.db.add_release(&fp, &signed).await?;
                    tx.send(hash).await.ok();
                }
                Query::IndexFromScan(query, tx) => {
                    let ret = self.db.index_from_scan(&query).await?;
                    tx.send(ret).await.ok();
                }
                Query::Spill(prefix, tx) => {
                    let ret = self.db.spill(&prefix).await?;
                    tx.send(ret).await.ok();
                }
                Query::GetValue(key, tx) => {
                    let ret = self.db.get_value(&key).await?;
                    tx.send(ret).await.ok();
                }
                Query::Count(key, tx) => {
                    let ret = self.db.count(&key).await?;
                    tx.send(ret).await.ok();
                }
            }
        }
        Ok(())
    }
}

#[derive(Debug, Clone)]
pub struct DatabaseServerClient {
    tx: mpsc::Sender<Query>,
}

impl DatabaseServerClient {
    async fn request<T>(&self, query: Query, mut rx: mpsc::Receiver<T>) -> Result<T> {
        self.tx
            .send(query)
            .await
            .map_err(|_| anyhow!("Database server disconnected"))?;
        let ret = rx.recv().await.context("Database server disconnected")?;
        Ok(ret)
    }
}

#[async_trait]
impl DatabaseClient for DatabaseServerClient {
    async fn add_release(&mut self, fp: &Fingerprint, signed: &Signed) -> Result<String> {
        let (tx, rx) = mpsc::channel(1);
        let query = Query::AddRelease(fp.clone(), signed.clone(), tx);
        self.request(query, rx).await
    }

    async fn index_from_scan(&mut self, query: &sync::TreeQuery) -> Result<(String, usize)> {
        let (tx, rx) = mpsc::channel(1);
        let query = Query::IndexFromScan(query.clone(), tx);
        self.request(query, rx).await
    }

    async fn spill(&self, prefix: &[u8]) -> Result<Vec<(db::Key, db::Value)>> {
        let (tx, rx) = mpsc::channel(1);
        let query = Query::Spill(prefix.to_vec(), tx);
        self.request(query, rx).await
    }

    async fn get_value(&self, key: &[u8]) -> Result<db::Value> {
        let (tx, rx) = mpsc::channel(1);
        let query = Query::GetValue(key.to_vec(), tx);
        self.request(query, rx).await
    }

    async fn count(&mut self, key: &[u8]) -> Result<u64> {
        let (tx, rx) = mpsc::channel(1);
        let query = Query::Count(key.to_vec(), tx);
        self.request(query, rx).await
    }
}