into_future again (#670)

* ListDocuments::into_stream

* ListPermissions::into_stream

* DeleteStoredProcedure::into_future

* fmt

* ListAttachments::into_stream

Author: rylev

sdk/data_cosmos/examples/attachments_00.rs

@@ -1,4 +1,5 @@
 use azure_data_cosmos::prelude::*;
+use futures::StreamExt;
 use serde::{Deserialize, Serialize};
 use std::error::Error;
 
@@ -62,7 +63,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let document_client = client.into_document_client(doc.id.clone(), &doc.id)?;
 
     // list attachments
-    let ret = document_client.list_attachments().execute().await?;
+    let ret = document_client
+        .list_attachments()
+        .into_stream()
+        .next()
+        .await
+        .unwrap()?;
     println!("list attachments == {:#?}", ret);
 
     // reference attachment

sdk/data_cosmos/examples/database_00.rs

@@ -80,8 +80,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
             let documents = collection_client
                 .list_documents()
-                .execute::<Value>()
-                .await?;
+                .into_stream::<Value>()
+                .next()
+                .await
+                .unwrap()?;
             println!("\ndocuments as json == {:?}", documents);
         }
     }

sdk/data_cosmos/examples/document_00.rs

@@ -145,8 +145,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     println!("Listing documents...");
     let list_documents_response = collection_client
         .list_documents()
-        .execute::<MySampleStruct>()
-        .await?;
+        .into_stream::<MySampleStruct>()
+        .next()
+        .await
+        .unwrap()?;
     println!(
         "list_documents_response contains {} documents",
         list_documents_response.documents.len()

sdk/data_cosmos/examples/document_entries_00.rs

@@ -58,12 +58,13 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     println!("Created 5 documents.");
 
     // Let's get 3 entries at a time.
-    let response = client
+    let mut paged = client
         .list_documents()
         .consistency_level(response.unwrap())
         .max_item_count(3i32)
-        .execute::<MySampleStruct>()
-        .await?;
+        .into_stream::<MySampleStruct>();
+
+    let response = paged.next().await.unwrap()?;
 
     assert_eq!(response.documents.len(), 3);
     println!("response == {:#?}", response);
@@ -72,16 +73,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // continuation_token must be present
     assert!(response.continuation_token.is_some());
 
-    let session_token = &response;
-    let ct = response.continuation_token.clone().unwrap();
-    println!("ct == {}", ct);
-
-    let response = client
-        .list_documents()
-        .consistency_level(session_token)
-        .continuation(ct.as_str())
-        .execute::<MySampleStruct>()
-        .await?;
+    let response = paged.next().await.unwrap()?;
 
     assert_eq!(response.documents.len(), 2);
     println!("response == {:#?}", response);
@@ -101,7 +93,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
             .list_documents()
             .consistency_level(session_token.clone())
             .max_item_count(3);
-        let mut stream = Box::pin(stream.stream::<MySampleStruct>());
+        let mut stream = stream.into_stream::<MySampleStruct>();
         // TODO: As soon as the streaming functionality is completed
         // in Rust substitute this while let Some... into
         // for each (or whatever the Rust team picks).

sdk/data_cosmos/examples/document_entries_01.rs

@@ -1,4 +1,5 @@
 use azure_data_cosmos::prelude::*;
+use futures::StreamExt;
 use serde::{Deserialize, Serialize};
 use std::error::Error;
 
@@ -82,8 +83,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let list_documents_response = client
         .list_documents()
         .consistency_level(&get_document_response)
-        .execute::<serde_json::Value>()
-        .await?;
+        .into_stream::<serde_json::Value>()
+        .next()
+        .await
+        .unwrap()?;
     println!("list_documents_response == {:#?}", list_documents_response);
 
     let query_documents_response = client

sdk/data_cosmos/examples/permission_00.rs

@@ -1,4 +1,5 @@
 use azure_data_cosmos::prelude::*;
+use futures::StreamExt;
 use std::error::Error;
 
 #[tokio::main]
@@ -92,8 +93,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .consistency_level(ConsistencyLevel::Session(
             create_permission2_response.session_token,
         ))
-        .execute()
-        .await?;
+        .into_stream()
+        .next()
+        .await
+        .unwrap()?;
     println!(
         "list_permissions_response == {:#?}",
         list_permissions_response

sdk/data_cosmos/examples/readme.rs

@@ -99,7 +99,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
             .list_documents()
             .consistency_level(session_token.clone())
             .max_item_count(3);
-        let mut stream = Box::pin(stream.stream::<MySampleStruct>());
+        let mut stream = stream.into_stream::<MySampleStruct>();
         // TODO: As soon as the streaming functionality is stabilized
         // in Rust we can substitute this while let Some... into
         // for each (or whatever the Rust team picks).
@@ -158,8 +158,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let list_documents_response = collection_client
         .list_documents()
         .consistency_level(session_token)
-        .execute::<serde_json::Value>() // you can use this if you don't know/care about the return type!
-        .await?;
+        .into_stream::<serde_json::Value>() // you can use this if you don't know/care about the return type!
+        .next()
+        .await
+        .unwrap()?;
     assert_eq!(list_documents_response.documents.len(), 4);
 
     Ok(())

sdk/data_cosmos/examples/remove_all_documents.rs

@@ -36,7 +36,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let mut documents = Vec::new();
 
     let stream = client.list_documents();
-    let mut stream = Box::pin(stream.stream::<serde_json::Value>());
+    let mut stream = stream.into_stream::<serde_json::Value>();
     while let Some(res) = stream.next().await {
         for doc in res?.documents {
             documents.push(doc);

sdk/data_cosmos/examples/stored_proc_01.rs

@@ -79,7 +79,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     let delete_stored_procedure_response = stored_procedure_client
         .delete_stored_procedure()
-        .execute()
+        .into_future()
         .await?;
     println!(
         "delete_stored_procedure_response == {:#?}",

sdk/data_cosmos/examples/user_permission_token.rs

@@ -1,4 +1,5 @@
 use azure_data_cosmos::prelude::*;
+use futures::StreamExt;
 use std::error::Error;
 
 #[tokio::main]
@@ -42,9 +43,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // test list documents
     let list_documents_response = collection_client
         .list_documents()
-        .execute::<serde_json::Value>()
+        .into_stream::<serde_json::Value>()
+        .next()
         .await
-        .unwrap();
+        .unwrap()?;
     println!(
         "list_documents_response got {} document(s).",
         list_documents_response.documents.len()
@@ -86,9 +88,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .into_database_client(database_name.clone())
         .into_collection_client(collection_name.clone())
         .list_documents()
-        .execute::<serde_json::Value>()
+        .into_stream::<serde_json::Value>()
+        .next()
         .await
-        .unwrap();
+        .unwrap()?;
     println!(
         "second list_documents_response got {} document(s).",
         list_documents_response.documents.len()