And the into_future changes continue (#669)

* ReplacePermission::into_future

* ReplaceUser::into_future

* StoredProcedure create and replace

Author: rylev

sdk/data_cosmos/examples/permission_00.rs

@@ -1,4 +1,3 @@
-use azure_core::prelude::*;
 use azure_data_cosmos::prelude::*;
 use std::error::Error;
 
@@ -110,19 +109,16 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .unwrap();
     println!("get_permission_response == {:#?}", get_permission_response);
 
-    let permission_mode = &get_permission_response.permission.permission_mode;
+    let permission_mode = get_permission_response.permission.permission_mode;
 
     // renew permission extending its validity for 60 seconds more.
     let replace_permission_response = permission_client
-        .replace_permission(
-            Context::new(),
-            ReplacePermissionOptions::new()
-                .expiry_seconds(600u64)
-                .consistency_level(ConsistencyLevel::Session(
-                    get_permission_response.session_token,
-                )),
-            permission_mode,
-        )
+        .replace_permission(permission_mode)
+        .expiry_seconds(600u64)
+        .consistency_level(ConsistencyLevel::Session(
+            get_permission_response.session_token,
+        ))
+        .into_future()
         .await
         .unwrap();
     println!(

sdk/data_cosmos/examples/stored_proc_01.rs

@@ -54,8 +54,8 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     );
 
     let create_stored_procedure_response = stored_procedure_client
-        .create_stored_procedure()
-        .execute(function_body)
+        .create_stored_procedure(function_body)
+        .into_future()
         .await?;
     println!(
         "create_stored_procedure_response == {:#?}",

sdk/data_cosmos/examples/user_00.rs

@@ -1,4 +1,3 @@
-use azure_core::Context;
 use azure_data_cosmos::prelude::*;
 use futures::StreamExt;
 use std::error::Error;
@@ -47,7 +46,8 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let new_user = format!("{}replaced", user_name);
 
     let replace_user_response = user_client
-        .replace_user(Context::new(), &new_user, ReplaceUserOptions::new())
+        .replace_user(new_user.clone())
+        .into_future()
         .await?;
     println!("replace_user_response == {:#?}", replace_user_response);
 

sdk/data_cosmos/src/clients/permission_client.rs

@@ -1,9 +1,8 @@
 use super::*;
 use crate::prelude::*;
-use crate::resources::permission::{PermissionMode, PermissionResponse};
-use crate::resources::ResourceType;
+use crate::resources::permission::PermissionMode;
 use crate::ReadonlyString;
-use azure_core::{Context, Pipeline, Request};
+use azure_core::{Pipeline, Request};
 
 /// A client for Cosmos permission resources.
 #[derive(Debug, Clone)]
@@ -49,22 +48,8 @@ impl PermissionClient {
     }
 
     /// Replace the permission
-    pub async fn replace_permission(
-        &self,
-        ctx: Context,
-        options: ReplacePermissionOptions,
-        permission_mode: &PermissionMode,
-    ) -> crate::Result<PermissionResponse> {
-        let mut request = self.prepare_request_with_permission_name(http::Method::PUT);
-
-        options.decorate_request(&mut request, self.permission_name(), permission_mode)?;
-
-        let response = self
-            .pipeline()
-            .send(ctx.clone().insert(ResourceType::Permissions), &mut request)
-            .await?;
-
-        Ok(PermissionResponse::try_from(response).await?)
+    pub fn replace_permission(&self, permission_mode: PermissionMode) -> ReplacePermissionBuilder {
+        ReplacePermissionBuilder::new(self.clone(), permission_mode)
     }
 
     /// Get the permission

sdk/data_cosmos/src/clients/stored_procedure_client.rs

@@ -1,7 +1,8 @@
 use super::*;
+use crate::prelude::*;
 use crate::resources::ResourceType;
 use crate::{requests, ReadonlyString};
-use azure_core::HttpClient;
+use azure_core::{HttpClient, Pipeline, Request};
 
 /// A client for Cosmos stored procedure resources.
 #[derive(Debug, Clone)]
@@ -42,13 +43,19 @@ impl StoredProcedureClient {
     }
 
     /// Create the stored procedure
-    pub fn create_stored_procedure(&self) -> requests::CreateStoredProcedureBuilder<'_, '_> {
-        requests::CreateStoredProcedureBuilder::new(self)
+    pub fn create_stored_procedure<S: Into<String>>(
+        &self,
+        function_body: S,
+    ) -> CreateStoredProcedureBuilder {
+        CreateStoredProcedureBuilder::new(self.clone(), function_body.into())
     }
 
     /// Replace the stored procedure
-    pub fn replace_stored_procedure(&self) -> requests::ReplaceStoredProcedureBuilder<'_, '_> {
-        requests::ReplaceStoredProcedureBuilder::new(self)
+    pub fn replace_stored_procedure<S: Into<String>>(
+        &self,
+        function_body: S,
+    ) -> ReplaceStoredProcedureBuilder {
+        ReplaceStoredProcedureBuilder::new(self.clone(), function_body.into())
     }
 
     /// Execute the stored procedure
@@ -61,34 +68,52 @@ impl StoredProcedureClient {
         requests::DeleteStoredProcedureBuilder::new(self)
     }
 
-    pub(crate) fn prepare_request(&self, method: http::Method) -> http::request::Builder {
+    pub(crate) fn prepare_request_with_stored_procedure_name(
+        &self,
+        method: http::Method,
+    ) -> http::request::Builder {
         self.cosmos_client().prepare_request(
             &format!(
-                "dbs/{}/colls/{}/sprocs",
+                "dbs/{}/colls/{}/sprocs/{}",
                 self.database_client().database_name(),
                 self.collection_client().collection_name(),
+                self.stored_procedure_name()
             ),
             method,
             ResourceType::StoredProcedures,
         )
     }
 
-    pub(crate) fn prepare_request_with_stored_procedure_name(
+    pub(crate) fn prepare_pipeline_with_stored_procedure_name(
         &self,
         method: http::Method,
-    ) -> http::request::Builder {
-        self.cosmos_client().prepare_request(
+    ) -> Request {
+        self.cosmos_client().prepare_request_pipeline(
             &format!(
                 "dbs/{}/colls/{}/sprocs/{}",
                 self.database_client().database_name(),
                 self.collection_client().collection_name(),
                 self.stored_procedure_name()
             ),
             method,
-            ResourceType::StoredProcedures,
         )
     }
 
+    pub(crate) fn prepare_request_pipeline(&self, method: http::Method) -> Request {
+        self.cosmos_client().prepare_request_pipeline(
+            &format!(
+                "dbs/{}/colls/{}/sprocs",
+                self.database_client().database_name(),
+                self.collection_client().collection_name(),
+            ),
+            method,
+        )
+    }
+
+    pub(crate) fn pipeline(&self) -> &Pipeline {
+        self.cosmos_client().pipeline()
+    }
+
     pub(crate) fn http_client(&self) -> &dyn HttpClient {
         self.cosmos_client().http_client()
     }

sdk/data_cosmos/src/clients/user_client.rs

@@ -1,9 +1,7 @@
 use super::*;
 use crate::prelude::*;
-use crate::resources::user::UserResponse;
-use crate::resources::ResourceType;
 use crate::{requests, ReadonlyString};
-use azure_core::{Context, HttpClient, Pipeline, Request};
+use azure_core::{HttpClient, Pipeline, Request};
 
 /// A client for Cosmos user resources.
 #[derive(Debug, Clone)]
@@ -49,21 +47,8 @@ impl UserClient {
     }
 
     /// Replace the user
-    pub async fn replace_user<S: AsRef<str>>(
-        &self,
-        ctx: Context,
-        user_name: S,
-        options: ReplaceUserOptions,
-    ) -> crate::Result<UserResponse> {
-        let mut request = self.prepare_request_with_user_name(http::Method::PUT);
-
-        options.decorate_request(&mut request, user_name.as_ref())?;
-        let response = self
-            .pipeline()
-            .send(ctx.clone().insert(ResourceType::Users), &mut request)
-            .await?;
-
-        Ok(UserResponse::try_from(response).await?)
+    pub fn replace_user<S: Into<String>>(&self, user_name: S) -> ReplaceUserBuilder {
+        ReplaceUserBuilder::new(self.clone(), user_name.into())
     }
 
     /// Delete the user

sdk/data_cosmos/src/operations/create_stored_procedure.rs

@@ -0,0 +1,111 @@
+use crate::headers::from_headers::*;
+use crate::prelude::*;
+use crate::resources::StoredProcedure;
+use crate::ResourceQuota;
+use azure_core::headers::{etag_from_headers, session_token_from_headers};
+use azure_core::{collect_pinned_stream, Context, Response as HttpResponse};
+use chrono::{DateTime, Utc};
+
+#[derive(Debug, Clone)]
+pub struct CreateStoredProcedureBuilder {
+    client: StoredProcedureClient,
+    function_body: String,
+    consistency_level: Option<ConsistencyLevel>,
+    context: Context,
+}
+
+impl CreateStoredProcedureBuilder {
+    pub(crate) fn new(client: StoredProcedureClient, body: String) -> Self {
+        Self {
+            client,
+            function_body: body,
+            consistency_level: None,
+            context: Context::new(),
+        }
+    }
+
+    setters! {
+        consistency_level: ConsistencyLevel => Some(consistency_level),
+        context: Context => context,
+    }
+
+    pub fn into_future(self) -> CreateStoredProcedure {
+        Box::pin(async move {
+            let mut req = self.client.prepare_request_pipeline(http::Method::POST);
+
+            azure_core::headers::add_optional_header2(&self.consistency_level, &mut req)?;
+
+            #[derive(Debug, Serialize)]
+            struct Request<'a> {
+                body: &'a str,
+                id: &'a str,
+            }
+            let body = Request {
+                body: &self.function_body,
+                id: self.client.stored_procedure_name(),
+            };
+
+            req.set_body(bytes::Bytes::from(serde_json::to_string(&body)?).into());
+
+            let response = self
+                .client
+                .pipeline()
+                .send(
+                    self.context.clone().insert(ResourceType::StoredProcedures),
+                    &mut req,
+                )
+                .await?;
+            CreateStoredProcedureResponse::try_from(response).await
+        })
+    }
+}
+
+#[cfg(feature = "into_future")]
+impl std::future::IntoFuture for CreateStoredProcedureBuilder {
+    type Future = CreateStoredProcedure;
+    type Output = <CreateStoredProcedure as std::future::Future>::Output;
+    fn into_future(self) -> Self::Future {
+        Self::into_future(self)
+    }
+}
+
+/// The future returned by calling `into_future` on the builder.
+pub type CreateStoredProcedure =
+    futures::future::BoxFuture<'static, crate::Result<CreateStoredProcedureResponse>>;
+
+/// A stored procedure response
+#[derive(Debug, Clone, PartialEq)]
+pub struct CreateStoredProcedureResponse {
+    pub stored_procedure: StoredProcedure,
+    pub charge: f64,
+    pub activity_id: uuid::Uuid,
+    pub etag: String,
+    pub session_token: String,
+    pub last_change: DateTime<Utc>,
+    pub resource_quota: Vec<ResourceQuota>,
+    pub resource_usage: Vec<ResourceQuota>,
+    pub quorum_acked_lsn: u64,
+    pub current_write_quorum: u64,
+    pub current_replica_set_size: u64,
+}
+
+impl CreateStoredProcedureResponse {
+    pub async fn try_from(response: HttpResponse) -> crate::Result<Self> {
+        let (_status_code, headers, pinned_stream) = response.deconstruct();
+        let body = collect_pinned_stream(pinned_stream).await?;
+
+        Ok(Self {
+            stored_procedure: serde_json::from_slice(&body)?,
+            charge: request_charge_from_headers(&headers)?,
+            activity_id: activity_id_from_headers(&headers)?,
+            etag: etag_from_headers(&headers)?,
+            session_token: session_token_from_headers(&headers)?,
+            last_change: last_state_change_from_headers(&headers)?,
+            resource_quota: resource_quota_from_headers(&headers)?,
+            resource_usage: resource_usage_from_headers(&headers)?,
+            quorum_acked_lsn: quorum_acked_lsn_from_headers(&headers)?,
+            current_write_quorum: current_write_quorum_from_headers(&headers)?,
+            current_replica_set_size: current_replica_set_size_from_headers(&headers)?,
+        })
+    }
+}

sdk/data_cosmos/src/operations/mod.rs

@@ -6,6 +6,7 @@ mod create_collection;
 mod create_database;
 mod create_document;
 mod create_permission;
+mod create_stored_procedure;
 mod create_user;
 mod delete_collection;
 mod delete_database;
@@ -23,12 +24,14 @@ mod list_users;
 mod replace_collection;
 mod replace_document;
 mod replace_permission;
+mod replace_stored_procedure;
 mod replace_user;
 
 pub use create_collection::*;
 pub use create_database::*;
 pub use create_document::*;
 pub use create_permission::*;
+pub use create_stored_procedure::*;
 pub use create_user::*;
 pub use delete_collection::*;
 pub use delete_database::*;
@@ -46,4 +49,5 @@ pub use list_users::*;
 pub use replace_collection::*;
 pub use replace_document::*;
 pub use replace_permission::*;
+pub use replace_stored_procedure::*;
 pub use replace_user::*;