refactor(oma-fetch,oma-pm,oma-refresh,oma-topics)!: use async closure (#370)

Author: eatradish

oma-fetch/examples/downloads.rs

@@ -64,7 +64,7 @@ async fn main() {
         .unwrap();
 
     let summary = download_manager
-        .start_download(|event| async {
+        .start_download(async |event| {
             if let Err(e) = tx.send_async(event).await {
                 eprintln!("Got Error: {:#?}", e);
             }

oma-fetch/src/download.rs

@@ -1,7 +1,6 @@
 use crate::{CompressFile, DownloadSource, Event, checksum::ChecksumValidator};
 use std::{
     fs::Permissions,
-    future::Future,
     io::{self, ErrorKind, SeekFrom},
     os::unix::fs::PermissionsExt,
     path::Path,
@@ -101,11 +100,7 @@ pub enum SingleDownloadError {
 }
 
 impl SingleDownloader<'_> {
-    pub(crate) async fn try_download<F, Fut>(self, callback: &F) -> DownloadResult
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+    pub(crate) async fn try_download(self, callback: &impl AsyncFn(Event)) -> DownloadResult {
         let mut sources = self.entry.source.clone();
         assert!(!sources.is_empty());
 
@@ -164,16 +159,12 @@ impl SingleDownloader<'_> {
     }
 
     /// Download file with retry (http)
-    async fn try_http_download<F, Fut>(
+    async fn try_http_download(
         &self,
         source: &DownloadSource,
         auth: &Option<(String, String)>,
-        callback: &F,
-    ) -> Result<bool, SingleDownloadError>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+        callback: &impl AsyncFn(Event),
+    ) -> Result<bool, SingleDownloadError> {
         let mut times = 1;
         let mut allow_resume = self.entry.allow_resume;
         loop {
@@ -210,17 +201,13 @@ impl SingleDownloader<'_> {
         }
     }
 
-    async fn http_download<F, Fut>(
+    async fn http_download(
         &self,
         allow_resume: bool,
         source: &DownloadSource,
         auth: &Option<(String, String)>,
-        callback: &F,
-    ) -> Result<bool, SingleDownloadError>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+        callback: &impl AsyncFn(Event),
+    ) -> Result<bool, SingleDownloadError> {
         let file = self.entry.dir.join(&*self.entry.filename);
         let file_exist = file.exists();
         let mut file_size = file.metadata().ok().map(|x| x.len()).unwrap_or(0);
@@ -588,16 +575,12 @@ impl SingleDownloader<'_> {
     }
 
     /// Download local source file
-    async fn download_local<F, Fut>(
+    async fn download_local(
         &self,
         source: &DownloadSource,
         as_symlink: bool,
-        callback: &F,
-    ) -> Result<bool, SingleDownloadError>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+        callback: &impl AsyncFn(Event),
+    ) -> Result<bool, SingleDownloadError> {
         debug!("{:?}", self.entry);
         let msg = self.progress_msg();
 
@@ -701,16 +684,12 @@ impl SingleDownloader<'_> {
         Ok(true)
     }
 
-    async fn checksum_local<F, Fut>(
+    async fn checksum_local(
         &self,
-        callback: &F,
+        callback: &impl AsyncFn(Event),
         url_path: &Path,
         hash: &crate::checksum::Checksum,
-    ) -> Result<(), SingleDownloadError>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+    ) -> Result<(), SingleDownloadError> {
         let mut f = fs::File::open(url_path).await.context(OpenSnafu)?;
         let (size, finish) = checksum(callback, &mut f, &mut hash.get_validator()).await;
 
@@ -724,11 +703,11 @@ impl SingleDownloader<'_> {
     }
 }
 
-async fn checksum<F, Fut>(callback: &F, f: &mut File, v: &mut ChecksumValidator) -> (u64, bool)
-where
-    F: Fn(Event) -> Fut,
-    Fut: Future<Output = ()>,
-{
+async fn checksum(
+    callback: &impl AsyncFn(Event),
+    f: &mut File,
+    v: &mut ChecksumValidator,
+) -> (u64, bool) {
     let mut reader = tokio::io::BufReader::with_capacity(READ_FILE_BUFSIZE, f);
 
     let mut read = 0;

oma-fetch/src/lib.rs

@@ -3,7 +3,7 @@ use std::{cmp::Ordering, path::PathBuf, time::Duration};
 use bon::{Builder, builder};
 use checksum::Checksum;
 use download::{EmptySource, SingleDownloader, SuccessSummary};
-use futures::{Future, StreamExt};
+use futures::StreamExt;
 
 use reqwest::{Client, Method, RequestBuilder};
 use tracing::debug;
@@ -203,11 +203,10 @@ impl Summary {
 
 impl DownloadManager<'_> {
     /// Start download
-    pub async fn start_download<F, Fut>(&self, callback: F) -> Result<Summary, EmptySource>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+    pub async fn start_download(
+        &self,
+        callback: impl AsyncFn(Event),
+    ) -> Result<Summary, EmptySource> {
         let mut tasks = Vec::new();
         let mut list = vec![];
         for (i, c) in self.download_list.iter().enumerate() {

oma-pm/examples/install_fish.rs

@@ -123,7 +123,7 @@ fn main() -> Result<(), OmaAptError> {
             network_thread: None,
             auth_config: Some(&AuthConfig::system("/").unwrap()),
         },
-        |event| async {
+        async |event| {
             if let Err(e) = tx.send_async(event).await {
                 eprintln!("{:#?}", e);
             }

oma-pm/src/apt.rs

@@ -10,7 +10,6 @@ use ahash::HashSet;
 use apt_auth_config::AuthConfig;
 use bon::{Builder, builder};
 pub use oma_apt::cache::Upgrade;
-use std::future::Future;
 use tokio::runtime::Runtime;
 use zbus::Connection;
 
@@ -418,16 +417,15 @@ impl OmaApt {
     }
 
     /// Download packages
-    pub fn download<F, Fut>(
+    pub fn download<F>(
         &self,
         client: &Client,
         pkgs: Vec<OmaPackage>,
         config: DownloadConfig<'_>,
         callback: F,
     ) -> OmaAptResult<Summary>
     where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
+        F: AsyncFn(Event),
     {
         let mut download_list = vec![];
         for pkg in pkgs {
@@ -551,18 +549,14 @@ impl OmaApt {
     }
 
     /// Commit changes
-    pub fn commit<F, Fut>(
+    pub fn commit(
         self,
         install_progress_manager: Box<dyn InstallProgressManager>,
         op: &OmaOperation,
         client: &Client,
         config: CommitNetworkConfig,
-        callback: F,
-    ) -> OmaAptResult<()>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+        callback: impl AsyncFn(Event),
+    ) -> OmaAptResult<()> {
         let sysroot = self.config.get("Dir").unwrap_or("/".to_string());
 
         if self.dry_run {

oma-pm/src/commit.rs

@@ -47,16 +47,12 @@ impl<'a> DoInstall<'a> {
         })
     }
 
-    pub fn commit<F, Fut>(
+    pub fn commit(
         self,
         op: &OmaOperation,
         install_progress_manager: Box<dyn InstallProgressManager>,
-        callback: F,
-    ) -> OmaAptResult<()>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: std::future::Future<Output = ()>,
-    {
+        callback: impl AsyncFn(Event),
+    ) -> OmaAptResult<()> {
         let summary = self.download_pkgs(&op.install, callback)?;
 
         if !summary.failed.is_empty() {
@@ -68,15 +64,11 @@ impl<'a> DoInstall<'a> {
         Ok(())
     }
 
-    fn download_pkgs<F, Fut>(
+    fn download_pkgs(
         &self,
         download_pkg_list: &[InstallEntry],
-        callback: F,
-    ) -> OmaAptResult<Summary>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: std::future::Future<Output = ()>,
-    {
+        callback: impl AsyncFn(Event),
+    ) -> OmaAptResult<Summary> {
         let path = self.apt.get_archive_dir();
         create_dir_all(path)
             .map_err(|e| OmaAptError::FailedOperateDirOrFile(path.display().to_string(), e))?;

oma-pm/src/download.rs

@@ -1,4 +1,4 @@
-use std::{borrow::Cow, future::Future, path::Path};
+use std::{borrow::Cow, path::Path};
 
 use oma_console::console;
 use oma_fetch::{
@@ -12,17 +12,13 @@ use tracing::debug;
 use crate::apt::{DownloadConfig, OmaAptError, OmaAptResult};
 
 /// Download packages (inner)
-pub async fn download_pkgs<F, Fut>(
+pub async fn download_pkgs(
     client: &Client,
     download_pkg_list: &[InstallEntry],
     config: DownloadConfig<'_>,
     download_only: bool,
-    callback: F,
-) -> OmaAptResult<Summary>
-where
-    F: Fn(Event) -> Fut,
-    Fut: Future<Output = ()>,
-{
+    callback: impl AsyncFn(Event),
+) -> OmaAptResult<Summary> {
     let DownloadConfig {
         network_thread,
         download_dir,

oma-refresh/src/db.rs

@@ -1,4 +1,3 @@
-use std::future::Future;
 use std::{
     borrow::Cow,
     fs::Permissions,
@@ -192,11 +191,7 @@ pub enum Event {
 }
 
 impl<'a> OmaRefresh<'a> {
-    pub async fn start<F, Fut>(mut self, callback: F) -> Result<()>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+    pub async fn start(mut self, callback: impl AsyncFn(Event)) -> Result<()> {
         let arch = dpkg_arch(&self.source)?;
         let sourcelist = sources_lists(&self.source, &arch, &callback)
             .await
@@ -261,16 +256,12 @@ impl<'a> OmaRefresh<'a> {
         Ok(())
     }
 
-    async fn download_release_data<F, Fut>(
+    async fn download_release_data(
         &self,
-        callback: &F,
+        callback: &impl AsyncFn(Event),
         tasks: &[DownloadEntry],
         total: u64,
-    ) -> Result<Summary>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: futures::Future,
-    {
+    ) -> Result<Summary> {
         let dm = DownloadManager::builder()
             .client(self.client)
             .download_list(tasks)
@@ -330,16 +321,12 @@ impl<'a> OmaRefresh<'a> {
         }
     }
 
-    async fn download_releases<'b, F, Fut>(
+    async fn download_releases<'b>(
         &mut self,
         sourcelist: &'b [OmaSourceEntry<'b>],
         replacer: &DatabaseFilenameReplacer,
-        callback: &F,
-    ) -> Result<MirrorSources<'b, 'a>>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+        callback: &impl AsyncFn(Event),
+    ) -> Result<MirrorSources<'b, 'a>> {
         #[cfg(feature = "aosc")]
         let mut not_found = vec![];
 
@@ -394,16 +381,12 @@ impl<'a> OmaRefresh<'a> {
     }
 
     #[cfg(feature = "aosc")]
-    async fn refresh_topics<'b, F, Fut>(
+    async fn refresh_topics<'b>(
         &self,
-        callback: &F,
+        callback: &impl AsyncFn(Event),
         not_found: Vec<url::Url>,
         sources: &mut MirrorSources<'b, 'a>,
-    ) -> Result<()>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+    ) -> Result<()> {
         if !self.refresh_topics || not_found.is_empty() {
             return Ok(());
         }
@@ -434,12 +417,8 @@ impl<'a> OmaRefresh<'a> {
         }
 
         tm.write_enabled().await?;
-        tm.write_sources_list(self.topic_msg, false, |topic, mirror| async move {
-            callback(Event::TopicNotInMirror {
-                topic: topic.to_string(),
-                mirror: mirror.to_string(),
-            })
-            .await
+        tm.write_sources_list(self.topic_msg, false, async move |topic, mirror| {
+            callback(Event::TopicNotInMirror { topic, mirror }).await
         })
         .await?;
 
@@ -449,16 +428,12 @@ impl<'a> OmaRefresh<'a> {
     }
 
     #[cfg(not(feature = "aosc"))]
-    async fn refresh_topics<'b, F, Fut>(
+    async fn refresh_topics<'b>(
         &self,
-        _callback: &F,
+        _callback: &impl AsyncFn(Event),
         _not_found: Vec<url::Url>,
         _sources: &mut MirrorSources<'b, 'a>,
-    ) -> Result<()>
-    where
-        F: Fn(Event) -> Fut,
-        Fut: Future<Output = ()>,
-    {
+    ) -> Result<()> {
         Ok(())
     }