Intercept during routing

commons/zenoh-protocol/src/network/declare.rs

@@ -148,6 +148,21 @@
             body,
         }
     }
+
+    #[inline]
+    pub fn wire_expr(&self) -> Option<&WireExpr> {
+        match &self.body {
+            DeclareBody::DeclareKeyExpr(m) => Some(&m.wire_expr),
+            DeclareBody::UndeclareKeyExpr(_) => None,
+            DeclareBody::DeclareSubscriber(m) => Some(&m.wire_expr),
+            DeclareBody::UndeclareSubscriber(m) => Some(&m.ext_wire_expr.wire_expr),
+            DeclareBody::DeclareQueryable(m) => Some(&m.wire_expr),
+            DeclareBody::UndeclareQueryable(m) => Some(&m.ext_wire_expr.wire_expr),
+            DeclareBody::DeclareToken(m) => Some(&m.wire_expr),
+            DeclareBody::UndeclareToken(m) => Some(&m.ext_wire_expr.wire_expr),
+            DeclareBody::DeclareFinal(_) => None,
+        }
+    }
 }
 
 pub mod common {

zenoh/src/net/primitives/demux.rs

@@ -11,104 +11,30 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
-use std::{any::Any, sync::Arc};
+use std::any::Any;
 
-use arc_swap::ArcSwap;
 use zenoh_link::Link;
-use zenoh_protocol::network::{
-    ext, response, Declare, DeclareBody, DeclareFinal, NetworkBodyMut, NetworkMessageMut,
-    ResponseFinal,
-};
+use zenoh_protocol::network::{NetworkBodyMut, NetworkMessageMut};
 use zenoh_result::ZResult;
 use zenoh_transport::{unicast::TransportUnicast, TransportPeerEventHandler};
 
 use super::Primitives;
-use crate::net::routing::{
-    dispatcher::face::Face,
-    interceptor::{InterceptorTrait, InterceptorsChain},
-    RoutingContext,
-};
+use crate::net::routing::dispatcher::face::Face;
 
 pub struct DeMux {
     face: Face,
     pub(crate) transport: Option<TransportUnicast>,
-    pub(crate) interceptor: Arc<ArcSwap<InterceptorsChain>>,
 }
 
 impl DeMux {
-    pub(crate) fn new(
-        face: Face,
-        transport: Option<TransportUnicast>,
-        interceptor: Arc<ArcSwap<InterceptorsChain>>,
-    ) -> Self {
-        Self {
-            face,
-            transport,
-            interceptor,
-        }
+    pub(crate) fn new(face: Face, transport: Option<TransportUnicast>) -> Self {
+        Self { face, transport }
     }
 }
 
 impl TransportPeerEventHandler for DeMux {
     #[inline]
-    fn handle_message(&self, mut msg: NetworkMessageMut) -> ZResult<()> {
-        let interceptor = self.interceptor.load();
-        if !interceptor.interceptors.is_empty() {
-            let mut ctx = RoutingContext::new_in(msg.as_mut(), self.face.clone());
-            let prefix = ctx
-                .wire_expr()
-                .and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
-                .flatten()
-                .cloned();
-            let cache_guard = prefix
-                .as_ref()
-                .and_then(|p| p.get_ingress_cache(&self.face, &interceptor));
-            let cache = cache_guard.as_ref().and_then(|c| c.get_ref().as_ref());
-
-            match &ctx.msg.body {
-                NetworkBodyMut::Request(request) => {
-                    let request_id = request.id;
-                    if !interceptor.intercept(&mut ctx, cache) {
-                        // request was blocked by an interceptor, we need to send response final to avoid timeout error
-                        self.face
-                            .state
-                            .primitives
-                            .send_response_final(&mut ResponseFinal {
-                                rid: request_id,
-                                ext_qos: response::ext::QoSType::RESPONSE_FINAL,
-                                ext_tstamp: None,
-                            });
-                        return Ok(());
-                    }
-                }
-                NetworkBodyMut::Interest(interest) => {
-                    let interest_id = interest.id;
-                    if !interceptor.intercept(&mut ctx, cache) {
-                        // request was blocked by an interceptor, we need to send declare final to avoid timeout error
-                        self.face
-                            .state
-                            .primitives
-                            .send_declare(RoutingContext::new_in(
-                                &mut Declare {
-                                    interest_id: Some(interest_id),
-                                    ext_qos: ext::QoSType::DECLARE,
-                                    ext_tstamp: None,
-                                    ext_nodeid: ext::NodeIdType::DEFAULT,
-                                    body: DeclareBody::DeclareFinal(DeclareFinal),
-                                },
-                                self.face.clone(),
-                            ));
-                        return Ok(());
-                    }
-                }
-                _ => {
-                    if !interceptor.intercept(&mut ctx, cache) {
-                        return Ok(());
-                    }
-                }
-            };
-        }
-
+    fn handle_message(&self, msg: NetworkMessageMut) -> ZResult<()> {
         match msg.body {
             NetworkBodyMut::Push(m) => self.face.send_push(m, msg.reliability),
             NetworkBodyMut::Declare(m) => self.face.send_declare(m),
@@ -126,12 +52,12 @@ impl TransportPeerEventHandler for DeMux {
                         &self.face.tables,
                         m,
                         transport,
-                        &mut |p, m| declares.push((p.clone(), m)),
+                        &mut |p, m, r| declares.push((p.clone(), m, r)),
                     )?;
                     drop(tables);
                     drop(ctrl_lock);
-                    for (p, m) in declares {
-                        m.with_mut(|m| p.send_declare(m));
+                    for (p, mut m, r) in declares {
+                        p.intercept_declare(&mut m, r.as_ref());
                     }
                 }
             }