add service bus topic support (#1155)

Author: Marek Pavelka

sdk/messaging_servicebus/examples/service_bus00.rs

@@ -16,7 +16,7 @@ async fn main() {
 
     let http_client = azure_core::new_http_client();
 
-    let client = Client::new(
+    let client = QueueClient::new(
         http_client,
         service_bus_namespace,
         queue_name,

sdk/messaging_servicebus/examples/service_bus01.rs

@@ -0,0 +1,49 @@
+use azure_messaging_servicebus::prelude::*;
+
+#[tokio::main]
+async fn main() {
+    let service_bus_namespace = std::env::var("AZURE_SERVICE_BUS_NAMESPACE")
+        .expect("Please set AZURE_SERVICE_BUS_NAMESPACE env variable first!");
+
+    let topic_name =
+        std::env::var("AZURE_TOPIC_NAME").expect("Please set AZURE_TOPIC_NAME env variable first!");
+
+    let subscription_name =
+        std::env::var("SUBSCRIPTION").expect("Please set SUBSCRIPTION env variable first!");
+
+    let policy_name = std::env::var("AZURE_POLICY_NAME")
+        .expect("Please set AZURE_POLICY_NAME env variable first!");
+
+    let policy_key =
+        std::env::var("AZURE_POLICY_KEY").expect("Please set AZURE_POLICY_KEY env variable first!");
+
+    let http_client = azure_core::new_http_client();
+
+    let client = TopicClient::new(
+        http_client,
+        service_bus_namespace,
+        topic_name,
+        policy_name,
+        policy_key,
+    )
+    .expect("Failed to create client");
+
+    let sender = client.topic_sender();
+    let receiver = client.subscription_receiver(&subscription_name);
+
+    let message_to_send = "hello, world!";
+
+    sender
+        .send_message(message_to_send)
+        .await
+        .expect("Failed to send message while testing receive");
+
+    println!("Sent Message: {}", message_to_send);
+
+    let received_message = receiver
+        .receive_and_delete_message()
+        .await
+        .expect("Failed to receive message");
+
+    println!("Received Message: {}", received_message);
+}

sdk/messaging_servicebus/src/prelude.rs

@@ -1 +1 @@
-pub use crate::service_bus::Client;
+pub use crate::service_bus::{QueueClient, SubscriptionReceiver, TopicClient, TopicSender};

sdk/messaging_servicebus/src/service_bus/mod.rs

@@ -6,11 +6,13 @@ use std::{ops::Add, sync::Arc};
 use time::OffsetDateTime;
 use url::form_urlencoded::{self, Serializer};
 
-mod client;
+mod queue_client;
+mod topic_client;
 
-use crate::utils::craft_peek_lock_url;
+use crate::utils::{craft_peek_lock_url, get_head_url};
 
-pub use self::client::Client;
+pub use self::queue_client::QueueClient;
+pub use self::topic_client::{SubscriptionReceiver, TopicClient, TopicSender};
 
 /// Default duration for the SAS token in days — We might want to make this configurable at some point
 const DEFAULT_SAS_DURATION: u64 = 3_600; // seconds = 1 hour
@@ -77,18 +79,18 @@ fn generate_signature(
     )
 }
 
-/// Sends a message to the queue
+/// Sends a message to the queue or topic
 async fn send_message(
     http_client: &Arc<dyn HttpClient>,
     namespace: &str,
-    queue: &str,
+    queue_or_topic: &str,
     policy_name: &str,
     signing_key: &hmac::Key,
     msg: &str,
 ) -> azure_core::Result<()> {
     let url = format!(
         "https://{}.servicebus.windows.net/{}/messages",
-        namespace, queue
+        namespace, queue_or_topic
     );
 
     let req = finalize_request(
@@ -110,15 +112,12 @@ async fn send_message(
 async fn receive_and_delete_message(
     http_client: &Arc<dyn HttpClient>,
     namespace: &str,
-    queue: &str,
+    queue_or_topic: &str,
     policy_name: &str,
     signing_key: &hmac::Key,
+    subscription: Option<&str>,
 ) -> azure_core::Result<CollectedResponse> {
-    let url = format!(
-        "https://{}.servicebus.windows.net/{}/messages/head",
-        namespace, queue
-    );
-
+    let url = get_head_url(namespace, queue_or_topic, subscription);
     let req = finalize_request(&url, Method::Delete, None, policy_name, signing_key)?;
 
     http_client
@@ -138,12 +137,13 @@ async fn receive_and_delete_message(
 async fn peek_lock_message(
     http_client: &Arc<dyn HttpClient>,
     namespace: &str,
-    queue: &str,
+    queue_or_topic: &str,
     policy_name: &str,
     signing_key: &hmac::Key,
     lock_expiry: Option<Duration>,
+    subscription: Option<&str>,
 ) -> azure_core::Result<CollectedResponse> {
-    let url = craft_peek_lock_url(namespace, queue, lock_expiry)?;
+    let url = craft_peek_lock_url(namespace, queue_or_topic, lock_expiry, subscription)?;
 
     let req = finalize_request(url.as_ref(), Method::Post, None, policy_name, signing_key)?;
 
@@ -160,12 +160,13 @@ async fn peek_lock_message(
 async fn peek_lock_message2(
     http_client: &Arc<dyn HttpClient>,
     namespace: &str,
-    queue: &str,
+    queue_or_topic: &str,
     policy_name: &str,
     signing_key: &hmac::Key,
     lock_expiry: Option<Duration>,
+    subscription: Option<&str>,
 ) -> azure_core::Result<PeekLockResponse> {
-    let url = craft_peek_lock_url(namespace, queue, lock_expiry)?;
+    let url = craft_peek_lock_url(namespace, queue_or_topic, lock_expiry, subscription)?;
 
     let req = finalize_request(url.as_ref(), Method::Post, None, policy_name, signing_key)?;
 

…ing_servicebus/src/service_bus/client.rs …rvicebus/src/service_bus/queue_client.rssdk/messaging_servicebus/src/service_bus/client.rs renamed to sdk/messaging_servicebus/src/service_bus/queue_client.rs sdk/messaging_servicebus/src/service_bus/client.rs renamed to sdk/messaging_servicebus/src/service_bus/queue_client.rs

@@ -14,23 +14,23 @@ use azure_core::{error::Error, HttpClient};
 
 /// Client object that allows interaction with the ServiceBus API
 #[derive(Debug, Clone)]
-pub struct Client {
+pub struct QueueClient {
     http_client: Arc<dyn HttpClient>,
     namespace: String,
     queue: String,
     policy_name: String,
     signing_key: Key,
 }
 
-impl Client {
-    /// Creates a new client instance
+impl QueueClient {
+    /// Creates a new queue client instance
     pub fn new<N, Q, P, K>(
         http_client: Arc<dyn HttpClient>,
         namespace: N,
         queue: Q,
         policy_name: P,
         policy_key: K,
-    ) -> Result<Client, Error>
+    ) -> Result<QueueClient, Error>
     where
         N: Into<String>,
         Q: Into<String>,
@@ -39,7 +39,7 @@ impl Client {
     {
         let signing_key = Key::new(ring::hmac::HMAC_SHA256, policy_key.as_ref().as_bytes());
 
-        Ok(Client {
+        Ok(QueueClient {
             http_client,
             namespace: namespace.into(),
             queue: queue.into(),
@@ -70,6 +70,7 @@ impl Client {
                 &self.queue,
                 &self.policy_name,
                 &self.signing_key,
+                None,
             )
             .await?
             .body(),
@@ -93,6 +94,7 @@ impl Client {
                 &self.policy_name,
                 &self.signing_key,
                 lock_expiry,
+                None,
             )
             .await?
             .body(),
@@ -114,6 +116,7 @@ impl Client {
             &self.policy_name,
             &self.signing_key,
             timeout,
+            None,
         )
         .await
     }

sdk/messaging_servicebus/src/service_bus/topic_client.rs

@@ -0,0 +1,159 @@
+use std::sync::Arc;
+
+use crate::{
+    service_bus::{
+        peek_lock_message, peek_lock_message2, receive_and_delete_message, send_message,
+        PeekLockResponse,
+    },
+    utils::body_bytes_to_utf8,
+};
+use ring::hmac::Key;
+use std::time::Duration;
+
+use azure_core::{error::Error, HttpClient};
+
+/// Client object that allows interaction with the ServiceBus API
+#[derive(Debug, Clone)]
+pub struct TopicClient {
+    http_client: Arc<dyn HttpClient>,
+    namespace: String,
+    topic: String,
+    policy_name: String,
+    signing_key: Key,
+}
+
+#[derive(Debug, Clone)]
+pub struct TopicSender {
+    topic_client: TopicClient,
+}
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionReceiver {
+    topic_client: TopicClient,
+    subscription: String,
+}
+
+impl TopicClient {
+    /// Creates a new topic client instance
+    pub fn new<N, T, P, K>(
+        http_client: Arc<dyn HttpClient>,
+        namespace: N,
+        topic: T,
+        policy_name: P,
+        policy_key: K,
+    ) -> Result<TopicClient, Error>
+    where
+        N: Into<String>,
+        T: Into<String>,
+        P: Into<String>,
+        K: AsRef<str>,
+    {
+        let signing_key = Key::new(ring::hmac::HMAC_SHA256, policy_key.as_ref().as_bytes());
+
+        Ok(Self {
+            http_client,
+            namespace: namespace.into(),
+            topic: topic.into(),
+            policy_name: policy_name.into(),
+            signing_key,
+        })
+    }
+
+    pub fn topic_sender(&self) -> TopicSender {
+        TopicSender::new(self.clone())
+    }
+
+    pub fn subscription_receiver(&self, subscription: &str) -> SubscriptionReceiver {
+        SubscriptionReceiver::new(self.clone(), subscription)
+    }
+}
+
+impl TopicSender {
+    pub fn new(topic_client: TopicClient) -> TopicSender {
+        Self { topic_client }
+    }
+    /// Sends a message to the topic
+    pub async fn send_message(&self, msg: &str) -> Result<(), Error> {
+        send_message(
+            &self.topic_client.http_client,
+            &self.topic_client.namespace,
+            &self.topic_client.topic,
+            &self.topic_client.policy_name,
+            &self.topic_client.signing_key,
+            msg,
+        )
+        .await
+    }
+}
+
+impl SubscriptionReceiver {
+    pub fn new<S>(topic_client: TopicClient, subscription: S) -> SubscriptionReceiver
+    where
+        S: Into<String>,
+    {
+        Self {
+            topic_client,
+            subscription: subscription.into(),
+        }
+    }
+
+    /// Receive and delete a message
+    pub async fn receive_and_delete_message(&self) -> Result<String, Error> {
+        body_bytes_to_utf8(
+            receive_and_delete_message(
+                &self.topic_client.http_client,
+                &self.topic_client.namespace,
+                &self.topic_client.topic,
+                &self.topic_client.policy_name,
+                &self.topic_client.signing_key,
+                Some(&self.subscription),
+            )
+            .await?
+            .body(),
+        )
+    }
+
+    /// Non-destructively read a message
+    ///
+    /// Note: This function does not return the delete location
+    /// of the message, so, after reading, you will lose
+    /// "track" of it until the lock expiry runs out and
+    /// the message can be consumed by others. If you want to keep
+    /// track of this message (i.e., have the possibility of deletion),
+    /// use `peek_lock_message2`.
+    pub async fn peek_lock_message(&self, lock_expiry: Option<Duration>) -> Result<String, Error> {
+        body_bytes_to_utf8(
+            peek_lock_message(
+                &self.topic_client.http_client,
+                &self.topic_client.namespace,
+                &self.topic_client.topic,
+                &self.topic_client.policy_name,
+                &self.topic_client.signing_key,
+                lock_expiry,
+                Some(&self.subscription),
+            )
+            .await?
+            .body(),
+        )
+    }
+
+    /// Non-destructively read a message but track it
+    ///
+    /// Note: This function returns a `PeekLockResponse`
+    /// that contains a helper `delete_message` function.
+    pub async fn peek_lock_message2(
+        &self,
+        timeout: Option<Duration>,
+    ) -> Result<PeekLockResponse, Error> {
+        peek_lock_message2(
+            &self.topic_client.http_client,
+            &self.topic_client.namespace,
+            &self.topic_client.topic,
+            &self.topic_client.policy_name,
+            &self.topic_client.signing_key,
+            timeout,
+            Some(&self.subscription),
+        )
+        .await
+    }
+}