Remove Box::pin requirement in usage of Pageable (#668)

Author: yoshuawuyts

sdk/core/Cargo.toml

@@ -32,6 +32,7 @@ serde_json = "1.0"
 thiserror = "1.0"
 url = "2.2"
 uuid = { version = "0.8" }
+pin-project = "1.0.10"
 
 # Add dependency to getrandom to enable WASM support
 [target.'cfg(target_arch = "wasm32")'.dependencies]

sdk/core/src/pageable.rs

@@ -1,5 +1,6 @@
 use futures::stream::unfold;
 use futures::Stream;
+use pin_project::pin_project;
 
 macro_rules! r#try {
     ($expr:expr $(,)?) => {
@@ -16,7 +17,9 @@ macro_rules! r#try {
 ///
 /// Internally uses the Azure specific continuation header to
 /// make repeated requests to Azure yielding a new page each time.
+#[pin_project]
 pub struct Pageable<T, E> {
+    #[pin]
     stream: std::pin::Pin<Box<dyn Stream<Item = Result<T, E>>>>,
 }
 
@@ -54,10 +57,11 @@ impl<T, E> Stream for Pageable<T, E> {
     type Item = Result<T, E>;
 
     fn poll_next(
-        mut self: std::pin::Pin<&mut Self>,
+        self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Option<Self::Item>> {
-        std::pin::Pin::new(&mut self.stream).poll_next(cx)
+        let this = self.project();
+        this.stream.poll_next(cx)
     }
 }
 

sdk/data_cosmos/examples/collection.rs

@@ -31,7 +31,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     );
 
     // The Cosmos' client exposes a lot of methods. This one lists the databases in the specified account.
-    let databases = Box::pin(client.list_databases().into_stream())
+    let databases = client
+        .list_databases()
+        .into_stream()
         .next()
         .await
         .unwrap()?;
@@ -59,7 +61,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     for db in databases.databases {
         let database_client = client.clone().into_database_client(db.id.clone());
-        let collections = Box::pin(database_client.list_collections().into_stream())
+        let collections = database_client
+            .list_collections()
+            .into_stream()
             .next()
             .await
             .unwrap()?;

sdk/data_cosmos/examples/create_delete_database.rs

@@ -33,7 +33,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // account. Database do not implement Display but deref to &str so you can pass it to methods
     // both as struct or id.
 
-    let mut list_databases_stream = Box::pin(client.list_databases().into_stream());
+    let mut list_databases_stream = client.list_databases().into_stream();
     while let Some(list_databases_response) = list_databases_stream.next().await {
         println!("list_databases_response = {:#?}", list_databases_response?);
     }
@@ -60,8 +60,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         let get_collection_response = db_collection.get_collection().into_future().await?;
         println!("get_collection_response == {:#?}", get_collection_response);
 
-        let stream = db_client.list_collections().into_stream();
-        let mut stream = Box::pin(stream);
+        let mut stream = db_client.list_collections().into_stream();
         while let Some(res) = stream.next().await {
             let res = res?;
             println!("res == {:#?}", res);

sdk/data_cosmos/examples/database_00.rs

@@ -15,7 +15,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     let client = CosmosClient::new(account, authorization_token, CosmosOptions::default());
 
-    let dbs = Box::pin(client.list_databases().into_stream())
+    let dbs = client
+        .list_databases()
+        .into_stream()
         .next()
         .await
         .unwrap()?;
@@ -24,7 +26,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         println!("database == {:?}", db);
         let database = client.clone().into_database_client(db.name().to_owned());
 
-        let collections = Box::pin(database.list_collections().into_stream())
+        let collections = database
+            .list_collections()
+            .into_stream()
             .next()
             .await
             .unwrap()?;

sdk/data_cosmos/examples/database_01.rs

@@ -17,7 +17,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let database_client = client.into_database_client("pollo");
     println!("database_name == {}", database_client.database_name());
 
-    let collections = Box::pin(database_client.list_collections().into_stream())
+    let collections = database_client
+        .list_collections()
+        .into_stream()
         .next()
         .await
         .unwrap()?;

sdk/data_cosmos/examples/document_00.rs

@@ -56,7 +56,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // an error (for example, the given key is not valid) you will receive a
     // specific azure_data_cosmos::Error. In this example we will look for a specific database
     // so we chain a filter operation.
-    let db = Box::pin(client.list_databases().into_stream())
+    let db = client
+        .list_databases()
+        .into_stream()
         .next()
         .await
         .unwrap()?
@@ -81,16 +83,14 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // we will create it. The collection creation is more complex and
     // has many options (such as indexing and so on).
     let collection = {
-        let collections = Box::pin(
-            client
-                .clone()
-                .into_database_client(database.id.clone())
-                .list_collections()
-                .into_stream(),
-        )
-        .next()
-        .await
-        .unwrap()?;
+        let collections = client
+            .clone()
+            .into_database_client(database.id.clone())
+            .list_collections()
+            .into_stream()
+            .next()
+            .await
+            .unwrap()?;
 
         if let Some(collection) = collections
             .collections

sdk/data_cosmos/examples/user_00.rs

@@ -32,7 +32,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let create_user_response = user_client.create_user().into_future().await?;
     println!("create_user_response == {:#?}", create_user_response);
 
-    let users = Box::pin(database_client.list_users().into_stream())
+    let users = database_client
+        .list_users()
+        .into_stream()
         .next()
         .await
         .unwrap()?;