Migrate iothub to pipelines (#972)

* migrate iothub to pipelines

* add token authorization and harmonize constructir names

* pr comments

Author: roeap

sdk/core/src/policies/mod.rs

@@ -1,11 +1,13 @@
 mod custom_headers_policy;
 mod retry_policies;
 mod telemetry_policy;
+mod timeout_policy;
 mod transport;
 
 pub use custom_headers_policy::{CustomHeaders, CustomHeadersPolicy};
 pub use retry_policies::*;
 pub use telemetry_policy::*;
+pub use timeout_policy::*;
 pub use transport::*;
 
 use crate::{Context, Request, Response};

sdk/storage/src/core/timeout_policy.rs renamed to sdk/core/src/policies/timeout_policy.rs

@@ -1,7 +1,7 @@
+use crate::request_options::Timeout;
+use crate::{AppendToUrlQuery, Context, Policy, PolicyResult, Request};
 use std::sync::Arc;
 
-use azure_core::{prelude::*, Context, Policy, PolicyResult, Request};
-
 #[derive(Debug, Clone, Default)]
 pub struct TimeoutPolicy {
     default_timeout: Option<Timeout>,

sdk/iot_hub/Cargo.toml

@@ -7,6 +7,7 @@ description = "Azure IoT Hub"
 license = "MIT"
 
 [dependencies]
+async-trait = "0.1"
 azure_core = { path = "../core", version = "0.3", default_features = false }
 base64 = "0.13"
 bytes = "1.0"
@@ -28,4 +29,4 @@ reqwest = "0.11.0"
 tokio = { version = "1.0", features = ["macros"] }
 
 [features]
-default = ["azure_core/default"]
+default = ["azure_core/default"]

sdk/iot_hub/examples/apply_configuration_on_edge_device.rs

@@ -56,9 +56,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         }
     });
 
-    let http_client = azure_core::new_http_client();
-    let service_client =
-        ServiceClient::from_connection_string(http_client, iot_hub_connection_string, 3600)?;
+    let service_client = ServiceClient::new_connection_string(iot_hub_connection_string, 3600)?;
     service_client
         .apply_on_edge_device(device_id)
         .modules_content(modules_content)

sdk/iot_hub/examples/configuration.rs

@@ -1,4 +1,4 @@
-use azure_iot_hub::service::{resources::Configuration, ServiceClient};
+use azure_iot_hub::service::ServiceClient;
 use std::error::Error;
 
 #[tokio::main]
@@ -10,9 +10,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .nth(1)
         .expect("Please pass the configuration id as the first parameter");
 
-    let http_client = azure_core::new_http_client();
-    let service_client =
-        ServiceClient::from_connection_string(http_client, iot_hub_connection_string, 3600)?;
+    let service_client = ServiceClient::new_connection_string(iot_hub_connection_string, 3600)?;
 
     println!("Creating a new configuration with id: {}", configuration_id);
 
@@ -42,7 +40,6 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .get_configuration(configuration_id)
         .into_future()
         .await?;
-    let configuration: Configuration = configuration.try_into()?;
 
     println!(
         "Successfully retrieved the new configuration '{:?}'",

sdk/iot_hub/examples/device_identity.rs

@@ -13,9 +13,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .expect("Please pass the device id as the first parameter");
 
     println!("Getting device twin for device '{}'", device_id);
-    let http_client = azure_core::new_http_client();
-    let service_client =
-        ServiceClient::from_connection_string(http_client, iot_hub_connection_string, 3600)?;
+    let service_client = ServiceClient::new_connection_string(iot_hub_connection_string, 3600)?;
     let device = service_client
         .create_device_identity(
             &device_id,
@@ -27,6 +25,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         )
         .into_future()
         .await?;
+    let device: DeviceIdentityResponse = device.try_into()?;
 
     println!("Successfully created a new device '{}'", device.device_id);
 

sdk/iot_hub/examples/directmethod.rs

@@ -23,9 +23,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .nth(4)
         .expect("Please pass the payload as the fourth parameter");
 
-    let http_client = azure_core::new_http_client();
-    let service_client =
-        ServiceClient::from_connection_string(http_client, iot_hub_connection_string, 3600)?;
+    let service_client = ServiceClient::new_connection_string(iot_hub_connection_string, 3600)?;
     println!(
         "Sending direct method {} to {}:{} on: {}",
         method_name, device_id, module_id, service_client.iot_hub_name

sdk/iot_hub/examples/gettwin.rs

@@ -12,9 +12,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     println!("Getting device twin for device: {}", device_id);
 
-    let http_client = azure_core::new_http_client();
-    let service_client =
-        ServiceClient::from_connection_string(http_client, iot_hub_connection_string, 3600)?;
+    let service_client = ServiceClient::new_connection_string(iot_hub_connection_string, 3600)?;
     let twin = service_client
         .get_device_twin(device_id)
         .into_future()

sdk/iot_hub/examples/module_identity.rs

@@ -16,9 +16,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .nth(2)
         .expect("Please pass the module id as the second parameter");
 
-    let http_client = azure_core::new_http_client();
-    let service_client =
-        ServiceClient::from_connection_string(http_client, iot_hub_connection_string, 3600)?;
+    let service_client = ServiceClient::new_connection_string(iot_hub_connection_string, 3600)?;
     let module = service_client
         .create_module_identity(
             &device_id,

sdk/iot_hub/examples/query_iothub.rs

@@ -11,9 +11,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let query = "SELECT * FROM devices";
     println!("Invoking query '{}' on the IoT Hub", query);
 
-    let http_client = azure_core::new_http_client();
-    let service_client =
-        ServiceClient::from_connection_string(http_client, iot_hub_connection_string, 3600)?;
+    let service_client = ServiceClient::new_connection_string(iot_hub_connection_string, 3600)?;
 
     let response = service_client
         .query(query)

sdk/iot_hub/examples/updatetwin.rs

@@ -16,10 +16,8 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .expect("Please pass the payload as the second parameter");
 
     println!("Updating device twin for device: {}", device_id);
-    let http_client = azure_core::new_http_client();
 
-    let service_client =
-        ServiceClient::from_connection_string(http_client, iot_hub_connection_string, 3600)?;
+    let service_client = ServiceClient::new_connection_string(iot_hub_connection_string, 3600)?;
     let updated_twin = service_client
         .update_device_twin(device_id)
         .desired_properties(serde_json::from_str(&payload)?)

sdk/iot_hub/src/authorization_policy.rs

@@ -0,0 +1,60 @@
+use crate::service::IoTHubCredentials;
+use azure_core::error::{ErrorKind, ResultExt};
+use azure_core::{
+    headers::{self, *},
+    Context, Policy, PolicyResult, Request,
+};
+use std::sync::Arc;
+
+const IOTHUB_TOKEN_SCOPE: &str = "https://iothubs.azure.net";
+
+#[derive(Debug, Clone)]
+pub struct AuthorizationPolicy {
+    credentials: IoTHubCredentials,
+}
+
+impl AuthorizationPolicy {
+    pub(crate) fn new(credentials: IoTHubCredentials) -> Self {
+        Self { credentials }
+    }
+}
+
+#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
+impl Policy for AuthorizationPolicy {
+    async fn send(
+        &self,
+        ctx: &Context,
+        request: &mut Request,
+        next: &[Arc<dyn Policy>],
+    ) -> PolicyResult {
+        assert!(
+            !next.is_empty(),
+            "Authorization policies cannot be the last policy of a pipeline"
+        );
+        let request = match &self.credentials {
+            IoTHubCredentials::SASToken(sas_token) => {
+                request.insert_header(headers::AUTHORIZATION, sas_token);
+                request
+            }
+            IoTHubCredentials::BearerToken(token) => {
+                request.insert_header(AUTHORIZATION, format!("Bearer {}", token));
+                request
+            }
+            IoTHubCredentials::TokenCredential(token_credential) => {
+                let bearer_token = token_credential
+                    .get_token(IOTHUB_TOKEN_SCOPE)
+                    .await
+                    .context(ErrorKind::Credential, "failed to get bearer token")?;
+
+                request.insert_header(
+                    AUTHORIZATION,
+                    format!("Bearer {}", bearer_token.token.secret()),
+                );
+                request
+            }
+        };
+
+        next[0].send(ctx, request, &next[1..]).await
+    }
+}

sdk/iot_hub/src/lib.rs

@@ -2,5 +2,6 @@
 #![deny(missing_docs)]
 //! The IoT Hub crate contains a client that can be used to manage the IoT Hub.
 
+mod authorization_policy;
 /// The service module contains the IoT Hub Service Client that can be used to manage the IoT Hub.
 pub mod service;