iroh-bitswap 0.2.0

Implementation of the bitswap protocol
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

use ahash::AHashMap;
use anyhow::Result;
use cid::Cid;
use tokio::sync::mpsc;

use crate::{block::Block, Store};

/// Maintains a pool of workers that make requests to the blockstore.
#[derive(Debug)]
pub struct BlockstoreManager<S: Store> {
    store: S,
    // pending_gauge -> iroh-metrics
    // active_gauge -> iroh-metrics
}

impl<S: Store> BlockstoreManager<S> {
    /// Creates a new manager.
    pub async fn new(store: S, _worker_count: usize) -> Self {
        BlockstoreManager { store }
    }

    pub async fn stop(self) -> Result<()> {
        Ok(())
    }

    pub async fn get_block_sizes(&self, keys: &[Cid]) -> Result<AHashMap<Cid, usize>> {
        let mut res = AHashMap::new();
        if keys.is_empty() {
            return Ok(res);
        }
        let (s, mut r) = mpsc::channel(1);

        let store = self.store.clone();
        let keys = keys.to_vec();
        tokio::task::spawn(async move {
            for cid in keys {
                if let Ok(size) = store.get_size(&cid).await {
                    s.send(Some((cid, size))).await.ok();
                } else {
                    s.send(None).await.ok();
                }
            }
        });

        while let Some(r) = r.recv().await {
            if let Some((cid, block)) = r {
                res.insert(cid, block);
            }
        }

        Ok(res)
    }

    pub async fn get_blocks(&self, keys: &[Cid]) -> Result<AHashMap<Cid, Block>> {
        let mut res = AHashMap::new();
        if keys.is_empty() {
            return Ok(res);
        }
        let (s, mut r) = mpsc::channel(1);

        let store = self.store.clone();
        let keys = keys.to_vec();
        tokio::task::spawn(async move {
            for cid in keys {
                if let Ok(block) = store.get(&cid).await {
                    s.send(Some((cid, block))).await.ok();
                } else {
                    s.send(None).await.ok();
                }
            }
        });

        while let Some(r) = r.recv().await {
            if let Some((cid, block)) = r {
                res.insert(cid, block);
            }
        }

        Ok(res)
    }
}