From bae67cf43aacce1489f7a95dcd34b98b510860f0 Mon Sep 17 00:00:00 2001
From: Jake Goulding <jake.goulding@integer32.com>
Date: Fri, 26 Jun 2020 20:54:16 -0400
Subject: [PATCH] refactor: Use InternalResult on the implementations of `get`
 streams

---
 delorean_object_store/src/lib.rs | 28 ++++++++++++++++++++--------
 1 file changed, 20 insertions(+), 8 deletions(-)

diff --git a/delorean_object_store/src/lib.rs b/delorean_object_store/src/lib.rs
index 4ae55db688..ee69275fb6 100644
--- a/delorean_object_store/src/lib.rs
+++ b/delorean_object_store/src/lib.rs
@@ -65,7 +65,8 @@ impl ObjectStore {
             AmazonS3(s3) => s3.get(location).await?.boxed(),
             GoogleCloudStorage(gcs) => gcs.get(location).await?.boxed(),
             InMemory(in_mem) => in_mem.get(location).await?.boxed(),
-        })
+        }
+        .err_into())
     }
 
     /// Delete the object at the specified location.
@@ -87,10 +88,12 @@ impl ObjectStore {
     ) -> Result<impl Stream<Item = Result<Vec<String>>> + 'a> {
         use ObjectStoreIntegration::*;
         Ok(match &self.0 {
-            AmazonS3(s3) => s3.list(prefix).await?.err_into().boxed(),
-            GoogleCloudStorage(gcs) => gcs.list(prefix).await?.err_into().boxed(),
-            InMemory(in_mem) => in_mem.list(prefix).await?.err_into().boxed(),
-        })
+            AmazonS3(s3) => s3.list(prefix).await?.boxed(),
+            GoogleCloudStorage(gcs) => gcs.list(prefix).await?.boxed(),
+            InMemory(in_mem) => in_mem.list(prefix).await?.boxed(),
+            File(file) => file.list(prefix).await?.boxed(),
+        }
+        .err_into())
     }
 }
 
@@ -153,7 +156,10 @@ impl GoogleCloudStorage {
         Ok(())
     }
 
-    async fn get(&self, location: &str) -> InternalResult<impl Stream<Item = Result<Bytes>>> {
+    async fn get(
+        &self,
+        location: &str,
+    ) -> InternalResult<impl Stream<Item = InternalResult<Bytes>>> {
         let location = location.to_string();
         let bucket_name = self.bucket_name.clone();
 
@@ -257,7 +263,10 @@ impl AmazonS3 {
     }
 
     /// Return the bytes that are stored at the specified location.
-    async fn get(&self, location: &str) -> InternalResult<impl Stream<Item = Result<Bytes>>> {
+    async fn get(
+        &self,
+        location: &str,
+    ) -> InternalResult<impl Stream<Item = InternalResult<Bytes>>> {
         let get_request = rusoto_s3::GetObjectRequest {
             bucket: self.bucket_name.clone(),
             key: location.to_string(),
@@ -380,7 +389,10 @@ impl InMemory {
     }
 
     /// Return the bytes that are stored at the specified location.
-    async fn get(&self, location: &str) -> InternalResult<impl Stream<Item = Result<Bytes>>> {
+    async fn get(
+        &self,
+        location: &str,
+    ) -> InternalResult<impl Stream<Item = InternalResult<Bytes>>> {
         let data = self
             .storage
             .read()