From c494a47235506a848cd399460d8557301ce1a327 Mon Sep 17 00:00:00 2001 From: William Myers Date: Wed, 13 May 2020 15:01:25 -0600 Subject: [PATCH 01/11] Add experimental support for async generated code `IterableMethod::execute` & `ResumableUpload` cause compilation errors. --- .../gen_include/error.rs | 12 ----- google_rest_api_generator/gen_include/iter.rs | 4 +- .../gen_include/resumable_upload.rs | 15 +++--- google_rest_api_generator/src/cargo.rs | 23 ++++++--- google_rest_api_generator/src/lib.rs | 6 +-- .../src/method_builder.rs | 51 ++++++++++--------- .../src/resource_builder.rs | 2 +- 7 files changed, 58 insertions(+), 55 deletions(-) diff --git a/google_rest_api_generator/gen_include/error.rs b/google_rest_api_generator/gen_include/error.rs index 5c6052b..8800a5a 100644 --- a/google_rest_api_generator/gen_include/error.rs +++ b/google_rest_api_generator/gen_include/error.rs @@ -53,15 +53,3 @@ impl From<::reqwest::Error> for Error { } } } - -/// Check the response to see if the status code represents an error. If so -/// convert it into the Reqwest variant of Error. -fn error_from_response(response: ::reqwest::blocking::Response) -> Result<::reqwest::blocking::Response, Error> { - match response.error_for_status_ref() { - Err(reqwest_err) => { - let body = response.text().ok(); - Err(Error::Reqwest { reqwest_err, body }) - } - Ok(_) => Ok(response), - } -} diff --git a/google_rest_api_generator/gen_include/iter.rs b/google_rest_api_generator/gen_include/iter.rs index ffd9a88..7c70d97 100644 --- a/google_rest_api_generator/gen_include/iter.rs +++ b/google_rest_api_generator/gen_include/iter.rs @@ -1,7 +1,7 @@ pub mod iter { pub trait IterableMethod { fn set_page_token(&mut self, value: String); - fn execute(&mut self) -> Result + async fn execute(&mut self) -> Result where T: ::serde::de::DeserializeOwned; } @@ -107,4 +107,4 @@ pub mod iter { } } } -} \ No newline at end of file +} diff --git a/google_rest_api_generator/gen_include/resumable_upload.rs b/google_rest_api_generator/gen_include/resumable_upload.rs index 84b5c8b..07211ba 100644 --- a/google_rest_api_generator/gen_include/resumable_upload.rs +++ b/google_rest_api_generator/gen_include/resumable_upload.rs @@ -1,11 +1,11 @@ pub struct ResumableUpload { - reqwest: ::reqwest::blocking::Client, + reqwest: ::reqwest::Client, url: String, progress: Option, } impl ResumableUpload { - pub fn new(reqwest: ::reqwest::blocking::Client, url: String) -> Self { + pub fn new(reqwest: ::reqwest::Client, url: String) -> Self { ResumableUpload { reqwest, url, @@ -17,7 +17,8 @@ impl ResumableUpload { &self.url } - pub fn upload(&mut self, mut reader: R) -> Result<(), Box> + // FIXME: fix compile issues `ResumableUpload::upload` to be async + pub async fn upload(&mut self, mut reader: R) -> Result<(), Box> where R: ::std::io::Read + ::std::io::Seek + Send + 'static, { @@ -36,8 +37,8 @@ impl ResumableUpload { ::reqwest::header::CONTENT_RANGE, format!("bytes */{}", reader_len), ); - let resp = req.send()?.error_for_status()?; - match resp.headers().get(::reqwest::header::RANGE) { + let response = req.send().await?.error_for_status()?; + match response.headers().get(::reqwest::header::RANGE) { Some(range_header) => { let (_, progress) = parse_range_header(range_header) .map_err(|e| format!("invalid RANGE header: {}", e))?; @@ -53,8 +54,8 @@ impl ResumableUpload { let content_range = format!("bytes {}-{}/{}", progress, reader_len - 1, reader_len); let req = self.reqwest.request(::reqwest::Method::PUT, &self.url); let req = req.header(::reqwest::header::CONTENT_RANGE, content_range); - let req = req.body(::reqwest::blocking::Body::sized(reader, content_length)); - req.send()?.error_for_status()?; + let req = req.body(::reqwest::Body::sized(reader, content_length)); + req.send().await?.error_for_status()?; Ok(()) } } diff --git a/google_rest_api_generator/src/cargo.rs b/google_rest_api_generator/src/cargo.rs index 39bcfec..f94baa9 100644 --- a/google_rest_api_generator/src/cargo.rs +++ b/google_rest_api_generator/src/cargo.rs @@ -8,15 +8,15 @@ edition = "2018" publish = false [dependencies] -serde = { version = "1", features = ["derive"] } -serde_json = "1" chrono = { version = "0.4", features = ["serde"] } -reqwest = { version = "0.10", default-features = false, features = ['rustls-tls', 'blocking', 'json'] } -google_field_selector = { git = "https://github.com/google-apis-rs/generator" } google_api_auth = { git = "https://github.com/google-apis-rs/generator" } +google_field_selector = { git = "https://github.com/google-apis-rs/generator" } mime = "0.3" -textnonce = "0.6" percent-encoding = "2" +reqwest = { version = "0.10", default-features = false, features = ["rustls-tls", "json"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +textnonce = "0.6" "#; pub(crate) fn cargo_toml(crate_name: &str, include_bytes_dep: bool, api: &shared::Api) -> String { @@ -30,9 +30,20 @@ pub(crate) fn cargo_toml(crate_name: &str, include_bytes_dep: bool, api: &shared .expect("available crate version"), ); + // TODO: figure out a better way to determine if we should add futures as a dep & include stream reqwest feature + if crate_name.contains("storage") { + doc = doc.replace( + r#"features = ["rustls-tls", "json"]"#, + r#"features = ["stream", "rustls-tls", "json"]"#, + ); + + doc.push_str("\nfutures = \"0.3\""); + } + if include_bytes_dep { - doc.push_str("\n[dependencies.google_api_bytes]\n"); + doc.push_str("\n\n[dependencies.google_api_bytes]\n"); doc.push_str("git = \"https://github.com/google-apis-rs/generator\"\n"); } + doc } diff --git a/google_rest_api_generator/src/lib.rs b/google_rest_api_generator/src/lib.rs index 165fef0..027f5e2 100644 --- a/google_rest_api_generator/src/lib.rs +++ b/google_rest_api_generator/src/lib.rs @@ -290,7 +290,7 @@ impl APIDesc { #(#param_type_defs)* } pub struct Client { - reqwest: ::reqwest::blocking::Client, + reqwest: ::reqwest::Client, auth: Box, } impl Client { @@ -300,13 +300,13 @@ impl APIDesc { { Client::with_reqwest_client( auth, - ::reqwest::blocking::Client::builder().timeout(None).build().unwrap() + ::reqwest::Client::builder().build().unwrap() ) } // Not necessarily the best API. If we have a need for anymore // configuration knobs we should switch to a builder pattern. - pub fn with_reqwest_client(auth: A, reqwest: ::reqwest::blocking::Client) -> Self + pub fn with_reqwest_client(auth: A, reqwest: ::reqwest::Client) -> Self where A: Into>, { diff --git a/google_rest_api_generator/src/method_builder.rs b/google_rest_api_generator/src/method_builder.rs index 5e4ae0b..c4ba675 100644 --- a/google_rest_api_generator/src/method_builder.rs +++ b/google_rest_api_generator/src/method_builder.rs @@ -113,7 +113,7 @@ pub(crate) fn generate( #[doc = #builder_doc] #[derive(Debug,Clone)] pub struct #builder_name<'a> { - pub(crate) reqwest: &'a ::reqwest::blocking::Client, + pub(crate) reqwest: &'a ::reqwest::Client, pub(crate) auth: &'a dyn ::google_api_auth::GetAccessToken, #(#builder_fields,)* } @@ -154,7 +154,7 @@ fn exec_method( /// are not generic over the return type and deserialize the /// response into an auto-generated struct will all possible /// fields. - pub fn execute(self) -> Result + pub async fn execute(self) -> Result where T: ::serde::de::DeserializeOwned + ::google_field_selector::FieldSelector, { @@ -164,7 +164,7 @@ fn exec_method( } else { Some(fields) }; - self.execute_with_fields(fields) + self.execute_with_fields(fields).await } /// Execute the given operation. This will not provide any @@ -172,46 +172,46 @@ fn exec_method( /// the fields returned. This typically includes the most common /// fields, but it will not include every possible attribute of /// the response resource. - pub fn execute_with_default_fields(self) -> Result<#resp_type_path, crate::Error> { - self.execute_with_fields(None::<&str>) + pub async fn execute_with_default_fields(self) -> Result<#resp_type_path, crate::Error> { + self.execute_with_fields(None::<&str>).await } /// Execute the given operation. This will provide a `fields` /// selector of `*`. This will include every attribute of the /// response resource and should be limited to use during /// development or debugging. - pub fn execute_with_all_fields(self) -> Result<#resp_type_path, crate::Error> { - self.execute_with_fields(Some("*")) + pub async fn execute_with_all_fields(self) -> Result<#resp_type_path, crate::Error> { + self.execute_with_fields(Some("*")).await } /// Execute the given operation. This will use the `fields` /// selector provided and will deserialize the response into /// whatever return value is provided. - pub fn execute_with_fields(mut self, fields: Option) -> Result + pub async fn execute_with_fields(mut self, fields: Option) -> Result where T: ::serde::de::DeserializeOwned, F: Into, { self.fields = fields.map(Into::into); - self._execute() + self._execute().await } - fn _execute(&mut self) -> Result + async fn _execute(&mut self) -> Result where T: ::serde::de::DeserializeOwned, { let req = self._request(&self._path())?; #set_body - Ok(crate::error_from_response(req.send()?)?.json()?) + Ok(req.send().await?.error_for_status()?.json().await?) } } } None => { quote! { - pub fn execute(self) -> Result<(), crate::Error> { + pub async fn execute(self) -> Result<(), crate::Error> { let req = self._request(&self._path())?; #set_body - crate::error_from_response(req.send()?)?; + req.send().await?.error_for_status()?; Ok(()) } } @@ -378,7 +378,7 @@ fn request_method<'a>(http_method: &str, params: impl Iterator .expect(format!("unknown http method: {}", http_method).as_str()); let reqwest_method = reqwest_http_method(&http_method); quote! { - fn _request(&self, path: &str) -> Result<::reqwest::blocking::RequestBuilder, crate::Error> { + fn _request(&self, path: &str) -> Result<::reqwest::RequestBuilder, crate::Error> { let req = self.reqwest.request(#reqwest_method, path); #(let req = req.query(&[#query_params]);)* let req = req.bearer_auth(self.auth.access_token().map_err(|err| crate::Error::OAuth2(err))?); @@ -399,6 +399,7 @@ fn iterable_method_impl<'a>(method: &Method) -> TokenStream { where T: ::serde::de::DeserializeOwned, { + // FIXME: make `IterableMethod` async/refactor `Iter` impl to async `Stream` self._execute() } } @@ -558,12 +559,13 @@ fn download_method(base_url: &str, method: &Method) -> TokenStream { ); quote! { #download_path_method - pub fn download(mut self, output: &mut W) -> Result + pub async fn download(mut self, output: &mut W) -> Result where - W: ::std::io::Write + ?Sized, + W: futures::io::AsyncWriteExt + std::marker::Unpin + ?Sized, { self.alt = Some(crate::params::Alt::Media); - Ok(crate::error_from_response(self._request(&self._path())?.send()?)?.copy_to(output)?) + let response = self._request(&self._path())?.send().await?.error_for_status()?; + Ok(futures::io::copy(response.bytes_stream(), output).await?) } } } @@ -581,7 +583,7 @@ fn upload_methods(base_url: &str, method: &Method) -> TokenStream { let upload_fn = match &method.response { Some(_response) => { quote!{ - pub fn upload(mut self, content: R, mime_type: ::mime::Mime) -> Result + pub async fn upload(mut self, content: R, mime_type: ::mime::Mime) -> Result where T: ::serde::de::DeserializeOwned + ::google_field_selector::FieldSelector, R: ::std::io::Read + ::std::io::Seek + Send + 'static, @@ -599,8 +601,9 @@ fn upload_methods(base_url: &str, method: &Method) -> TokenStream { #add_request_part multipart.new_part(Part::new(mime_type, Box::new(content))); let req = req.header(::reqwest::header::CONTENT_TYPE, format!("multipart/related; boundary={}", multipart.boundary())); - let req = req.body(reqwest::blocking::Body::new(multipart.into_reader())); - Ok(crate::error_from_response(req.send()?)?.json()?) + let req = req.body(reqwest::Body::new(multipart.into_reader())); + let response = req.send().await?.error_for_status()?; + Ok(response) } } }, @@ -617,8 +620,8 @@ fn upload_methods(base_url: &str, method: &Method) -> TokenStream { #add_request_part multipart.new_part(Part::new(mime_type, Box::new(content))); let req = req.header(::reqwest::header::CONTENT_TYPE, format!("multipart/related; boundary={}", multipart.boundary())); - let req = req.body(reqwest::blocking::Body::new(multipart.into_reader())); - crate::error_from_response(req.send()?)?; + let req = req.body(reqwest::Body::new(multipart.into_reader())); + req.send().await?.error_for_status()?; Ok(()) } } @@ -644,12 +647,12 @@ fn upload_methods(base_url: &str, method: &Method) -> TokenStream { &method.params, ); let upload_fn = quote!{ - pub fn start_resumable_upload(self, mime_type: ::mime::Mime) -> Result { + pub async fn start_resumable_upload(self, mime_type: ::mime::Mime) -> Result { let req = self._request(&self._resumable_upload_path())?; let req = req.query(&[("uploadType", "resumable")]); let req = req.header(::reqwest::header::HeaderName::from_static("x-upload-content-type"), mime_type.to_string()); #set_body - let resp = crate::error_from_response(req.send()?)?; + let resp = req.send().await?.error_for_status()?; let location_header = resp.headers().get(::reqwest::header::LOCATION).ok_or_else(|| crate::Error::Other(format!("No LOCATION header returned when initiating resumable upload").into()))?; let upload_url = ::std::str::from_utf8(location_header.as_bytes()).map_err(|_| crate::Error::Other(format!("Non UTF8 LOCATION header returned").into()))?.to_owned(); Ok(crate::ResumableUpload::new(self.reqwest.clone(), upload_url)) diff --git a/google_rest_api_generator/src/resource_builder.rs b/google_rest_api_generator/src/resource_builder.rs index 4975716..63587b8 100644 --- a/google_rest_api_generator/src/resource_builder.rs +++ b/google_rest_api_generator/src/resource_builder.rs @@ -54,7 +54,7 @@ pub(crate) fn generate( } pub struct #action_ident<'a> { - pub(crate) reqwest: &'a reqwest::blocking::Client, + pub(crate) reqwest: &'a reqwest::Client, pub(crate) auth: &'a dyn ::google_api_auth::GetAccessToken, } impl<'a> #action_ident<'a> { From c96a73b1af60e422877265d62b72a080b35fa0cf Mon Sep 17 00:00:00 2001 From: William Myers Date: Fri, 29 May 2020 23:29:26 -0600 Subject: [PATCH 02/11] Make access_token async --- google_api_auth/Cargo.toml | 1 + google_api_auth/src/lib.rs | 3 ++- google_api_auth/src/yup_oauth2.rs | 7 +++---- google_rest_api_generator/src/cargo.rs | 2 +- google_rest_api_generator/src/method_builder.rs | 11 +++++++---- 5 files changed, 14 insertions(+), 10 deletions(-) diff --git a/google_api_auth/Cargo.toml b/google_api_auth/Cargo.toml index c0d3bf1..fce3084 100644 --- a/google_api_auth/Cargo.toml +++ b/google_api_auth/Cargo.toml @@ -14,3 +14,4 @@ with-yup-oauth2 = ["yup-oauth2", "tokio"] yup-oauth2 = { version = "^4.1", optional = true } tokio = { version = "0.2", optional = true } hyper = "^0.13" +async-trait = "^0.1" diff --git a/google_api_auth/src/lib.rs b/google_api_auth/src/lib.rs index 913321f..3dd4c25 100644 --- a/google_api_auth/src/lib.rs +++ b/google_api_auth/src/lib.rs @@ -2,8 +2,9 @@ /// client libraries to retrieve access tokens when making http requests. This /// library optionally provides a variety of implementations, but users are also /// free to implement whatever logic they want for retrieving a token. +#[async_trait::async_trait] pub trait GetAccessToken: ::std::fmt::Debug + Send + Sync { - fn access_token(&self) -> Result>; + async fn access_token(&self) -> Result>; } impl From for Box diff --git a/google_api_auth/src/yup_oauth2.rs b/google_api_auth/src/yup_oauth2.rs index 602372a..40d065c 100644 --- a/google_api_auth/src/yup_oauth2.rs +++ b/google_api_auth/src/yup_oauth2.rs @@ -24,14 +24,13 @@ impl ::std::fmt::Debug for YupAuthenticator { } } +#[async_trait::async_trait] impl crate::GetAccessToken for YupAuthenticator where C: Connect + Clone + Send + Sync + 'static, { - fn access_token(&self) -> Result> { - let fut = self.auth.token(&self.scopes); - let mut runtime = ::tokio::runtime::Runtime::new().expect("unable to start tokio runtime"); - Ok(runtime.block_on(fut)?.as_str().to_string()) + async fn access_token(&self) -> Result> { + Ok(self.auth.token(&self.scopes).await?.as_str().to_string()) } } diff --git a/google_rest_api_generator/src/cargo.rs b/google_rest_api_generator/src/cargo.rs index f94baa9..f01b798 100644 --- a/google_rest_api_generator/src/cargo.rs +++ b/google_rest_api_generator/src/cargo.rs @@ -9,7 +9,7 @@ publish = false [dependencies] chrono = { version = "0.4", features = ["serde"] } -google_api_auth = { git = "https://github.com/google-apis-rs/generator" } +google_api_auth = { git = "https://github.com/google-apis-rs/generator", branch = "refactor/async" } google_field_selector = { git = "https://github.com/google-apis-rs/generator" } mime = "0.3" percent-encoding = "2" diff --git a/google_rest_api_generator/src/method_builder.rs b/google_rest_api_generator/src/method_builder.rs index c4ba675..6736d62 100644 --- a/google_rest_api_generator/src/method_builder.rs +++ b/google_rest_api_generator/src/method_builder.rs @@ -200,7 +200,7 @@ fn exec_method( where T: ::serde::de::DeserializeOwned, { - let req = self._request(&self._path())?; + let req = self._request(&self._path()).await?; #set_body Ok(req.send().await?.error_for_status()?.json().await?) } @@ -209,7 +209,7 @@ fn exec_method( None => { quote! { pub async fn execute(self) -> Result<(), crate::Error> { - let req = self._request(&self._path())?; + let req = self._request(&self._path()).await?; #set_body req.send().await?.error_for_status()?; Ok(()) @@ -378,10 +378,13 @@ fn request_method<'a>(http_method: &str, params: impl Iterator .expect(format!("unknown http method: {}", http_method).as_str()); let reqwest_method = reqwest_http_method(&http_method); quote! { - fn _request(&self, path: &str) -> Result<::reqwest::RequestBuilder, crate::Error> { + async fn _request(&self, path: &str) -> Result<::reqwest::RequestBuilder, crate::Error> { let req = self.reqwest.request(#reqwest_method, path); #(let req = req.query(&[#query_params]);)* - let req = req.bearer_auth(self.auth.access_token().map_err(|err| crate::Error::OAuth2(err))?); + let access_token = self.auth.access_token() + .await + .map_err(|err| crate::Error::OAuth2(err))?; + let req = req.bearer_auth(access_token); Ok(req) } } From 3ddcc3b27604f36d8834e839b126c1d83ef7792d Mon Sep 17 00:00:00 2001 From: William Myers Date: Sat, 30 May 2020 00:18:22 -0600 Subject: [PATCH 03/11] Refactor storage download method to be async --- google_rest_api_generator/src/method_builder.rs | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/google_rest_api_generator/src/method_builder.rs b/google_rest_api_generator/src/method_builder.rs index 6736d62..0cbe1bd 100644 --- a/google_rest_api_generator/src/method_builder.rs +++ b/google_rest_api_generator/src/method_builder.rs @@ -564,11 +564,22 @@ fn download_method(base_url: &str, method: &Method) -> TokenStream { #download_path_method pub async fn download(mut self, output: &mut W) -> Result where - W: futures::io::AsyncWriteExt + std::marker::Unpin + ?Sized, + W: futures::io::AsyncWrite + std::marker::Unpin + ?Sized, { + use futures::io::AsyncWriteExt; + self.alt = Some(crate::params::Alt::Media); - let response = self._request(&self._path())?.send().await?.error_for_status()?; - Ok(futures::io::copy(response.bytes_stream(), output).await?) + let request = self._request(&self._path()).await?; + + let mut response = request.send().await?.error_for_status()?; + + let mut num_bytes_written: usize = 0; + while let Some(chunk) = response.chunk().await? { + output.write(&chunk); + num_bytes_written += chunk.len(); + } + + Ok(num_bytes_written as u64) } } } From 0fd60d18892243706b2dc9c8ebd95cf083f6c374 Mon Sep 17 00:00:00 2001 From: William Myers Date: Sat, 30 May 2020 18:27:21 -0600 Subject: [PATCH 04/11] Refactor upload to be async It reads the entire body into memory instead of implementing Stream. --- .../gen_include/error.rs | 9 ++++ .../gen_include/multipart.rs | 42 ++++++++++----- .../src/method_builder.rs | 54 ++++++++++++++----- 3 files changed, 80 insertions(+), 25 deletions(-) diff --git a/google_rest_api_generator/gen_include/error.rs b/google_rest_api_generator/gen_include/error.rs index 8800a5a..8ad9273 100644 --- a/google_rest_api_generator/gen_include/error.rs +++ b/google_rest_api_generator/gen_include/error.rs @@ -6,6 +6,7 @@ pub enum Error { reqwest_err: ::reqwest::Error, body: Option, }, + IO(std::io::Error), Other(Box), } @@ -15,6 +16,7 @@ impl Error { Error::OAuth2(_) => None, Error::JSON(err) => Some(err), Error::Reqwest { .. } => None, + Error::IO(_) => None, Error::Other(_) => None, } } @@ -32,6 +34,7 @@ impl ::std::fmt::Display for Error { } Ok(()) } + Error::IO(err) => write!(f, "IO Error: {}", err), Error::Other(err) => write!(f, "Uknown Error: {}", err), } } @@ -53,3 +56,9 @@ impl From<::reqwest::Error> for Error { } } } + +impl From for Error { + fn from(err: std::io::Error) -> Error { + Error::IO(err) + } +} diff --git a/google_rest_api_generator/gen_include/multipart.rs b/google_rest_api_generator/gen_include/multipart.rs index f324196..829d6d0 100644 --- a/google_rest_api_generator/gen_include/multipart.rs +++ b/google_rest_api_generator/gen_include/multipart.rs @@ -37,13 +37,13 @@ mod multipart { pub(crate) struct Part { content_type: ::mime::Mime, - body: Box, + body: Box, } impl Part { pub(crate) fn new( content_type: ::mime::Mime, - body: Box, + body: Box, ) -> Part { Part { content_type, body } } @@ -52,26 +52,32 @@ mod multipart { pub(crate) struct RelatedMultiPartReader { state: RelatedMultiPartReaderState, boundary: String, - next_body: Option>, + next_body: Option>, parts: std::vec::IntoIter, } enum RelatedMultiPartReaderState { WriteBoundary { - start: usize, boundary: String, + start: usize, + boundary: String, }, WriteContentType { start: usize, content_type: Vec, }, WriteBody { - body: Box, + body: Box, }, } - impl ::std::io::Read for RelatedMultiPartReader { - fn read(&mut self, buf: &mut [u8]) -> ::std::io::Result { + impl futures::io::AsyncRead for RelatedMultiPartReader { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + ctx: &mut futures::task::Context, + buf: &mut [u8], + ) -> futures::task::Poll> { use RelatedMultiPartReaderState::*; + let mut bytes_written: usize = 0; loop { let rem_buf = &mut buf[bytes_written..]; @@ -90,8 +96,11 @@ mod multipart { self.next_body = Some(next_part.body); self.state = WriteContentType { start: 0, - content_type: format!("Content-Type: {}\r\n\r\n", next_part.content_type) - .into_bytes(), + content_type: format!( + "Content-Type: {}\r\n\r\n", + next_part.content_type + ) + .into_bytes(), }; } else { break; @@ -101,7 +110,8 @@ mod multipart { start, content_type, } => { - let bytes_to_copy = std::cmp::min(content_type.len() - *start, rem_buf.len()); + let bytes_to_copy = + std::cmp::min(content_type.len() - *start, rem_buf.len()); rem_buf[..bytes_to_copy] .copy_from_slice(&content_type[*start..*start + bytes_to_copy]); *start += bytes_to_copy; @@ -115,7 +125,14 @@ mod multipart { } } WriteBody { body } => { - let written = body.read(rem_buf)?; + let body = std::pin::Pin::new(body); + let written = match futures::io::AsyncRead::poll_read(body, ctx, rem_buf) { + futures::task::Poll::Ready(Ok(n)) => n, + futures::task::Poll::Ready(Err(err)) => { + return futures::task::Poll::Ready(Err(err)); + } + futures::task::Poll::Pending => return futures::task::Poll::Pending, + }; bytes_written += written; if written == 0 { self.state = WriteBoundary { @@ -128,7 +145,8 @@ mod multipart { } } } - Ok(bytes_written) + + futures::task::Poll::Ready(Ok(bytes_written)) } } diff --git a/google_rest_api_generator/src/method_builder.rs b/google_rest_api_generator/src/method_builder.rs index 0cbe1bd..9e31695 100644 --- a/google_rest_api_generator/src/method_builder.rs +++ b/google_rest_api_generator/src/method_builder.rs @@ -402,7 +402,7 @@ fn iterable_method_impl<'a>(method: &Method) -> TokenStream { where T: ::serde::de::DeserializeOwned, { - // FIXME: make `IterableMethod` async/refactor `Iter` impl to async `Stream` + // FIXME: refactor `Iter` impl to async `Stream` self._execute() } } @@ -591,51 +591,79 @@ fn upload_methods(base_url: &str, method: &Method) -> TokenStream { let add_request_part = method.request.as_ref().map(|_| { quote!{ let request_json = ::serde_json::to_vec(&self.request)?; - multipart.new_part(Part::new(::mime::APPLICATION_JSON, Box::new(::std::io::Cursor::new(request_json)))); + multipart.new_part(Part::new( + ::mime::APPLICATION_JSON, + Box::new(futures::io::Cursor::new(request_json)), + )); } }); + // TODO: We either have to use `read_to_end` and read the whole body + // into memory or implement Stream for RelatedMultiPartReader, but + // that would require allocating a lot of intermediate Vecs; see: + // http://smallcultfollowing.com/babysteps/blog/2019/12/10/async-interview-2-cramertj-part-2/#the-natural-way-to-write-attached-streams-is-with-gats let upload_fn = match &method.response { Some(_response) => { quote!{ pub async fn upload(mut self, content: R, mime_type: ::mime::Mime) -> Result where T: ::serde::de::DeserializeOwned + ::google_field_selector::FieldSelector, - R: ::std::io::Read + ::std::io::Seek + Send + 'static, + R: futures::io::AsyncRead + std::marker::Unpin + Send + 'static, { + use crate::multipart::{RelatedMultiPart, Part}; + use futures::io::AsyncReadExt; + let fields = ::google_field_selector::to_string::(); self.fields = if fields.is_empty() { None } else { Some(fields) }; - let req = self._request(&self._simple_upload_path())?; + + let req = self._request(&self._simple_upload_path()).await?; let req = req.query(&[("uploadType", "multipart")]); - use crate::multipart::{RelatedMultiPart, Part}; + let mut multipart = RelatedMultiPart::new(); #add_request_part multipart.new_part(Part::new(mime_type, Box::new(content))); + let req = req.header(::reqwest::header::CONTENT_TYPE, format!("multipart/related; boundary={}", multipart.boundary())); - let req = req.body(reqwest::Body::new(multipart.into_reader())); + + let mut body: Vec = vec![]; + let mut reader = multipart.into_reader(); + let _num_bytes = reader.read_to_end(&mut body).await?; + let req = req.body(body); + let response = req.send().await?.error_for_status()?; - Ok(response) + + Ok(response.json().await?) } } }, None => { quote!{ - pub fn upload(self, content: R, mime_type: ::mime::Mime) -> Result<(), crate::Error> + pub async fn upload(self, content: R, mime_type: ::mime::Mime) -> Result<(), crate::Error> where - R: ::std::io::Read + ::std::io::Seek + Send + 'static, + R: futures::io::AsyncRead + std::marker::Unpin + Send + 'static, { - let req = self._request(&self._simple_upload_path())?; - let req = req.query(&[("uploadType", "multipart")]); use crate::multipart::{RelatedMultiPart, Part}; + use futures::io::AsyncReadExt; + + let req = self._request(&self._simple_upload_path()).await?; + let req = req.query(&[("uploadType", "multipart")]); + let mut multipart = RelatedMultiPart::new(); #add_request_part multipart.new_part(Part::new(mime_type, Box::new(content))); + let req = req.header(::reqwest::header::CONTENT_TYPE, format!("multipart/related; boundary={}", multipart.boundary())); - let req = req.body(reqwest::Body::new(multipart.into_reader())); + + let mut body: Vec = vec![]; + let mut reader = multipart.into_reader(); + let _num_bytes = reader.read_to_end(&mut body).await?; + let req = req.body(body); + req.send().await?.error_for_status()?; + Ok(()) } } @@ -662,7 +690,7 @@ fn upload_methods(base_url: &str, method: &Method) -> TokenStream { ); let upload_fn = quote!{ pub async fn start_resumable_upload(self, mime_type: ::mime::Mime) -> Result { - let req = self._request(&self._resumable_upload_path())?; + let req = self._request(&self._resumable_upload_path()).await?; let req = req.query(&[("uploadType", "resumable")]); let req = req.header(::reqwest::header::HeaderName::from_static("x-upload-content-type"), mime_type.to_string()); #set_body From 2c92c176f615ab1d6013de968fc7bdc490c9fe8b Mon Sep 17 00:00:00 2001 From: William Myers Date: Sat, 30 May 2020 18:40:10 -0600 Subject: [PATCH 05/11] Temporarily remove PageIter & ResumableUpload We need to refactor them to be async. --- google_rest_api_generator/gen_include/iter.rs | 2 +- .../gen_include/resumable_upload.rs | 1 - google_rest_api_generator/src/lib.rs | 14 ++++++++------ google_rest_api_generator/src/method_builder.rs | 9 +++++---- 4 files changed, 14 insertions(+), 12 deletions(-) diff --git a/google_rest_api_generator/gen_include/iter.rs b/google_rest_api_generator/gen_include/iter.rs index 7c70d97..839c1ae 100644 --- a/google_rest_api_generator/gen_include/iter.rs +++ b/google_rest_api_generator/gen_include/iter.rs @@ -1,7 +1,7 @@ pub mod iter { pub trait IterableMethod { fn set_page_token(&mut self, value: String); - async fn execute(&mut self) -> Result + fn execute(&mut self) -> Result where T: ::serde::de::DeserializeOwned; } diff --git a/google_rest_api_generator/gen_include/resumable_upload.rs b/google_rest_api_generator/gen_include/resumable_upload.rs index 07211ba..394b4b6 100644 --- a/google_rest_api_generator/gen_include/resumable_upload.rs +++ b/google_rest_api_generator/gen_include/resumable_upload.rs @@ -17,7 +17,6 @@ impl ResumableUpload { &self.url } - // FIXME: fix compile issues `ResumableUpload::upload` to be async pub async fn upload(&mut self, mut reader: R) -> Result<(), Box> where R: ::std::io::Read + ::std::io::Seek + Send + 'static, diff --git a/google_rest_api_generator/src/lib.rs b/google_rest_api_generator/src/lib.rs index 027f5e2..8cb372c 100644 --- a/google_rest_api_generator/src/lib.rs +++ b/google_rest_api_generator/src/lib.rs @@ -92,12 +92,14 @@ pub fn generate( rustfmt_writer.write_all(include_bytes!("../gen_include/percent_encode_consts.rs"))?; rustfmt_writer.write_all(include_bytes!("../gen_include/multipart.rs"))?; rustfmt_writer.write_all(include_bytes!("../gen_include/parsed_string.rs"))?; - if any_resumable_upload_methods { - rustfmt_writer.write_all(include_bytes!("../gen_include/resumable_upload.rs"))?; - } - if any_iterable_methods { - rustfmt_writer.write_all(include_bytes!("../gen_include/iter.rs"))?; - } + // FIXME: refactor ResumableUpload to be async + // if any_resumable_upload_methods { + // rustfmt_writer.write_all(include_bytes!("../gen_include/resumable_upload.rs"))?; + // } + // FIXME: implement Stream instead of Iter so it is async + // if any_iterable_methods { + // rustfmt_writer.write_all(include_bytes!("../gen_include/iter.rs"))?; + // } rustfmt_writer.close()?; info!("api: generated and formatted in {:?}", time.elapsed()); info!("api: done in {:?}", total_time.elapsed()); diff --git a/google_rest_api_generator/src/method_builder.rs b/google_rest_api_generator/src/method_builder.rs index 9e31695..b1a9d6f 100644 --- a/google_rest_api_generator/src/method_builder.rs +++ b/google_rest_api_generator/src/method_builder.rs @@ -109,6 +109,7 @@ pub(crate) fn generate( let upload_methods = upload_methods(root_url, method); let builder_doc = builder_doc(method, creator_ident); + // FIXME: implement Stream instead of Iter so it is async quote! { #[doc = #builder_doc] #[derive(Debug,Clone)] @@ -121,7 +122,7 @@ pub(crate) fn generate( impl<'a> #builder_name<'a> { #(#param_methods)* - #iter_methods + // #iter_methods #download_method #upload_methods #exec_method @@ -130,7 +131,7 @@ pub(crate) fn generate( #request_method } - #iter_types_and_impls + // #iter_types_and_impls } } @@ -402,7 +403,6 @@ fn iterable_method_impl<'a>(method: &Method) -> TokenStream { where T: ::serde::de::DeserializeOwned, { - // FIXME: refactor `Iter` impl to async `Stream` self._execute() } } @@ -706,9 +706,10 @@ fn upload_methods(base_url: &str, method: &Method) -> TokenStream { } }); + // FIXME: refactor ResumableUpload to be async quote! { #simple_fns - #resumable_fns + // #resumable_fns } } else { quote! {} From 6fffa16c52b75b20e3ba4225db2318668732dc32 Mon Sep 17 00:00:00 2001 From: William Myers Date: Sat, 30 May 2020 18:45:51 -0600 Subject: [PATCH 06/11] Add missing await to download chunk write call --- google_rest_api_generator/src/method_builder.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/google_rest_api_generator/src/method_builder.rs b/google_rest_api_generator/src/method_builder.rs index b1a9d6f..f0542ce 100644 --- a/google_rest_api_generator/src/method_builder.rs +++ b/google_rest_api_generator/src/method_builder.rs @@ -560,6 +560,7 @@ fn download_method(base_url: &str, method: &Method) -> TokenStream { &method.path, &method.params, ); + // TODO: faster to use tokio_util::compat? see: https://github.com/seanmonstar/reqwest/issues/482 quote! { #download_path_method pub async fn download(mut self, output: &mut W) -> Result @@ -575,7 +576,7 @@ fn download_method(base_url: &str, method: &Method) -> TokenStream { let mut num_bytes_written: usize = 0; while let Some(chunk) = response.chunk().await? { - output.write(&chunk); + output.write(&chunk).await?; num_bytes_written += chunk.len(); } From 2a4981ca958bf9dd91204d0fabb2301b4e6ac34a Mon Sep 17 00:00:00 2001 From: William Myers Date: Sat, 30 May 2020 18:51:39 -0600 Subject: [PATCH 07/11] Run rustfmt --- google_rest_api_generator/gen_include/iter.rs | 42 ++++++++++++------- 1 file changed, 27 insertions(+), 15 deletions(-) diff --git a/google_rest_api_generator/gen_include/iter.rs b/google_rest_api_generator/gen_include/iter.rs index 839c1ae..ae140ee 100644 --- a/google_rest_api_generator/gen_include/iter.rs +++ b/google_rest_api_generator/gen_include/iter.rs @@ -6,7 +6,7 @@ pub mod iter { T: ::serde::de::DeserializeOwned; } - pub struct PageIter{ + pub struct PageIter { pub method: M, pub finished: bool, pub _phantom: ::std::marker::PhantomData, @@ -18,7 +18,7 @@ pub mod iter { T: ::serde::de::DeserializeOwned, { pub(crate) fn new(method: M) -> Self { - PageIter{ + PageIter { method, finished: false, _phantom: ::std::marker::PhantomData, @@ -37,24 +37,30 @@ pub mod iter { if self.finished { return None; } - let paginated_result: ::serde_json::Map = match self.method.execute() { - Ok(r) => r, - Err(err) => return Some(Err(err)), - }; - if let Some(next_page_token) = paginated_result.get("nextPageToken").and_then(|t| t.as_str()) { + let paginated_result: ::serde_json::Map = + match self.method.execute() { + Ok(r) => r, + Err(err) => return Some(Err(err)), + }; + if let Some(next_page_token) = paginated_result + .get("nextPageToken") + .and_then(|t| t.as_str()) + { self.method.set_page_token(next_page_token.to_owned()); } else { self.finished = true; } - Some(match ::serde_json::from_value(::serde_json::Value::Object(paginated_result)) { - Ok(resp) => Ok(resp), - Err(err) => Err(err.into()) - }) + Some( + match ::serde_json::from_value(::serde_json::Value::Object(paginated_result)) { + Ok(resp) => Ok(resp), + Err(err) => Err(err.into()), + }, + ) } } - pub struct PageItemIter{ + pub struct PageItemIter { items_field: &'static str, page_iter: PageIter>, items: ::std::vec::IntoIter, @@ -66,7 +72,7 @@ pub mod iter { T: ::serde::de::DeserializeOwned, { pub(crate) fn new(method: M, items_field: &'static str) -> Self { - PageItemIter{ + PageItemIter { items_field, page_iter: PageIter::new(method), items: Vec::new().into_iter(), @@ -92,10 +98,16 @@ pub mod iter { None => return None, Some(Err(err)) => return Some(Err(err)), Some(Ok(next_page)) => { - let mut next_page: ::serde_json::Map = next_page; + let mut next_page: ::serde_json::Map = + next_page; let items_array = match next_page.remove(self.items_field) { Some(items) => items, - None => return Some(Err(crate::Error::Other(format!("no {} field found in iter response", self.items_field).into()))), + None => { + return Some(Err(crate::Error::Other( + format!("no {} field found in iter response", self.items_field) + .into(), + ))) + } }; let items_vec: Result, _> = ::serde_json::from_value(items_array); match items_vec { From cc3961f50605b462a120a3b8d0ac1a05267dc7bd Mon Sep 17 00:00:00 2001 From: William Myers Date: Sat, 6 Jun 2020 13:30:54 -0600 Subject: [PATCH 08/11] Add missing futures dependency to cargo.rs --- google_rest_api_generator/src/cargo.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/google_rest_api_generator/src/cargo.rs b/google_rest_api_generator/src/cargo.rs index f01b798..8a87c23 100644 --- a/google_rest_api_generator/src/cargo.rs +++ b/google_rest_api_generator/src/cargo.rs @@ -9,6 +9,7 @@ publish = false [dependencies] chrono = { version = "0.4", features = ["serde"] } +futures = "0.3" google_api_auth = { git = "https://github.com/google-apis-rs/generator", branch = "refactor/async" } google_field_selector = { git = "https://github.com/google-apis-rs/generator" } mime = "0.3" @@ -30,14 +31,12 @@ pub(crate) fn cargo_toml(crate_name: &str, include_bytes_dep: bool, api: &shared .expect("available crate version"), ); - // TODO: figure out a better way to determine if we should add futures as a dep & include stream reqwest feature + // TODO: figure out a better way to determine if we should include stream reqwest feature if crate_name.contains("storage") { doc = doc.replace( r#"features = ["rustls-tls", "json"]"#, r#"features = ["stream", "rustls-tls", "json"]"#, ); - - doc.push_str("\nfutures = \"0.3\""); } if include_bytes_dep { From 2d5887365397732ab34c15d5aae71a0213890806 Mon Sep 17 00:00:00 2001 From: William Myers Date: Fri, 19 Jun 2020 17:27:40 -0600 Subject: [PATCH 09/11] Add rustls-tls & native-tls features These features will be passed to reqwest and defaults to rustls-tls to maintain backwards compatibility. --- google_rest_api_generator/src/cargo.rs | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/google_rest_api_generator/src/cargo.rs b/google_rest_api_generator/src/cargo.rs index 8a87c23..eac4b3b 100644 --- a/google_rest_api_generator/src/cargo.rs +++ b/google_rest_api_generator/src/cargo.rs @@ -7,14 +7,20 @@ edition = "2018" # for now, let's not even accidentally publish these publish = false +[features] +default = ["rustls-tls"] + +native-tls = ["reqwest/native-tls"] +rustls-tls = ["reqwest/rustls-tls"] + [dependencies] chrono = { version = "0.4", features = ["serde"] } futures = "0.3" google_api_auth = { git = "https://github.com/google-apis-rs/generator", branch = "refactor/async" } -google_field_selector = { git = "https://github.com/google-apis-rs/generator" } +google_field_selector = { git = "https://github.com/google-apis-rs/generator", branch = "refactor/async" } mime = "0.3" percent-encoding = "2" -reqwest = { version = "0.10", default-features = false, features = ["rustls-tls", "json"] } +reqwest = { version = "0.10", default-features = false, features = ["json"] } serde = { version = "1", features = ["derive"] } serde_json = "1" textnonce = "0.6" @@ -33,10 +39,7 @@ pub(crate) fn cargo_toml(crate_name: &str, include_bytes_dep: bool, api: &shared // TODO: figure out a better way to determine if we should include stream reqwest feature if crate_name.contains("storage") { - doc = doc.replace( - r#"features = ["rustls-tls", "json"]"#, - r#"features = ["stream", "rustls-tls", "json"]"#, - ); + doc = doc.replace(r#"features = ["json"]"#, r#"features = ["stream", "json"]"#); } if include_bytes_dep { From bfc9cf3f51bcac17a09766d13f40013b5343e9b1 Mon Sep 17 00:00:00 2001 From: William Myers Date: Sun, 14 Mar 2021 00:38:34 -0800 Subject: [PATCH 10/11] Update dependencies in cargo.rs --- google_rest_api_generator/src/cargo.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/google_rest_api_generator/src/cargo.rs b/google_rest_api_generator/src/cargo.rs index eac4b3b..606bcba 100644 --- a/google_rest_api_generator/src/cargo.rs +++ b/google_rest_api_generator/src/cargo.rs @@ -20,10 +20,10 @@ google_api_auth = { git = "https://github.com/google-apis-rs/generator", branch google_field_selector = { git = "https://github.com/google-apis-rs/generator", branch = "refactor/async" } mime = "0.3" percent-encoding = "2" -reqwest = { version = "0.10", default-features = false, features = ["json"] } +reqwest = { version = "0.11", default-features = false, features = ["json"] } serde = { version = "1", features = ["derive"] } serde_json = "1" -textnonce = "0.6" +textnonce = "1" "#; pub(crate) fn cargo_toml(crate_name: &str, include_bytes_dep: bool, api: &shared::Api) -> String { From ef73c95c7f948f8a59dc96e51abc8593b766c814 Mon Sep 17 00:00:00 2001 From: William Myers Date: Sun, 14 Mar 2021 01:28:36 -0800 Subject: [PATCH 11/11] Upgrade google_api_auth dependencies --- google_api_auth/Cargo.toml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/google_api_auth/Cargo.toml b/google_api_auth/Cargo.toml index fce3084..56c0dd2 100644 --- a/google_api_auth/Cargo.toml +++ b/google_api_auth/Cargo.toml @@ -11,7 +11,7 @@ default = [] with-yup-oauth2 = ["yup-oauth2", "tokio"] [dependencies] -yup-oauth2 = { version = "^4.1", optional = true } -tokio = { version = "0.2", optional = true } -hyper = "^0.13" -async-trait = "^0.1" +yup-oauth2 = { version = "5", optional = true } +tokio = { version = "1", optional = true } +hyper = "0.14" +async-trait = "0.1"