Make download_to_path_with_backend async

Author: rbtcollins

download/src/lib.rs

@@ -5,7 +5,6 @@ use std::path::Path;
 
 use anyhow::Context;
 pub use anyhow::Result;
-use tokio::{runtime::Handle, task};
 use url::Url;
 
 mod errors;
@@ -50,7 +49,7 @@ async fn download_with_backend(
 
 type DownloadCallback<'a> = &'a dyn Fn(Event<'_>) -> Result<()>;
 
-pub fn download_to_path_with_backend(
+pub async fn download_to_path_with_backend(
     backend: Backend,
     url: &Url,
     path: &Path,
@@ -62,100 +61,80 @@ pub fn download_to_path_with_backend(
     use std::fs::OpenOptions;
     use std::io::{Read, Seek, SeekFrom, Write};
 
-    || -> Result<()> {
-        let (file, resume_from) = if resume_from_partial {
-            let possible_partial = OpenOptions::new().read(true).open(path);
-
-            let downloaded_so_far = if let Ok(mut partial) = possible_partial {
-                if let Some(cb) = callback {
-                    cb(Event::ResumingPartialDownload)?;
-
-                    let mut buf = vec![0; 32768];
-                    let mut downloaded_so_far = 0;
-                    loop {
-                        let n = partial.read(&mut buf)?;
-                        downloaded_so_far += n as u64;
-                        if n == 0 {
-                            break;
+    (|| {
+        async move {
+            let (file, resume_from) = if resume_from_partial {
+                // TODO: blocking call
+                let possible_partial = OpenOptions::new().read(true).open(path);
+
+                let downloaded_so_far = if let Ok(mut partial) = possible_partial {
+                    if let Some(cb) = callback {
+                        cb(Event::ResumingPartialDownload)?;
+
+                        let mut buf = vec![0; 32768];
+                        let mut downloaded_so_far = 0;
+                        loop {
+                            let n = partial.read(&mut buf)?;
+                            downloaded_so_far += n as u64;
+                            if n == 0 {
+                                break;
+                            }
+                            cb(Event::DownloadDataReceived(&buf[..n]))?;
                         }
-                        cb(Event::DownloadDataReceived(&buf[..n]))?;
-                    }
 
-                    downloaded_so_far
+                        downloaded_so_far
+                    } else {
+                        let file_info = partial.metadata()?;
+                        file_info.len()
+                    }
                 } else {
-                    let file_info = partial.metadata()?;
-                    file_info.len()
-                }
-            } else {
-                0
-            };
-
-            let mut possible_partial = OpenOptions::new()
-                .write(true)
-                .create(true)
-                .open(path)
-                .context("error opening file for download")?;
+                    0
+                };
 
-            possible_partial.seek(SeekFrom::End(0))?;
-
-            (possible_partial, downloaded_so_far)
-        } else {
-            (
-                OpenOptions::new()
+                let mut possible_partial = OpenOptions::new()
                     .write(true)
                     .create(true)
                     .open(path)
-                    .context("error creating file for download")?,
-                0,
-            )
-        };
+                    .context("error opening file for download")?;
 
-        let file = RefCell::new(file);
-
-        match Handle::try_current() {
-            Ok(current) => {
-                // hide the asyncness for now.
-                task::block_in_place(|| {
-                    current.block_on(download_with_backend(backend, url, resume_from, &|event| {
-                        if let Event::DownloadDataReceived(data) = event {
-                            file.borrow_mut()
-                                .write_all(data)
-                                .context("unable to write download to disk")?;
-                        }
-                        match callback {
-                            Some(cb) => cb(event),
-                            None => Ok(()),
-                        }
-                    }))
-                })
-            }
-            Err(_) => {
-                // Make a runtime to hide the asyncness.
-                tokio::runtime::Runtime::new()?.block_on(download_with_backend(
-                    backend,
-                    url,
-                    resume_from,
-                    &|event| {
-                        if let Event::DownloadDataReceived(data) = event {
-                            file.borrow_mut()
-                                .write_all(data)
-                                .context("unable to write download to disk")?;
-                        }
-                        match callback {
-                            Some(cb) => cb(event),
-                            None => Ok(()),
-                        }
-                    },
-                ))
-            }
-        }?;
+                possible_partial.seek(SeekFrom::End(0))?;
 
-        file.borrow_mut()
-            .sync_data()
-            .context("unable to sync download to disk")?;
+                (possible_partial, downloaded_so_far)
+            } else {
+                (
+                    OpenOptions::new()
+                        .write(true)
+                        .create(true)
+                        .open(path)
+                        .context("error creating file for download")?,
+                    0,
+                )
+            };
 
-        Ok(())
-    }()
+            let file = RefCell::new(file);
+
+            // TODO: the sync callback will stall the async runtime if IO calls block, which is OS dependent. Rearrange.
+            download_with_backend(backend, url, resume_from, &|event| {
+                if let Event::DownloadDataReceived(data) = event {
+                    file.borrow_mut()
+                        .write_all(data)
+                        .context("unable to write download to disk")?;
+                }
+                match callback {
+                    Some(cb) => cb(event),
+                    None => Ok(()),
+                }
+            })
+            .await?;
+
+            file.borrow_mut()
+                .sync_data()
+                .context("unable to sync download to disk")?;
+
+            Ok::<(), anyhow::Error>(())
+        }
+    })()
+    .await
     .map_err(|e| {
         // TODO: We currently clear up the cached download on any error, should we restrict it to a subset?
         if let Err(file_err) = remove_file(path).context("cleaning up cached downloads") {

download/tests/download-curl-resume.rs

@@ -10,8 +10,8 @@ use download::*;
 mod support;
 use crate::support::{serve_file, tmp_dir, write_file};
 
-#[test]
-fn partially_downloaded_file_gets_resumed_from_byte_offset() {
+#[tokio::test]
+async fn partially_downloaded_file_gets_resumed_from_byte_offset() {
     let tmpdir = tmp_dir();
     let from_path = tmpdir.path().join("download-source");
     write_file(&from_path, "xxx45");
@@ -21,13 +21,14 @@ fn partially_downloaded_file_gets_resumed_from_byte_offset() {
 
     let from_url = Url::from_file_path(&from_path).unwrap();
     download_to_path_with_backend(Backend::Curl, &from_url, &target_path, true, None)
+        .await
         .expect("Test download failed");
 
     assert_eq!(std::fs::read_to_string(&target_path).unwrap(), "12345");
 }
 
-#[test]
-fn callback_gets_all_data_as_if_the_download_happened_all_at_once() {
+#[tokio::test]
+async fn callback_gets_all_data_as_if_the_download_happened_all_at_once() {
     let tmpdir = tmp_dir();
     let target_path = tmpdir.path().join("downloaded");
     write_file(&target_path, "123");
@@ -66,6 +67,7 @@ fn callback_gets_all_data_as_if_the_download_happened_all_at_once() {
             Ok(())
         }),
     )
+    .await
     .expect("Test download failed");
 
     assert!(callback_partial.into_inner());

download/tests/download-reqwest-resume.rs

@@ -10,8 +10,8 @@ use download::*;
 mod support;
 use crate::support::{serve_file, tmp_dir, write_file};
 
-#[test]
-fn resume_partial_from_file_url() {
+#[tokio::test]
+async fn resume_partial_from_file_url() {
     let tmpdir = tmp_dir();
     let from_path = tmpdir.path().join("download-source");
     write_file(&from_path, "xxx45");
@@ -27,13 +27,14 @@ fn resume_partial_from_file_url() {
         true,
         None,
     )
+    .await
     .expect("Test download failed");
 
     assert_eq!(std::fs::read_to_string(&target_path).unwrap(), "12345");
 }
 
-#[test]
-fn callback_gets_all_data_as_if_the_download_happened_all_at_once() {
+#[tokio::test]
+async fn callback_gets_all_data_as_if_the_download_happened_all_at_once() {
     let tmpdir = tmp_dir();
     let target_path = tmpdir.path().join("downloaded");
     write_file(&target_path, "123");
@@ -72,6 +73,7 @@ fn callback_gets_all_data_as_if_the_download_happened_all_at_once() {
             Ok(())
         }),
     )
+    .await
     .expect("Test download failed");
 
     assert!(callback_partial.into_inner());

src/utils/utils.rs

@@ -1,5 +1,6 @@
 use std::env;
 use std::fs::{self, File};
+use std::future::Future;
 use std::io::{self, BufReader, Write};
 use std::path::{Path, PathBuf};
 
@@ -8,6 +9,8 @@ use home::env as home;
 use retry::delay::{jitter, Fibonacci};
 use retry::{retry, OperationResult};
 use sha2::Sha256;
+use tokio::runtime::Handle;
+use tokio::task;
 use url::Url;
 
 use crate::currentprocess::{cwdsource::CurrentDirSource, varsource::VarSource};
@@ -238,14 +241,37 @@ fn download_file_(
         (Backend::Reqwest(tls_backend), Notification::UsingReqwest)
     };
     notify_handler(notification);
-    let res =
-        download_to_path_with_backend(backend, url, path, resume_from_partial, Some(callback));
+    let res = run_future(download_to_path_with_backend(
+        backend,
+        url,
+        path,
+        resume_from_partial,
+        Some(callback),
+    ));
 
     notify_handler(Notification::DownloadFinished);
 
     res
 }
 
+/// Temporary thunk to support asyncifying from underneath.
+pub(crate) fn run_future<F, R, E>(f: F) -> Result<R, E>
+where
+    F: Future<Output = Result<R, E>>,
+    E: std::convert::From<std::io::Error>,
+{
+    match Handle::try_current() {
+        Ok(current) => {
+            // hide the asyncness for now.
+            task::block_in_place(|| current.block_on(f))
+        }
+        Err(_) => {
+            // Make a runtime to hide the asyncness.
+            tokio::runtime::Runtime::new()?.block_on(f)
+        }
+    }
+}
+
 pub(crate) fn parse_url(url: &str) -> Result<Url> {
     Url::parse(url).with_context(|| format!("failed to parse url: {url}"))
 }