feat: cleanup devicename handlers -wip

core/main/src/bootstrap/start_communication_broker.rs

@@ -38,7 +38,7 @@ impl Bootstep<BootstrapState> for StartCommunicationBroker {
         state
             .platform_state
             .get_client()
-            .add_request_processor(RpcGatewayProcessor::new(state.platform_state.get_client()));
+            .add_request_processor(RpcGatewayProcessor::new(state.platform_state.clone()));
 
         // Start the Broker Reciever
         if let Ok(rx) = state.channels_state.get_broker_receiver() {

core/main/src/broker/broker_utils.rs

@@ -15,15 +15,20 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
-use std::time::Duration;
-
+use crate::state::platform_state::PlatformState;
 use crate::utils::rpc_utils::extract_tcp_port;
 use futures::stream::{SplitSink, SplitStream};
 use futures_util::StreamExt;
+use jsonrpsee::{core::RpcResult, types::error::CallError};
+use ripple_sdk::api::firebolt::fb_capabilities::CAPABILITY_NOT_AVAILABLE;
 use ripple_sdk::{
+    api::gateway::rpc_gateway_api::{ApiProtocol, CallContext, RpcRequest},
+    extn::extn_client_message::ExtnResponse,
     log::{error, info},
     tokio::{self, net::TcpStream},
 };
+use serde_json::from_value;
+use std::time::Duration;
 use tokio_tungstenite::{client_async, tungstenite::Message, WebSocketStream};
 
 pub struct BrokerUtils;
@@ -66,4 +71,35 @@ impl BrokerUtils {
             tokio::time::sleep(Duration::from_secs(1)).await;
         }
     }
+
+    pub async fn get_device_name(ctx: &CallContext, state: &PlatformState) -> RpcResult<String> {
+        let mut new_ctx = ctx.clone();
+        new_ctx.protocol = ApiProtocol::Extn;
+
+        let rpc_request = RpcRequest {
+            ctx: new_ctx.clone(),
+            method: "device.name".into(),
+            params_json: RpcRequest::prepend_ctx(None, &new_ctx),
+        };
+
+        let resp = state
+            .get_client()
+            .get_extn_client()
+            .main_internal_request(rpc_request.clone())
+            .await;
+
+        if let Ok(res) = resp {
+            if let Some(ExtnResponse::Value(val)) = res.payload.extract::<ExtnResponse>() {
+                if let Ok(v) = from_value::<String>(val) {
+                    return Ok(v);
+                }
+            }
+        }
+
+        Err(jsonrpsee::core::Error::Call(CallError::Custom {
+            code: CAPABILITY_NOT_AVAILABLE,
+            message: "device.name is not available".into(),
+            data: None,
+        }))
+    }
 }

core/main/src/broker/endpoint_broker.rs

@@ -232,7 +232,7 @@ pub struct EndpointBrokerState {
     callback: BrokerCallback,
     request_map: Arc<RwLock<HashMap<u64, BrokerRequest>>>,
     extension_request_map: Arc<RwLock<HashMap<u64, ExtnMessage>>>,
-    rule_engine: RuleEngine,
+    pub rule_engine: RuleEngine,
     cleaner_list: Arc<RwLock<Vec<BrokerCleaner>>>,
     reconnect_tx: Sender<BrokerConnectRequest>,
 }
@@ -605,6 +605,7 @@ impl BrokerOutputForwarder {
 
                         // Step 3: Handle Non Extension
                         if matches!(rpc_request.ctx.protocol, ApiProtocol::Extn) {
+                            is_event = rpc_request.is_subscription();
                             if let Ok(extn_message) =
                                 platform_state.endpoint_state.get_extn_message(id, is_event)
                             {

core/main/src/broker/websocket_broker.rs

@@ -222,7 +222,6 @@ mod tests {
     ) -> WebsocketBroker {
         // setup mock websocket server
         let port = MockWebsocket::start(send_data, Vec::new(), tx, on_close).await;
-
         let endpoint = RuleEndpoint {
             url: format!("ws://127.0.0.1:{}", port),
             protocol: crate::broker::rules_engine::RuleEndpointProtocol::Websocket,

core/main/src/firebolt/handlers/device_rpc.rs

@@ -18,11 +18,9 @@
 use std::{collections::HashMap, env, time::Duration};
 
 use crate::{
-    firebolt::rpc::RippleRPCProvider,
-    processor::storage::storage_manager::StorageManager,
-    service::apps::app_events::AppEvents,
-    state::platform_state::PlatformState,
-    utils::rpc_utils::{rpc_add_event_listener, rpc_err},
+    firebolt::rpc::RippleRPCProvider, processor::storage::storage_manager::StorageManager,
+    service::apps::app_events::AppEvents, state::platform_state::PlatformState,
+    utils::rpc_utils::rpc_err,
 };
 
 use jsonrpsee::{
@@ -41,7 +39,6 @@ use ripple_sdk::{
             },
             device_info_request::{DeviceInfoRequest, DeviceResponse, FirmwareInfo},
             device_operator::DEFAULT_DEVICE_OPERATION_TIMEOUT_SECS,
-            device_peristence::SetStringProperty,
             device_request::{
                 AudioProfile, DeviceVersionResponse, HdcpProfile, HdrProfile, NetworkResponse,
             },
@@ -53,10 +50,7 @@ use ripple_sdk::{
         },
         gateway::rpc_gateway_api::CallContext,
         session::{AccountSessionRequest, ProvisionRequest},
-        storage_property::{
-            StorageProperty, StoragePropertyData, EVENT_DEVICE_DEVICE_NAME_CHANGED,
-            EVENT_DEVICE_NAME_CHANGED, KEY_FIREBOLT_DEVICE_UID, SCOPE_DEVICE,
-        },
+        storage_property::{StoragePropertyData, KEY_FIREBOLT_DEVICE_UID, SCOPE_DEVICE},
     },
     extn::extn_client_message::ExtnResponse,
     log::error,
@@ -88,30 +82,10 @@ pub const DEVICE_UID: &str = "device.uid";
 
 #[rpc(server)]
 pub trait Device {
-    #[method(name = "device.name")]
-    async fn name(&self, ctx: CallContext) -> RpcResult<String>;
-    #[method(name = "device.setName")]
-    async fn set_name(
-        &self,
-        ctx: CallContext,
-        _setname_request: SetStringProperty,
-    ) -> RpcResult<()>;
     #[method(name = "device.id")]
     async fn id(&self, ctx: CallContext) -> RpcResult<String>;
     #[method(name = "device.uid")]
     async fn uid(&self, ctx: CallContext) -> RpcResult<String>;
-    #[method(name = "device.onNameChanged")]
-    async fn on_name_changed(
-        &self,
-        ctx: CallContext,
-        request: ListenRequest,
-    ) -> RpcResult<ListenerResponse>;
-    #[method(name = "device.onDeviceNameChanged")]
-    async fn on_device_name_changed(
-        &self,
-        ctx: CallContext,
-        request: ListenRequest,
-    ) -> RpcResult<ListenerResponse>;
     #[method(name = "device.model")]
     async fn model(&self, ctx: CallContext) -> RpcResult<String>;
     #[method(name = "device.sku")]
@@ -275,15 +249,6 @@ pub async fn get_ll_mac_addr(state: PlatformState) -> RpcResult<String> {
         ))),
     }
 }
-
-pub async fn set_device_name(state: &PlatformState, prop: SetStringProperty) -> RpcResult<()> {
-    StorageManager::set_string(state, StorageProperty::DeviceName, prop.value, None).await
-}
-
-pub async fn get_device_name(state: &PlatformState) -> RpcResult<String> {
-    StorageManager::get_string(state, StorageProperty::DeviceName).await
-}
-
 #[derive(Debug)]
 pub struct DeviceImpl {
     pub state: PlatformState,
@@ -313,34 +278,6 @@ impl DeviceImpl {
 
 #[async_trait]
 impl DeviceServer for DeviceImpl {
-    async fn name(&self, _ctx: CallContext) -> RpcResult<String> {
-        get_device_name(&self.state).await
-    }
-
-    async fn set_name(
-        &self,
-        _ctx: CallContext,
-        setname_request: SetStringProperty,
-    ) -> RpcResult<()> {
-        set_device_name(&self.state, setname_request).await
-    }
-
-    async fn on_name_changed(
-        &self,
-        ctx: CallContext,
-        request: ListenRequest,
-    ) -> RpcResult<ListenerResponse> {
-        rpc_add_event_listener(&self.state, ctx, request, EVENT_DEVICE_NAME_CHANGED).await
-    }
-
-    async fn on_device_name_changed(
-        &self,
-        ctx: CallContext,
-        request: ListenRequest,
-    ) -> RpcResult<ListenerResponse> {
-        rpc_add_event_listener(&self.state, ctx, request, EVENT_DEVICE_DEVICE_NAME_CHANGED).await
-    }
-
     async fn id(&self, _ctx: CallContext) -> RpcResult<String> {
         get_device_id(&self.state).await
     }

core/main/src/firebolt/handlers/second_screen_rpc.rs

@@ -34,17 +34,14 @@ use ripple_sdk::api::{
     gateway::rpc_gateway_api::CallContext,
 };
 
-use super::device_rpc::{get_device_id, get_device_name};
+use super::device_rpc::get_device_id;
 
 pub const EVENT_SECOND_SCREEN_ON_CLOSE_REQUEST: &str = "secondscreen.onCloseRequest";
-pub const EVENT_SECOND_SCREEN_ON_FRIENDLY_NAME_CHANGED: &str = "secondscreen.onFriendlyNameChanged";
 
 #[rpc(server)]
 pub trait SecondScreen {
     #[method(name = "secondscreen.device")]
     async fn device(&self, ctx: CallContext, param: SecondScreenDeviceInfo) -> RpcResult<String>;
-    #[method(name = "secondscreen.friendlyName")]
-    async fn friendly_name(&self, ctx: CallContext) -> RpcResult<String>;
     #[method(name = "secondscreen.protocols")]
     async fn protocols(&self, _ctx: CallContext) -> RpcResult<HashMap<String, bool>>;
     #[method(name = "secondscreen.onLaunchRequest")]
@@ -59,12 +56,6 @@ pub trait SecondScreen {
         ctx: CallContext,
         request: ListenRequest,
     ) -> RpcResult<ListenerResponse>;
-    #[method(name = "secondscreen.onFriendlyNameChanged")]
-    async fn on_friendly_name_changed(
-        &self,
-        ctx: CallContext,
-        request: ListenRequest,
-    ) -> RpcResult<ListenerResponse>;
 }
 
 pub struct SecondScreenImpl {
@@ -77,10 +68,6 @@ impl SecondScreenServer for SecondScreenImpl {
         get_device_id(&self.state).await
     }
 
-    async fn friendly_name(&self, _ctx: CallContext) -> RpcResult<String> {
-        get_device_name(&self.state).await
-    }
-
     async fn protocols(&self, _ctx: CallContext) -> RpcResult<HashMap<String, bool>> {
         let mut protocols = HashMap::<String, bool>::new();
         protocols.insert("dial2".into(), true);
@@ -114,20 +101,6 @@ impl SecondScreenServer for SecondScreenImpl {
         )
         .await
     }
-
-    async fn on_friendly_name_changed(
-        &self,
-        ctx: CallContext,
-        request: ListenRequest,
-    ) -> RpcResult<ListenerResponse> {
-        rpc_add_event_listener(
-            &self.state,
-            ctx,
-            request,
-            EVENT_SECOND_SCREEN_ON_FRIENDLY_NAME_CHANGED,
-        )
-        .await
-    }
 }
 
 pub struct SecondScreenRPCProvider;

core/main/src/processor/rpc_gateway_processor.rs

@@ -24,34 +24,35 @@ use ripple_sdk::{
         },
         extn_client_message::ExtnMessage,
     },
+    log::debug,
     tokio::sync::mpsc::Sender,
 };
 
 use crate::{
-    firebolt::firebolt_gateway::FireboltGatewayCommand, service::extn::ripple_client::RippleClient,
+    firebolt::firebolt_gateway::FireboltGatewayCommand, state::platform_state::PlatformState,
 };
 
 /// Processor to service incoming RPC Requests used by extensions and other local rpc handlers for aliasing.
 #[derive(Debug)]
 pub struct RpcGatewayProcessor {
-    client: RippleClient,
+    state: PlatformState,
     streamer: DefaultExtnStreamer,
 }
 
 impl RpcGatewayProcessor {
-    pub fn new(client: RippleClient) -> RpcGatewayProcessor {
+    pub fn new(state: PlatformState) -> RpcGatewayProcessor {
         RpcGatewayProcessor {
-            client,
+            state,
             streamer: DefaultExtnStreamer::new(),
         }
     }
 }
 
 impl ExtnStreamProcessor for RpcGatewayProcessor {
-    type STATE = RippleClient;
+    type STATE = PlatformState;
     type VALUE = RpcRequest;
     fn get_state(&self) -> Self::STATE {
-        self.client.clone()
+        self.state.clone()
     }
 
     fn sender(&self) -> Sender<ExtnMessage> {
@@ -66,23 +67,34 @@ impl ExtnStreamProcessor for RpcGatewayProcessor {
 #[async_trait]
 impl ExtnRequestProcessor for RpcGatewayProcessor {
     fn get_client(&self) -> ripple_sdk::extn::client::extn_client::ExtnClient {
-        self.client.get_extn_client()
+        self.state.get_client().get_extn_client()
     }
 
     async fn process_request(state: Self::STATE, msg: ExtnMessage, request: Self::VALUE) -> bool {
+        debug!("Inside RPC gateway processor");
         match request.ctx.protocol {
             ApiProtocol::Extn => {
                 // Notice how this processor is different from others where it doesnt respond to
                 // Self::respond this processor delegates the request down
                 // to the gateway which does more complex inter connected operations. The design for
                 // Extn Processor is built in such a way to support transient processors which do not
                 // necessarily need to provide response
-                if let Err(e) =
-                    state.send_gateway_command(FireboltGatewayCommand::HandleRpcForExtn {
-                        msg: msg.clone(),
-                    })
-                {
-                    return Self::handle_error(state.get_extn_client(), msg, e).await;
+                if state.endpoint_state.rule_engine.has_rule(&request) {
+                    if !state
+                        .endpoint_state
+                        .handle_brokerage(request, Some(msg.clone()))
+                    {
+                        return Self::handle_error(
+                            state.get_client().get_extn_client(),
+                            msg,
+                            ripple_sdk::utils::error::RippleError::InvalidAccess,
+                        )
+                        .await;
+                    }
+                } else if let Err(e) = state.get_client().send_gateway_command(
+                    FireboltGatewayCommand::HandleRpcForExtn { msg: msg.clone() },
+                ) {
+                    return Self::handle_error(state.get_client().get_extn_client(), msg, e).await;
                 }
             }
             _ =>
@@ -92,10 +104,11 @@ impl ExtnRequestProcessor for RpcGatewayProcessor {
             // Extn Processor is built in such a way to support transient processors which do not
             // necessarily need to provide response
             {
-                if let Err(e) =
-                    state.send_gateway_command(FireboltGatewayCommand::HandleRpc { request })
+                if let Err(e) = state
+                    .get_client()
+                    .send_gateway_command(FireboltGatewayCommand::HandleRpc { request })
                 {
-                    return Self::handle_error(state.get_extn_client(), msg, e).await;
+                    return Self::handle_error(state.get_client().get_extn_client(), msg, e).await;
                 }
             }
         }