feat: thunderBroker migration changes with config switch

core/main/src/broker/broker_utils.rs

@@ -15,14 +15,15 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
-use crate::{state::platform_state::PlatformState, utils::rpc_utils::extract_tcp_port};
+use crate::state::platform_state::PlatformState;
 use futures::stream::{SplitSink, SplitStream};
 use futures_util::StreamExt;
 use jsonrpsee::core::RpcResult;
 use ripple_sdk::{
     api::gateway::rpc_gateway_api::{JsonRpcApiError, RpcRequest},
     log::{error, info},
     tokio::{self, net::TcpStream},
+    utils::rpc_utils::extract_tcp_port,
 };
 use serde_json::Value;
 use std::time::Duration;
@@ -46,12 +47,14 @@ impl BrokerUtils {
         };
         let url = url::Url::parse(&url_path).unwrap();
         let port = extract_tcp_port(endpoint);
+        let tcp_port = port.unwrap();
+
         info!("Url host str {}", url.host_str().unwrap());
         let mut index = 0;
 
         loop {
             // Try connecting to the tcp port first
-            if let Ok(v) = TcpStream::connect(&port).await {
+            if let Ok(v) = TcpStream::connect(&tcp_port).await {
                 // Setup handshake for websocket with the tcp port
                 // Some WS servers lock on to the Port but not setup handshake till they are fully setup
                 if let Ok((stream, _)) = client_async(url_path.clone(), v).await {
@@ -61,7 +64,7 @@ impl BrokerUtils {
             if (index % 10).eq(&0) {
                 error!(
                     "Broker with {} failed with retry for last {} secs in {}",
-                    url_path, index, port
+                    url_path, index, tcp_port
                 );
             }
             index += 1;

core/main/src/broker/thunder/user_data_migrator.rs

@@ -41,15 +41,14 @@ use ripple_sdk::{
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
 
-use crate::broker::endpoint_broker::{
-    self, BrokerCallback, BrokerOutput, BrokerRequest, EndpointBrokerState,
+use crate::broker::{
+    endpoint_broker::{self, BrokerCallback, BrokerOutput, BrokerRequest, EndpointBrokerState},
+    rules_engine::{Rule, RuleTransformType},
+    thunder_broker::ThunderBroker,
 };
-use crate::broker::rules_engine::{Rule, RuleTransformType};
 
 use futures::stream::SplitSink;
 use futures_util::SinkExt;
-
-use crate::broker::thunder_broker::ThunderBroker;
 use tokio_tungstenite::{tungstenite::Message, WebSocketStream};
 
 // TBD get the storage dir from manifest or other Ripple config file

core/main/src/utils/rpc_utils.rs

@@ -109,13 +109,3 @@ pub fn get_base_method(method: &str) -> String {
     let method_vec: Vec<&str> = method.split('.').collect();
     method_vec.first().unwrap().to_string().to_lowercase()
 }
-
-pub fn extract_tcp_port(url: &str) -> String {
-    let url_split: Vec<&str> = url.split("://").collect();
-    if let Some(domain) = url_split.get(1) {
-        let domain_split: Vec<&str> = domain.split('/').collect();
-        domain_split.first().unwrap().to_string()
-    } else {
-        url.to_owned()
-    }
-}

core/sdk/src/api/device/mod.rs

@@ -21,7 +21,6 @@ pub mod device_apps;
 pub mod device_browser;
 pub mod device_events;
 pub mod device_info_request;
-pub mod device_operator;
 pub mod device_peristence;
 pub mod device_request;
 pub mod device_user_grants_data;

core/sdk/src/utils/rpc_utils.rs

@@ -20,3 +20,13 @@ use jsonrpsee::core::Error;
 pub fn rpc_err(msg: impl Into<String>) -> Error {
     Error::Custom(msg.into())
 }
+
+pub fn extract_tcp_port(url: &str) -> Result<String, Error> {
+    let url_split: Vec<&str> = url.split("://").collect();
+    if let Some(domain) = url_split.get(1) {
+        let domain_split: Vec<&str> = domain.split('/').collect();
+        Ok(domain_split.first().unwrap().to_string())
+    } else {
+        Err(Error::Custom("Invalid URL format".to_string()))
+    }
+}

device/thunder_ripple_sdk/Cargo.toml

@@ -44,6 +44,9 @@ regex.workspace = true
 jsonrpsee = { workspace = true, features = ["macros", "ws-client"] }
 serde.workspace = true
 url.workspace = true
+serde_json.workspace = true
+futures-channel.workspace = true
+futures.workspace = true
 
 strum = { version = "0.24", default-features = false }
 strum_macros = "0.24"
@@ -56,8 +59,9 @@ csv = "1.1"  # Allowing minor updates
 home = { version = "=0.5.5", optional = true }
 tree_magic_mini = { version = "=3.0.3", optional = true }
 rstest = { version = "0.18.2", optional = true, default-features = false }
+tokio-tungstenite = { workspace = true, features = ["handshake"] }
+futures-util = { version = "0.3.28", features = ["sink", "std"], default-features = false}
 
 [dev-dependencies]
-tokio-tungstenite = { workspace = true, features = ["native-tls"] }
 ripple_sdk = { path = "../../core/sdk", features = ["tdk"] }
-
+rstest = "0.18.0"

device/thunder_ripple_sdk/src/bootstrap/boot_thunder.rs

@@ -21,25 +21,120 @@ use crate::{
 };
 use ripple_sdk::{
     extn::client::extn_client::ExtnClient,
-    log::{error, info},
+    log::{debug, error, info, warn},
 };
 
 use super::{get_config_step::ThunderGetConfigStep, setup_thunder_pool_step::ThunderPoolStep};
+use crate::client::thunder_client::ThunderClientBuilder;
+use crate::thunder_state::ThunderBootstrapStateWithConfig;
+use crate::thunder_state::ThunderState;
+use ripple_sdk::api::config::Config;
+use ripple_sdk::utils::error::RippleError;
+use serde::Deserialize;
+
+use ripple_sdk::extn::extn_client_message::{ExtnMessage, ExtnResponse};
+
+const GATEWAY_DEFAULT: &str = "ws://127.0.0.1:9998/jsonrpc";
+
+#[derive(Deserialize, Clone)]
+pub struct ThunderPlatformParams {
+    #[serde(default = "gateway_default")]
+    gateway: String,
+}
+
+fn gateway_default() -> String {
+    String::from(GATEWAY_DEFAULT)
+}
 
 pub async fn boot_thunder(
-    state: ExtnClient,
+    ext_client: ExtnClient,
     plugin_param: ThunderPluginBootParam,
 ) -> Option<ThunderBootstrapStateWithClient> {
-    info!("Booting thunder");
-    if let Ok(state) = ThunderGetConfigStep::setup(state, plugin_param).await {
+    info!("Booting thunder initiated");
+    //by default enabling the thunderBroker
+    let state = if ext_client.get_bool_config("use_with_thunder_async_client") {
+        info!("Using thunder_async_clinet");
+        let mut extn_client = ext_client.clone();
+        let mut gateway_url = url::Url::parse(GATEWAY_DEFAULT).unwrap();
+        let extn_message_response: Result<ExtnMessage, RippleError> =
+            extn_client.request(Config::PlatformParameters).await;
+
+        if let Ok(message) = extn_message_response {
+            if let Some(_response) = message.payload.extract().map(|response| {
+                if let ExtnResponse::Value(v) = response {
+                    serde_json::from_value::<ThunderPlatformParams>(v)
+                        .map(|thunder_parameters| {
+                            url::Url::parse(&thunder_parameters.gateway).map_or_else(
+                                |_| {
+                                    warn!(
+                                        "Could not parse thunder gateway '{}', using default {}",
+                                        thunder_parameters.gateway, GATEWAY_DEFAULT
+                                    );
+                                },
+                                |gtway_url| {
+                                    debug!("Got url from device manifest");
+                                    gateway_url = gtway_url;
+                                },
+                            );
+                        })
+                        .unwrap_or_else(|_| {
+                            warn!(
+                                "Could not read thunder platform parameters, using default {}",
+                                GATEWAY_DEFAULT
+                            );
+                        });
+                }
+                if let Ok(host_override) = std::env::var("DEVICE_HOST") {
+                    gateway_url.set_host(Some(&host_override)).ok();
+                }
+            }) {}
+        }
+
+        if let Ok(thndr_client) = ThunderClientBuilder::start_thunder_client(
+            gateway_url.clone(),
+            None,
+            None,
+            None,
+            None,
+            true,
+        )
+        .await
+        {
+            let thunder_state = ThunderState::new(ext_client.clone(), thndr_client);
+
+            let thndr_boot_statecfg = ThunderBootstrapStateWithConfig {
+                extn_client: ext_client,
+                url: gateway_url,
+                pool_size: None,
+                plugin_param: None,
+                thunder_connection_state: None,
+            };
+
+            let thndr_boot_stateclient = ThunderBootstrapStateWithClient {
+                prev: thndr_boot_statecfg,
+                state: thunder_state,
+            };
+
+            thndr_boot_stateclient.clone().state.start_event_thread();
+
+            Some(thndr_boot_stateclient)
+        } else {
+            None
+        }
+    } else if let Ok(state) = ThunderGetConfigStep::setup(ext_client, plugin_param).await {
         if let Ok(state) = ThunderPoolStep::setup(state).await {
-            SetupThunderProcessor::setup(state.clone()).await;
-            return Some(state);
+            Some(state)
         } else {
             error!("Unable to connect to Thunder, error in ThunderPoolStep");
+            None
         }
     } else {
         error!("Unable to connect to Thunder, error in ThunderGetConfigStep");
+        None
+    };
+
+    if let Some(s) = state.clone() {
+        SetupThunderProcessor::setup(s).await;
     }
-    None
+    state
 }

device/thunder_ripple_sdk/src/bootstrap/get_config_step.rs

@@ -93,9 +93,9 @@ impl ThunderGetConfigStep {
                 return Ok(ThunderBootstrapStateWithConfig {
                     extn_client: state,
                     url: gateway_url,
-                    pool_size,
-                    plugin_param: expected_plugins,
-                    thunder_connection_state: Arc::new(ThunderConnectionState::new()),
+                    pool_size: Some(pool_size),
+                    plugin_param: Some(expected_plugins),
+                    thunder_connection_state: Some(Arc::new(ThunderConnectionState::new())),
                 });
             }
         }

device/thunder_ripple_sdk/src/bootstrap/setup_thunder_pool_step.rs

@@ -40,13 +40,16 @@ impl ThunderPoolStep {
     pub async fn setup(
         state: ThunderBootstrapStateWithConfig,
     ) -> Result<ThunderBootstrapStateWithClient, RippleError> {
-        let pool_size = state.pool_size;
         let url = state.url.clone();
         let thunder_connection_state = state.thunder_connection_state.clone();
-        if pool_size < 2 {
-            warn!("Pool size of 1 is not recommended, there will be no dedicated connection for Controller events");
-            return Err(RippleError::BootstrapError);
-        }
+        let pool_size = match state.pool_size {
+            Some(s) => s,
+            None => {
+                warn!("Pool size of 1 is not recommended, there will be no dedicated connection for Controller events");
+                return Err(RippleError::BootstrapError);
+            }
+        };
+
         let controller_pool = ripple_sdk::tokio::time::timeout(
             Duration::from_secs(10),
             ThunderClientPool::start(url.clone(), None, thunder_connection_state.clone(), 1),
@@ -68,7 +71,13 @@ impl ThunderPoolStep {
         };
 
         info!("Received Controller pool");
-        let expected_plugins = state.plugin_param.clone();
+        let expected_plugins = match state.plugin_param.clone() {
+            Some(plugins) => plugins,
+            None => {
+                error!("Expected plugins are not provided.");
+                return Err(RippleError::BootstrapError);
+            }
+        };
         let tc = Box::new(controller_pool);
         let (plugin_manager_tx, failed_plugins) =
             PluginManager::start(tc, expected_plugins.clone()).await;
@@ -98,7 +107,7 @@ impl ThunderPoolStep {
         }
 
         let client = ThunderClientPool::start(
-            url,
+            url.clone(),
             Some(plugin_manager_tx),
             thunder_connection_state.clone(),
             pool_size - 1,