Use Sink and Stream instead of manual buffering. Moar formatting

Signed-off-by: itowlson <[email protected]>

Author: itowlson

crates/blobstore-fs/src/lib.rs

@@ -206,23 +206,17 @@ impl spin_factor_blobstore::Container for FileSystemContainer {
 
 impl FileSystemContainer {
     async fn write_data_core(
-        mut data: tokio::io::ReadHalf<tokio::io::SimplexStream>,
-        mut file: tokio::fs::File,
+        data: tokio::io::ReadHalf<tokio::io::SimplexStream>,
+        file: tokio::fs::File,
     ) -> anyhow::Result<()> {
-        use tokio::io::AsyncReadExt;
-        use tokio::io::AsyncWriteExt;
+        use futures::SinkExt;
+        use tokio_util::codec::{BytesCodec, FramedWrite};
 
-        const BUF_SIZE: usize = 8192;
+        // Ceremonies to turn `file` and `data` into Sink and Stream
+        let mut file_sink = FramedWrite::new(file, BytesCodec::new());
+        let mut data_stm = tokio_util::io::ReaderStream::new(data);
 
-        loop {
-            let mut buf = vec![0; BUF_SIZE];
-            let count = data.read(&mut buf).await?;
-            if count == 0 {
-                _ = file.flush().await;
-                break;
-            }
-            file.write_all(&buf[0..count]).await?;
-        }
+        file_sink.send_all(&mut data_stm).await?;
 
         Ok(())
     }

crates/blobstore-s3/src/store.rs

@@ -292,11 +292,13 @@ impl S3Container {
     ) -> anyhow::Result<()> {
         use object_store::ObjectStore;
 
+        const BUF_SIZE: usize = 5 * 1024 * 1024;
+
         let mupload = store.put_multipart(&path).await?;
         let mut writer = object_store::WriteMultipart::new(mupload);
         loop {
             use tokio::io::AsyncReadExt;
-            let mut buf = vec![0; 5 * 1024 * 1024];
+            let mut buf = vec![0; BUF_SIZE];
             let read_amount = data.read(&mut buf).await?;
             if read_amount == 0 {
                 break;

crates/factor-blobstore/src/host/object_names.rs

@@ -28,10 +28,7 @@ impl HostStreamObjectNames for BlobStoreDispatch<'_> {
         object_names.skip(num).await.map_err(|e| e.to_string())
     }
 
-    async fn drop(
-        &mut self,
-        rep: Resource<StreamObjectNames>,
-    ) -> anyhow::Result<()> {
+    async fn drop(&mut self, rep: Resource<StreamObjectNames>) -> anyhow::Result<()> {
         self.object_names.write().await.remove(rep.rep());
         Ok(())
     }

crates/factor-blobstore/src/lib.rs

@@ -1,22 +1,22 @@
 //! Example usage:
-//! 
+//!
 //! --------------------
-//! 
+//!
 //! spin.toml:
-//! 
+//!
 //! [component.foo]
 //! blob_containers = ["default"]
 //!
 //! --------------------
-//! 
+//!
 //! runtime-config.toml
-//! 
+//!
 //! [blob_store.default]
 //! type = "file_system" | "s3" | "azure_blob"
 //! # further config settings per type
-//! 
+//!
 //! --------------------
-//! 
+//!
 //! TODO: the naming here is not very consistent and we should make a more conscious
 //! decision about whether these things are "blob stores" or "containers" or what
 
@@ -35,10 +35,7 @@ use spin_factors::{ConfigureAppContext, Factor, InitContext, PrepareContext, Run
 use spin_locked_app::MetadataKey;
 use spin_resource_table::Table;
 
-pub use host::{
-    BlobStoreDispatch, Container, ContainerManager, Error, IncomingData,
-    ObjectNames,
-};
+pub use host::{BlobStoreDispatch, Container, ContainerManager, Error, IncomingData, ObjectNames};
 pub use runtime_config::RuntimeConfig;
 pub use spin_world::wasi::blobstore::types::{ContainerMetadata, ObjectMetadata};
 pub use stream::AsyncWriteStream;
@@ -170,15 +167,15 @@ pub struct InstanceBuilder {
     /// There are multiple WASI interfaces in play here. The factor adds each of them
     /// to the linker, passing a closure that derives the interface implementation
     /// from the InstanceBuilder.
-    /// 
+    ///
     /// For the different interfaces to agree on their resource tables, each closure
     /// needs to derive the same resource table from the InstanceBuilder.
     /// The only* way that works is for the InstanceBuilder to set up all
     /// the resource tables, and Arc-RwLock them so that each clone gets
     /// the same one.
-    /// 
+    ///
     /// * TODO: for 'only', read 'or maybe we can do some shenanigans with borrowing
-    /// from the InstanceBuilder/instance state'
+    ///   from the InstanceBuilder/instance state'
     containers: Arc<RwLock<Table<Arc<dyn Container>>>>,
     incoming_values: Arc<RwLock<Table<Box<dyn IncomingData>>>>,
     outgoing_values: Arc<RwLock<Table<host::OutgoingValue>>>,