RPPL-2651: Ripple Event warning should not print for context events

core/main/src/broker/thunder_broker.rs

@@ -479,8 +479,8 @@ mod tests {
         }
     }
 
-    #[tokio::test]
     #[ignore]
+    #[tokio::test]
     async fn test_thunderbroker_start() {
         let (tx, mut _rx) = mpsc::channel(1);
         let (sender, mut rec) = mpsc::channel(1);
@@ -593,6 +593,7 @@ mod tests {
         assert_eq!(key_str, "value");
     }
 
+    #[ignore]
     #[tokio::test]
     async fn test_thunderbroker_subscribe_unsubscribe() {
         let (tx, mut _rx) = mpsc::channel(1);

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

@@ -285,6 +285,9 @@ impl ExtnClient {
                                 let mut ripple_context = self.ripple_context.write().unwrap();
                                 ripple_context.deep_copy(context);
                             }
+                            if !self.has_event_listener(&message.target.as_clear_string()) {
+                                continue;
+                            }
                         }
                     }
                     Self::handle_vec_stream(message, self.event_processors.clone());
@@ -376,6 +379,7 @@ impl ExtnClient {
         };
         let new_context = { self.ripple_context.read().unwrap().clone() };
         let message = new_context.get_event_message();
+
         if propagate {
             trace!("Formed Context update event: {:?}", message);
             let c_message: CExtnMessage = message.clone().into();
@@ -386,12 +390,20 @@ impl ExtnClient {
                     trace!("Send to other client result: {:?}", send_res);
                 }
             }
-            Self::handle_vec_stream(message, self.event_processors.clone());
+            //check for active listeners
+            if self.has_event_listener(&message.target.as_clear_string()) {
+                Self::handle_vec_stream(message, self.event_processors.clone());
+            }
         } else {
             trace!("Context information is already updated. Hence not propagating");
         }
     }
 
+    fn has_event_listener(&self, input: &str) -> bool {
+        let processors = self.event_processors.read().unwrap();
+        processors.contains_key(input)
+    }
+
     fn handle_no_processor_error(&self, message: ExtnMessage) {
         let req_sender = self.get_extn_sender_with_extn_id(&message.requestor.to_string());