Http ws nonjsonrpc support

core/main/src/broker/http_broker.rs

@@ -22,19 +22,22 @@ use ripple_sdk::{
 };
 
 use super::{
-    endpoint_broker::{BrokerCallback, BrokerSender, EndpointBroker},
+    endpoint_broker::{
+        BrokerCallback, BrokerCleaner, BrokerOutputForwarder, BrokerSender, EndpointBroker,
+    },
     rules_engine::RuleEndpoint,
 };
 
 pub struct HttpBroker {
     sender: BrokerSender,
+    cleaner: BrokerCleaner,
 }
 
 impl EndpointBroker for HttpBroker {
     fn get_broker(endpoint: RuleEndpoint, callback: BrokerCallback) -> Self {
         let (tx, mut tr) = mpsc::channel(10);
         let broker = BrokerSender { sender: tx };
-
+        let is_json_rpc = endpoint.jsonrpc;
         let uri: Uri = endpoint.url.parse().unwrap();
         // let mut headers = HeaderMap::new();
         // headers.insert("Content-Type", "application/json".parse().unwrap());
@@ -69,16 +72,33 @@ impl EndpointBroker for HttpBroker {
                         if let Ok(bytes) = hyper::body::to_bytes(body).await {
                             let value: Vec<u8> = bytes.into();
                             let value = value.as_slice();
-                            Self::handle_response(value, callback.clone());
+                            if is_json_rpc {
+                                Self::handle_jsonrpc_response(value, callback.clone());
+                            } else if let Err(e) =
+                                BrokerOutputForwarder::handle_non_jsonrpc_response(
+                                    value,
+                                    callback.clone(),
+                                    request.clone(),
+                                )
+                            {
+                                error!("Error forwarding {:?}", e)
+                            }
                         }
                     }
                 }
             }
         });
-        Self { sender: broker }
+        Self {
+            sender: broker,
+            cleaner: BrokerCleaner { cleaner: None },
+        }
     }
 
     fn get_sender(&self) -> BrokerSender {
         self.sender.clone()
     }
+
+    fn get_cleaner(&self) -> BrokerCleaner {
+        self.cleaner.clone()
+    }
 }

core/main/src/broker/rules_engine.rs

@@ -45,6 +45,12 @@ impl RuleSet {
 pub struct RuleEndpoint {
     pub protocol: RuleEndpointProtocol,
     pub url: String,
+    #[serde(default = "default_autostart")]
+    pub jsonrpc: bool,
+}
+
+fn default_autostart() -> bool {
+    true
 }
 
 #[derive(Deserialize, Debug, Clone)]

core/main/src/broker/thunder_broker.rs

@@ -17,7 +17,9 @@ use crate::utils::rpc_utils::extract_tcp_port;
 // SPDX-License-Identifier: Apache-2.0
 //
 use super::{
-    endpoint_broker::{BrokerCallback, BrokerOutput, BrokerRequest, BrokerSender, EndpointBroker},
+    endpoint_broker::{
+        BrokerCallback, BrokerCleaner, BrokerOutput, BrokerRequest, BrokerSender, EndpointBroker,
+    },
     rules_engine::RuleEndpoint,
 };
 use futures_util::{SinkExt, StreamExt};
@@ -40,18 +42,24 @@ use tokio_tungstenite::client_async;
 pub struct ThunderBroker {
     sender: BrokerSender,
     subscription_map: Arc<RwLock<HashMap<String, Vec<BrokerRequest>>>>,
+    cleaner: BrokerCleaner,
 }
 
 impl ThunderBroker {
     fn start(endpoint: RuleEndpoint, callback: BrokerCallback) -> Self {
         let (tx, mut tr) = mpsc::channel(10);
+        let (c_tx, mut c_tr) = mpsc::channel(2);
         let sender = BrokerSender { sender: tx };
         let subscription_map = Arc::new(RwLock::new(HashMap::new()));
         let broker = Self {
             sender,
             subscription_map,
+            cleaner: BrokerCleaner {
+                cleaner: Some(c_tx.clone()),
+            },
         };
         let broker_c = broker.clone();
+        let broker_for_cleanup = broker.clone();
         let callback_for_sender = callback.clone();
         tokio::spawn(async move {
             info!("Broker Endpoint url {}", endpoint.url);
@@ -88,7 +96,7 @@ impl ThunderBroker {
                             Ok(v) => {
                                 if let tokio_tungstenite::tungstenite::Message::Text(t) = v {
                                     // send the incoming text without context back to the sender
-                                    Self::handle_response(t.as_bytes(),callback.clone())
+                                    Self::handle_jsonrpc_response(t.as_bytes(),callback.clone())
                                 }
                             },
                             Err(e) => {
@@ -105,11 +113,27 @@ impl ThunderBroker {
                                 debug!("Sending request to broker {:?}", updated_request);
                                 for r in updated_request {
                                     let _feed = ws_tx.feed(tokio_tungstenite::tungstenite::Message::Text(r)).await;
-                                let _flush = ws_tx.flush().await;
+                                    let _flush = ws_tx.flush().await;
                                 }
                             }
                             Err(e) => callback_for_sender.send_error(request,e).await
                         }
+                    },
+                    Some(cleanup_request) = c_tr.recv() => {
+                        let value = {
+                            broker_for_cleanup.subscription_map.write().unwrap().remove(&cleanup_request)
+                        };
+                        if let Some(mut cleanup) = value {
+                            let sender = broker_for_cleanup.get_sender();
+                            while let Some(mut v) = cleanup.pop() {
+                                v.rpc = v.rpc.get_unsubscribe();
+                                if (sender.send(v).await).is_err() {
+                                    error!("Cleanup Error for {}",&cleanup_request);
+                                }
+                            }
+
+                        }
+
                     }
                 }
             }
@@ -178,6 +202,10 @@ impl EndpointBroker for ThunderBroker {
         self.sender.clone()
     }
 
+    fn get_cleaner(&self) -> BrokerCleaner {
+        self.cleaner.clone()
+    }
+
     fn prepare_request(
         &self,
         rpc_request: &super::endpoint_broker::BrokerRequest,
@@ -236,7 +264,7 @@ impl EndpointBroker for ThunderBroker {
 
     /// Default handler method for the broker to remove the context and send it back to the
     /// client for consumption
-    fn handle_response(result: &[u8], callback: BrokerCallback) {
+    fn handle_jsonrpc_response(result: &[u8], callback: BrokerCallback) {
         let mut final_result = Err(RippleError::ParseError);
         if let Ok(data) = serde_json::from_slice::<JsonRpcApiResponse>(result) {
             let updated_data = Self::update_response(&data);