Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions src/flow_service/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand All @@ -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,
Expand All @@ -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,
Expand Down
10 changes: 7 additions & 3 deletions src/flow_service/retry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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!(
Expand Down