Address feedback from Storage Queue pipeline pr (#858)

* instead of passing around Bytes, use u8 slice

* fix clippy

* address feedback from Pipeline PR

* remove message_id and pop_receipt direct exposure

* address clippy

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

Author: bmc-msft

sdk/storage_queues/Cargo.toml

@@ -15,7 +15,6 @@ edition = "2021"
 [dependencies]
 azure_core = { path = "../core", version = "0.3", default-features=false }
 azure_storage = { path = "../storage", version = "0.4", default-features=false, features=["account"] }
-bytes = "1.0"
 chrono = { version = "0.4", features = ["serde"] }
 futures = "0.3"
 http = "0.2"

sdk/storage_queues/src/clients/pop_receipt_client.rs

@@ -64,13 +64,9 @@ impl PopReceiptClient {
         Ok(url)
     }
 
-    /// 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.clone())
-    }
-
-    /// Updates the message.  The message must not have been made visible again
+    /// Updates the message.
+    ///
+    /// The message must not have been made visible again
     /// or this call would fail.
     pub fn update(
         &self,
@@ -79,4 +75,12 @@ impl PopReceiptClient {
     ) -> UpdateMessageBuilder {
         UpdateMessageBuilder::new(self.clone(), body.into(), visibility_timeout.into())
     }
+
+    /// 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.clone())
+    }
 }

sdk/storage_queues/src/clients/queue_client.rs

@@ -94,16 +94,18 @@ impl QueueClient {
         GetQueueMetadataBuilder::new(self.clone())
     }
 
-    /// Get the queue ACL. This call returns
-    /// all the stored access policies associated
-    /// to the current queue.
+    /// 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.clone())
     }
 
-    /// 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`.
+    /// 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`.
     ///
     /// 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
@@ -112,8 +114,10 @@ impl QueueClient {
         SetQueueACLBuilder::new(self.clone(), policies)
     }
 
-    /// Puts a message in the queue. The body will be passed
-    /// to the `execute` function of the returned struct.
+    /// Puts a message in the queue.
+    ///
+    /// The body will be passed to the `execute` function of the returned
+    /// struct.
     pub fn put_message<S: Into<String>>(&self, message: S) -> PutMessageBuilder {
         PutMessageBuilder::new(self.clone(), message.into())
     }

sdk/storage_queues/src/operations/get_messages.rs

@@ -2,12 +2,12 @@ use crate::{clients::QueueClient, prelude::*, PopReceipt};
 use azure_core::{
     collect_pinned_stream,
     error::{ErrorKind, ResultExt},
-    headers::utc_date_from_rfc2822,
     prelude::*,
     Context, Response as AzureResponse,
 };
 use azure_storage::core::{headers::CommonStorageResponseHeaders, xml::read_xml};
 use chrono::{DateTime, Utc};
+use serde::Deserialize;
 use std::convert::TryInto;
 
 #[derive(Debug, Clone)]
@@ -25,7 +25,6 @@ impl GetMessagesBuilder {
             queue_client,
             number_of_messages: None,
             visibility_timeout: None,
-
             timeout: None,
             context: Context::new(),
         }
@@ -78,68 +77,105 @@ pub struct GetMessagesResponse {
     pub messages: Vec<Message>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Deserialize)]
 pub struct Message {
-    pub pop_receipt: PopReceipt,
+    #[serde(rename = "MessageId")]
+    message_id: String,
+    #[serde(rename = "PopReceipt")]
+    pop_receipt: String,
+    #[serde(rename = "InsertionTime", deserialize_with = "deserialize_utc")]
     pub insertion_time: DateTime<Utc>,
+    #[serde(rename = "ExpirationTime", deserialize_with = "deserialize_utc")]
     pub expiration_time: DateTime<Utc>,
+    #[serde(rename = "TimeNextVisible", deserialize_with = "deserialize_utc")]
     pub time_next_visible: DateTime<Utc>,
+    #[serde(rename = "DequeueCount")]
     pub dequeue_count: u64,
+    #[serde(rename = "MessageText")]
     pub message_text: String,
 }
 
-impl From<Message> for PopReceipt {
-    fn from(message: Message) -> Self {
-        message.pop_receipt
+impl Message {
+    pub fn pop_receipt(&self) -> PopReceipt {
+        PopReceipt::new(self.message_id.clone(), self.pop_receipt.clone())
     }
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
-struct MessageInternal {
-    #[serde(rename = "MessageId")]
-    pub message_id: String,
-    #[serde(rename = "InsertionTime")]
-    pub insertion_time: String,
-    #[serde(rename = "ExpirationTime")]
-    pub expiration_time: String,
-    #[serde(rename = "PopReceipt")]
-    pub pop_receipt: String,
-    #[serde(rename = "TimeNextVisible")]
-    pub time_next_visible: String,
-    #[serde(rename = "DequeueCount")]
-    pub dequeue_count: u64,
-    #[serde(rename = "MessageText")]
-    pub message_text: String,
+impl From<Message> for PopReceipt {
+    fn from(message: Message) -> Self {
+        PopReceipt::new(message.message_id, message.pop_receipt)
+    }
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
-struct MessagesInternal {
+#[derive(Debug, Clone, Deserialize)]
+struct MessageList {
     #[serde(rename = "QueueMessage")]
-    pub messages: Option<Vec<MessageInternal>>,
+    pub messages: Option<Vec<Message>>,
 }
 
 impl GetMessagesResponse {
+    fn parse_messages(body: &[u8]) -> azure_core::Result<Vec<Message>> {
+        let response: MessageList = read_xml(body).map_kind(ErrorKind::DataConversion)?;
+        Ok(response.messages.unwrap_or_default())
+    }
+
     async fn try_from(response: AzureResponse) -> azure_core::Result<Self> {
         let (_, headers, body) = response.deconstruct();
         let body = collect_pinned_stream(body).await?;
 
-        let response: MessagesInternal = read_xml(&body).map_kind(ErrorKind::DataConversion)?;
-
-        let mut messages = Vec::new();
-        for message in response.messages.unwrap_or_default().into_iter() {
-            messages.push(Message {
-                pop_receipt: PopReceipt::new(message.message_id, message.pop_receipt),
-                insertion_time: utc_date_from_rfc2822(&message.insertion_time)?,
-                expiration_time: utc_date_from_rfc2822(&message.expiration_time)?,
-                time_next_visible: utc_date_from_rfc2822(&message.time_next_visible)?,
-                dequeue_count: message.dequeue_count,
-                message_text: message.message_text,
-            })
-        }
+        let messages = Self::parse_messages(&body)?;
 
         Ok(GetMessagesResponse {
             common_storage_response_headers: (&headers).try_into()?,
             messages,
         })
     }
 }
+
+fn deserialize_utc<'de, D>(deserializer: D) -> std::result::Result<DateTime<Utc>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let s = String::deserialize(deserializer)?;
+    let date = DateTime::parse_from_rfc2822(&s).map_err(serde::de::Error::custom)?;
+    Ok(DateTime::from_utc(date.naive_utc(), Utc))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_messages() -> azure_core::Result<()> {
+        let message = b"\xef\xbb\xbf\
+             <?xml version=\"1.0\" encoding=\"utf-8\"?>\
+            <QueueMessagesList><QueueMessage>\
+            <MessageId>00000000-0000-0000-0000-000000000000</MessageId>\
+            <InsertionTime>Mon, 27 Jun 2022 13:38:48 GMT</InsertionTime>\
+            <ExpirationTime>Mon, 04 Jul 2022 13:38:48 GMT</ExpirationTime>\
+            <PopReceipt>REDACTED1</PopReceipt>\
+            <TimeNextVisible>Mon, 27 Jun 2022 13:38:53 GMT</TimeNextVisible>\
+            <DequeueCount>1</DequeueCount>\
+            <MessageText>test1</MessageText>\
+            </QueueMessage>\
+            <QueueMessage>\
+            <MessageId>11111111-1111-1111-1111-111111111111</MessageId>\
+            <InsertionTime>Mon, 27 Jun 2022 13:38:48 GMT</InsertionTime>\
+            <ExpirationTime>Mon, 04 Jul 2022 13:38:48 GMT</ExpirationTime>\
+            <PopReceipt>REDACTED2</PopReceipt>\
+            <TimeNextVisible>Mon, 27 Jun 2022 13:38:53 GMT</TimeNextVisible>\
+            <DequeueCount>1</DequeueCount>\
+            <MessageText>test2</MessageText>\
+            </QueueMessage>\
+            </QueueMessagesList>\
+            ";
+
+        let messages = GetMessagesResponse::parse_messages(message)?;
+
+        assert_eq!(messages.len(), 2);
+        assert_eq!(messages[0].message_text, "test1");
+        assert_eq!(messages[1].message_text, "test2");
+
+        Ok(())
+    }
+}