Passthru with mock

Cargo.toml

@@ -24,6 +24,7 @@ members = [
     "core/main",
     "core/launcher",
     "device/thunder",
+    "device/mock_device",
     "distributor/general",
     "examples/rpc_extn",
     "examples/tm_extn",

core/main/Cargo.toml

@@ -49,7 +49,6 @@ sd-notify = { version = "0.4.1", optional = true }
 exitcode = "1.1.2"
 rand = "0.8"
 
-
 [build-dependencies]
 vergen = "1"
 

core/main/src/state/bootstrap_state.rs

@@ -47,6 +47,7 @@ impl ChannelsState {
         let (gateway_tx, gateway_tr) = mpsc::channel(32);
         let (app_req_tx, app_req_tr) = mpsc::channel(32);
         let (ctx, ctr) = unbounded();
+
         ChannelsState {
             gateway_channel: TransientChannel::new(gateway_tx, gateway_tr),
             app_req_channel: TransientChannel::new(app_req_tx, app_req_tr),

core/main/src/state/platform_state.rs

@@ -113,7 +113,6 @@ impl PlatformState {
         app_library: Vec<AppLibraryEntry>,
     ) -> PlatformState {
         let exclusory = ExclusoryImpl::get(&manifest);
-
         Self {
             extn_manifest,
             cap_state: CapState::new(manifest.clone()),

core/main/src/utils/rpc_utils.rs

@@ -32,14 +32,12 @@ use crate::{
     state::platform_state::PlatformState,
 };
 
+pub use ripple_sdk::utils::rpc_utils::rpc_err;
+
 pub const FIRE_BOLT_DEEPLINK_ERROR_CODE: i32 = -40400;
 pub const DOWNSTREAM_SERVICE_UNAVAILABLE_ERROR_CODE: i32 = -50200;
 pub const SESSION_NO_INTENT_ERROR_CODE: i32 = -40000;
 
-pub fn rpc_err(msg: impl Into<String>) -> Error {
-    Error::Custom(msg.into())
-}
-
 /// Awaits a oneshot to respond. If the oneshot fails to repond, creates a generic
 /// RPC internal error
 pub async fn rpc_await_oneshot<T>(rx: oneshot::Receiver<T>) -> RpcResult<T> {

core/sdk/src/extn/client/extn_client.rs

@@ -283,6 +283,22 @@ impl ExtnClient {
                     {
                         self.context_update(request);
                     }
+                    // if its a request coming as an extn provider the extension is calling on itself.
+                    // for eg an extension has a RPC Method provider and also a channel to process the
+                    // requests this below impl will take care of sending the data back to the Extension
+                    else if let Some(extn_id) = target_contract.is_extn_provider() {
+                        if let Some(s) = self.get_extn_sender_with_extn_id(&extn_id) {
+                            let new_message = message.clone();
+                            tokio::spawn(async move {
+                                if let Err(e) = s.send(new_message.into()).await {
+                                    error!("Error forwarding request {:?}", e)
+                                }
+                            });
+                        } else {
+                            error!("couldn't find the extension id registered the extn channel {:?} is not available", extn_id);
+                            self.handle_no_processor_error(message);
+                        }
+                    }
                     // Forward the message to an extn sender
                     else if let Some(sender) = self.get_extn_sender_with_contract(target_contract)
                     {
@@ -544,7 +560,7 @@ impl ExtnClient {
 
     /// Request method which accepts a impl [ExtnPayloadProvider] and uses the capability provided by the trait to send the request.
     /// As part of the send process it adds a callback to asynchronously respond back to the caller when the response does get
-    /// received. This method can be called synchrnously with a timeout
+    /// received. This method can be called synchronously with a timeout
     ///
     /// # Arguments
     /// `payload` - impl [ExtnPayloadProvider]

core/sdk/src/extn/client/extn_processor.rs

@@ -154,10 +154,9 @@ pub trait ExtnRequestProcessor: ExtnStreamProcessor + Send + Sync + 'static {
     ///
     /// # Returns
     ///
-    /// `Option<bool>` -> Used by [ExtnClient] to handle post processing
-    /// None - means not processed
-    /// Some(true) - Successful processing with status success
-    /// Some(false) - Successful processing with status error
+    /// `bool` -> Used by [ExtnClient] to handle post processing
+    /// `true` - Successful processing with status success
+    /// `false` - Successful processing with status error
     async fn process_request(
         state: Self::STATE,
         msg: ExtnMessage,

core/sdk/src/extn/client/extn_sender.rs

@@ -78,10 +78,12 @@ impl ExtnSender {
     }
 
     pub fn check_contract_fulfillment(&self, contract: RippleContract) -> bool {
-        if self.id.is_main() {
+        if self.id.is_main() || self.fulfills.contains(&contract.as_clear_string()) {
             true
+        } else if let Ok(extn_id) = ExtnId::try_from(contract.as_clear_string()) {
+            self.id.eq(&extn_id)
         } else {
-            self.fulfills.contains(&contract.as_clear_string())
+            false
         }
     }
 

core/sdk/src/extn/extn_id.rs

@@ -15,7 +15,14 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
-use crate::utils::error::RippleError;
+use crate::{
+    framework::ripple_contract::{ContractAdjective, RippleContract},
+    utils::error::RippleError,
+};
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use serde_json::Value;
+
+use super::extn_client_message::{ExtnPayload, ExtnPayloadProvider, ExtnRequest, ExtnResponse};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ExtnClassId {
@@ -140,6 +147,37 @@ pub struct ExtnId {
     service: String,
 }
 
+impl Serialize for ExtnId {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&self.to_string())
+    }
+}
+
+impl<'de> Deserialize<'de> for ExtnId {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        if let Ok(str) = String::deserialize(deserializer) {
+            if let Ok(id) = ExtnId::try_from(str) {
+                return Ok(id);
+            }
+        }
+        Err(serde::de::Error::unknown_variant("unknown", &["unknown"]))
+    }
+}
+
+impl PartialEq for ExtnId {
+    fn eq(&self, other: &Self) -> bool {
+        self._type.eq(&other._type)
+            && self.class.eq(&other.class)
+            && self.service.eq(&other.service)
+    }
+}
+
 impl ToString for ExtnId {
     fn to_string(&self) -> String {
         let r = format!(
@@ -406,8 +444,99 @@ impl ExtnId {
     }
 }
 
-impl PartialEq for ExtnId {
-    fn eq(&self, other: &ExtnId) -> bool {
-        self._type == other._type && self.class == other.class
+#[derive(Debug, Clone, PartialEq)]
+pub struct ExtnProviderAdjective {
+    pub id: ExtnId,
+}
+
+impl Serialize for ExtnProviderAdjective {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&self.id.to_string())
+    }
+}
+
+impl<'de> Deserialize<'de> for ExtnProviderAdjective {
+    fn deserialize<D>(deserializer: D) -> Result<ExtnProviderAdjective, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        if let Ok(str) = String::deserialize(deserializer) {
+            if let Ok(id) = ExtnId::try_from(str) {
+                return Ok(ExtnProviderAdjective { id });
+            }
+        }
+        Err(serde::de::Error::unknown_variant("unknown", &["unknown"]))
+    }
+}
+
+impl ContractAdjective for ExtnProviderAdjective {
+    fn get_contract(&self) -> RippleContract {
+        RippleContract::ExtnProvider(self.clone())
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ExtnProviderRequest {
+    pub value: Value,
+    pub id: ExtnId,
+}
+
+impl ExtnPayloadProvider for ExtnProviderRequest {
+    fn get_from_payload(payload: ExtnPayload) -> Option<Self> {
+        if let ExtnPayload::Request(ExtnRequest::Extn(value)) = payload {
+            if let Ok(v) = serde_json::from_value::<ExtnProviderRequest>(value) {
+                return Some(v);
+            }
+        }
+
+        None
+    }
+
+    fn get_extn_payload(&self) -> ExtnPayload {
+        ExtnPayload::Request(ExtnRequest::Extn(
+            serde_json::to_value(self.clone()).unwrap(),
+        ))
+    }
+
+    fn contract() -> RippleContract {
+        // Will be replaced by the IEC before CExtnMessage conversion
+        RippleContract::ExtnProvider(ExtnProviderAdjective {
+            id: ExtnId::get_main_target("default".into()),
+        })
+    }
+
+    fn get_contract(&self) -> RippleContract {
+        RippleContract::ExtnProvider(ExtnProviderAdjective {
+            id: self.id.clone(),
+        })
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ExtnProviderResponse {
+    pub value: Value,
+}
+
+impl ExtnPayloadProvider for ExtnProviderResponse {
+    fn get_from_payload(payload: ExtnPayload) -> Option<Self> {
+        if let ExtnPayload::Response(ExtnResponse::Value(value)) = payload {
+            return Some(ExtnProviderResponse { value });
+        }
+
+        None
+    }
+
+    fn get_extn_payload(&self) -> ExtnPayload {
+        ExtnPayload::Response(ExtnResponse::Value(self.value.clone()))
+    }
+
+    fn contract() -> RippleContract {
+        // Will be replaced by the IEC before CExtnMessage conversion
+        RippleContract::ExtnProvider(ExtnProviderAdjective {
+            id: ExtnId::get_main_target("default".into()),
+        })
     }
 }

core/sdk/src/extn/ffi/ffi_library.rs

@@ -46,7 +46,7 @@ pub struct ExtnSymbolMetadata {
 }
 
 #[repr(C)]
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct CExtnMetadata {
     pub name: String,
     pub metadata: String,

core/sdk/src/framework/ripple_contract.rs

@@ -20,8 +20,10 @@ use crate::{
         session::{EventAdjective, PubSubAdjective, SessionAdjective},
         storage_property::StorageAdjective,
     },
+    extn::extn_id::ExtnProviderAdjective,
     utils::{error::RippleError, serde_utils::SerdeClearString},
 };
+use jsonrpsee_core::DeserializeOwned;
 use log::error;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
@@ -122,13 +124,14 @@ pub enum RippleContract {
     /// the Session information based on their policies. Used by [crate::api::session::AccountSession]
     Session(SessionAdjective),
     RippleContext,
+    ExtnProvider(ExtnProviderAdjective),
     AppCatalog,
     Apps,
     // Runtime ability for a given distributor to turn off a certian feature
     RemoteFeatureControl,
 }
 
-pub trait ContractAdjective: serde::ser::Serialize {
+pub trait ContractAdjective: serde::ser::Serialize + DeserializeOwned {
     fn as_string(&self) -> String {
         let adjective = SerdeClearString::as_clear_string(self);
         if let Some(contract) = self.get_contract().get_adjective_contract() {
@@ -185,13 +188,27 @@ impl RippleContract {
             Self::Session(adj) => Some(adj.as_string()),
             Self::PubSub(adj) => Some(adj.as_string()),
             Self::DeviceEvents(adj) => Some(adj.as_string()),
+            Self::ExtnProvider(adj) => Some(adj.id.to_string()),
             _ => None,
         }
     }
 
+    fn get_contract_from_adjective<T: ContractAdjective>(str: &str) -> Option<RippleContract> {
+        match serde_json::from_str::<T>(str) {
+            Ok(v) => Some(v.get_contract()),
+            Err(e) => {
+                error!("contract parser_error={:?}", e);
+                None
+            }
+        }
+    }
+
     pub fn from_adjective_string(contract: &str, adjective: &str) -> Option<Self> {
         let adjective = format!("\"{}\"", adjective);
         match contract {
+            "extn_provider" => {
+                return Self::get_contract_from_adjective::<ExtnProviderAdjective>(&adjective)
+            }
             "storage" => match serde_json::from_str::<StorageAdjective>(&adjective) {
                 Ok(v) => return Some(v.get_contract()),
                 Err(e) => error!("contract parser_error={:?}", e),
@@ -217,6 +234,7 @@ impl RippleContract {
         match self {
             Self::Storage(_) => Some("storage".to_owned()),
             Self::Session(_) => Some("session".to_owned()),
+            Self::ExtnProvider(_) => Some("extn_provider".to_owned()),
             Self::PubSub(_) => Some("pubsub".to_owned()),
             Self::DeviceEvents(_) => Some("device_events".to_owned()),
             _ => None,
@@ -238,6 +256,14 @@ impl RippleContract {
         }
         None
     }
+
+    pub fn is_extn_provider(&self) -> Option<String> {
+        if let RippleContract::ExtnProvider(e) = self {
+            Some(e.id.to_string())
+        } else {
+            None
+        }
+    }
 }
 
 #[derive(Debug, Clone)]

core/sdk/src/lib.rs

@@ -35,3 +35,7 @@ pub extern crate serde_json;
 pub extern crate serde_yaml;
 pub extern crate tokio;
 pub extern crate uuid;
+
+pub trait Mockable {
+    fn mock() -> Self;
+}

core/sdk/src/utils/mod.rs

@@ -20,6 +20,7 @@ pub mod error;
 pub mod extn_utils;
 pub mod logger;
 pub mod mock_utils;
+pub mod rpc_utils;
 pub mod serde_utils;
 pub mod test_utils;
 pub mod time_utils;

core/sdk/src/utils/rpc_utils.rs

@@ -0,0 +1,22 @@
+// Copyright 2023 Comcast Cable Communications Management, LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+use jsonrpsee_core::Error;
+
+pub fn rpc_err(msg: impl Into<String>) -> Error {
+    Error::Custom(msg.into())
+}