Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Protocol interest #870

Merged
merged 5 commits into from
Mar 28, 2024
Merged

Protocol interest #870

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b13ff86
Add InterestId in Declare message
Mallets Mar 27, 2024
07fef43
Improve comments
Mallets Mar 28, 2024
444ad9e
Merge branch 'protocol_changes' into protocol_interest
Mallets Mar 28, 2024
b452cac
Update commons/zenoh-protocol/src/network/declare.rs
Mallets Mar 28, 2024
68199c3
Update commons/zenoh-protocol/src/network/declare.rs
Mallets Mar 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletion commons/zenoh-codec/src/network/declare.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,7 @@ where

fn write(self, writer: &mut W, x: &Declare) -> Self::Output {
let Declare {
interest_id,
ext_qos,
ext_tstamp,
ext_nodeid,
Expand All @@ -103,6 +104,9 @@ where

// Header
let mut header = id::DECLARE;
if x.interest_id.is_some() {
header |= declare::flag::I;
}
let mut n_exts = ((ext_qos != &declare::ext::QoSType::DEFAULT) as u8)
+ (ext_tstamp.is_some() as u8)
+ ((ext_nodeid != &declare::ext::NodeIdType::DEFAULT) as u8);
Expand All @@ -111,6 +115,11 @@ where
}
self.write(&mut *writer, header)?;

// Body
if let Some(interest_id) = interest_id {
self.write(&mut *writer, interest_id)?;
}

// Extensions
if ext_qos != &declare::ext::QoSType::DEFAULT {
n_exts -= 1;
Expand Down Expand Up @@ -157,6 +166,11 @@ where
return Err(DidntRead);
}

let mut interest_id = None;
if imsg::has_flag(self.header, declare::flag::I) {
interest_id = Some(self.codec.read(&mut *reader)?);
}

// Extensions
let mut ext_qos = declare::ext::QoSType::DEFAULT;
let mut ext_tstamp = None;
Expand Down Expand Up @@ -192,10 +206,11 @@ where
let body: DeclareBody = self.codec.read(&mut *reader)?;

Ok(Declare {
body,
interest_id,
ext_qos,
ext_tstamp,
ext_nodeid,
body,
})
}
}
Expand Down
17 changes: 11 additions & 6 deletions commons/zenoh-protocol/src/network/declare.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,27 +25,30 @@ pub use subscriber::*;
pub use token::*;

pub mod flag {
// pub const X: u8 = 1 << 5; // 0x20 Reserved
// pub const X: u8 = 1 << 6; // 0x40 Reserved
pub const I: u8 = 1 << 5; // 0x20 Interest if I==1 then the declare is in a response to an Interest with future==false
// pub const X: u8 = 1 << 6; // 0x40 Reserved
pub const Z: u8 = 1 << 7; // 0x80 Extensions if Z==1 then an extension will follow
}

/// Flags:
/// - X: Reserved
/// - I: Interest If I==1 then the declare is in a response to an Interest with future==false
/// - X: Reserved
/// - Z: Extension If Z==1 then at least one extension is present
///
/// 7 6 5 4 3 2 1 0
/// +-+-+-+-+-+-+-+-+
/// |Z|X|X| DECLARE |
/// |Z|X|I| DECLARE |
/// +-+-+-+---------+
/// ~interest_id:z32~ if I==1
/// +---------------+
/// ~ [decl_exts] ~ if Z==1
/// +---------------+
/// ~ declaration ~
/// +---------------+
///
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Declare {
pub interest_id: Option<InterestId>,
pub ext_qos: ext::QoSType,
pub ext_tstamp: Option<ext::TimestampType>,
pub ext_nodeid: ext::NodeIdType,
Expand Down Expand Up @@ -132,16 +135,18 @@ impl Declare {

let mut rng = rand::thread_rng();

let body = DeclareBody::rand();
let interest_id = rng.gen_bool(0.5).then_some(rng.gen::<InterestId>());
let ext_qos = ext::QoSType::rand();
let ext_tstamp = rng.gen_bool(0.5).then(ext::TimestampType::rand);
let ext_nodeid = ext::NodeIdType::rand();
let body = DeclareBody::rand();

Self {
body,
interest_id,
ext_qos,
ext_tstamp,
ext_nodeid,
body,
}
}
}
Expand Down
1 change: 1 addition & 0 deletions zenoh/src/key_expr.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -664,6 +664,7 @@ impl SyncResolve for KeyExprUndeclaration<'_> {
let primitives = state.primitives.as_ref().unwrap().clone();
drop(state);
primitives.send_declare(zenoh_protocol::network::Declare {
interest_id: None,
ext_qos: declare::ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: declare::ext::NodeIdType::DEFAULT,
Expand Down
1 change: 1 addition & 0 deletions zenoh/src/net/routing/dispatcher/resource.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -452,6 +452,7 @@ impl Resource {
.insert(expr_id, nonwild_prefix.clone());
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down
4 changes: 4 additions & 0 deletions zenoh/src/net/routing/hat/client/pubsub.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ fn propagate_simple_subscription_to(
let key_expr = Resource::decl_key(res, dst_face);
dst_face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -136,6 +137,7 @@ fn declare_client_subscription(
.primitives
.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -169,6 +171,7 @@ fn propagate_forget_simple_subscription(tables: &mut Tables, res: &Arc<Resource>
if let Some(id) = face_hat_mut!(face).local_subs.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -203,6 +206,7 @@ pub(super) fn undeclare_client_subscription(
if let Some(id) = face_hat_mut!(face).local_subs.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down
3 changes: 3 additions & 0 deletions zenoh/src/net/routing/hat/client/queries.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@ fn propagate_simple_queryable(
let key_expr = Resource::decl_key(res, &mut dst_face);
dst_face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -164,6 +165,7 @@ fn propagate_forget_simple_queryable(tables: &mut Tables, res: &mut Arc<Resource
if let Some((id, _)) = face_hat_mut!(face).local_qabls.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -203,6 +205,7 @@ pub(super) fn undeclare_client_queryable(
if let Some((id, _)) = face_hat_mut!(face).local_qabls.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down
6 changes: 6 additions & 0 deletions zenoh/src/net/routing/hat/linkstate_peer/pubsub.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ fn send_sourced_subscription_to_net_childs(

someface.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType {
Expand Down Expand Up @@ -95,6 +96,7 @@ fn propagate_simple_subscription_to(
let key_expr = Resource::decl_key(res, dst_face);
dst_face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -284,6 +286,7 @@ fn send_forget_sourced_subscription_to_net_childs(

someface.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType {
Expand All @@ -309,6 +312,7 @@ fn propagate_forget_simple_subscription(tables: &mut Tables, res: &Arc<Resource>
if let Some(id) = face_hat_mut!(face).local_subs.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -414,6 +418,7 @@ pub(super) fn undeclare_client_subscription(
if let Some(id) = face_hat_mut!(face).local_subs.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -455,6 +460,7 @@ pub(super) fn pubsub_new_face(tables: &mut Tables, face: &mut Arc<FaceState>) {
let key_expr = Resource::decl_key(sub, face);
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down
6 changes: 6 additions & 0 deletions zenoh/src/net/routing/hat/linkstate_peer/queries.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,7 @@ fn send_sourced_queryable_to_net_childs(

someface.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType {
Expand Down Expand Up @@ -169,6 +170,7 @@ fn propagate_simple_queryable(
let key_expr = Resource::decl_key(res, &mut dst_face);
dst_face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -337,6 +339,7 @@ fn send_forget_sourced_queryable_to_net_childs(

someface.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType {
Expand All @@ -362,6 +365,7 @@ fn propagate_forget_simple_queryable(tables: &mut Tables, res: &mut Arc<Resource
if let Some((id, _)) = face_hat_mut!(face).local_qabls.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -475,6 +479,7 @@ pub(super) fn undeclare_client_queryable(
if let Some((id, _)) = face_hat_mut!(face).local_qabls.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -515,6 +520,7 @@ pub(super) fn queries_new_face(tables: &mut Tables, face: &mut Arc<FaceState>) {
let key_expr = Resource::decl_key(qabl, face);
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down
4 changes: 4 additions & 0 deletions zenoh/src/net/routing/hat/p2p_peer/pubsub.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ fn propagate_simple_subscription_to(
let key_expr = Resource::decl_key(res, dst_face);
dst_face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -136,6 +137,7 @@ fn declare_client_subscription(
.primitives
.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -169,6 +171,7 @@ fn propagate_forget_simple_subscription(tables: &mut Tables, res: &Arc<Resource>
if let Some(id) = face_hat_mut!(face).local_subs.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -203,6 +206,7 @@ pub(super) fn undeclare_client_subscription(
if let Some(id) = face_hat_mut!(face).local_subs.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down
3 changes: 3 additions & 0 deletions zenoh/src/net/routing/hat/p2p_peer/queries.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@ fn propagate_simple_queryable(
let key_expr = Resource::decl_key(res, &mut dst_face);
dst_face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -164,6 +165,7 @@ fn propagate_forget_simple_queryable(tables: &mut Tables, res: &mut Arc<Resource
if let Some((id, _)) = face_hat_mut!(face).local_qabls.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down Expand Up @@ -203,6 +205,7 @@ pub(super) fn undeclare_client_queryable(
if let Some((id, _)) = face_hat_mut!(face).local_qabls.remove(res) {
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
Expand Down
Loading