Use a struct to keep track of the Stream state

Integrates some changes from Azure#3279 to make merging that PR a little later easier.

Author: heaths

sdk/core/azure_core/src/http/pager.rs

@@ -6,7 +6,7 @@ mod signature_target;
 
 pub use authorization_policy::AuthorizationPolicy;
 use azure_core::http::{
-    pager::PagerState,
+    pager::{PagerOptions, PagerState},
     request::{options::ContentType, Request},
     response::Response,
     ClientOptions, Context, Method, RawResponse, RetryOptions,
@@ -115,25 +115,30 @@ impl CosmosPipeline {
         // We have to double-clone here.
         // First we clone the pipeline to pass it in to the closure
         let pipeline = self.pipeline.clone();
-        let ctx = ctx.with_value(resource_link).into_owned();
-        Ok(FeedPager::from_callback(move |continuation| {
-            // Then we have to clone it again to pass it in to the async block.
-            // This is because Pageable can't borrow any data, it has to own it all.
-            // That's probably good, because it means a Pageable can outlive the client that produced it, but it requires some extra cloning.
-            let pipeline = pipeline.clone();
-            let mut req = base_request.clone();
-            let ctx = ctx.clone();
-            async move {
-                if let PagerState::More(continuation) = continuation {
-                    req.insert_header(constants::CONTINUATION, continuation);
+        let options = PagerOptions {
+            context: ctx.with_value(resource_link).into_owned(),
+        };
+        Ok(FeedPager::from_callback(
+            move |continuation, ctx| {
+                // Then we have to clone it again to pass it in to the async block.
+                // This is because Pageable can't borrow any data, it has to own it all.
+                // That's probably good, because it means a Pageable can outlive the client that produced it, but it requires some extra cloning.
+                let pipeline = pipeline.clone();
+                let mut req = base_request.clone();
+                let ctx = ctx.clone();
+                async move {
+                    if let PagerState::More(continuation) = continuation {
+                        req.insert_header(constants::CONTINUATION, continuation);
+                    }
+
+                    let resp = pipeline.send(&ctx, &mut req, None).await?;
+                    let page = FeedPage::<T>::from_response(resp).await?;
+
+                    Ok(page.into())
                 }
-
-                let resp = pipeline.send(&ctx, &mut req, None).await?;
-                let page = FeedPage::<T>::from_response(resp).await?;
-
-                Ok(page.into())
-            }
-        }))
+            },
+            Some(options),
+        ))
     }
 
     /// Helper function to read a throughput offer given a resource ID.

sdk/cosmos/azure_data_cosmos/src/pipeline/mod.rs

@@ -7,6 +7,10 @@
 ### Breaking Changes
 
 - `CertificateClient::create_certificate()` now returns a `Poller<CertificateOperation>`.
+- `CertificateClientListCertificatePropertiesOptions::method_options` is now `PagerOptions`.
+- `CertificateClientListCertificatePropertiesVersionsOptions::method_options` is now `PagerOptions`.
+- `CertificateClientListDeletedCertificatePropertiesOptions::method_options` is now `PagerOptions`.
+- `CertificateClientListIssuerPropertiesOptions::method_options` is now `PagerOptions`.
 - Removed `CertificateClient::begin_create_certificate()`.
 - Removed `CertificateClient::resume_create_certificate()`.
 - Removed `wait()` function from `Poller<CertificateOperation>`.