cargo fmt

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

Author: itowlson

crates/blobstore-azure/src/lib.rs

@@ -3,8 +3,8 @@ mod store;
 use serde::Deserialize;
 use spin_factor_blobstore::runtime_config::spin::MakeBlobStore;
 use store::{
-    AzureContainerManager,
     auth::{AzureBlobAuthOptions, AzureKeyAuth},
+    AzureContainerManager,
 };
 
 /// A key-value store that uses Azure Cosmos as the backend.
@@ -41,7 +41,10 @@ impl MakeBlobStore for AzureBlobStoreBuilder {
         runtime_config: Self::RuntimeConfig,
     ) -> anyhow::Result<Self::ContainerManager> {
         let auth = match &runtime_config.key {
-            Some(key) => AzureBlobAuthOptions::AccountKey(AzureKeyAuth::new(runtime_config.account.clone(), key.clone())),
+            Some(key) => AzureBlobAuthOptions::AccountKey(AzureKeyAuth::new(
+                runtime_config.account.clone(),
+                key.clone(),
+            )),
             None => AzureBlobAuthOptions::Environmental,
         };
 

crates/blobstore-azure/src/store.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use anyhow::Result;
 use azure_storage_blobs::prelude::{BlobServiceClient, ContainerClient};
 use spin_core::async_trait;
-use spin_factor_blobstore::{Error, Container, ContainerManager};
+use spin_factor_blobstore::{Container, ContainerManager, Error};
 
 pub mod auth;
 mod incoming_data;
@@ -18,21 +18,25 @@ pub struct AzureContainerManager {
 }
 
 impl AzureContainerManager {
-    pub fn new(
-        auth_options: AzureBlobAuthOptions,
-    ) -> Result<Self> {
+    pub fn new(auth_options: AzureBlobAuthOptions) -> Result<Self> {
         let (account, credentials) = match auth_options {
-            AzureBlobAuthOptions::AccountKey(config) => {
-                (config.account.clone(), azure_storage::StorageCredentials::access_key(&config.account, config.key.clone()))
-            },
+            AzureBlobAuthOptions::AccountKey(config) => (
+                config.account.clone(),
+                azure_storage::StorageCredentials::access_key(&config.account, config.key.clone()),
+            ),
             AzureBlobAuthOptions::Environmental => {
                 let account = std::env::var("STORAGE_ACCOUNT").expect("missing STORAGE_ACCOUNT");
-                let access_key = std::env::var("STORAGE_ACCESS_KEY").expect("missing STORAGE_ACCOUNT_KEY");
-                (account.clone(), azure_storage::StorageCredentials::access_key(account, access_key))
-            },
+                let access_key =
+                    std::env::var("STORAGE_ACCESS_KEY").expect("missing STORAGE_ACCOUNT_KEY");
+                (
+                    account.clone(),
+                    azure_storage::StorageCredentials::access_key(account, access_key),
+                )
+            }
         };
 
-        let client = azure_storage_blobs::prelude::ClientBuilder::new(account, credentials).blob_service_client();
+        let client = azure_storage_blobs::prelude::ClientBuilder::new(account, credentials)
+            .blob_service_client();
         Ok(Self { client })
     }
 }
@@ -51,7 +55,10 @@ impl ContainerManager for AzureContainerManager {
     }
 
     fn summary(&self, _store_name: &str) -> Option<String> {
-        Some(format!("Azure blob storage account {}", self.client.account()))
+        Some(format!(
+            "Azure blob storage account {}",
+            self.client.account()
+        ))
     }
 }
 
@@ -101,17 +108,31 @@ impl Container for AzureContainer {
         Ok(self.client.blob_client(name).exists().await?)
     }
 
-    async fn object_info(&self, name: &str) -> anyhow::Result<spin_factor_blobstore::ObjectMetadata> {
+    async fn object_info(
+        &self,
+        name: &str,
+    ) -> anyhow::Result<spin_factor_blobstore::ObjectMetadata> {
         let response = self.client.blob_client(name).get_properties().await?;
         Ok(spin_factor_blobstore::ObjectMetadata {
             name: name.to_string(),
             container: self.client.container_name().to_string(),
-            created_at: response.blob.properties.creation_time.unix_timestamp().try_into().unwrap(),
+            created_at: response
+                .blob
+                .properties
+                .creation_time
+                .unix_timestamp()
+                .try_into()
+                .unwrap(),
             size: response.blob.properties.content_length,
         })
     }
 
-    async fn get_data(&self, name: &str, start: u64, end: u64) -> anyhow::Result<Box<dyn spin_factor_blobstore::IncomingData>> {
+    async fn get_data(
+        &self,
+        name: &str,
+        start: u64,
+        end: u64,
+    ) -> anyhow::Result<Box<dyn spin_factor_blobstore::IncomingData>> {
         // We can't use a Rust range because the Azure type does not accept inclusive ranges,
         // and we don't want to add 1 to `end` if it's already at MAX!
         let range = if end == u64::MAX {
@@ -123,12 +144,20 @@ impl Container for AzureContainer {
         Ok(Box::new(AzureIncomingData::new(client, range)))
     }
 
-    async fn connect_stm(&self, name: &str, stm: tokio::io::ReadHalf<tokio::io::SimplexStream>, finished_tx: tokio::sync::mpsc::Sender<anyhow::Result<()>>) -> anyhow::Result<()> {
+    async fn connect_stm(
+        &self,
+        name: &str,
+        stm: tokio::io::ReadHalf<tokio::io::SimplexStream>,
+        finished_tx: tokio::sync::mpsc::Sender<anyhow::Result<()>>,
+    ) -> anyhow::Result<()> {
         let client = self.client.blob_client(name);
 
         tokio::spawn(async move {
             let result = Self::connect_stm_core(stm, client).await;
-            finished_tx.send(result).await.expect("should sent finish tx");
+            finished_tx
+                .send(result)
+                .await
+                .expect("should sent finish tx");
         });
 
         Ok(())
@@ -141,7 +170,10 @@ impl Container for AzureContainer {
 }
 
 impl AzureContainer {
-    async fn connect_stm_core(mut stm: tokio::io::ReadHalf<tokio::io::SimplexStream>, client: azure_storage_blobs::prelude::BlobClient) -> anyhow::Result<()> {
+    async fn connect_stm_core(
+        mut stm: tokio::io::ReadHalf<tokio::io::SimplexStream>,
+        client: azure_storage_blobs::prelude::BlobClient,
+    ) -> anyhow::Result<()> {
         use tokio::io::AsyncReadExt;
 
         // Azure limits us to 50k blocks per blob.  At 2MB/block that allows 100GB, which will be
@@ -162,22 +194,24 @@ impl AzureContainer {
                     let id_bytes = uuid::Uuid::new_v4().as_bytes().to_vec();
                     let block_id = azure_storage_blobs::prelude::BlockId::new(id_bytes);
                     client.put_block(block_id.clone(), bytes).await?;
-                    blocks.push(azure_storage_blobs::blob::BlobBlockType::Uncommitted(block_id));
+                    blocks.push(azure_storage_blobs::blob::BlobBlockType::Uncommitted(
+                        block_id,
+                    ));
                     break 'put_blocks;
                 }
                 if len >= BLOCK_SIZE {
                     let id_bytes = uuid::Uuid::new_v4().as_bytes().to_vec();
                     let block_id = azure_storage_blobs::prelude::BlockId::new(id_bytes);
                     client.put_block(block_id.clone(), bytes).await?;
-                    blocks.push(azure_storage_blobs::blob::BlobBlockType::Uncommitted(block_id));
+                    blocks.push(azure_storage_blobs::blob::BlobBlockType::Uncommitted(
+                        block_id,
+                    ));
                     break;
                 }
             }
         }
 
-        let block_list = azure_storage_blobs::blob::BlockList {
-            blocks
-        };
+        let block_list = azure_storage_blobs::blob::BlockList { blocks };
         client.put_block_list(block_list).await?;
 
         Ok(())

crates/blobstore-azure/src/store/auth.rs

@@ -18,10 +18,10 @@ pub enum AzureBlobAuthOptions {
     AccountKey(AzureKeyAuth),
     /// Spin should use the environment variables of the process to
     /// create the StorageCredentials for the storage client. For now this uses old school credentials:
-    /// 
+    ///
     /// STORAGE_ACCOUNT
     /// STORAGE_ACCESS_KEY
-    /// 
+    ///
     /// TODO: Thorsten pls make this proper with *hand waving* managed identity and stuff!
     Environmental,
 }

crates/blobstore-azure/src/store/incoming_data.rs

@@ -1,19 +1,14 @@
-use tokio::sync::Mutex;
 use anyhow::Result;
-use futures::StreamExt;
-use spin_core::async_trait;
 use azure_core::Pageable;
 use azure_storage_blobs::blob::operations::GetBlobResponse;
 use azure_storage_blobs::prelude::BlobClient;
+use futures::StreamExt;
+use spin_core::async_trait;
+use tokio::sync::Mutex;
 
 pub struct AzureIncomingData {
     // The Mutex is used to make it Send
-    stm: Mutex<Option<
-        Pageable<
-            GetBlobResponse,
-            azure_core::error::Error
-        >
-    >>,
+    stm: Mutex<Option<Pageable<GetBlobResponse, azure_core::error::Error>>>,
     client: BlobClient,
 }
 
@@ -35,14 +30,18 @@ impl AzureIncomingData {
         wasmtime_wasi::pipe::AsyncReadStream::new(arr)
     }
 
-    fn consume_as_stream(&mut self) -> impl futures::stream::Stream<Item = Result<Vec<u8>, std::io::Error>> {
+    fn consume_as_stream(
+        &mut self,
+    ) -> impl futures::stream::Stream<Item = Result<Vec<u8>, std::io::Error>> {
         let opt_stm = self.stm.get_mut();
         let stm = opt_stm.take().unwrap();
         stm.flat_map(|chunk| streamify_chunk(chunk.unwrap().data))
     }
 }
 
-fn streamify_chunk(chunk: azure_core::ResponseBody) -> impl futures::stream::Stream<Item = Result<Vec<u8>, std::io::Error>> {
+fn streamify_chunk(
+    chunk: azure_core::ResponseBody,
+) -> impl futures::stream::Stream<Item = Result<Vec<u8>, std::io::Error>> {
     chunk.map(|c| Ok(c.unwrap().to_vec()))
 }
 
@@ -76,6 +75,12 @@ impl spin_factor_blobstore::IncomingData for AzureIncomingData {
         // we do the first read. So that would force us to either pre-fetch the
         // first chunk or to issue a properties request *just in case* size() was
         // called. So I'm making it fallible for now.
-        Ok(self.client.get_properties().await?.blob.properties.content_length)
+        Ok(self
+            .client
+            .get_properties()
+            .await?
+            .blob
+            .properties
+            .content_length)
     }
 }

crates/blobstore-azure/src/store/object_names.rs

@@ -6,29 +6,21 @@ use spin_core::async_trait;
 
 pub struct AzureObjectNames {
     // The Mutex is used to make it Send
-    stm: Mutex<
-        Pageable<
-            ListBlobsResponse,
-            azure_core::error::Error
-        >
-    >,
+    stm: Mutex<Pageable<ListBlobsResponse, azure_core::error::Error>>,
     read_but_not_yet_returned: Vec<String>,
     end_stm_after_read_but_not_yet_returned: bool,
 }
 
 impl AzureObjectNames {
-    pub fn new(stm: Pageable<
-        ListBlobsResponse,
-        azure_core::error::Error
-    >) -> Self {
+    pub fn new(stm: Pageable<ListBlobsResponse, azure_core::error::Error>) -> Self {
         Self {
             stm: Mutex::new(stm),
             read_but_not_yet_returned: Default::default(),
             end_stm_after_read_but_not_yet_returned: false,
         }
     }
 
-    async fn read_impl(&mut self, len: u64) -> anyhow::Result<(Vec<String>,bool)> {
+    async fn read_impl(&mut self, len: u64) -> anyhow::Result<(Vec<String>, bool)> {
         use futures::StreamExt;
 
         let len: usize = len.try_into().unwrap();
@@ -78,7 +70,7 @@ impl AzureObjectNames {
 #[async_trait]
 impl spin_factor_blobstore::ObjectNames for AzureObjectNames {
     async fn read(&mut self, len: u64) -> anyhow::Result<(Vec<String>, bool)> {
-        self.read_impl(len).await  // Separate function because rust-analyser gives better intellisense when async_trait isn't in the picture!
+        self.read_impl(len).await // Separate function because rust-analyser gives better intellisense when async_trait isn't in the picture!
     }
 
     async fn skip(&mut self, num: u64) -> anyhow::Result<(u64, bool)> {