rsynko_download/
program.rs

1use crate::*;
2use alux_ext::ext;
3use std::fmt::Display;
4use std::path::Path;
5use thiserror::Error;
6
7/// Derives complete one-resource download meaning from three primitive capabilities.
8#[ext(name = DownloadExt)]
9pub impl<This, Source, FetchError, PublishError, Event, Progress> This
10where
11    Source: ?Sized,
12    This: FetchStreamAlg<Source, Error = FetchError>
13        + AtomicPublishAlg<Error = PublishError>
14        + DownloadObservationAlg<Event = Event, Progress = Progress>
15        + DownloadProgressAlg<Progress = Progress>
16        + DownloadReportAlg<Event = Event>,
17    FetchError: Display,
18    PublishError: Display,
19{
20    /// Fetches, atomically publishes, and reports exactly one terminal event.
21    ///
22    /// # Errors
23    ///
24    /// Returns the fetch or publication failure after reporting it.
25    fn download_resource(
26        &self,
27        source: &Source,
28        destination: &Path,
29    ) -> Result<u64, DownloadError<FetchError, PublishError>> {
30        let mut opened = match self.open_fetch(source) {
31            Ok(opened) => opened,
32            Err(error) => {
33                self.report_download(self.download_failed(destination, error.to_string()));
34                return Err(DownloadError::Fetch(error));
35            }
36        };
37        let mut publication = match self.begin_publication(destination) {
38            Ok(publication) => publication,
39            Err(error) => {
40                self.report_download(self.download_failed(destination, error.to_string()));
41                return Err(DownloadError::Publish(error));
42            }
43        };
44        let mut byte_count = 0_u64;
45        self.report_progress(self.download_progress(destination, byte_count, opened.total));
46        let mut buffer = vec![0_u8; 64 * 1024];
47        loop {
48            let read = match self.read_fetch(&mut opened.stream, &mut buffer) {
49                Ok(read) => read,
50                Err(error) => {
51                    self.abort_publication(publication);
52                    self.report_download(self.download_failed(destination, error.to_string()));
53                    return Err(DownloadError::Fetch(error));
54                }
55            };
56            if read == 0 {
57                break;
58            }
59            if let Err(error) = self.write_publication(&mut publication, &buffer[..read]) {
60                self.abort_publication(publication);
61                self.report_download(self.download_failed(destination, error.to_string()));
62                return Err(DownloadError::Publish(error));
63            }
64            byte_count = byte_count.saturating_add(u64::try_from(read).unwrap_or(u64::MAX));
65            self.report_progress(self.download_progress(destination, byte_count, opened.total));
66        }
67        if let Err(error) = self.commit_publication(publication) {
68            self.report_download(self.download_failed(destination, error.to_string()));
69            return Err(DownloadError::Publish(error));
70        }
71        self.report_download(self.download_succeeded(destination, byte_count));
72        Ok(byte_count)
73    }
74}
75
76/// Denotes failure to fetch or atomically publish one resource.
77#[derive(Debug, Error)]
78pub enum DownloadError<FetchError, PublishError> {
79    /// Denotes resource retrieval failure.
80    #[error("resource fetch failed: {0}")]
81    Fetch(FetchError),
82    /// Denotes atomic publication failure.
83    #[error("atomic publication failed: {0}")]
84    Publish(PublishError),
85}
rsynko_download/program.rs

rsynko_download/
program.rs
