Refactoring

Author: fuzzypixelz

zenoh/src/net/routing/dispatcher/face.rs

@@ -298,6 +298,26 @@ impl Face {
             interest.rejection_token.cancel();
         }
     }
+
+    pub(crate) fn load_egress_interceptors(
+        &self,
+    ) -> Option<arc_swap::Guard<Arc<InterceptorsChain>>> {
+        self.state
+            .eg_interceptors
+            .as_ref()
+            .map(|i| i.load())
+            .and_then(|i| i.is_empty().not().then_some(i))
+    }
+
+    pub(crate) fn load_ingress_interceptors(
+        &self,
+    ) -> Option<arc_swap::Guard<Arc<InterceptorsChain>>> {
+        self.state
+            .in_interceptors
+            .as_ref()
+            .map(|i| i.load())
+            .and_then(|i| i.is_empty().not().then_some(i))
+    }
 }
 
 impl Primitives for Face {

zenoh/src/net/routing/dispatcher/pubsub.rs

@@ -14,7 +14,7 @@
 
 #[zenoh_macros::unstable]
 use std::collections::HashMap;
-use std::{ops::Not, sync::Arc};
+use std::sync::Arc;
 
 use zenoh_core::zread;
 use zenoh_protocol::{
@@ -33,7 +33,6 @@ use super::{
 use crate::key_expr::KeyExpr;
 use crate::net::routing::{
     hat::{HatTrait, SendDeclare},
-    interceptor::InterceptorTrait,
     router::get_or_set_route,
     RoutingContext,
 };
@@ -300,23 +299,15 @@ pub fn route_data(
                 msg.wire_expr.suffix.as_ref()
             );
 
-            if let Some(interceptor) = face
-                .state
-                .in_interceptors
-                .as_ref()
-                .map(|i| i.load())
-                .and_then(|i| i.is_empty().not().then_some(i))
-            {
-                let cache_guard = prefix.get_ingress_cache(face, &interceptor);
-                let cache = cache_guard.as_ref().and_then(|c| c.get_ref().as_ref());
+            if let Some(interceptor) = face.load_ingress_interceptors() {
                 let ctx = &mut RoutingContext::new(NetworkMessageMut {
                     body: NetworkBodyMut::Push(msg),
                     reliability,
                     #[cfg(feature = "stats")]
                     size: None,
                 });
 
-                if !interceptor.intercept(ctx, cache) {
+                if !interceptor.intercept_with_face(ctx, face, &prefix) {
                     return;
                 }
             };
@@ -365,24 +356,15 @@ pub fn route_data(
                             msg.wire_expr = key_expr.into();
                             msg.ext_nodeid = ext::NodeIdType { node_id: *context };
 
-                            if let Some(interceptor) = face
-                                .state
-                                .eg_interceptors
-                                .as_ref()
-                                .map(|i| i.load())
-                                .and_then(|i| i.is_empty().not().then_some(i))
-                            {
-                                let cache_guard = prefix.get_ingress_cache(face, &interceptor);
-                                let cache = cache_guard.as_ref().and_then(|c| c.get_ref().as_ref());
-
+                            if let Some(interceptor) = face.load_egress_interceptors() {
                                 let ctx = &mut RoutingContext::new(NetworkMessageMut {
                                     body: NetworkBodyMut::Push(msg),
                                     reliability,
                                     #[cfg(feature = "stats")]
                                     size: None,
                                 });
 
-                                if !interceptor.intercept(ctx, cache) {
+                                if !interceptor.intercept_with_face(ctx, face, &prefix) {
                                     return;
                                 }
                             };
@@ -420,24 +402,15 @@ pub fn route_data(
                                 payload: msg.payload.clone(),
                             };
 
-                            if let Some(interceptor) = face
-                                .state
-                                .eg_interceptors
-                                .as_ref()
-                                .map(|i| i.load())
-                                .and_then(|i| i.is_empty().not().then_some(i))
-                            {
-                                let cache_guard = prefix.get_ingress_cache(face, &interceptor);
-                                let cache = cache_guard.as_ref().and_then(|c| c.get_ref().as_ref());
-
+                            if let Some(interceptor) = face.load_egress_interceptors() {
                                 let ctx = &mut RoutingContext::new(NetworkMessageMut {
                                     body: NetworkBodyMut::Push(msg),
                                     reliability,
                                     #[cfg(feature = "stats")]
                                     size: None,
                                 });
 
-                                if !interceptor.intercept(ctx, cache) {
+                                if !interceptor.intercept_with_face(ctx, face, &prefix) {
                                     continue;
                                 }
                             };

zenoh/src/net/routing/interceptor/mod.rs

@@ -34,6 +34,8 @@ use zenoh_protocol::network::NetworkMessageMut;
 use zenoh_result::ZResult;
 use zenoh_transport::{multicast::TransportMulticast, unicast::TransportUnicast};
 
+use super::dispatcher::face::Face;
+use super::router::Resource;
 use super::RoutingContext;
 
 pub mod downsampling;
@@ -151,6 +153,18 @@ impl InterceptorsChain {
             version,
         }
     }
+
+    pub(crate) fn intercept_with_face(
+        &self,
+        ctx: &mut RoutingContext<NetworkMessageMut>,
+        face: &Face,
+        prefix: &Resource,
+    ) -> bool {
+        let cache_guard = prefix.get_ingress_cache(face, self);
+        let cache = cache_guard.as_ref().and_then(|c| c.get_ref().as_ref());
+
+        self.intercept(ctx, cache)
+    }
 }
 
 impl InterceptorTrait for InterceptorsChain {