feat(core): call multiple connectors in parallel (#128)

crates/router/src/configs/defaults.toml

@@ -54,3 +54,7 @@ batch_size = 200
 [drainer]
 stream_name = "DRAINER_STREAM"
 num_partitions = 4
+
+[connectors.supported]
+wallets = ["klarna","braintree"]
+cards = ["stripe","adyen","authorizedotnet","checkout","braintree"]

crates/router/src/core/payments.rs

@@ -6,6 +6,7 @@ pub mod transformers;
 use std::{fmt::Debug, marker::PhantomData, time::Instant};
 
 use error_stack::{IntoReport, ResultExt};
+use futures::future::join_all;
 use router_env::{tracing, tracing::instrument};
 use time;
 
@@ -323,7 +324,7 @@ where
     let res = router_data
         .decide_flows(
             state,
-            connector,
+            &connector,
             customer,
             call_connector_action,
             merchant_account.storage_scheme,
@@ -377,30 +378,30 @@ where
     PaymentResponse: Operation<F, Req>,
 {
     let call_connectors_start_time = Instant::now();
-    let mut router_data_list = Vec::with_capacity(connectors.len());
+    let mut join_handlers = Vec::with_capacity(connectors.len());
 
-    for connector in connectors {
+    for connector in connectors.iter() {
         let connector_id = connector.connector.id();
         let router_data = payment_data
             .construct_router_data(state, connector_id, merchant_account)
             .await?;
 
-        router_data_list.push((connector, router_data));
+        let res = router_data.decide_flows(
+            state,
+            connector,
+            customer,
+            CallConnectorAction::Trigger,
+            merchant_account.storage_scheme,
+        );
+
+        join_handlers.push(res);
     }
 
-    for (connector, router_data) in router_data_list {
+    let result = join_all(join_handlers).await;
-        let connector_name = connector.connector_name.to_owned().to_string();
-        let res = router_data
-            .decide_flows(
-                state,
-                connector,
-                customer,
-                CallConnectorAction::Trigger,
-                merchant_account.storage_scheme,
-            )
-            .await?; //FIXME: remove this error propogation
 
-        match res.response {
+    for (connector_res, connector) in result.into_iter().zip(connectors) {
+        let connector_name = connector.connector_name.to_string();
+        match connector_res?.response {
             Ok(connector_response) => {
                 if let types::PaymentsResponseData::SessionResponse {
                     session_token,
@@ -416,6 +417,7 @@ where
                         });
                 }
             }
+
             Err(connector_error) => {
                 logger::debug!(
                     "sessions_connector_error {} {:?}",

crates/router/src/core/payments/flows.rs

@@ -32,7 +32,7 @@ pub trait Feature<F, T> {
     async fn decide_flows<'a>(
         self,
         state: &AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         maybe_customer: &Option<storage::Customer>,
         call_connector_action: payments::CallConnectorAction,
         storage_scheme: enums::MerchantStorageScheme,

crates/router/src/core/payments/flows/authorize_flow.rs

@@ -52,7 +52,7 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
     async fn decide_flows<'a>(
         self,
         state: &AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         customer: &Option<storage::Customer>,
         call_connector_action: payments::CallConnectorAction,
         storage_scheme: storage_enums::MerchantStorageScheme,
@@ -78,7 +78,7 @@ impl PaymentsAuthorizeRouterData {
     pub async fn decide_flow<'a, 'b>(
         &'b self,
         state: &'a AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         maybe_customer: &Option<storage::Customer>,
         confirm: Option<bool>,
         call_connector_action: payments::CallConnectorAction,

crates/router/src/core/payments/flows/cancel_flow.rs

@@ -42,7 +42,7 @@ impl Feature<api::Void, types::PaymentsCancelData>
     async fn decide_flows<'a>(
         self,
         state: &AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         customer: &Option<storage::Customer>,
         call_connector_action: payments::CallConnectorAction,
         _storage_scheme: enums::MerchantStorageScheme,
@@ -63,7 +63,7 @@ impl PaymentsCancelRouterData {
     pub async fn decide_flow<'a, 'b>(
         &'b self,
         state: &AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         _maybe_customer: &Option<storage::Customer>,
         _confirm: Option<bool>,
         call_connector_action: payments::CallConnectorAction,

crates/router/src/core/payments/flows/capture_flow.rs

@@ -43,7 +43,7 @@ impl Feature<api::Capture, types::PaymentsCaptureData>
     async fn decide_flows<'a>(
         self,
         state: &AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         customer: &Option<storage::Customer>,
         call_connector_action: payments::CallConnectorAction,
         _storage_scheme: enums::MerchantStorageScheme,
@@ -64,7 +64,7 @@ impl PaymentsCaptureRouterData {
     pub async fn decide_flow<'a, 'b>(
         &'b self,
         state: &'a AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         _maybe_customer: &Option<storage::Customer>,
         _confirm: Option<bool>,
         call_connector_action: payments::CallConnectorAction,

crates/router/src/core/payments/flows/psync_flow.rs

@@ -44,7 +44,7 @@ impl Feature<api::PSync, types::PaymentsSyncData>
     async fn decide_flows<'a>(
         self,
         state: &AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         customer: &Option<storage::Customer>,
         call_connector_action: payments::CallConnectorAction,
         _storage_scheme: enums::MerchantStorageScheme,
@@ -64,7 +64,7 @@ impl PaymentsSyncRouterData {
     pub async fn decide_flow<'a, 'b>(
         &'b self,
         state: &'a AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         _maybe_customer: &Option<storage::Customer>,
         _confirm: Option<bool>,
         call_connector_action: payments::CallConnectorAction,

crates/router/src/core/payments/flows/session_flow.rs

@@ -39,7 +39,7 @@ impl Feature<api::Session, types::PaymentsSessionData> for types::PaymentsSessio
     async fn decide_flows<'a>(
         self,
         state: &routes::AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         customer: &Option<storage::Customer>,
         call_connector_action: payments::CallConnectorAction,
         _storage_schema: enums::MerchantStorageScheme,
@@ -59,7 +59,7 @@ impl types::PaymentsSessionRouterData {
     pub async fn decide_flow<'a, 'b>(
         &'b self,
         state: &'a routes::AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         _customer: &Option<storage::Customer>,
         _confirm: Option<bool>,
         call_connector_action: payments::CallConnectorAction,

crates/router/src/core/payments/flows/verfiy_flow.rs

@@ -40,7 +40,7 @@ impl Feature<api::Verify, types::VerifyRequestData> for types::VerifyRouterData
     async fn decide_flows<'a>(
         self,
         state: &AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         customer: &Option<storage::Customer>,
         call_connector_action: payments::CallConnectorAction,
         storage_scheme: enums::MerchantStorageScheme,
@@ -61,7 +61,7 @@ impl types::VerifyRouterData {
     pub async fn decide_flow<'a, 'b>(
         &'b self,
         state: &'a AppState,
-        connector: api::ConnectorData,
+        connector: &api::ConnectorData,
         maybe_customer: &Option<storage::Customer>,
         confirm: Option<bool>,
         call_connector_action: payments::CallConnectorAction,