vielpork 0.1.0

A multi-threaded downloader with multi-reporter built in Rust
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

use crate::base::enums::{DownloadResult, OperationType, ProgressEvent};
use crate::base::structs::DownloadProgress;
use crate::base::traits::{ProgressReporter, ResultReporter};
use crate::error::Result;
use async_trait::async_trait;

#[derive(Debug, Clone)]
pub struct CliReporterBoardcastMpsc {
    inner_tx: tokio::sync::broadcast::Sender<ProgressEvent>,
    buffer_size: usize,
}
impl CliReporterBoardcastMpsc {
    pub fn new(buffer_size: usize) -> Self {
        let (inner_tx, _) = tokio::sync::broadcast::channel(buffer_size);
        Self {
            inner_tx,
            buffer_size,
        }
    }

    pub fn subscribe_mpsc(&self) -> tokio::sync::mpsc::Receiver<ProgressEvent> {
        let (tx, rx) = tokio::sync::mpsc::channel(self.buffer_size);
        let mut inner_rx = self.inner_tx.subscribe();

        tokio::spawn(async move {
            loop {
                match inner_rx.recv().await {
                    Ok(event) => {
                        if tx.send(event).await.is_err() {
                            break;
                        }
                    }
                    Err(_) => {
                        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                    }
                }
            }
        });

        rx
    }

    // 发送事件的方法
    pub async fn send(&self, event: ProgressEvent) -> Result<usize> {
        self.inner_tx.send(event)?;
        Ok(self.inner_tx.receiver_count())
    }

    // 创建新订阅者
    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<ProgressEvent> {
        self.inner_tx.subscribe()
    }
}

#[async_trait]
impl ProgressReporter for CliReporterBoardcastMpsc {
    async fn start_task(&self, task_id: u32, total: u64) -> Result<()> {
        self.send(ProgressEvent::Start { task_id, total }).await?;
        Ok(())
    }

    async fn update_progress(&self, task_id: u32, progress: &DownloadProgress) -> Result<()> {
        self.send(ProgressEvent::Update {
            task_id,
            progress: progress.clone(),
        })
        .await?;
        Ok(())
    }

    async fn finish_task(&self, task_id: u32, finish: DownloadResult) -> Result<()> {
        self.send(ProgressEvent::Finish { task_id, finish }).await?;
        Ok(())
    }
}

#[async_trait]
impl ResultReporter for CliReporterBoardcastMpsc {
    async fn operation_result(
        &self,
        operation: OperationType,
        task_id: u32,
        code: u32,
        message: String,
    ) -> Result<()> {
        self.send(ProgressEvent::OperationResult {
            operation,
            task_id,
            code,
            message,
        })
        .await?;
        Ok(())
    }
}