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;

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/device_operator.rs

@@ -44,6 +44,33 @@ pub enum DeviceChannelRequest {
     Unsubscribe(DeviceUnsubscribeRequest),
 }
 
+impl DeviceChannelRequest {
+    pub fn get_callsign_method(&self) -> (String, String) {
+        match self {
+            DeviceChannelRequest::Call(c) => {
+                let mut collection: Vec<&str> = c.method.split('.').collect();
+                let method = collection.pop().unwrap_or_default();
+                let callsign = collection.join(".");
+                (callsign, method.into())
+            }
+            DeviceChannelRequest::Subscribe(s) => (s.module.clone(), s.event_name.clone()),
+            DeviceChannelRequest::Unsubscribe(u) => (u.module.clone(), u.event_name.clone()),
+        }
+    }
+
+    pub fn is_subscription(&self) -> bool {
+        !matches!(self, DeviceChannelRequest::Call(_))
+    }
+
+    pub fn is_unsubscribe(&self) -> Option<DeviceUnsubscribeRequest> {
+        if let DeviceChannelRequest::Unsubscribe(u) = self {
+            Some(u.clone())
+        } else {
+            None
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DeviceCallRequest {
     pub method: String,
@@ -116,6 +143,16 @@ impl DeviceResponseMessage {
             sub_id: Some(sub_id),
         }
     }
+
+    pub fn new(message: Value, sub_id: Option<String>) -> DeviceResponseMessage {
+        DeviceResponseMessage { message, sub_id }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct DeviceResponseSubscription {
+    pub sub_id: Option<String>,
+    pub handlers: Vec<tokio::sync::mpsc::Sender<DeviceResponseMessage>>,
 }
 
 #[cfg(test)]

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) -> 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()
+    }
+}

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"] }
 

device/thunder_ripple_sdk/src/bootstrap/boot_thunder.rs

@@ -21,25 +21,105 @@ use crate::{
 };
 use ripple_sdk::{
     extn::client::extn_client::ExtnClient,
-    log::{error, info},
+    log::{debug, error, info, warn},
+    serde_json::{self, Error},
 };
 
 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 {
-        if let Ok(state) = ThunderPoolStep::setup(state).await {
-            SetupThunderProcessor::setup(state.clone()).await;
-            return Some(state);
-        } else {
-            error!("Unable to connect to Thunder, error in ThunderPoolStep");
+    //by default enabling the thunderBroker
+    if ext_client.get_bool_config("use_with_thunder_broker") {
+        info!("Using thunder broker");
+        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(ExtnResponse::Value(v)) = message.payload.extract() {
+                let tp_res: Result<ThunderPlatformParams, Error> = serde_json::from_value(v);
+                if let Ok(thunder_parameters) = tp_res {
+                    if let Ok(gtway_url) = url::Url::parse(&thunder_parameters.gateway) {
+                        debug!("Got url from device manifest");
+                        gateway_url = gtway_url
+                    } else {
+                        warn!(
+                            "Could not parse thunder gateway '{}', using default {}",
+                            thunder_parameters.gateway, GATEWAY_DEFAULT
+                        );
+                    }
+                } 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::get_client(Some(gateway_url), 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: None,
+                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();
+
+            SetupThunderProcessor::setup(thndr_boot_stateclient.clone()).await;
+            return Some(thndr_boot_stateclient);
+        }
+        None
     } else {
-        error!("Unable to connect to Thunder, error in ThunderGetConfigStep");
+        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);
+            } else {
+                error!("Unable to connect to Thunder, error in ThunderPoolStep");
+            }
+        } else {
+            error!("Unable to connect to Thunder, error in ThunderGetConfigStep");
+        }
+        None
     }
-    None
 }

device/thunder_ripple_sdk/src/bootstrap/get_config_step.rs

@@ -92,10 +92,10 @@ impl ThunderGetConfigStep {
                 }
                 return Ok(ThunderBootstrapStateWithConfig {
                     extn_client: state,
-                    url: gateway_url,
-                    pool_size,
-                    plugin_param: expected_plugins,
-                    thunder_connection_state: Arc::new(ThunderConnectionState::new()),
+                    url: Some(gateway_url),
+                    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

@@ -43,13 +43,18 @@ impl ThunderPoolStep {
         let pool_size = state.pool_size;
         let url = state.url.clone();
         let thunder_connection_state = state.thunder_connection_state.clone();
-        if pool_size < 2 {
+        if pool_size < Some(2) {
             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),
+            ThunderClientPool::start(
+                url.clone().unwrap(),
+                None,
+                thunder_connection_state.clone(),
+                1,
+            ),
         )
         .await;
 
@@ -71,7 +76,7 @@ impl ThunderPoolStep {
         let expected_plugins = state.plugin_param.clone();
         let tc = Box::new(controller_pool);
         let (plugin_manager_tx, failed_plugins) =
-            PluginManager::start(tc, expected_plugins.clone()).await;
+            PluginManager::start(tc, expected_plugins.clone().unwrap()).await;
 
         if !failed_plugins.is_empty() {
             error!(
@@ -80,7 +85,7 @@ impl ThunderPoolStep {
             );
             loop {
                 let failed_plugins = PluginManager::activate_mandatory_plugins(
-                    expected_plugins.clone(),
+                    expected_plugins.as_ref().unwrap().clone(),
                     plugin_manager_tx.clone(),
                 )
                 .await;
@@ -98,10 +103,10 @@ impl ThunderPoolStep {
         }
 
         let client = ThunderClientPool::start(
-            url,
+            url.clone().unwrap(),
             Some(plugin_manager_tx),
             thunder_connection_state.clone(),
-            pool_size - 1,
+            pool_size.unwrap() - 1,
         )
         .await;
 

device/thunder_ripple_sdk/src/client/plugin_manager.rs

@@ -496,7 +496,7 @@ mod tests {
         let controller_pool = ThunderClientPool::start(
             url.clone(),
             None,
-            Arc::new(ThunderConnectionState::new()),
+            Some(Arc::new(ThunderConnectionState::new())),
             1,
         )
         .await;
@@ -519,7 +519,7 @@ mod tests {
         let client = ThunderClientPool::start(
             url,
             Some(plugin_manager_tx_clone),
-            Arc::new(ThunderConnectionState::new()),
+            Some(Arc::new(ThunderConnectionState::new())),
             4,
         )
         .await;