Merge pull request #168 from code0-tech/#167-aquila-grpc-channel-timeout

made timeout configurable

Author: raphael-goetz

src/flow_service/mod.rs

@@ -2,6 +2,7 @@ use crate::{
     flow_definition::Reader,
     flow_service::{auth::get_authorization_metadata, retry::create_channel_with_retry},
 };
+use tokio::time::Duration;
 use tonic::{Extensions, Request, transport::Channel};
 use tucana::{
     aquila::{ModuleUpdateRequest, module_service_client::ModuleServiceClient},
@@ -28,7 +29,13 @@ impl FlowUpdateService {
     ///
     /// This reads the definition files from the given path as modules and initializes the
     /// service with those module definitions.
-    pub async fn from_url(aquila_url: String, definition_path: &str, aquila_token: String) -> Self {
+    pub async fn from_url(
+        aquila_url: String,
+        definition_path: &str,
+        aquila_token: String,
+        connect_timeout: Duration,
+        request_timeout: Duration,
+    ) -> Self {
         let reader = Reader::configure(definition_path.to_string(), true, vec![], None);
         let modules = match reader.read_modules() {
             Ok(modules) => modules,
@@ -38,7 +45,8 @@ impl FlowUpdateService {
             }
         };
 
-        let channel = create_channel_with_retry("Aquila", aquila_url).await;
+        let channel =
+            create_channel_with_retry("Aquila", aquila_url, connect_timeout, request_timeout).await;
 
         Self {
             modules,

src/flow_service/retry.rs

@@ -5,16 +5,20 @@ const MAX_BACKOFF: u64 = 2000 * 60;
 const MAX_RETRIES: i8 = 10;
 
 // Will create a channel and retry if its not possible
-pub async fn create_channel_with_retry(channel_name: &str, url: String) -> Channel {
+pub async fn create_channel_with_retry(
+    channel_name: &str,
+    url: String,
+    connect_timeout: Duration,
+    request_timeout: Duration,
+) -> Channel {
     let mut backoff = 100;
     let mut retries = 0;
 
     loop {
         let channel = match Endpoint::from_shared(url.clone()) {
             Ok(c) => {
                 log::debug!("Creating a new endpoint for the: {} Service", channel_name);
-                c.connect_timeout(Duration::from_secs(2))
-                    .timeout(Duration::from_secs(10))
+                c.connect_timeout(connect_timeout).timeout(request_timeout)
             }
             Err(err) => {
                 panic!(