From 268ed3c8a034bb038e329561bb5a1db9a9aaed79 Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Mon, 7 Oct 2024 16:25:14 +0000
Subject: [PATCH 01/10] initial pager refactor

---
 sdk/core/azure_core/src/lib.rs                |   4 +-
 .../src/clients/container_client.rs           |   4 +-
 .../src/options/query_databases_options.rs    |  37 +++++
 .../typespec_client_core/src/http/pageable.rs | 142 ++++++------------
 4 files changed, 87 insertions(+), 100 deletions(-)
 create mode 100644 sdk/cosmos/azure_data_cosmos/src/options/query_databases_options.rs

diff --git a/sdk/core/azure_core/src/lib.rs b/sdk/core/azure_core/src/lib.rs
index b0dd236578..719021f226 100644
--- a/sdk/core/azure_core/src/lib.rs
+++ b/sdk/core/azure_core/src/lib.rs
@@ -51,8 +51,8 @@ pub use typespec_client_core::xml;
 pub use typespec_client_core::{
     base64, date,
     http::{
-        headers::Header, new_http_client, AppendToUrlQuery, Body, Context, Continuable, HttpClient,
-        Method, Pageable, Request, RequestContent, StatusCode, Url,
+        headers::Header, new_http_client, AppendToUrlQuery, Body, Context, HttpClient, Method,
+        Pager, Request, RequestContent, StatusCode, Url,
     },
     json, parsing,
     sleep::{self, sleep},
diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
index e2af9f2848..ae93d068c2 100644
--- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
@@ -10,7 +10,7 @@ use crate::{
     ItemOptions, PartitionKey, Query, QueryPartitionStrategy,
 };
 
-use azure_core::{Context, Request, Response};
+use azure_core::{Context, Pager, Request, Response};
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use url::Url;
 
@@ -444,7 +444,7 @@ impl ContainerClientMethods for ContainerClient {
         #[allow(unused_variables)]
         // REASON: This is a documented public API so prefixing with '_' is undesirable.
         options: Option<QueryOptions>,
-    ) -> azure_core::Result<azure_core::Pageable<QueryResults<T>, azure_core::Error>> {
+    ) -> azure_core::Result<impl Pager<T>> {
         // Represents the raw response model from the server.
         // We'll use this to deserialize the response body and then convert it to a more user-friendly model.
         #[derive(Deserialize)]
diff --git a/sdk/cosmos/azure_data_cosmos/src/options/query_databases_options.rs b/sdk/cosmos/azure_data_cosmos/src/options/query_databases_options.rs
new file mode 100644
index 0000000000..c64b0cd5c2
--- /dev/null
+++ b/sdk/cosmos/azure_data_cosmos/src/options/query_databases_options.rs
@@ -0,0 +1,37 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+#[cfg(doc)]
+use crate::CosmosClientMethods;
+
+/// Options to be passed to [`CosmosClient::query_databases()`](crate::CosmosClient::query_databases()).
+#[derive(Clone, Debug, Default)]
+pub struct QueryDatabasesOptions {}
+
+impl QueryDatabasesOptions {
+    /// Creates a new [`QueryDatabasesOptionsBuilder`](QueryDatabasesOptionsBuilder) that can be used to construct a [`QueryDatabasesOptions`].
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// let options = azure_data_cosmos::QueryDatabasesOptions::builder().build();
+    /// ```
+    pub fn builder() -> QueryDatabasesOptionsBuilder {
+        QueryDatabasesOptionsBuilder::default()
+    }
+}
+
+/// Builder used to construct a [`QueryDatabasesOptions`].
+///
+/// Obtain a [`QueryDatabasesOptionsBuilder`] by calling [`QueryDatabasesOptions::builder()`]
+#[derive(Default)]
+pub struct QueryDatabasesOptionsBuilder(QueryDatabasesOptions);
+
+impl QueryDatabasesOptionsBuilder {
+    /// Builds a [`QueryDatabasesOptions`] from the builder.
+    ///
+    /// This does not consume the builder, and can be called multiple times.
+    pub fn build(&self) -> QueryDatabasesOptions {
+        self.0.clone()
+    }
+}
diff --git a/sdk/typespec/typespec_client_core/src/http/pageable.rs b/sdk/typespec/typespec_client_core/src/http/pageable.rs
index 398648d36d..5781171813 100644
--- a/sdk/typespec/typespec_client_core/src/http/pageable.rs
+++ b/sdk/typespec/typespec_client_core/src/http/pageable.rs
@@ -1,112 +1,62 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-use futures::{stream::unfold, Stream};
-
-/// Helper macro for unwrapping `Result`s into the right types that `futures::stream::unfold` expects.
-macro_rules! r#try {
-    ($expr:expr $(,)?) => {
-        match $expr {
-            ::std::result::Result::Ok(val) => val,
-            ::std::result::Result::Err(err) => {
-                return Some((Err(err.into()), State::Done));
-            }
-        }
-    };
-}
-
-/// Helper macro for declaring the `Pageable` and `Continuable` types which easily allows
-/// for conditionally compiling with a `Send` constraint or not.
-macro_rules! declare {
-    ($($extra:tt)*) => {
-        // The use of a module here is a hack to get around the fact that `pin_project`
-        // generates a method `project_ref` which is never used and generates a warning.
-        // The module allows us to declare that `dead_code` is allowed but only for
-        // the `Pageable` type.
-        mod pageable {
-            #![allow(dead_code)]
-
-            /// A pageable stream that yields items of type `T`
-            ///
-            /// Internally uses a specific continuation header to
-            /// make repeated requests to the service yielding a new page each time.
-            #[pin_project::pin_project]
-            // This is to suppress the unused `project_ref` warning
-            pub struct Pageable<T, E> {
-                #[pin]
-                pub(crate) stream: ::std::pin::Pin<Box<dyn ::futures::Stream<Item = ::std::result::Result<T, E>> $($extra)*>>,
-            }
-        }
-        pub use pageable::Pageable;
+use std::{future::Future, pin::Pin};
 
-        impl<T, E> Pageable<T, E>
-        where
-            T: Continuable,
-        {
-            pub fn new<F>(
-                make_request: impl Fn(Option<T::Continuation>) -> F + Clone $($extra)* + 'static,
-            ) -> Self
-            where
-                F: ::std::future::Future<Output = Result<T, E>> $($extra)* + 'static,
-            {
-                let stream = unfold(State::Init, move |state: State<T::Continuation>| {
-                    let make_request = make_request.clone();
-                    async move {
-                        let response = match state {
-                            State::Init => {
-                                let request = make_request(None);
-                                r#try!(request.await)
-                            }
-                            State::Continuation(token) => {
-                                let request = make_request(Some(token));
-                                r#try!(request.await)
-                            }
-                            State::Done => {
-                                return None;
-                            }
-                        };
+use futures::{stream::unfold, Stream};
+use typespec::Error;
 
-                        let next_state = response
-                            .continuation()
-                            .map_or(State::Done, State::Continuation);
+use crate::http::Response;
 
-                        Some((Ok(response), next_state))
-                    }
-                });
-                Self {
-                    stream: Box::pin(stream),
-                }
-            }
-        }
+#[pin_project::pin_project]
+pub struct Pager<T> {
+    #[pin]
+    #[cfg(not(target_arch = "wasm32"))]
+    stream: Pin<Box<dyn Stream<Item = Result<Response<T>, Error>> + Send>>,
 
-        /// A type that can yield an optional continuation token
-        pub trait Continuable {
-            type Continuation: 'static $($extra)*;
-            fn continuation(&self) -> Option<Self::Continuation>;
-        }
-    };
+    #[pin]
+    #[cfg(target_arch = "wasm32")]
+    stream: Pin<Box<dyn Stream<Item = Result<Response<T>, Error>>>>,
 }
 
-#[cfg(not(target_arch = "wasm32"))]
-declare!(+ Send);
-#[cfg(target_arch = "wasm32")]
-declare!();
-
-impl<T, E> Stream for Pageable<T, E> {
-    type Item = Result<T, E>;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Option<Self::Item>> {
-        let this = self.project();
-        this.stream.poll_next(cx)
+impl<T> Pager<T> {
+    pub fn from_fn<
+        // This is a bit gnarly, but the only thing that differs between the WASM/non-WASM configs is the presence of Send bounds.
+        #[cfg(not(target_arch = "wasm32"))] C: Send + 'static,
+        #[cfg(target_arch = "wasm32")] C: 'static,
+        E: Into<typespec::Error>,
+        #[cfg(not(target_arch = "wasm32"))] F: Fn(Option<C>) -> Fut + Send + 'static,
+        #[cfg(target_arch = "wasm32")] F: Fn(Option<C>) -> Fut + 'static,
+        #[cfg(not(target_arch = "wasm32"))] Fut: Future<Output = Result<(Response<T>, Option<C>), E>> + Send + 'static,
+        #[cfg(target_arch = "wasm32")] Fut: Future<Output = Result<(Response<T>, Option<C>), E>> + 'static,
+    >(
+        make_request: F,
+    ) -> Self {
+        let stream = unfold(
+            (State::Init, make_request),
+            |(state, make_request)| async move {
+                let result = match state {
+                    State::Init => make_request(None).await,
+                    State::Continuation(c) => make_request(Some(c)).await,
+                    State::Done => return None,
+                };
+                let (response, continuation) = match result {
+                    Err(e) => return Some((Err(e.into()), (State::Done, make_request))),
+                    Ok(r) => r,
+                };
+                let next_state = continuation.map_or(State::Done, State::Continuation);
+                Some((Ok(response), (next_state, make_request)))
+            },
+        );
+        Self {
+            stream: Box::pin(stream),
+        }
     }
 }
 
-impl<T, O> std::fmt::Debug for Pageable<T, O> {
+impl<T> std::fmt::Debug for Pager<T> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Pageable").finish_non_exhaustive()
+        f.debug_struct("Pager").finish_non_exhaustive()
     }
 }
 

From d2a9158146c61b8e4e0184be8975fcf56bdd640b Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Mon, 7 Oct 2024 17:07:53 +0000
Subject: [PATCH 02/10] update ContainerClient::query_items for pager refactor

---
 .../examples/cosmos/query.rs                  |  4 +-
 .../src/clients/container_client.rs           | 73 +++++--------------
 sdk/cosmos/azure_data_cosmos/src/constants.rs |  2 -
 .../azure_data_cosmos/src/models/mod.rs       | 20 +++--
 .../typespec_client_core/src/http/mod.rs      |  4 +-
 .../src/http/{pageable.rs => pager.rs}        | 18 ++++-
 6 files changed, 46 insertions(+), 75 deletions(-)
 rename sdk/typespec/typespec_client_core/src/http/{pageable.rs => pager.rs} (81%)

diff --git a/sdk/cosmos/azure_data_cosmos/examples/cosmos/query.rs b/sdk/cosmos/azure_data_cosmos/examples/cosmos/query.rs
index c084392624..43383d1a76 100644
--- a/sdk/cosmos/azure_data_cosmos/examples/cosmos/query.rs
+++ b/sdk/cosmos/azure_data_cosmos/examples/cosmos/query.rs
@@ -34,10 +34,8 @@ impl QueryCommand {
             container_client.query_items::<serde_json::Value>(&self.query, pk, None)?;
 
         while let Some(page) = items_pager.next().await {
-            let response = page?;
+            let response = page?.deserialize_body().await?;
             println!("Results Page");
-            println!("  Query Metrics: {:?}", response.query_metrics);
-            println!("  Index Metrics: {:?}", response.index_metrics);
             println!("  Items:");
             for item in response.items {
                 println!("    * {:#?}", item);
diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
index ae93d068c2..b9f43bd2de 100644
--- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
@@ -11,7 +11,7 @@ use crate::{
 };
 
 use azure_core::{Context, Pager, Request, Response};
-use serde::{de::DeserializeOwned, Deserialize, Serialize};
+use serde::{de::DeserializeOwned, Serialize};
 use url::Url;
 
 #[cfg(doc)]
@@ -44,7 +44,7 @@ pub trait ContainerClientMethods {
     async fn read(
         &self,
         options: Option<ReadContainerOptions>,
-    ) -> azure_core::Result<azure_core::Response<ContainerProperties>>;
+    ) -> azure_core::Result<Response<ContainerProperties>>;
 
     /// Creates a new item in the container.
     ///
@@ -87,7 +87,7 @@ pub trait ContainerClientMethods {
         partition_key: impl Into<PartitionKey>,
         item: T,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response<Item<T>>>;
+    ) -> azure_core::Result<Response<Item<T>>>;
 
     /// Replaces an existing item in the container.
     ///
@@ -132,7 +132,7 @@ pub trait ContainerClientMethods {
         item_id: impl AsRef<str>,
         item: T,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response<Item<T>>>;
+    ) -> azure_core::Result<Response<Item<T>>>;
 
     /// Creates or replaces an item in the container.
     ///
@@ -178,7 +178,7 @@ pub trait ContainerClientMethods {
         partition_key: impl Into<PartitionKey>,
         item: T,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response<Item<T>>>;
+    ) -> azure_core::Result<Response<Item<T>>>;
 
     /// Reads a specific item from the container.
     ///
@@ -216,7 +216,7 @@ pub trait ContainerClientMethods {
         partition_key: impl Into<PartitionKey>,
         item_id: impl AsRef<str>,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response<Item<T>>>;
+    ) -> azure_core::Result<Response<Item<T>>>;
 
     /// Deletes an item from the container.
     ///
@@ -243,7 +243,7 @@ pub trait ContainerClientMethods {
         partition_key: impl Into<PartitionKey>,
         item_id: impl AsRef<str>,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response>;
+    ) -> azure_core::Result<Response>;
 
     /// Executes a single-partition query against items in the container.
     ///
@@ -304,7 +304,7 @@ pub trait ContainerClientMethods {
         query: impl Into<Query>,
         partition_key: impl Into<QueryPartitionStrategy>,
         options: Option<QueryOptions>,
-    ) -> azure_core::Result<azure_core::Pageable<QueryResults<T>, azure_core::Error>>;
+    ) -> azure_core::Result<Pager<QueryResults<T>>>;
 }
 
 /// A client for working with a specific container in a Cosmos DB account.
@@ -333,7 +333,7 @@ impl ContainerClientMethods for ContainerClient {
         #[allow(unused_variables)]
         // REASON: This is a documented public API so prefixing with '_' is undesirable.
         options: Option<ReadContainerOptions>,
-    ) -> azure_core::Result<azure_core::Response<ContainerProperties>> {
+    ) -> azure_core::Result<Response<ContainerProperties>> {
         let mut req = Request::new(self.container_url.clone(), azure_core::Method::Get);
         self.pipeline
             .send(Context::new(), &mut req, ResourceType::Containers)
@@ -348,7 +348,7 @@ impl ContainerClientMethods for ContainerClient {
         #[allow(unused_variables)]
         // REASON: This is a documented public API so prefixing with '_' is undesirable.
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response<Item<T>>> {
+    ) -> azure_core::Result<Response<Item<T>>> {
         let url = self.container_url.with_path_segments(["docs"]);
         let mut req = Request::new(url, azure_core::Method::Post);
         req.insert_headers(&partition_key.into())?;
@@ -367,7 +367,7 @@ impl ContainerClientMethods for ContainerClient {
         #[allow(unused_variables)]
         // REASON: This is a documented public API so prefixing with '_' is undesirable.
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response<Item<T>>> {
+    ) -> azure_core::Result<Response<Item<T>>> {
         let url = self
             .container_url
             .with_path_segments(["docs", item_id.as_ref()]);
@@ -387,7 +387,7 @@ impl ContainerClientMethods for ContainerClient {
         #[allow(unused_variables)]
         // REASON: This is a documented public API so prefixing with '_' is undesirable.
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response<Item<T>>> {
+    ) -> azure_core::Result<Response<Item<T>>> {
         let url = self.container_url.with_path_segments(["docs"]);
         let mut req = Request::new(url, azure_core::Method::Post);
         req.insert_header(constants::IS_UPSERT, "true");
@@ -406,7 +406,7 @@ impl ContainerClientMethods for ContainerClient {
         #[allow(unused_variables)]
         // REASON: This is a documented public API so prefixing with '_' is undesirable.
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response<Item<T>>> {
+    ) -> azure_core::Result<Response<Item<T>>> {
         let url = self
             .container_url
             .with_path_segments(["docs", item_id.as_ref()]);
@@ -425,7 +425,7 @@ impl ContainerClientMethods for ContainerClient {
         #[allow(unused_variables)]
         // REASON: This is a documented public API so prefixing with '_' is undesirable.
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<azure_core::Response> {
+    ) -> azure_core::Result<Response> {
         let url = self
             .container_url
             .with_path_segments(["docs", item_id.as_ref()]);
@@ -444,26 +444,9 @@ impl ContainerClientMethods for ContainerClient {
         #[allow(unused_variables)]
         // REASON: This is a documented public API so prefixing with '_' is undesirable.
         options: Option<QueryOptions>,
-    ) -> azure_core::Result<impl Pager<T>> {
-        // Represents the raw response model from the server.
-        // We'll use this to deserialize the response body and then convert it to a more user-friendly model.
-        #[derive(Deserialize)]
-        struct QueryResponseModel<M> {
-            #[serde(rename = "Documents")]
-            documents: Vec<M>,
-        }
-
-        // We have to manually implement Model, because the derive macro doesn't support auto-inferring type and lifetime bounds.
-        // See https://github.com/Azure/azure-sdk-for-rust/issues/1803
-        impl<M: DeserializeOwned> azure_core::Model for QueryResponseModel<M> {
-            async fn from_response_body(
-                body: azure_core::ResponseBody,
-            ) -> typespec_client_core::Result<Self> {
-                body.json().await
-            }
-        }
-
-        let url = self.container_url.with_path_segments(["docs"]);
+    ) -> azure_core::Result<Pager<QueryResults<T>>> {
+        let mut url = self.container_url.clone();
+        url.append_path_segments(["docs"]);
         let mut base_req = Request::new(url, azure_core::Method::Post);
 
         base_req.insert_header(constants::QUERY, "True");
@@ -477,7 +460,7 @@ impl ContainerClientMethods for ContainerClient {
         // We have to double-clone here.
         // First we clone the pipeline to pass it in to the closure
         let pipeline = self.pipeline.clone();
-        Ok(azure_core::Pageable::new(move |continuation| {
+        Ok(Pager::from_fn(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.
@@ -488,29 +471,13 @@ impl ContainerClientMethods for ContainerClient {
                     req.insert_header(constants::CONTINUATION, continuation);
                 }
 
-                let resp: Response<QueryResponseModel<T>> = pipeline
+                let resp = pipeline
                     .send(Context::new(), &mut req, ResourceType::Items)
                     .await?;
-
-                let query_metrics = resp
-                    .headers()
-                    .get_optional_string(&constants::QUERY_METRICS);
-                let index_metrics = resp
-                    .headers()
-                    .get_optional_string(&constants::INDEX_METRICS);
                 let continuation_token =
                     resp.headers().get_optional_string(&constants::CONTINUATION);
 
-                let query_response: QueryResponseModel<T> = resp.deserialize_body().await?;
-
-                let query_results = QueryResults {
-                    items: query_response.documents,
-                    query_metrics,
-                    index_metrics,
-                    continuation_token,
-                };
-
-                Ok(query_results)
+                Ok((resp, continuation_token))
             }
         }))
     }
diff --git a/sdk/cosmos/azure_data_cosmos/src/constants.rs b/sdk/cosmos/azure_data_cosmos/src/constants.rs
index e42f88926e..f578ffbe61 100644
--- a/sdk/cosmos/azure_data_cosmos/src/constants.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/constants.rs
@@ -9,8 +9,6 @@ use azure_core::{headers::HeaderName, request_options::ContentType};
 pub const QUERY: HeaderName = HeaderName::from_static("x-ms-documentdb-query");
 pub const PARTITION_KEY: HeaderName = HeaderName::from_static("x-ms-documentdb-partitionkey");
 pub const CONTINUATION: HeaderName = HeaderName::from_static("x-ms-continuation");
-pub const INDEX_METRICS: HeaderName = HeaderName::from_static("x-ms-cosmos-index-utilization");
-pub const QUERY_METRICS: HeaderName = HeaderName::from_static("x-ms-documentdb-query-metrics");
 pub const IS_UPSERT: HeaderName = HeaderName::from_static("x-ms-documentdb-is-upsert");
 
 pub const QUERY_CONTENT_TYPE: ContentType = ContentType::from_static("application/query+json");
diff --git a/sdk/cosmos/azure_data_cosmos/src/models/mod.rs b/sdk/cosmos/azure_data_cosmos/src/models/mod.rs
index 0c55691749..63e191e6fc 100644
--- a/sdk/cosmos/azure_data_cosmos/src/models/mod.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/models/mod.rs
@@ -3,8 +3,8 @@
 
 //! Model types sent to and received from the Cosmos DB API.
 
-use azure_core::{date::OffsetDateTime, Continuable, Model};
-use serde::{Deserialize, Deserializer};
+use azure_core::{date::OffsetDateTime, Model};
+use serde::{de::DeserializeOwned, Deserialize, Deserializer};
 
 #[cfg(doc)]
 use crate::{
@@ -37,19 +37,17 @@ where
 
 /// A page of query results, where each item is a document of type `T`.
 #[non_exhaustive]
-#[derive(Clone, Default, Debug)]
+#[derive(Clone, Default, Debug, Deserialize)]
 pub struct QueryResults<T> {
+    #[serde(rename = "Documents")]
     pub items: Vec<T>,
-    pub query_metrics: Option<String>,
-    pub index_metrics: Option<String>,
-    pub continuation_token: Option<String>,
 }
 
-impl<T> Continuable for QueryResults<T> {
-    type Continuation = String;
-
-    fn continuation(&self) -> Option<Self::Continuation> {
-        self.continuation_token.clone()
+impl<T: DeserializeOwned> azure_core::Model for QueryResults<T> {
+    async fn from_response_body(
+        body: azure_core::ResponseBody,
+    ) -> typespec_client_core::Result<Self> {
+        body.json().await
     }
 }
 
diff --git a/sdk/typespec/typespec_client_core/src/http/mod.rs b/sdk/typespec/typespec_client_core/src/http/mod.rs
index 2e906b06e4..cf8ea37c94 100644
--- a/sdk/typespec/typespec_client_core/src/http/mod.rs
+++ b/sdk/typespec/typespec_client_core/src/http/mod.rs
@@ -8,7 +8,7 @@ mod context;
 pub mod headers;
 mod models;
 mod options;
-mod pageable;
+mod pager;
 mod pipeline;
 pub mod policies;
 pub mod request;
@@ -19,7 +19,7 @@ pub use context::*;
 pub use headers::Header;
 pub use models::*;
 pub use options::*;
-pub use pageable::*;
+pub use pager::*;
 pub use pipeline::*;
 pub use request::{Body, Request, RequestContent};
 pub use response::{Model, Response};
diff --git a/sdk/typespec/typespec_client_core/src/http/pageable.rs b/sdk/typespec/typespec_client_core/src/http/pager.rs
similarity index 81%
rename from sdk/typespec/typespec_client_core/src/http/pageable.rs
rename to sdk/typespec/typespec_client_core/src/http/pager.rs
index 5781171813..5ee01bb332 100644
--- a/sdk/typespec/typespec_client_core/src/http/pageable.rs
+++ b/sdk/typespec/typespec_client_core/src/http/pager.rs
@@ -24,11 +24,10 @@ impl<T> Pager<T> {
         // This is a bit gnarly, but the only thing that differs between the WASM/non-WASM configs is the presence of Send bounds.
         #[cfg(not(target_arch = "wasm32"))] C: Send + 'static,
         #[cfg(target_arch = "wasm32")] C: 'static,
-        E: Into<typespec::Error>,
         #[cfg(not(target_arch = "wasm32"))] F: Fn(Option<C>) -> Fut + Send + 'static,
         #[cfg(target_arch = "wasm32")] F: Fn(Option<C>) -> Fut + 'static,
-        #[cfg(not(target_arch = "wasm32"))] Fut: Future<Output = Result<(Response<T>, Option<C>), E>> + Send + 'static,
-        #[cfg(target_arch = "wasm32")] Fut: Future<Output = Result<(Response<T>, Option<C>), E>> + 'static,
+        #[cfg(not(target_arch = "wasm32"))] Fut: Future<Output = Result<(Response<T>, Option<C>), typespec::Error>> + Send + 'static,
+        #[cfg(target_arch = "wasm32")] Fut: Future<Output = Result<(Response<T>, Option<C>), typespec::Error>> + 'static,
     >(
         make_request: F,
     ) -> Self {
@@ -41,7 +40,7 @@ impl<T> Pager<T> {
                     State::Done => return None,
                 };
                 let (response, continuation) = match result {
-                    Err(e) => return Some((Err(e.into()), (State::Done, make_request))),
+                    Err(e) => return Some((Err(e), (State::Done, make_request))),
                     Ok(r) => r,
                 };
                 let next_state = continuation.map_or(State::Done, State::Continuation);
@@ -54,6 +53,17 @@ impl<T> Pager<T> {
     }
 }
 
+impl<T> futures::Stream for Pager<T> {
+    type Item = Result<Response<T>, Error>;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        self.project().stream.poll_next(cx)
+    }
+}
+
 impl<T> std::fmt::Debug for Pager<T> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Pager").finish_non_exhaustive()

From 106aefe182682a51b47659c451576ed9c6456268 Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Mon, 7 Oct 2024 18:00:49 +0000
Subject: [PATCH 03/10] docs and renaming

---
 .../src/clients/container_client.rs           |   2 +-
 .../typespec_client_core/src/http/pager.rs    | 179 +++++++++++++++++-
 2 files changed, 179 insertions(+), 2 deletions(-)

diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
index b9f43bd2de..975db86219 100644
--- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
@@ -460,7 +460,7 @@ impl ContainerClientMethods for ContainerClient {
         // We have to double-clone here.
         // First we clone the pipeline to pass it in to the closure
         let pipeline = self.pipeline.clone();
-        Ok(Pager::from_fn(move |continuation| {
+        Ok(Pager::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.
diff --git a/sdk/typespec/typespec_client_core/src/http/pager.rs b/sdk/typespec/typespec_client_core/src/http/pager.rs
index 5ee01bb332..ea0e0e9fe4 100644
--- a/sdk/typespec/typespec_client_core/src/http/pager.rs
+++ b/sdk/typespec/typespec_client_core/src/http/pager.rs
@@ -8,6 +8,7 @@ use typespec::Error;
 
 use crate::http::Response;
 
+/// Represents a paginated result across multiple requests.
 #[pin_project::pin_project]
 pub struct Pager<T> {
     #[pin]
@@ -20,7 +21,43 @@ pub struct Pager<T> {
 }
 
 impl<T> Pager<T> {
-    pub fn from_fn<
+    /// Creates a [`Pager<T>`] from a callback that will be called repeatedly to request each page.
+    ///
+    /// This method expect a callback that accepts a single `Option<C>` parameter, and returns a `(Response<T>, Option<C>)` tuple, asynchronously.
+    /// The `C` type parameter is the type of the continuation. It may be any [`Send`]able type.
+    /// The result will be an asynchronous stream of [`Result<Response<T>>`](typespec::Result<Response<T>>) values.
+    ///
+    /// The first time your callback is called, it will be called with [`Option::None`], indicating no continuation value is present.
+    /// Your callback must return one of:
+    /// * `Ok((response, Some(continuation)))` - The request succeeded, and return a response `response` and a continuation value `continuation`. The response will be yielded to the stream and the callback will be called again immediately with `Some(continuation)`.
+    /// * `Ok((response, None))` - The request succeeded, and there are no more pages. The response will be yielded to the stream, the stream will end, and the callback will not be called again.
+    /// * `Err(..)` - The request failed. The error will be yielded to the stream, the stream will end, and the callback will not be called again.
+    ///
+    /// ## Examples
+    ///
+    /// ```rust,no_run
+    /// # use typespec_client_core::http::{Context, Pager, Pipeline, Request, Response, Method, headers::HeaderName};
+    /// # let pipeline: Pipeline = panic!("Not a runnable example");
+    /// # struct MyModel;
+    /// let url = "https://example.com/my_paginated_api".parse().unwrap();
+    /// let mut base_req = Request::new(url, Method::Get);
+    /// let pager = Pager::from_callback(move |continuation| {
+    ///     // The callback must be 'static, so you have to clone and move any values you want to use.
+    ///     let pipeline = pipeline.clone();
+    ///     let mut req = base_req.clone();
+    ///     async move {
+    ///         if let Some(continuation) = continuation {
+    ///             req.insert_header("x-continuation", continuation);
+    ///         }
+    ///         let resp: Response<MyModel> = pipeline
+    ///           .send(&Context::new(), &mut req)
+    ///           .await?;
+    ///         let continuation_token = resp.headers().get_optional_string(&HeaderName::from_static("x-next-continuation"));
+    ///         Ok((resp, continuation_token))
+    ///     }
+    /// });
+    /// ```
+    pub fn from_callback<
         // This is a bit gnarly, but the only thing that differs between the WASM/non-WASM configs is the presence of Send bounds.
         #[cfg(not(target_arch = "wasm32"))] C: Send + 'static,
         #[cfg(target_arch = "wasm32")] C: 'static,
@@ -76,3 +113,143 @@ enum State<T> {
     Continuation(T),
     Done,
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use futures::StreamExt;
+    use serde::Deserialize;
+
+    use crate::http::{
+        headers::{HeaderName, HeaderValue},
+        Model, Pager, Response, StatusCode,
+    };
+
+    #[tokio::test]
+    pub async fn standard_pagination() {
+        #[derive(Model, Deserialize, Debug, PartialEq, Eq)]
+        #[typespec(crate = "crate")]
+        struct Page {
+            pub page: usize,
+        }
+
+        let pager: Pager<Page> = Pager::from_callback(|continuation| async move {
+            match continuation {
+                None => Ok((
+                    Response::from_bytes(
+                        StatusCode::Ok,
+                        HashMap::from([(
+                            HeaderName::from_static("x-test-header"),
+                            HeaderValue::from_static("page-1"),
+                        )])
+                        .into(),
+                        r#"{"page":1}"#,
+                    ),
+                    Some("1"),
+                )),
+                Some("1") => Ok((
+                    Response::from_bytes(
+                        StatusCode::Ok,
+                        HashMap::from([(
+                            HeaderName::from_static("x-test-header"),
+                            HeaderValue::from_static("page-2"),
+                        )])
+                        .into(),
+                        r#"{"page":2}"#,
+                    ),
+                    Some("2"),
+                )),
+                Some("2") => Ok((
+                    Response::from_bytes(
+                        StatusCode::Ok,
+                        HashMap::from([(
+                            HeaderName::from_static("x-test-header"),
+                            HeaderValue::from_static("page-3"),
+                        )])
+                        .into(),
+                        r#"{"page":3}"#,
+                    ),
+                    None,
+                )),
+                _ => {
+                    panic!("Unexpected continuation value")
+                }
+            }
+        });
+        let pages: Vec<(String, Page)> = pager
+            .then(|r| async move {
+                let r = r.unwrap();
+                let header = r
+                    .headers()
+                    .get_optional_string(&HeaderName::from_static("x-test-header"))
+                    .unwrap();
+                let body = r.deserialize_body().await.unwrap();
+                (header, body)
+            })
+            .collect()
+            .await;
+        assert_eq!(
+            &[
+                ("page-1".to_string(), Page { page: 1 }),
+                ("page-2".to_string(), Page { page: 2 }),
+                ("page-3".to_string(), Page { page: 3 }),
+            ],
+            pages.as_slice()
+        )
+    }
+
+    #[tokio::test]
+    pub async fn error_stops_pagination() {
+        #[derive(Model, Deserialize, Debug, PartialEq, Eq)]
+        #[typespec(crate = "crate")]
+        struct Page {
+            pub page: usize,
+        }
+
+        let pager: Pager<Page> = Pager::from_callback(|continuation| async move {
+            match continuation {
+                None => Ok((
+                    Response::from_bytes(
+                        StatusCode::Ok,
+                        HashMap::from([(
+                            HeaderName::from_static("x-test-header"),
+                            HeaderValue::from_static("page-1"),
+                        )])
+                        .into(),
+                        r#"{"page":1}"#,
+                    ),
+                    Some("1"),
+                )),
+                Some("1") => Err(typespec::Error::message(
+                    typespec::error::ErrorKind::Other,
+                    "yon request didst fail",
+                )),
+                _ => {
+                    panic!("Unexpected continuation value")
+                }
+            }
+        });
+        let pages: Vec<Result<(String, Page), typespec::Error>> = pager
+            .then(|r| async move {
+                let r = r?;
+                let header = r
+                    .headers()
+                    .get_optional_string(&HeaderName::from_static("x-test-header"))
+                    .unwrap();
+                let body = r.deserialize_body().await?;
+                Ok((header, body))
+            })
+            .collect()
+            .await;
+        assert_eq!(2, pages.len());
+        assert_eq!(
+            &("page-1".to_string(), Page { page: 1 }),
+            pages[0].as_ref().unwrap()
+        );
+
+        let err = pages[1].as_ref().unwrap_err();
+        assert_eq!(&typespec::error::ErrorKind::Other, err.kind());
+        assert_eq!("yon request didst fail", format!("{}", err));
+    }
+}

From 658d87af8446823ef4afdc31a7e101931556f4e7 Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Mon, 7 Oct 2024 18:09:51 +0000
Subject: [PATCH 04/10] remove extra file

---
 .../src/options/query_databases_options.rs    | 37 -------------------
 1 file changed, 37 deletions(-)
 delete mode 100644 sdk/cosmos/azure_data_cosmos/src/options/query_databases_options.rs

diff --git a/sdk/cosmos/azure_data_cosmos/src/options/query_databases_options.rs b/sdk/cosmos/azure_data_cosmos/src/options/query_databases_options.rs
deleted file mode 100644
index c64b0cd5c2..0000000000
--- a/sdk/cosmos/azure_data_cosmos/src/options/query_databases_options.rs
+++ /dev/null
@@ -1,37 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-
-#[cfg(doc)]
-use crate::CosmosClientMethods;
-
-/// Options to be passed to [`CosmosClient::query_databases()`](crate::CosmosClient::query_databases()).
-#[derive(Clone, Debug, Default)]
-pub struct QueryDatabasesOptions {}
-
-impl QueryDatabasesOptions {
-    /// Creates a new [`QueryDatabasesOptionsBuilder`](QueryDatabasesOptionsBuilder) that can be used to construct a [`QueryDatabasesOptions`].
-    ///
-    /// # Examples
-    ///
-    /// ```rust
-    /// let options = azure_data_cosmos::QueryDatabasesOptions::builder().build();
-    /// ```
-    pub fn builder() -> QueryDatabasesOptionsBuilder {
-        QueryDatabasesOptionsBuilder::default()
-    }
-}
-
-/// Builder used to construct a [`QueryDatabasesOptions`].
-///
-/// Obtain a [`QueryDatabasesOptionsBuilder`] by calling [`QueryDatabasesOptions::builder()`]
-#[derive(Default)]
-pub struct QueryDatabasesOptionsBuilder(QueryDatabasesOptions);
-
-impl QueryDatabasesOptionsBuilder {
-    /// Builds a [`QueryDatabasesOptions`] from the builder.
-    ///
-    /// This does not consume the builder, and can be called multiple times.
-    pub fn build(&self) -> QueryDatabasesOptions {
-        self.0.clone()
-    }
-}

From 6c8241eea5bb66bb4e032b56c24f5c61a22a86db Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Mon, 7 Oct 2024 18:12:52 +0000
Subject: [PATCH 05/10] fix doc links

---
 .../azure_data_cosmos/src/clients/container_client.rs       | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
index 975db86219..ad9d89b66f 100644
--- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
@@ -50,7 +50,7 @@ pub trait ContainerClientMethods {
     ///
     /// # Arguments
     /// * `partition_key` - The partition key of the new item.
-    /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`]
+    /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`](serde::Deserialize)
     /// * `options` - Optional parameters for the request
     ///
     /// # Examples
@@ -94,7 +94,7 @@ pub trait ContainerClientMethods {
     /// # Arguments
     /// * `partition_key` - The partition key of the item to replace.
     /// * `item_id` - The id of the item to replace.
-    /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`]
+    /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`](serde::Deserialize)
     /// * `options` - Optional parameters for the request
     ///
     /// # Examples
@@ -141,7 +141,7 @@ pub trait ContainerClientMethods {
     ///
     /// # Arguments
     /// * `partition_key` - The partition key of the item to create or replace.
-    /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`]
+    /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`](serde::Deserialize)
     /// * `options` - Optional parameters for the request
     ///
     /// # Examples

From c4e325f468d76c1a00fc7c8a41521c445ce334ec Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Mon, 7 Oct 2024 18:42:55 +0000
Subject: [PATCH 06/10] import Model derive macro directly

---
 sdk/typespec/typespec_client_core/src/http/pager.rs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/sdk/typespec/typespec_client_core/src/http/pager.rs b/sdk/typespec/typespec_client_core/src/http/pager.rs
index ea0e0e9fe4..5224059ddc 100644
--- a/sdk/typespec/typespec_client_core/src/http/pager.rs
+++ b/sdk/typespec/typespec_client_core/src/http/pager.rs
@@ -120,10 +120,11 @@ mod tests {
 
     use futures::StreamExt;
     use serde::Deserialize;
+    use typespec_derive::Model;
 
     use crate::http::{
         headers::{HeaderName, HeaderValue},
-        Model, Pager, Response, StatusCode,
+        Pager, Response, StatusCode,
     };
 
     #[tokio::test]

From e8578c62fcbddc2ab3f04c87939803e5860eaae7 Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Tue, 8 Oct 2024 16:28:54 +0000
Subject: [PATCH 07/10] some pr feedback

---
 sdk/cosmos/azure_data_cosmos/src/constants.rs       | 2 ++
 sdk/cosmos/azure_data_cosmos/src/lib.rs             | 2 +-
 sdk/typespec/typespec_client_core/src/http/pager.rs | 4 ++--
 3 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/sdk/cosmos/azure_data_cosmos/src/constants.rs b/sdk/cosmos/azure_data_cosmos/src/constants.rs
index f578ffbe61..e42f88926e 100644
--- a/sdk/cosmos/azure_data_cosmos/src/constants.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/constants.rs
@@ -9,6 +9,8 @@ use azure_core::{headers::HeaderName, request_options::ContentType};
 pub const QUERY: HeaderName = HeaderName::from_static("x-ms-documentdb-query");
 pub const PARTITION_KEY: HeaderName = HeaderName::from_static("x-ms-documentdb-partitionkey");
 pub const CONTINUATION: HeaderName = HeaderName::from_static("x-ms-continuation");
+pub const INDEX_METRICS: HeaderName = HeaderName::from_static("x-ms-cosmos-index-utilization");
+pub const QUERY_METRICS: HeaderName = HeaderName::from_static("x-ms-documentdb-query-metrics");
 pub const IS_UPSERT: HeaderName = HeaderName::from_static("x-ms-documentdb-is-upsert");
 
 pub const QUERY_CONTENT_TYPE: ContentType = ContentType::from_static("application/query+json");
diff --git a/sdk/cosmos/azure_data_cosmos/src/lib.rs b/sdk/cosmos/azure_data_cosmos/src/lib.rs
index 50639d00fd..945847fabe 100644
--- a/sdk/cosmos/azure_data_cosmos/src/lib.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/lib.rs
@@ -11,7 +11,7 @@
 #![cfg_attr(docsrs, feature(doc_cfg_hide))]
 
 pub mod clients;
-pub(crate) mod constants;
+pub mod constants;
 mod options;
 mod partition_key;
 pub(crate) mod pipeline;
diff --git a/sdk/typespec/typespec_client_core/src/http/pager.rs b/sdk/typespec/typespec_client_core/src/http/pager.rs
index 5224059ddc..88f190317e 100644
--- a/sdk/typespec/typespec_client_core/src/http/pager.rs
+++ b/sdk/typespec/typespec_client_core/src/http/pager.rs
@@ -60,10 +60,10 @@ impl<T> Pager<T> {
     pub fn from_callback<
         // This is a bit gnarly, but the only thing that differs between the WASM/non-WASM configs is the presence of Send bounds.
         #[cfg(not(target_arch = "wasm32"))] C: Send + 'static,
-        #[cfg(target_arch = "wasm32")] C: 'static,
         #[cfg(not(target_arch = "wasm32"))] F: Fn(Option<C>) -> Fut + Send + 'static,
-        #[cfg(target_arch = "wasm32")] F: Fn(Option<C>) -> Fut + 'static,
         #[cfg(not(target_arch = "wasm32"))] Fut: Future<Output = Result<(Response<T>, Option<C>), typespec::Error>> + Send + 'static,
+        #[cfg(target_arch = "wasm32")] C: 'static,
+        #[cfg(target_arch = "wasm32")] F: Fn(Option<C>) -> Fut + 'static,
         #[cfg(target_arch = "wasm32")] Fut: Future<Output = Result<(Response<T>, Option<C>), typespec::Error>> + 'static,
     >(
         make_request: F,

From 8cfff3c2abc7276f6ea6c7a1274abdd755044420 Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Tue, 8 Oct 2024 18:59:32 +0000
Subject: [PATCH 08/10] add PagerResult

---
 .../src/clients/container_client.rs           | 11 +--
 .../typespec_client_core/src/http/pager.rs    | 84 +++++++++++++------
 2 files changed, 63 insertions(+), 32 deletions(-)

diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
index ad9d89b66f..c9f008e5bd 100644
--- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
@@ -12,6 +12,7 @@ use crate::{
 
 use azure_core::{Context, Pager, Request, Response};
 use serde::{de::DeserializeOwned, Serialize};
+use typespec_client_core::http::PagerResult;
 use url::Url;
 
 #[cfg(doc)]
@@ -471,13 +472,13 @@ impl ContainerClientMethods for ContainerClient {
                     req.insert_header(constants::CONTINUATION, continuation);
                 }
 
-                let resp = pipeline
+                let response = pipeline
                     .send(Context::new(), &mut req, ResourceType::Items)
                     .await?;
-                let continuation_token =
-                    resp.headers().get_optional_string(&constants::CONTINUATION);
-
-                Ok((resp, continuation_token))
+                Ok(PagerResult::from_response_header(
+                    response,
+                    &constants::CONTINUATION,
+                ))
             }
         }))
     }
diff --git a/sdk/typespec/typespec_client_core/src/http/pager.rs b/sdk/typespec/typespec_client_core/src/http/pager.rs
index 88f190317e..2df966c442 100644
--- a/sdk/typespec/typespec_client_core/src/http/pager.rs
+++ b/sdk/typespec/typespec_client_core/src/http/pager.rs
@@ -6,7 +6,33 @@ use std::{future::Future, pin::Pin};
 use futures::{stream::unfold, Stream};
 use typespec::Error;
 
-use crate::http::Response;
+use crate::http::{headers::HeaderName, Response};
+
+pub enum PagerResult<T, C> {
+    Continue {
+        response: Response<T>,
+        continuation: C,
+    },
+    Finish {
+        response: Response<T>,
+    },
+}
+
+impl<T> PagerResult<T, String> {
+    /// Creates a [`PagerResult<T, C>`] from the provided response, extracting the continuation value from the provided header.
+    ///
+    /// If the provided response has a header with the matching name, this returns [`PagerResult::Continue`], using the value from the header as the continuation.
+    /// If the provided response does not have a header with the matching name, this returns [`PagerResult::Finish`].
+    pub fn from_response_header(response: Response<T>, header_name: &HeaderName) -> Self {
+        match response.headers().get_optional_string(header_name) {
+            Some(continuation) => PagerResult::Continue {
+                response,
+                continuation,
+            },
+            None => PagerResult::Finish { response },
+        }
+    }
+}
 
 /// Represents a paginated result across multiple requests.
 #[pin_project::pin_project]
@@ -36,7 +62,7 @@ impl<T> Pager<T> {
     /// ## Examples
     ///
     /// ```rust,no_run
-    /// # use typespec_client_core::http::{Context, Pager, Pipeline, Request, Response, Method, headers::HeaderName};
+    /// # use typespec_client_core::http::{Context, Pager, PagerResult, Pipeline, Request, Response, Method, headers::HeaderName};
     /// # let pipeline: Pipeline = panic!("Not a runnable example");
     /// # struct MyModel;
     /// let url = "https://example.com/my_paginated_api".parse().unwrap();
@@ -52,8 +78,7 @@ impl<T> Pager<T> {
     ///         let resp: Response<MyModel> = pipeline
     ///           .send(&Context::new(), &mut req)
     ///           .await?;
-    ///         let continuation_token = resp.headers().get_optional_string(&HeaderName::from_static("x-next-continuation"));
-    ///         Ok((resp, continuation_token))
+    ///         Ok(PagerResult::from_response_header(resp, &HeaderName::from_static("x-next-continuation")))
     ///     }
     /// });
     /// ```
@@ -61,14 +86,15 @@ impl<T> Pager<T> {
         // This is a bit gnarly, but the only thing that differs between the WASM/non-WASM configs is the presence of Send bounds.
         #[cfg(not(target_arch = "wasm32"))] C: Send + 'static,
         #[cfg(not(target_arch = "wasm32"))] F: Fn(Option<C>) -> Fut + Send + 'static,
-        #[cfg(not(target_arch = "wasm32"))] Fut: Future<Output = Result<(Response<T>, Option<C>), typespec::Error>> + Send + 'static,
+        #[cfg(not(target_arch = "wasm32"))] Fut: Future<Output = Result<PagerResult<T, C>, typespec::Error>> + Send + 'static,
         #[cfg(target_arch = "wasm32")] C: 'static,
         #[cfg(target_arch = "wasm32")] F: Fn(Option<C>) -> Fut + 'static,
-        #[cfg(target_arch = "wasm32")] Fut: Future<Output = Result<(Response<T>, Option<C>), typespec::Error>> + 'static,
+        #[cfg(target_arch = "wasm32")] Fut: Future<Output = Result<PagerResult<T, C>, typespec::Error>> + 'static,
     >(
         make_request: F,
     ) -> Self {
         let stream = unfold(
+            // We flow the `make_request` callback through the state value so that we can avoid cloning.
             (State::Init, make_request),
             |(state, make_request)| async move {
                 let result = match state {
@@ -76,12 +102,17 @@ impl<T> Pager<T> {
                     State::Continuation(c) => make_request(Some(c)).await,
                     State::Done => return None,
                 };
-                let (response, continuation) = match result {
+                let (response, next_state) = match result {
                     Err(e) => return Some((Err(e), (State::Done, make_request))),
-                    Ok(r) => r,
+                    Ok(PagerResult::Continue {
+                        response,
+                        continuation,
+                    }) => (Ok(response), State::Continuation(continuation)),
+                    Ok(PagerResult::Finish { response }) => (Ok(response), State::Done),
                 };
-                let next_state = continuation.map_or(State::Done, State::Continuation);
-                Some((Ok(response), (next_state, make_request)))
+
+                // Flow 'make_request' through to avoid cloning
+                Some((response, (next_state, make_request)))
             },
         );
         Self {
@@ -124,7 +155,7 @@ mod tests {
 
     use crate::http::{
         headers::{HeaderName, HeaderValue},
-        Pager, Response, StatusCode,
+        Pager, PagerResult, Response, StatusCode,
     };
 
     #[tokio::test]
@@ -137,8 +168,8 @@ mod tests {
 
         let pager: Pager<Page> = Pager::from_callback(|continuation| async move {
             match continuation {
-                None => Ok((
-                    Response::from_bytes(
+                None => Ok(PagerResult::Continue {
+                    response: Response::from_bytes(
                         StatusCode::Ok,
                         HashMap::from([(
                             HeaderName::from_static("x-test-header"),
@@ -147,10 +178,10 @@ mod tests {
                         .into(),
                         r#"{"page":1}"#,
                     ),
-                    Some("1"),
-                )),
-                Some("1") => Ok((
-                    Response::from_bytes(
+                    continuation: "1",
+                }),
+                Some("1") => Ok(PagerResult::Continue {
+                    response: Response::from_bytes(
                         StatusCode::Ok,
                         HashMap::from([(
                             HeaderName::from_static("x-test-header"),
@@ -159,10 +190,10 @@ mod tests {
                         .into(),
                         r#"{"page":2}"#,
                     ),
-                    Some("2"),
-                )),
-                Some("2") => Ok((
-                    Response::from_bytes(
+                    continuation: "2",
+                }),
+                Some("2") => Ok(PagerResult::Finish {
+                    response: Response::from_bytes(
                         StatusCode::Ok,
                         HashMap::from([(
                             HeaderName::from_static("x-test-header"),
@@ -171,8 +202,7 @@ mod tests {
                         .into(),
                         r#"{"page":3}"#,
                     ),
-                    None,
-                )),
+                }),
                 _ => {
                     panic!("Unexpected continuation value")
                 }
@@ -210,8 +240,8 @@ mod tests {
 
         let pager: Pager<Page> = Pager::from_callback(|continuation| async move {
             match continuation {
-                None => Ok((
-                    Response::from_bytes(
+                None => Ok(PagerResult::Continue {
+                    response: Response::from_bytes(
                         StatusCode::Ok,
                         HashMap::from([(
                             HeaderName::from_static("x-test-header"),
@@ -220,8 +250,8 @@ mod tests {
                         .into(),
                         r#"{"page":1}"#,
                     ),
-                    Some("1"),
-                )),
+                    continuation: "1",
+                }),
                 Some("1") => Err(typespec::Error::message(
                     typespec::error::ErrorKind::Other,
                     "yon request didst fail",

From 7c7a6913e7b16848c045a5bc0344361c2410b177 Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Wed, 9 Oct 2024 17:15:53 +0000
Subject: [PATCH 09/10] rename PagerResult::Finish to PagerResult::Complete

---
 sdk/typespec/typespec_client_core/src/http/pager.rs | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/sdk/typespec/typespec_client_core/src/http/pager.rs b/sdk/typespec/typespec_client_core/src/http/pager.rs
index 2df966c442..7964a79252 100644
--- a/sdk/typespec/typespec_client_core/src/http/pager.rs
+++ b/sdk/typespec/typespec_client_core/src/http/pager.rs
@@ -13,7 +13,7 @@ pub enum PagerResult<T, C> {
         response: Response<T>,
         continuation: C,
     },
-    Finish {
+    Complete {
         response: Response<T>,
     },
 }
@@ -29,7 +29,7 @@ impl<T> PagerResult<T, String> {
                 response,
                 continuation,
             },
-            None => PagerResult::Finish { response },
+            None => PagerResult::Complete { response },
         }
     }
 }
@@ -108,7 +108,7 @@ impl<T> Pager<T> {
                         response,
                         continuation,
                     }) => (Ok(response), State::Continuation(continuation)),
-                    Ok(PagerResult::Finish { response }) => (Ok(response), State::Done),
+                    Ok(PagerResult::Complete { response }) => (Ok(response), State::Done),
                 };
 
                 // Flow 'make_request' through to avoid cloning
@@ -192,7 +192,7 @@ mod tests {
                     ),
                     continuation: "2",
                 }),
-                Some("2") => Ok(PagerResult::Finish {
+                Some("2") => Ok(PagerResult::Complete {
                     response: Response::from_bytes(
                         StatusCode::Ok,
                         HashMap::from([(

From 8ecd1712e65b6e19be9234acc8b0fcfe804c8056 Mon Sep 17 00:00:00 2001
From: Ashley Stanton-Nurse <github@analogrelay.net>
Date: Wed, 9 Oct 2024 19:53:53 +0000
Subject: [PATCH 10/10] fix docs

---
 sdk/typespec/typespec_client_core/src/http/pager.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sdk/typespec/typespec_client_core/src/http/pager.rs b/sdk/typespec/typespec_client_core/src/http/pager.rs
index 7964a79252..e21ff34467 100644
--- a/sdk/typespec/typespec_client_core/src/http/pager.rs
+++ b/sdk/typespec/typespec_client_core/src/http/pager.rs
@@ -22,7 +22,7 @@ impl<T> PagerResult<T, String> {
     /// Creates a [`PagerResult<T, C>`] from the provided response, extracting the continuation value from the provided header.
     ///
     /// If the provided response has a header with the matching name, this returns [`PagerResult::Continue`], using the value from the header as the continuation.
-    /// If the provided response does not have a header with the matching name, this returns [`PagerResult::Finish`].
+    /// If the provided response does not have a header with the matching name, this returns [`PagerResult::Complete`].
     pub fn from_response_header(response: Response<T>, header_name: &HeaderName) -> Self {
         match response.headers().get_optional_string(header_name) {
             Some(continuation) => PagerResult::Continue {