Add support for global and per call request timeouts (#132)

This commit adds the ability to support a global request timeout
on TransportBuilder, as well as per request timeouts on
builder structs.

Closes #116

Author: russcam

README.md

@@ -238,6 +238,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             HeaderMap::new(),
             Option::<&Value>::None,
             Some(body.as_ref()),
+            None,
         )
         .await?;
     Ok(())

api_generator/src/generator/code_gen/mod.rs

@@ -43,7 +43,10 @@ pub fn use_declarations() -> Tokens {
                 transport::Transport,
             },
         };
-        use std::borrow::Cow;
+        use std::{
+            borrow::Cow,
+            time::Duration
+        };
         use percent_encoding::percent_encode;
         use serde::Serialize;
     )

api_generator/src/generator/code_gen/request/request_builder.rs

@@ -350,6 +350,43 @@ impl<'a> RequestBuilder<'a> {
         }
     }
 
+    /// Creates the AST for a builder fn to add a request timeout
+    fn create_request_timeout_fn(field: &syn::Ident) -> syn::ImplItem {
+        let doc_attr = doc("Sets a request timeout for this API call.\n\nThe timeout is applied from when the request starts connecting until the response body has finished.");
+
+        syn::ImplItem {
+            ident: ident("request_timeout"),
+            vis: syn::Visibility::Public,
+            defaultness: syn::Defaultness::Final,
+            attrs: vec![doc_attr],
+            node: syn::ImplItemKind::Method(
+                syn::MethodSig {
+                    unsafety: syn::Unsafety::Normal,
+                    constness: syn::Constness::NotConst,
+                    abi: None,
+                    decl: syn::FnDecl {
+                        inputs: vec![
+                            syn::FnArg::SelfValue(syn::Mutability::Mutable),
+                            syn::FnArg::Captured(
+                                syn::Pat::Path(None, path_none("timeout")),
+                                syn::parse_type("Duration").unwrap(),
+                            ),
+                        ],
+                        output: syn::FunctionRetTy::Ty(code_gen::ty("Self")),
+                        variadic: false,
+                    },
+                    generics: generics_none(),
+                },
+                syn::Block {
+                    stmts: vec![
+                        syn::Stmt::Semi(Box::new(parse_expr(quote!(self.#field = Some(timeout))))),
+                        syn::Stmt::Expr(Box::new(parse_expr(quote!(self)))),
+                    ],
+                },
+            ),
+        }
+    }
+
     /// Creates the AST for a builder fn for a builder impl
     fn create_impl_fn(f: (&String, &Type)) -> syn::ImplItem {
         let name = valid_name(&f.0).to_lowercase();
@@ -456,6 +493,7 @@ impl<'a> RequestBuilder<'a> {
             .collect();
 
         let headers_field_ident = ident("headers");
+        let request_timeout_ident = ident("request_timeout");
 
         // add a field for HTTP headers
         fields.push(syn::Field {
@@ -464,6 +502,12 @@ impl<'a> RequestBuilder<'a> {
             attrs: vec![],
             ty: syn::parse_type("HeaderMap").unwrap(),
         });
+        fields.push(syn::Field {
+            ident: Some(request_timeout_ident.clone()),
+            vis: syn::Visibility::Inherited,
+            attrs: vec![],
+            ty: syn::parse_type("Option<Duration>").unwrap(),
+        });
 
         if supports_body {
             fields.push(syn::Field {
@@ -491,6 +535,7 @@ impl<'a> RequestBuilder<'a> {
             endpoint.params.iter().map(Self::create_impl_fn).collect();
 
         builder_fns.push(Self::create_header_fn(&headers_field_ident));
+        builder_fns.push(Self::create_request_timeout_fn(&request_timeout_ident));
 
         // add a body impl if supported
         if supports_body {
@@ -591,9 +636,10 @@ impl<'a> RequestBuilder<'a> {
                       let path = self.parts.url();
                       let method = #method_expr;
                       let headers = self.headers;
+                      let timeout = self.request_timeout;
                       let query_string = #query_string_expr;
                       let body = #body_expr;
-                      let response = self.transport.send(method, &path, headers, query_string.as_ref(), body).await?;
+                      let response = self.transport.send(method, &path, headers, query_string.as_ref(), body, timeout).await?;
                       Ok(response)
                 }
             }

elasticsearch/src/client.rs

@@ -22,6 +22,7 @@ use crate::{
 };
 
 use serde::{Serialize, Serializer};
+use std::time::Duration;
 
 /// Serializes an `Option<&[Serialize]>` with
 /// `Some(value)` to a comma separated string of values.
@@ -80,13 +81,14 @@ impl Elasticsearch {
         headers: HeaderMap,
         query_string: Option<&Q>,
         body: Option<B>,
+        timeout: Option<Duration>,
     ) -> Result<Response, Error>
     where
         B: Body,
         Q: Serialize + ?Sized,
     {
         self.transport
-            .send(method, path, headers, query_string, body)
+            .send(method, path, headers, query_string, body, timeout)
             .await
     }
 }

elasticsearch/src/generated/namespace_clients/async_search.rs

@@ -37,7 +37,7 @@ use crate::{
 };
 use percent_encoding::percent_encode;
 use serde::Serialize;
-use std::borrow::Cow;
+use std::{borrow::Cow, time::Duration};
 #[derive(Debug, Clone, PartialEq)]
 #[doc = "API parts for the Async Search Delete API"]
 pub enum AsyncSearchDeleteParts<'b> {
@@ -68,6 +68,7 @@ pub struct AsyncSearchDelete<'a, 'b> {
     headers: HeaderMap,
     human: Option<bool>,
     pretty: Option<bool>,
+    request_timeout: Option<Duration>,
     source: Option<&'b str>,
 }
 impl<'a, 'b> AsyncSearchDelete<'a, 'b> {
@@ -82,6 +83,7 @@ impl<'a, 'b> AsyncSearchDelete<'a, 'b> {
             filter_path: None,
             human: None,
             pretty: None,
+            request_timeout: None,
             source: None,
         }
     }
@@ -110,6 +112,11 @@ impl<'a, 'b> AsyncSearchDelete<'a, 'b> {
         self.pretty = Some(pretty);
         self
     }
+    #[doc = "Sets a request timeout for this API call.\n\nThe timeout is applied from when the request starts connecting until the response body has finished."]
+    pub fn request_timeout(mut self, timeout: Duration) -> Self {
+        self.request_timeout = Some(timeout);
+        self
+    }
     #[doc = "The URL-encoded request definition. Useful for libraries that do not accept a request body for non-POST requests."]
     pub fn source(mut self, source: &'b str) -> Self {
         self.source = Some(source);
@@ -120,6 +127,7 @@ impl<'a, 'b> AsyncSearchDelete<'a, 'b> {
         let path = self.parts.url();
         let method = Method::Delete;
         let headers = self.headers;
+        let timeout = self.request_timeout;
         let query_string = {
             #[serde_with::skip_serializing_none]
             #[derive(Serialize)]
@@ -150,7 +158,7 @@ impl<'a, 'b> AsyncSearchDelete<'a, 'b> {
         let body = Option::<()>::None;
         let response = self
             .transport
-            .send(method, &path, headers, query_string.as_ref(), body)
+            .send(method, &path, headers, query_string.as_ref(), body, timeout)
             .await?;
         Ok(response)
     }
@@ -186,6 +194,7 @@ pub struct AsyncSearchGet<'a, 'b> {
     human: Option<bool>,
     keep_alive: Option<&'b str>,
     pretty: Option<bool>,
+    request_timeout: Option<Duration>,
     source: Option<&'b str>,
     typed_keys: Option<bool>,
     wait_for_completion_timeout: Option<&'b str>,
@@ -203,6 +212,7 @@ impl<'a, 'b> AsyncSearchGet<'a, 'b> {
             human: None,
             keep_alive: None,
             pretty: None,
+            request_timeout: None,
             source: None,
             typed_keys: None,
             wait_for_completion_timeout: None,
@@ -238,6 +248,11 @@ impl<'a, 'b> AsyncSearchGet<'a, 'b> {
         self.pretty = Some(pretty);
         self
     }
+    #[doc = "Sets a request timeout for this API call.\n\nThe timeout is applied from when the request starts connecting until the response body has finished."]
+    pub fn request_timeout(mut self, timeout: Duration) -> Self {
+        self.request_timeout = Some(timeout);
+        self
+    }
     #[doc = "The URL-encoded request definition. Useful for libraries that do not accept a request body for non-POST requests."]
     pub fn source(mut self, source: &'b str) -> Self {
         self.source = Some(source);
@@ -258,6 +273,7 @@ impl<'a, 'b> AsyncSearchGet<'a, 'b> {
         let path = self.parts.url();
         let method = Method::Get;
         let headers = self.headers;
+        let timeout = self.request_timeout;
         let query_string = {
             #[serde_with::skip_serializing_none]
             #[derive(Serialize)]
@@ -297,7 +313,7 @@ impl<'a, 'b> AsyncSearchGet<'a, 'b> {
         let body = Option::<()>::None;
         let response = self
             .transport
-            .send(method, &path, headers, query_string.as_ref(), body)
+            .send(method, &path, headers, query_string.as_ref(), body, timeout)
             .await?;
         Ok(response)
     }
@@ -362,6 +378,7 @@ pub struct AsyncSearchSubmit<'a, 'b, B> {
     pretty: Option<bool>,
     q: Option<&'b str>,
     request_cache: Option<bool>,
+    request_timeout: Option<Duration>,
     routing: Option<&'b [&'b str]>,
     search_type: Option<SearchType>,
     seq_no_primary_term: Option<bool>,
@@ -421,6 +438,7 @@ where
             pretty: None,
             q: None,
             request_cache: None,
+            request_timeout: None,
             routing: None,
             search_type: None,
             seq_no_primary_term: None,
@@ -519,6 +537,7 @@ where
             pretty: self.pretty,
             q: self.q,
             request_cache: self.request_cache,
+            request_timeout: self.request_timeout,
             routing: self.routing,
             search_type: self.search_type,
             seq_no_primary_term: self.seq_no_primary_term,
@@ -640,6 +659,11 @@ where
         self.request_cache = Some(request_cache);
         self
     }
+    #[doc = "Sets a request timeout for this API call.\n\nThe timeout is applied from when the request starts connecting until the response body has finished."]
+    pub fn request_timeout(mut self, timeout: Duration) -> Self {
+        self.request_timeout = Some(timeout);
+        self
+    }
     #[doc = "A comma-separated list of specific routing values"]
     pub fn routing(mut self, routing: &'b [&'b str]) -> Self {
         self.routing = Some(routing);
@@ -740,6 +764,7 @@ where
         let path = self.parts.url();
         let method = Method::Post;
         let headers = self.headers;
+        let timeout = self.request_timeout;
         let query_string = {
             #[serde_with::skip_serializing_none]
             #[derive(Serialize)]
@@ -914,7 +939,7 @@ where
         let body = self.body;
         let response = self
             .transport
-            .send(method, &path, headers, query_string.as_ref(), body)
+            .send(method, &path, headers, query_string.as_ref(), body, timeout)
             .await?;
         Ok(response)
     }