Skip to content
Clone

refactor(storage): refresh local, repo and s3 drivers

zhenyi committed
parent 7a46188 commit bd3a6c8 Browse files

3 files changed

+52 −1

File tree 3
Modified lib/storage/local.rs View lib/storage/local.rs
@@ -70,6 +70,16 @@
7070 Ok(self.store.copy(from, to).await?)
7171 }
7272
73 async fn begin_multipart(
74 &self,
75 path: &Path,
76 ) -> Result<Box<dyn object_store::MultipartUpload>, StorageError> {
77 Ok(self
78 .store
79 .put_multipart_opts(path, object_store::PutMultipartOptions::default())
80 .await?)
81 }
82
7383 async fn presigned_upload(
7484 &self,
7585 _path: &Path,
@@ -145,6 +155,36 @@
145155 }
146156
147157 #[tokio::test]
158 async fn local_multipart_roundtrip() {
159 use object_store::{MultipartUpload, PutPayload};
160
161 let dir = tempdir().unwrap();
162 let settings = local_settings(dir.path());
163 let repo = from_config(&settings).unwrap();
164
165 let path = Path::from("test/multipart.bin");
166 let mut upload = repo.begin_multipart(&path).await.unwrap();
167 upload
168 .put_part(PutPayload::from_bytes(Bytes::from_static(b"hello ")))
169 .await
170 .unwrap();
171 upload
172 .put_part(PutPayload::from_bytes(Bytes::from_static(b"world ")))
173 .await
174 .unwrap();
175 upload
176 .put_part(PutPayload::from_bytes(Bytes::from_static(b"from multipart")))
177 .await
178 .unwrap();
179 upload.complete().await.unwrap();
180
181 let got = repo.get(&path).await.unwrap().bytes().await.unwrap();
182 assert_eq!(got, Bytes::from_static(b"hello world from multipart"));
183
184 dir.close().unwrap();
185 }
186
187 #[tokio::test]
148188 async fn local_presign_unsupported() {
149189 let dir = tempdir().unwrap();
150190 let settings = local_settings(dir.path());
Modified lib/storage/repo.rs View lib/storage/repo.rs
@@ -4,7 +4,7 @@
44use async_trait::async_trait;
55use bytes::Bytes;
66use object_store::path::Path;
7use object_store::{GetResult, ObjectMeta, PutResult};
7use object_store::{GetResult, MultipartUpload, ObjectMeta, PutResult};
88use url::Url;
99
1010use crate::config::{BackendType, StorageSettings};
@@ -28,6 +28,7 @@
2828 async fn exists(&self, path: &Path) -> Result<bool, StorageError>;
2929 async fn list(&self, prefix: Option<&Path>) -> Result<Vec<ObjectMeta>, StorageError>;
3030 async fn copy(&self, from: &Path, to: &Path) -> Result<(), StorageError>;
31 async fn begin_multipart(&self, path: &Path) -> Result<Box<dyn MultipartUpload>, StorageError>;
3132 async fn presigned_upload(
3233 &self,
3334 path: &Path,
Modified lib/storage/s3.rs View lib/storage/s3.rs
@@ -97,6 +97,16 @@
9797 Ok(self.store.copy(from, to).await?)
9898 }
9999
100 async fn begin_multipart(
101 &self,
102 path: &Path,
103 ) -> Result<Box<dyn object_store::MultipartUpload>, StorageError> {
104 Ok(self
105 .store
106 .put_multipart_opts(path, object_store::PutMultipartOptions::default())
107 .await?)
108 }
109
100110 async fn presigned_upload(
101111 &self,
102112 path: &Path,