p2panda_blobs/
download.rs

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
// SPDX-License-Identifier: MIT OR Apache-2.0

use anyhow::{ensure, Result};
use futures_lite::{Stream, StreamExt};
use iroh_base::rpc::RpcError;
use iroh_blobs::downloader::{DownloadRequest, Downloader};
use iroh_blobs::get::db::DownloadProgress;
use iroh_blobs::get::Stats;
use iroh_blobs::util::local_pool::LocalPoolHandle;
use iroh_blobs::util::progress::{AsyncChannelProgressSender, ProgressSender};
use iroh_blobs::{BlobFormat, Hash as IrohHash, HashAndFormat};
use p2panda_core::Hash;
use p2panda_net::{Network, TopicId};
use p2panda_sync::TopicQuery;
use serde::{Deserialize, Serialize};

/// Status of a blob download attempt.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum DownloadBlobEvent {
    Done,
    Abort(RpcError),
}

pub(crate) async fn download_blob<T: TopicQuery + TopicId + 'static>(
    network: Network<T>,
    downloader: Downloader,
    pool_handle: LocalPoolHandle,
    hash: Hash,
) -> impl Stream<Item = DownloadBlobEvent> {
    let (sender, receiver) = async_channel::bounded(1024);
    let progress = AsyncChannelProgressSender::new(sender);
    let hash_and_format = HashAndFormat {
        hash: IrohHash::from_bytes(*hash.as_bytes()),
        format: BlobFormat::Raw,
    };

    pool_handle.spawn_detached(move || async move {
        match download_queued(network, &downloader, hash_and_format, progress.clone()).await {
            Ok(stats) => {
                progress.send(DownloadProgress::AllDone(stats)).await.ok();
            }
            Err(err) => {
                progress
                    .send(DownloadProgress::Abort(err.into()))
                    .await
                    .ok();
            }
        }
    });

    receiver.filter_map(|event| match event {
        DownloadProgress::AllDone(_) => Some(DownloadBlobEvent::Done),
        // @TODO: Use own error type here
        DownloadProgress::Abort(err) => Some(DownloadBlobEvent::Abort(err)),
        _ => {
            // @TODO: Add more event types
            None
        }
    })
}

async fn download_queued<T: TopicQuery + TopicId + 'static>(
    network: Network<T>,
    downloader: &Downloader,
    hash_and_format: HashAndFormat,
    progress: AsyncChannelProgressSender<DownloadProgress>,
) -> Result<Stats> {
    let addrs = network.known_peers().await?;
    ensure!(!addrs.is_empty(), "no way to reach a node for download");

    let req = DownloadRequest::new(hash_and_format, addrs).progress_sender(progress);
    let handle = downloader.queue(req).await;

    let stats = handle.await?;
    Ok(stats)
}