Merge in changes to pipeline context

sdk/core/src/headers/mod.rs

@@ -83,8 +83,11 @@ pub fn add_mandatory_header<T: AddAsHeader>(item: &T, builder: Builder) -> Build
     item.add_as_header(builder)
 }
 
-pub fn add_mandatory_header2<T: AddAsHeader>(item: &T, request: &mut crate::Request) {
-    item.add_as_header2(request);
+pub fn add_mandatory_header2<T: AddAsHeader>(
+    item: &T,
+    request: &mut crate::Request,
+) -> Result<(), crate::errors::HTTPHeaderError> {
+    item.add_as_header2(request)
 }
 
 pub const SERVER: &str = "server";

sdk/cosmos/src/clients/collection_client.rs

@@ -1,9 +1,11 @@
 use super::{DatabaseClient, UserDefinedFunctionClient};
+use crate::authorization_policy::CosmosContext;
 use crate::clients::*;
 use crate::operations::*;
 use crate::requests;
 use crate::resources::ResourceType;
 use crate::ReadonlyString;
+use azure_core::PipelineContext;
 use azure_core::{pipeline::Pipeline, Context, HttpClient};
 use serde::Serialize;
 
@@ -67,13 +69,15 @@ impl CollectionClient {
         document: &D,
         options: CreateDocumentOptions<'_>,
     ) -> Result<CreateDocumentResponse, crate::Error> {
-        let request = self.prepare_request_with_collection_name(http::Method::POST);
-        let mut request = request.body(bytes::Bytes::new()).unwrap().into();
-        let mut ctx = ctx.clone();
+        let mut request = self
+            .cosmos_client()
+            .prepare_request_pipeline("collection", http::Method::POST);
+        let mut pipeline_context = PipelineContext::new(ctx, ResourceType::Collections.into());
+
         options.decorate_request(&mut request, document)?;
         let response = self
             .pipeline()
-            .send(&mut ctx, &mut request)
+            .send(&mut pipeline_context, &mut request)
             .await?
             .validate(http::StatusCode::CREATED)
             .await?;
@@ -155,7 +159,7 @@ impl CollectionClient {
         self.cosmos_client().http_client()
     }
 
-    pub(crate) fn pipeline(&self) -> &Pipeline {
+    pub(crate) fn pipeline(&self) -> &Pipeline<CosmosContext> {
         self.cosmos_client().pipeline()
     }
 }

sdk/cosmos/src/lib.rs

@@ -10,6 +10,7 @@ should also be possible with this crate.
 ```no_run
 // Using the prelude module of the Cosmos crate makes easier to use the Rust Azure SDK for Cosmos DB.
 use azure_cosmos::prelude::*;
+use azure_core::Context;
 use serde::{Deserialize, Serialize};
 use std::error::Error;
 
@@ -75,9 +76,11 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
         // insert it
         collection_client
-            .create_document()
-            .is_upsert(true) // this option will overwrite a preexisting document (if any)
-            .execute(&document_to_insert)
+            .create_document(
+                Context::new(),
+                &document_to_insert,
+                CreateDocumentOptions::new().is_upsert(true),
+            )
             .await?;
     }
     // wow that was easy and fast, wasn't it? :)

sdk/cosmos/src/operations/create_document.rs

@@ -68,12 +68,12 @@ impl<'a> CreateDocumentOptions<'a> {
             .unwrap_or_else(|| serialize_partition_key(document).unwrap());
 
         add_as_partition_key_header_serialized2(&partition_key, req);
-        azure_core::headers::add_optional_header2(&self.if_match_condition, req);
-        azure_core::headers::add_optional_header2(&self.if_modified_since, req);
-        azure_core::headers::add_optional_header2(&self.consistency_level, req);
-        azure_core::headers::add_mandatory_header2(&self.is_upsert, req);
-        azure_core::headers::add_mandatory_header2(&self.indexing_directive, req);
-        azure_core::headers::add_mandatory_header2(&self.allow_tentative_writes, req);
+        azure_core::headers::add_optional_header2(&self.if_match_condition, req)?;
+        azure_core::headers::add_optional_header2(&self.if_modified_since, req)?;
+        azure_core::headers::add_optional_header2(&self.consistency_level, req)?;
+        azure_core::headers::add_mandatory_header2(&self.is_upsert, req)?;
+        azure_core::headers::add_mandatory_header2(&self.indexing_directive, req)?;
+        azure_core::headers::add_mandatory_header2(&self.allow_tentative_writes, req)?;
 
         req.set_body(bytes::Bytes::from(serialized).into());
         Ok(())