move storage queues to pipeline architecture (#851)

* move storage queues to pipeline architecture

* fmt

Co-authored-by: Brian Caswell <[email protected]>

Author: bmc-msft

sdk/storage/src/core/clients/storage_account_client.rs

@@ -54,7 +54,7 @@ impl std::fmt::Debug for StorageCredentials {
 #[derive(Debug, Clone, Copy)]
 pub enum ServiceType {
     Blob,
-    // Queue,
+    Queue,
     // File,
     Table,
 }

sdk/storage_queues/Cargo.toml

@@ -29,6 +29,7 @@ url = "2.2"
 
 [dev-dependencies]
 tokio = { version = "1.0", features = ["full"] }
+uuid = { version = "1.0", features = ["v4"] }
 
 [features]
 default = ["enable_reqwest"]

sdk/storage_queues/examples/delete_message.rs

@@ -30,7 +30,7 @@ async fn main() -> azure_core::Result<()> {
         .get_messages()
         .number_of_messages(2)
         .visibility_timeout(Duration::from_secs(5)) // the message will become visible again after 5 secs
-        .execute()
+        .into_future()
         .await?;
 
     println!("get_response == {:#?}", get_response);
@@ -44,7 +44,7 @@ async fn main() -> azure_core::Result<()> {
             let delete_response = queue
                 .pop_receipt_client(message_to_delete)
                 .delete()
-                .execute()
+                .into_future()
                 .await?;
 
             println!("delete_response == {:#?}", delete_response);

sdk/storage_queues/examples/get_messages.rs

@@ -30,7 +30,7 @@ async fn main() -> azure_core::Result<()> {
         .get_messages()
         .number_of_messages(2)
         .visibility_timeout(Duration::from_secs(5)) // the message will become visible again after 5 secs
-        .execute()
+        .into_future()
         .await?;
 
     println!("response == {:#?}", response);
@@ -39,7 +39,7 @@ async fn main() -> azure_core::Result<()> {
         .get_messages()
         .number_of_messages(2)
         .visibility_timeout(Duration::from_secs(10)) // the message will become visible again after 10 secs
-        .execute()
+        .into_future()
         .await?;
     println!("get_messages_response == {:#?}", get_messages_response);
 
@@ -50,8 +50,11 @@ async fn main() -> azure_core::Result<()> {
         let pop_receipt = queue.pop_receipt_client(message_to_update);
 
         let response = pop_receipt
-            .update(Duration::from_secs(4))
-            .execute(format!("new body at {}", chrono::Utc::now()))
+            .update(
+                format!("new body at {}", chrono::Utc::now()),
+                Duration::from_secs(4),
+            )
+            .into_future()
             .await?;
         println!("response == {:#?}", response);
     }

sdk/storage_queues/examples/list_queues.rs

@@ -19,13 +19,16 @@ async fn main() -> azure_core::Result<()> {
     let queue_service = storage_account.queue_service_client();
 
     println!("getting service stats");
-    let response = queue_service.get_queue_service_stats().execute().await?;
+    let response = queue_service
+        .get_queue_service_stats()
+        .into_future()
+        .await?;
     println!("get_queue_service_properties.response == {:#?}", response);
 
     println!("getting service properties");
     let response = queue_service
         .get_queue_service_properties()
-        .execute()
+        .into_future()
         .await?;
     println!("get_queue_service_stats.response == {:#?}", response);
 
@@ -35,17 +38,16 @@ async fn main() -> azure_core::Result<()> {
         .prefix("a")
         .include_metadata(true)
         .max_results(NonZeroU32::new(2u32).unwrap())
-        .execute()
-        .await?;
+        .into_stream()
+        .next()
+        .await;
     println!("response == {:#?}", response);
 
     println!("streaming queues");
-    let mut stream = Box::pin(
-        queue_service
-            .list_queues()
-            .max_results(NonZeroU32::new(3u32).unwrap())
-            .stream(),
-    );
+    let mut stream = queue_service
+        .list_queues()
+        .max_results(NonZeroU32::new(3u32).unwrap())
+        .into_stream();
 
     while let Some(value) = stream.next().await {
         let value = value?;

sdk/storage_queues/examples/peek_messages.rs

@@ -29,7 +29,7 @@ async fn main() -> azure_core::Result<()> {
     let response = queue
         .peek_messages()
         .number_of_messages(2)
-        .execute()
+        .into_future()
         .await?;
 
     println!("response == {:#?}", response);

sdk/storage_queues/examples/put_message.rs

@@ -25,9 +25,8 @@ async fn main() -> azure_core::Result<()> {
 
     trace!("putting message");
     let response = queue
-        .put_message()
-        .client_request_id("optional correlation token")
-        .execute(format!("Azure SDK for Rust rocks! {}", chrono::Utc::now()))
+        .put_message(format!("Azure SDK for Rust rocks! {}", chrono::Utc::now()))
+        .into_future()
         .await?;
 
     println!("response == {:#?}", response);

sdk/storage_queues/examples/queue_create.rs

@@ -37,7 +37,11 @@ async fn main() -> azure_core::Result<()> {
         .as_mut()
         .insert("created".into(), format!("{:?}", Utc::now()).into());
 
-    let response = queue.create().metadata(&metadata).execute().await?;
+    let response = queue
+        .create()
+        .metadata(metadata.clone())
+        .into_future()
+        .await?;
     println!("response == {:#?}", response);
 
     // let's add some more metadata
@@ -46,15 +50,15 @@ async fn main() -> azure_core::Result<()> {
 
     println!("metadata == {:#?}", metadata);
 
-    let response = queue.set_metadata().execute(&metadata).await?;
+    let response = queue.set_metadata(metadata).into_future().await?;
     println!("response == {:#?}", response);
 
     // let's get back the metadata
-    let response = queue.get_metadata().execute().await?;
+    let response = queue.get_metadata().into_future().await?;
     println!("response == {:#?}", response);
 
     // use two queue stored access policies
-    let queue_stored_acess_policies = vec![
+    let policies = vec![
         QueueStoredAccessPolicy::new(
             "first_sap_read_process",
             Utc::now() - Duration::hours(1),
@@ -70,22 +74,15 @@ async fn main() -> azure_core::Result<()> {
         .enable_all(),
     ];
 
-    let response = queue
-        .set_acl()
-        .execute(&queue_stored_acess_policies)
-        .await?;
+    let response = queue.set_acl(policies).into_future().await?;
     println!("response == {:#?}", response);
 
     // get the queue ACL
-    let response = queue.get_acl().execute().await?;
+    let response = queue.get_acl().into_future().await?;
     println!("response == {:#?}", response);
 
     // now let's delete it
-    let response = queue
-        .delete()
-        .client_request_id("myclientid")
-        .execute()
-        .await?;
+    let response = queue.delete().into_future().await?;
     println!("response == {:#?}", response);
 
     Ok(())

sdk/storage_queues/src/clients/pop_receipt_client.rs

@@ -1,5 +1,5 @@
-use crate::prelude::*;
-use crate::requests::*;
+use crate::{operations::*, prelude::*};
+use azure_core::{Context, Request, Response};
 use azure_storage::core::clients::StorageClient;
 use std::sync::Arc;
 
@@ -39,6 +39,14 @@ impl PopReceiptClient {
         })
     }
 
+    pub(crate) async fn send(
+        &self,
+        context: &mut Context,
+        request: &mut Request,
+    ) -> azure_core::Result<Response> {
+        self.queue_client.send(context, request).await
+    }
+
     pub(crate) fn storage_client(&self) -> &StorageClient {
         self.queue_client.storage_client()
     }
@@ -56,16 +64,19 @@ impl PopReceiptClient {
         Ok(url)
     }
 
-    /// Deletes the message. The message must not have been
-    /// made visible again or this call would fail.
+    /// Deletes the message. The message must not have been made visible again
+    /// or this call would fail.
     pub fn delete(&self) -> DeleteMessageBuilder {
-        DeleteMessageBuilder::new(self)
+        DeleteMessageBuilder::new(self.clone())
     }
 
-    /// Updates the message.
-    /// The message must not have been
-    /// made visible again or this call would fail.
-    pub fn update(&self, visibility_timeout: impl Into<VisibilityTimeout>) -> UpdateMessageBuilder {
-        UpdateMessageBuilder::new(self, visibility_timeout)
+    /// Updates the message.  The message must not have been made visible again
+    /// or this call would fail.
+    pub fn update(
+        &self,
+        body: impl Into<String>,
+        visibility_timeout: impl Into<VisibilityTimeout>,
+    ) -> UpdateMessageBuilder {
+        UpdateMessageBuilder::new(self.clone(), body.into(), visibility_timeout.into())
     }
 }

sdk/storage_queues/src/clients/queue_client.rs

@@ -1,8 +1,13 @@
-use crate::requests::*;
-use azure_core::error::{ErrorKind, ResultExt};
-use azure_storage::core::clients::{AsStorageClient, StorageAccountClient, StorageClient};
-use std::fmt::Debug;
-use std::sync::Arc;
+use crate::{operations::*, QueueStoredAccessPolicy};
+use azure_core::{
+    error::{ErrorKind, ResultExt},
+    prelude::*,
+    Context, Request, Response,
+};
+use azure_storage::core::clients::{
+    AsStorageClient, ServiceType, StorageAccountClient, StorageClient,
+};
+use std::{fmt::Debug, sync::Arc};
 
 pub trait AsQueueClient<QN: Into<String>> {
     fn queue_client(&self, queue_name: QN) -> Arc<QueueClient>;
@@ -34,6 +39,16 @@ impl QueueClient {
         })
     }
 
+    pub(crate) async fn send(
+        &self,
+        context: &mut Context,
+        request: &mut Request,
+    ) -> azure_core::Result<Response> {
+        self.storage_client
+            .send(context, request, ServiceType::Queue)
+            .await
+    }
+
     pub(crate) fn storage_client(&self) -> &StorageClient {
         self.storage_client.as_ref()
     }
@@ -53,68 +68,77 @@ impl QueueClient {
 
     /// Creates the queue.
     pub fn create(&self) -> CreateQueueBuilder {
-        CreateQueueBuilder::new(self)
+        CreateQueueBuilder::new(self.clone())
     }
 
     /// Deletes the queue.
     pub fn delete(&self) -> DeleteQueueBuilder {
-        DeleteQueueBuilder::new(self)
+        DeleteQueueBuilder::new(self.clone())
     }
 
-    /// Sets or clears the queue metadata. The metadata
-    /// will be passed to the `execute` function of the returned struct.
-    pub fn set_metadata(&self) -> SetQueueMetadataBuilder {
-        SetQueueMetadataBuilder::new(self)
+    /// Sets or clears the queue metadata.
+    ///
+    /// Keep in mind that keys present on Azure but not included in the passed
+    /// metadata parameter will be deleted. If you want to keep the preexisting
+    /// key-value pairs, retrieve them with GetMetadata first and then
+    /// update/add to the received Metadata struct. Then pass the Metadata back
+    /// to SetQueueMetadata.  If you just want to clear the metadata, just pass
+    /// an empty Metadata struct.
+    pub fn set_metadata(&self, metadata: Metadata) -> SetQueueMetadataBuilder {
+        SetQueueMetadataBuilder::new(self.clone(), metadata)
     }
 
     /// Get the queue metadata.
+
     pub fn get_metadata(&self) -> GetQueueMetadataBuilder {
-        GetQueueMetadataBuilder::new(self)
+        GetQueueMetadataBuilder::new(self.clone())
     }
 
     /// Get the queue ACL. This call returns
     /// all the stored access policies associated
     /// to the current queue.
     pub fn get_acl(&self) -> GetQueueACLBuilder {
-        GetQueueACLBuilder::new(self)
+        GetQueueACLBuilder::new(self.clone())
     }
 
-    /// Set the queue ACL. You can call this function
-    /// to change or remove already existing stored
-    /// access policies by modifying the list returned
+    /// Set the queue ACL. You can call this function to change or remove
+    /// already existing stored access policies by modifying the list returned
     /// by `get_acl`.
-    pub fn set_acl(&self) -> SetQueueACLBuilder {
-        SetQueueACLBuilder::new(self)
+    ///
+    /// While this SDK does not enforce any limit, keep in mind Azure supports a
+    /// limited number of stored access policies for each queue.  More info here
+    /// [https://docs.microsoft.com/rest/api/storageservices/set-queue-acl#remarks](https://docs.microsoft.com/rest/api/storageservices/set-queue-acl#remarks).
+    pub fn set_acl(&self, policies: Vec<QueueStoredAccessPolicy>) -> SetQueueACLBuilder {
+        SetQueueACLBuilder::new(self.clone(), policies)
     }
 
     /// Puts a message in the queue. The body will be passed
     /// to the `execute` function of the returned struct.
-    pub fn put_message(&self) -> PutMessageBuilder {
-        PutMessageBuilder::new(self)
+    pub fn put_message<S: Into<String>>(&self, message: S) -> PutMessageBuilder {
+        PutMessageBuilder::new(self.clone(), message.into())
     }
 
     /// Peeks, without removing, one or more messages.
     pub fn peek_messages(&self) -> PeekMessagesBuilder {
-        PeekMessagesBuilder::new(self)
+        PeekMessagesBuilder::new(self.clone())
     }
 
     /// Gets, shadowing them, one or more messages.
     pub fn get_messages(&self) -> GetMessagesBuilder {
-        GetMessagesBuilder::new(self)
+        GetMessagesBuilder::new(self.clone())
     }
 
     /// Removes all messages from the queue.
     pub fn clear_messages(&self) -> ClearMessagesBuilder {
-        ClearMessagesBuilder::new(self)
+        ClearMessagesBuilder::new(self.clone())
     }
 }
 
 #[cfg(test)]
 #[cfg(feature = "test_integration")]
 mod integration_tests {
     use super::*;
-    use crate::core::prelude::*;
-    use crate::queue::clients::AsQueueClient;
+    use crate::{core::prelude::*, queue::clients::AsQueueClient};
 
     fn get_emulator_client(queue_name: &str) -> Arc<QueueClient> {
         let storage_account = StorageAccountClient::new_emulator_default().storage_client();