Add Advanced Pub/Sub feature #1582
Conversation
|
|
||
| /// Change the history size for each resource. | ||
| #[zenoh_macros::unstable] | ||
| pub fn cache(mut self, config: CacheConfig) -> Self { |
There was a problem hiding this comment.
i would like having some config: impl Into<CacheConfig>, together with impl From<usize> for CacheConfig, but I assume that's not for the first version, is it?
| } | ||
| } | ||
|
|
||
| macro_rules! spawn_periodoic_queries { |
There was a problem hiding this comment.
Why a macro and not just a function?
| let session = conf.session.clone(); | ||
| let key_expr = key_expr.clone().into_owned(); | ||
|
|
||
| move |s: Sample| { |
There was a problem hiding this comment.
too much indentation, extract the callback in a function please
| global_pending_queries: u64, | ||
| sequenced_states: HashMap<EntityGlobalId, SourceState<u32>>, | ||
| timestamped_states: HashMap<ID, SourceState<Timestamp>>, | ||
| session: Session, |
There was a problem hiding this comment.
State should not have a Session but a WeakSession, otherwise, keeping a reference to an advanced subscriber prevent the session to be closed, contrary to a normal subscriber.
| let mut lock = zlock!(statesref); | ||
| let states = &mut *lock; | ||
| let source_id = s.source_info().source_id().cloned(); | ||
| let new = handle_sample(states, s); |
There was a problem hiding this comment.
Samples are handled with lock held. If there are multiple connections, i.e. P2P, it samples from different publishers to be handled in parallel, and it will block the transports.
Having only one big lock, while publishers are independently managed is a big issue.
| let session = conf.session.clone(); | ||
| let statesref = statesref.clone(); | ||
| let key_expr = key_expr.clone().into_owned(); | ||
| move |s: Sample| { |
| let key_expr = key_expr.clone().into_owned(); | ||
| move |s: Sample| { | ||
| if s.kind() == SampleKind::Put { | ||
| if let Ok(parsed) = ke_liveliness::parse(s.key_expr().as_keyexpr()) { |
|
Please do not forget to remove zenoh/default features from zenoh-ext Cargo.toml. |
This PR adds:
An AdvancedPublisher that can:
An AdvancedSubscriber that can:
This PR also updates the Connectivity Status and Events to be retrievable through an AdvancedSubscriber.
This PR marks the FetchingSubscriber and PublicationCache as
deprecated.