From 2a5466bad77de0f964f1dabb9fc6ef33ef76e8d8 Mon Sep 17 00:00:00 2001 From: OlivierHecart Date: Mon, 13 Nov 2023 15:10:03 +0100 Subject: [PATCH] Move ingress/egress filters out of pubsub --- zenoh/src/net/routing/dispatcher/pubsub.rs | 4 +- zenoh/src/net/routing/dispatcher/queries.rs | 3 +- zenoh/src/net/routing/hat/mod.rs | 47 ++++++++++++++++++++- zenoh/src/net/routing/hat/pubsub.rs | 45 -------------------- 4 files changed, 49 insertions(+), 50 deletions(-) diff --git a/zenoh/src/net/routing/dispatcher/pubsub.rs b/zenoh/src/net/routing/dispatcher/pubsub.rs index 8dd3076b72..cd43a1ccab 100644 --- a/zenoh/src/net/routing/dispatcher/pubsub.rs +++ b/zenoh/src/net/routing/dispatcher/pubsub.rs @@ -15,8 +15,8 @@ use super::super::hat::pubsub::{compute_data_route, compute_data_routes, compute use super::face::FaceState; use super::resource::{DataRoutes, Direction, PullCaches, Resource}; use super::tables::{RoutingContext, RoutingExpr, Tables}; -use crate::net::routing::hat::map_routing_context; -use crate::net::routing::hat::pubsub::{compute_matching_pulls, egress_filter, ingress_filter}; +use crate::net::routing::hat::pubsub::compute_matching_pulls; +use crate::net::routing::hat::{egress_filter, ingress_filter, map_routing_context}; use std::sync::Arc; use std::sync::RwLock; use zenoh_core::zread; diff --git a/zenoh/src/net/routing/dispatcher/queries.rs b/zenoh/src/net/routing/dispatcher/queries.rs index 6ca3ed31c9..2a2a97e91a 100644 --- a/zenoh/src/net/routing/dispatcher/queries.rs +++ b/zenoh/src/net/routing/dispatcher/queries.rs @@ -19,8 +19,7 @@ use super::face::FaceState; use super::resource::{QueryRoute, QueryRoutes, QueryTargetQablSet, Resource}; use super::tables::RoutingContext; use super::tables::{RoutingExpr, Tables, TablesLock}; -use crate::net::routing::hat::pubsub::egress_filter; -use crate::net::routing::hat::pubsub::ingress_filter; +use crate::net::routing::hat::{egress_filter, ingress_filter}; use async_trait::async_trait; use std::collections::HashMap; use std::sync::{Arc, Weak}; diff --git a/zenoh/src/net/routing/hat/mod.rs b/zenoh/src/net/routing/hat/mod.rs index cf0fb26047..9eeca30c05 100644 --- a/zenoh/src/net/routing/hat/mod.rs +++ b/zenoh/src/net/routing/hat/mod.rs @@ -24,7 +24,7 @@ use self::{ }; use super::dispatcher::{ face::FaceState, - tables::{Resource, RoutingContext, Tables, TablesLock}, + tables::{Resource, RoutingContext, RoutingExpr, Tables, TablesLock}, }; use crate::{ hat, hat_mut, @@ -758,3 +758,48 @@ fn get_peer(tables: &Tables, face: &Arc, nodeid: RoutingContext) -> O } } } + +#[inline] +pub(crate) fn ingress_filter(tables: &Tables, face: &FaceState, expr: &mut RoutingExpr) -> bool { + tables.whatami != WhatAmI::Router + || face.whatami != WhatAmI::Peer + || hat!(tables).peers_net.is_none() + || tables.zid + == *hat!(tables).elect_router( + &tables.zid, + expr.full_expr(), + hat!(tables).get_router_links(face.zid), + ) +} + +#[inline] +pub(crate) fn egress_filter( + tables: &Tables, + src_face: &FaceState, + out_face: &Arc, + expr: &mut RoutingExpr, +) -> bool { + if src_face.id != out_face.id + && match (src_face.mcast_group.as_ref(), out_face.mcast_group.as_ref()) { + (Some(l), Some(r)) => l != r, + _ => true, + } + { + let dst_master = tables.whatami != WhatAmI::Router + || out_face.whatami != WhatAmI::Peer + || hat!(tables).peers_net.is_none() + || tables.zid + == *hat!(tables).elect_router( + &tables.zid, + expr.full_expr(), + hat!(tables).get_router_links(out_face.zid), + ); + + return dst_master + && (src_face.whatami != WhatAmI::Peer + || out_face.whatami != WhatAmI::Peer + || hat!(tables).full_net(WhatAmI::Peer) + || hat!(tables).failover_brokering(src_face.zid, out_face.zid)); + } + false +} diff --git a/zenoh/src/net/routing/hat/pubsub.rs b/zenoh/src/net/routing/hat/pubsub.rs index 76e82c7702..ea67068b9f 100644 --- a/zenoh/src/net/routing/hat/pubsub.rs +++ b/zenoh/src/net/routing/hat/pubsub.rs @@ -1608,48 +1608,3 @@ pub(crate) fn compute_data_routes(tables: &mut Tables, res: &mut Arc) res_mut.context_mut().matching_pulls = compute_matching_pulls(tables, &mut expr); } } - -#[inline] -pub(crate) fn ingress_filter(tables: &Tables, face: &FaceState, expr: &mut RoutingExpr) -> bool { - tables.whatami != WhatAmI::Router - || face.whatami != WhatAmI::Peer - || hat!(tables).peers_net.is_none() - || tables.zid - == *hat!(tables).elect_router( - &tables.zid, - expr.full_expr(), - hat!(tables).get_router_links(face.zid), - ) -} - -#[inline] -pub(crate) fn egress_filter( - tables: &Tables, - src_face: &FaceState, - out_face: &Arc, - expr: &mut RoutingExpr, -) -> bool { - if src_face.id != out_face.id - && match (src_face.mcast_group.as_ref(), out_face.mcast_group.as_ref()) { - (Some(l), Some(r)) => l != r, - _ => true, - } - { - let dst_master = tables.whatami != WhatAmI::Router - || out_face.whatami != WhatAmI::Peer - || hat!(tables).peers_net.is_none() - || tables.zid - == *hat!(tables).elect_router( - &tables.zid, - expr.full_expr(), - hat!(tables).get_router_links(out_face.zid), - ); - - return dst_master - && (src_face.whatami != WhatAmI::Peer - || out_face.whatami != WhatAmI::Peer - || hat!(tables).full_net(WhatAmI::Peer) - || hat!(tables).failover_brokering(src_face.zid, out_face.zid)); - } - false -}