Implement origin::Store::get_file, to deprecate read_file_into

src/domain/manager.rs

@@ -3,7 +3,8 @@ use crate::error::unexpected::{self, Mappable};
 use crate::origin;
 use crate::util;
 
-use std::sync;
+use futures::stream;
+use std::{io, sync};
 use tokio_util::sync::CancellationToken;
 
 #[derive(thiserror::Error, Debug)]
@@ -25,7 +26,7 @@ impl From<config::GetError> for GetConfigError {
 }
 
 #[derive(thiserror::Error, Debug)]
-pub enum ReadFileIntoError {
+pub enum GetFileError {
     #[error("domain not found")]
     DomainNotFound,
 
@@ -36,7 +37,7 @@ pub enum ReadFileIntoError {
     Unexpected(#[from] unexpected::Error),
 }
 
-impl From<config::GetError> for ReadFileIntoError {
+impl From<config::GetError> for GetFileError {
     fn from(e: config::GetError) -> Self {
         match e {
             config::GetError::NotFound => Self::DomainNotFound,
@@ -45,18 +46,20 @@ impl From<config::GetError> for ReadFileIntoError {
     }
 }
 
-impl From<origin::ReadFileIntoError> for ReadFileIntoError {
-    fn from(e: origin::ReadFileIntoError) -> Self {
+impl From<origin::GetFileError> for GetFileError {
+    fn from(e: origin::GetFileError) -> Self {
         match e {
-            origin::ReadFileIntoError::DescrNotSynced => {
+            origin::GetFileError::DescrNotSynced => {
                 Self::Unexpected(unexpected::Error::from("origin descr not synced"))
             }
-            origin::ReadFileIntoError::FileNotFound => Self::FileNotFound,
-            origin::ReadFileIntoError::Unexpected(e) => Self::Unexpected(e),
+            origin::GetFileError::FileNotFound => Self::FileNotFound,
+            origin::GetFileError::Unexpected(e) => Self::Unexpected(e),
         }
     }
 }
 
+pub type ReadFileIntoError = GetFileError;
+
 #[derive(thiserror::Error, Debug)]
 pub enum SyncError {
     #[error("not found")]
@@ -143,6 +146,12 @@ pub trait Manager: Sync + Send + rustls::server::ResolvesServerCert {
         into: &mut dyn std::io::Write,
     ) -> Result<(), ReadFileIntoError>;
 
+    fn get_file<'store>(
+        &'store self,
+        domain: &domain::Name,
+        path: &str,
+    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError>;
+
     fn sync_cert<'mgr>(
         &'mgr self,
         domain: domain::Name,
@@ -241,6 +250,16 @@ impl Manager for ManagerImpl {
         Ok(())
     }
 
+    fn get_file<'store>(
+        &'store self,
+        domain: &domain::Name,
+        path: &str,
+    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError> {
+        let config = self.domain_config_store.get(domain)?;
+        let f = self.origin_store.get_file(&config.origin_descr, path)?;
+        Ok(f)
+    }
+
     fn sync_cert<'mgr>(
         &'mgr self,
         domain: domain::Name,

src/origin.rs

@@ -1,5 +1,6 @@
 use crate::error::unexpected;
-use std::sync;
+use futures::stream;
+use std::{io, sync};
 
 pub mod git;
 pub mod mux;
@@ -29,7 +30,7 @@ pub enum AllDescrsError {
 }
 
 #[derive(thiserror::Error, Debug)]
-pub enum ReadFileIntoError {
+pub enum GetFileError {
     #[error("descr not synced")]
     DescrNotSynced,
 
@@ -40,6 +41,8 @@ pub enum ReadFileIntoError {
     Unexpected(#[from] unexpected::Error),
 }
 
+pub type ReadFileIntoError = GetFileError;
+
 #[mockall::automock]
 /// Describes a storage mechanism for Origins. Each Origin is uniquely identified by its Descr.
 pub trait Store {
@@ -47,6 +50,8 @@ pub trait Store {
     /// the origin into the storage.
     fn sync(&self, descr: &Descr) -> Result<(), SyncError>;
 
+    fn all_descrs(&self) -> Result<Vec<Descr>, AllDescrsError>;
+
     fn read_file_into(
         &self,
         descr: &Descr,
@@ -54,7 +59,11 @@ pub trait Store {
         into: &mut dyn std::io::Write,
     ) -> Result<(), ReadFileIntoError>;
 
-    fn all_descrs(&self) -> Result<Vec<Descr>, AllDescrsError>;
+    fn get_file(
+        &self,
+        descr: &Descr,
+        path: &str,
+    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError>;
 }
 
 pub fn new_mock() -> sync::Arc<sync::Mutex<MockStore>> {
@@ -70,6 +79,7 @@ impl Store for sync::Arc<sync::Mutex<MockStore>> {
         self.lock().unwrap().all_descrs()
     }
 
+    /// Deprecated, use get_file
     fn read_file_into(
         &self,
         descr: &Descr,
@@ -78,4 +88,12 @@ impl Store for sync::Arc<sync::Mutex<MockStore>> {
     ) -> Result<(), ReadFileIntoError> {
         self.lock().unwrap().read_file_into(descr, path, into)
     }
+
+    fn get_file<'store>(
+        &'store self,
+        descr: &Descr,
+        path: &str,
+    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError> {
+        self.lock().unwrap().get_file(descr, path)
+    }
 }

src/origin/git.rs

@@ -4,6 +4,8 @@ use crate::origin;
 use std::path::{Path, PathBuf};
 use std::{collections, fs, io, sync};
 
+use futures::stream;
+
 #[derive(Clone)]
 struct RepoSnapshot {
     repo: sync::Arc<gix::ThreadSafeRepository>,
@@ -331,6 +333,50 @@ 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> {
+        let repo_snapshot = match self.get_repo_snapshot(descr) {
+            Ok(Some(repo_snapshot)) => repo_snapshot,
+            Ok(None) => return Err(origin::GetFileError::DescrNotSynced),
+            Err(e) => return Err(e.into()),
+        };
+
+        let mut clean_path = Path::new(path);
+        clean_path = clean_path.strip_prefix("/").unwrap_or(clean_path);
+
+        let repo = repo_snapshot.repo.to_thread_local();
+
+        let file_object = repo
+            .find_object(repo_snapshot.tree_object_id)
+            .map_unexpected_while(|| {
+                format!("finding tree object {}", repo_snapshot.tree_object_id)
+            })?
+            .peel_to_tree()
+            .map_unexpected_while(|| {
+                format!("peeling tree object {}", repo_snapshot.tree_object_id)
+            })?
+            .lookup_entry_by_path(clean_path)
+            .map_unexpected_while(|| {
+                format!(
+                    "looking up {} in tree object {}",
+                    clean_path.display(),
+                    repo_snapshot.tree_object_id
+                )
+            })?
+            .ok_or(origin::GetFileError::FileNotFound)?
+            .object()
+            .or_unexpected()?;
+
+        // TODO this is very not ideal, the whole file is first read totally into memory, and then
+        // that is cloned.
+        let data = file_object.data.clone();
+        Ok(Box::pin(stream::once(async move { Ok(data) })))
+    }
 }
 
 #[cfg(test)]

src/origin/mux.rs

@@ -1,5 +1,7 @@
 use crate::error::unexpected::Mappable;
 use crate::origin;
+use futures::stream;
+use std::io;
 
 pub struct Store<F, S>
 where
@@ -51,6 +53,16 @@ where
             .or_unexpected_while(format!("mapping {:?} to store", &descr))?
             .read_file_into(descr, path, into)
     }
+
+    fn get_file<'store>(
+        &'store self,
+        descr: &origin::Descr,
+        path: &str,
+    ) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, origin::GetFileError> {
+        (self.mapping_fn)(descr)
+            .or_unexpected_while(format!("mapping {:?} to store", &descr))?
+            .get_file(descr, path)
+    }
 }
 
 #[cfg(test)]