Alias BoxByteStream, and test git get_file implementation

src/domain/acme/manager.rs

@@ -14,7 +14,7 @@ pub trait Manager {
     fn sync_domain<'mgr>(
         &'mgr self,
         domain: domain::Name,
-    ) -> util::BoxedFuture<'mgr, Result<(), unexpected::Error>>;
+    ) -> util::BoxFuture<'mgr, Result<(), unexpected::Error>>;
 
     fn get_http01_challenge_key(&self, token: &str) -> Result<String, GetHttp01ChallengeKeyError>;
 
@@ -87,7 +87,7 @@ impl Manager for ManagerImpl {
     fn sync_domain<'mgr>(
         &'mgr self,
         domain: domain::Name,
-    ) -> util::BoxedFuture<'mgr, Result<(), unexpected::Error>> {
+    ) -> util::BoxFuture<'mgr, Result<(), unexpected::Error>> {
         Box::pin(async move {
             // if there's an existing cert, and its expiry (determined by the soonest value of
             // not_after amongst its parts) is later than 30 days from now, then we consider it to be

src/domain/manager.rs

@@ -3,8 +3,7 @@ use crate::error::unexpected::{self, Mappable};
 use crate::origin;
 use crate::util;
 
-use futures::stream;
-use std::{io, sync};
+use std::sync;
 use tokio_util::sync::CancellationToken;
 
 #[derive(thiserror::Error, Debug)]
@@ -150,18 +149,18 @@ pub trait Manager: Sync + Send + rustls::server::ResolvesServerCert {
         &'store self,
         domain: &domain::Name,
         path: &str,
-    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError>;
+    ) -> Result<util::BoxByteStream, GetFileError>;
 
     fn sync_cert<'mgr>(
         &'mgr self,
         domain: domain::Name,
-    ) -> util::BoxedFuture<'mgr, Result<(), unexpected::Error>>;
+    ) -> util::BoxFuture<'mgr, Result<(), unexpected::Error>>;
 
     fn sync_with_config<'mgr>(
         &'mgr self,
         domain: domain::Name,
         config: config::Config,
-    ) -> util::BoxedFuture<'mgr, Result<(), SyncWithConfigError>>;
+    ) -> util::BoxFuture<'mgr, Result<(), SyncWithConfigError>>;
 
     fn get_acme_http01_challenge_key(
         &self,
@@ -254,7 +253,7 @@ impl Manager for ManagerImpl {
         &'store self,
         domain: &domain::Name,
         path: &str,
-    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError> {
+    ) -> Result<util::BoxByteStream, GetFileError> {
         let config = self.domain_config_store.get(domain)?;
         let f = self.origin_store.get_file(&config.origin_descr, path)?;
         Ok(f)
@@ -263,7 +262,7 @@ impl Manager for ManagerImpl {
     fn sync_cert<'mgr>(
         &'mgr self,
         domain: domain::Name,
-    ) -> util::BoxedFuture<'mgr, Result<(), unexpected::Error>> {
+    ) -> util::BoxFuture<'mgr, Result<(), unexpected::Error>> {
         Box::pin(async move {
             if let Some(ref acme_manager) = self.acme_manager {
                 acme_manager.sync_domain(domain.clone()).await?;
@@ -277,7 +276,7 @@ impl Manager for ManagerImpl {
         &'mgr self,
         domain: domain::Name,
         config: config::Config,
-    ) -> util::BoxedFuture<'mgr, Result<(), SyncWithConfigError>> {
+    ) -> util::BoxFuture<'mgr, Result<(), SyncWithConfigError>> {
         Box::pin(async move {
             let config_hash = config
                 .hash()

src/origin.rs

@@ -1,6 +1,6 @@
 use crate::error::unexpected;
-use futures::stream;
-use std::{io, sync};
+use crate::util;
+use std::sync;
 
 pub mod git;
 pub mod mux;
@@ -59,11 +59,7 @@ pub trait Store {
         into: &mut dyn std::io::Write,
     ) -> Result<(), ReadFileIntoError>;
 
-    fn get_file(
-        &self,
-        descr: &Descr,
-        path: &str,
-    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError>;
+    fn get_file(&self, descr: &Descr, path: &str) -> Result<util::BoxByteStream, GetFileError>;
 }
 
 pub fn new_mock() -> sync::Arc<sync::Mutex<MockStore>> {
@@ -93,7 +89,7 @@ impl Store for sync::Arc<sync::Mutex<MockStore>> {
         &'store self,
         descr: &Descr,
         path: &str,
-    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError> {
+    ) -> Result<util::BoxByteStream, GetFileError> {
         self.lock().unwrap().get_file(descr, path)
     }
 }

src/origin/git.rs

@@ -1,5 +1,5 @@
 use crate::error::unexpected::{self, Intoable, Mappable};
-use crate::origin;
+use crate::{origin, util};
 
 use std::path::{Path, PathBuf};
 use std::{collections, fs, io, sync};
@@ -334,12 +334,11 @@ impl super::Store for FSStore {
         Ok(())
     }
 
-    // TODO test this
     fn get_file<'store>(
         &'store self,
         descr: &origin::Descr,
         path: &str,
-    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, origin::GetFileError> {
+    ) -> Result<util::BoxByteStream, origin::GetFileError> {
         let repo_snapshot = match self.get_repo_snapshot(descr) {
             Ok(Some(repo_snapshot)) => repo_snapshot,
             Ok(None) => return Err(origin::GetFileError::DescrNotSynced),
@@ -382,10 +381,11 @@ impl super::Store for FSStore {
 #[cfg(test)]
 mod tests {
     use crate::origin::{self, Store};
+    use futures::StreamExt;
     use tempdir::TempDir;
 
-    #[test]
-    fn basic() {
+    #[tokio::test]
+    async fn basic() {
         let tmp_dir = TempDir::new("origin_store_git").unwrap();
 
         let curr_dir = format!("file://{}", std::env::current_dir().unwrap().display());
@@ -405,31 +405,32 @@ mod tests {
         store.sync(&descr).expect("sync should succeed");
         store.sync(&descr).expect("second sync should succeed");
 
-        let assert_write = |path: &str| {
+        {
+            // RepoSnapshot doesn't exist
             let mut into: Vec<u8> = vec![];
-            store
-                .read_file_into(&descr, path, &mut into)
-                .expect("write should succeed");
-            assert!(into.len() > 0);
-        };
-
-        assert_write("src/lib.rs");
-        assert_write("/src/lib.rs");
+            assert!(matches!(
+                store.read_file_into(&other_descr, "DNE", &mut into),
+                Err::<_, origin::ReadFileIntoError>(origin::ReadFileIntoError::DescrNotSynced),
+            ));
+        }
 
-        let mut into: Vec<u8> = vec![];
+        let assert_file_dne = |path: &str| match store.get_file(&descr, path) {
+            Err(origin::ReadFileIntoError::FileNotFound) => (),
+            _ => assert!(false, "file should have not been found"),
+        };
 
-        // RepoSnapshot doesn't exist
-        assert!(matches!(
-            store.read_file_into(&other_descr, "DNE", &mut into),
-            Err::<_, origin::ReadFileIntoError>(origin::ReadFileIntoError::DescrNotSynced),
-        ));
+        let assert_file_not_empty = |path: &str| {
+            let f = store.get_file(&descr, path).expect("file not retrieved");
+            async move {
+                let body = f.map(|r| r.unwrap()).concat().await;
+                assert!(body.len() > 0);
+            }
+        };
 
-        // File doesn't exist
-        assert!(matches!(
-            store.read_file_into(&descr, "DNE", &mut into),
-            Err::<(), origin::ReadFileIntoError>(origin::ReadFileIntoError::FileNotFound),
-        ));
-        assert_eq!(into.len(), 0);
+        assert_file_not_empty("src/lib.rs").await;
+        assert_file_not_empty("/src/lib.rs").await;
+        assert_file_dne("DNE");
+        assert_file_dne("src/../src/lib.rs");
 
         let descrs = store.all_descrs().expect("all_descrs called");
 

src/origin/mux.rs

@@ -1,7 +1,5 @@
 use crate::error::unexpected::Mappable;
-use crate::origin;
-use futures::stream;
-use std::io;
+use crate::{origin, util};
 
 pub struct Store<F, S>
 where
@@ -58,7 +56,7 @@ where
         &'store self,
         descr: &origin::Descr,
         path: &str,
-    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, origin::GetFileError> {
+    ) -> Result<util::BoxByteStream, origin::GetFileError> {
         (self.mapping_fn)(descr)
             .or_unexpected_while(format!("mapping {:?} to store", &descr))?
             .get_file(descr, path)

src/util.rs

@@ -12,13 +12,15 @@ pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> {
     }
 }
 
-pub type BoxedFuture<'a, O> = pin::Pin<Box<dyn futures::Future<Output = O> + Send + 'a>>;
+pub type BoxByteStream = futures::stream::BoxStream<'static, io::Result<Vec<u8>>>;
+
+pub type BoxFuture<'a, O> = pin::Pin<Box<dyn futures::Future<Output = O> + Send + 'a>>;
 
 pub struct TaskStack<E>
 where
     E: error::Error + Send + 'static,
 {
-    wait_group: Vec<BoxedFuture<'static, Result<(), E>>>,
+    wait_group: Vec<BoxFuture<'static, Result<(), E>>>,
 }
 
 impl<E> TaskStack<E>