zenoh_ext/
advanced_publisher.rs

1//
2// Copyright (c) 2023 ZettaScale Technology
3//
4// This program and the accompanying materials are made available under the
5// terms of the Eclipse Public License 2.0 which is available at
6// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
7// which is available at https://www.apache.org/licenses/LICENSE-2.0.
8//
9// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
10//
11// Contributors:
12//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
13//
14use std::{
15    fmt,
16    future::{IntoFuture, Ready},
17    sync::{
18        atomic::{AtomicU32, Ordering},
19        Arc,
20    },
21    time::Duration,
22};
23
24use zenoh::{
25    bytes::{Encoding, OptionZBytes, ZBytes},
26    internal::{
27        bail,
28        runtime::ZRuntime,
29        traits::{
30            EncodingBuilderTrait, QoSBuilderTrait, TimestampBuilderTrait,
31            TimestampInstrumentationBuilderTrait,
32        },
33        TerminatableTask,
34    },
35    key_expr::{keyexpr, KeyExpr},
36    liveliness::LivelinessToken,
37    pubsub::{
38        PublicationBuilder, PublicationBuilderDelete, PublicationBuilderPut, Publisher,
39        PublisherBuilder, PublisherUndeclaration,
40    },
41    qos::{CongestionControl, Priority, Reliability},
42    sample::{Locality, SourceInfo},
43    session::EntityGlobalId,
44    Resolvable, Resolve, Result as ZResult, Session, Wait, KE_ADV_PREFIX, KE_EMPTY,
45};
46use zenoh_macros::ke;
47
48use crate::{
49    advanced_cache::{AdvancedCache, AdvancedCacheBuilder, CacheConfig, KE_UHLC},
50    z_serialize,
51};
52
53pub(crate) static KE_PUB: &keyexpr = ke!("pub");
54
55#[derive(Debug, PartialEq)]
56#[zenoh_macros::unstable]
57pub(crate) enum Sequencing {
58    None,
59    Timestamp,
60    SequenceNumber,
61}
62
63/// Configuration for sample miss detection
64///
65/// Enabling [`sample_miss_detection`](crate::AdvancedPublisherBuilder::sample_miss_detection) in [`AdvancedPublisher`](crate::AdvancedPublisher)
66/// allows [`AdvancedSubscribers`](crate::AdvancedSubscriber) to detect missed samples
67/// through [`sample_miss_listener`](crate::AdvancedSubscriber::sample_miss_listener)
68/// and to recover missed samples through [`recovery`](crate::AdvancedSubscriberBuilder::recovery).
69#[zenoh_macros::unstable]
70#[derive(Debug, Default, Clone)]
71pub struct MissDetectionConfig {
72    pub(crate) state_publisher: Option<(Duration, bool)>,
73}
74
75#[zenoh_macros::unstable]
76impl MissDetectionConfig {
77    /// Allow last sample miss detection through periodic heartbeat.
78    ///
79    /// Periodically send the last published sample's sequence number to allow last sample recovery.
80    ///
81    /// [`heartbeat`](MissDetectionConfig::heartbeat) and [`sporadic_heartbeat`](MissDetectionConfig::sporadic_heartbeat)
82    /// are mutually exclusive. Enabling one will disable the other.
83    ///
84    /// [`AdvancedSubscribers`](crate::AdvancedSubscriber) can recover last sample with the
85    /// [`heartbeat`](crate::advanced_subscriber::RecoveryConfig::heartbeat) option.
86    #[zenoh_macros::unstable]
87    pub fn heartbeat(mut self, period: Duration) -> Self {
88        self.state_publisher = Some((period, false));
89        self
90    }
91
92    /// Allow last sample miss detection through sporadic heartbeat.
93    ///
94    /// Each period, the last published sample's sequence number is sent with [`CongestionControl::Block`]
95    /// but only if it changed since the last period.
96    ///
97    /// [`heartbeat`](MissDetectionConfig::heartbeat) and [`sporadic_heartbeat`](MissDetectionConfig::sporadic_heartbeat)
98    /// are mutually exclusive. Enabling one will disable the other.
99    ///
100    /// [`AdvancedSubscribers`](crate::AdvancedSubscriber) can recover last sample with the
101    /// [`heartbeat`](crate::advanced_subscriber::RecoveryConfig::heartbeat) option.
102    #[zenoh_macros::unstable]
103    pub fn sporadic_heartbeat(mut self, period: Duration) -> Self {
104        self.state_publisher = Some((period, true));
105        self
106    }
107}
108
109/// The builder of AdvancedPublisher, allowing to configure it.
110#[must_use = "Resolvables do nothing unless you resolve them using the `res` method from either `SyncResolve` or `AsyncResolve`"]
111#[zenoh_macros::unstable]
112pub struct AdvancedPublisherBuilder<'a, 'b, 'c> {
113    session: &'a Session,
114    pub_key_expr: ZResult<KeyExpr<'b>>,
115    encoding: Encoding,
116    destination: Locality,
117    reliability: Reliability,
118    congestion_control: CongestionControl,
119    priority: Priority,
120    is_express: bool,
121    meta_key_expr: Option<ZResult<KeyExpr<'c>>>,
122    sequencing: Sequencing,
123    miss_config: Option<MissDetectionConfig>,
124    liveliness: bool,
125    history: Option<CacheConfig>,
126}
127
128#[zenoh_macros::unstable]
129impl fmt::Debug for AdvancedPublisherBuilder<'_, '_, '_> {
130    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
131        f.debug_struct("AdvancedPublisherBuilder")
132            .field("session", &"..")
133            .field("pub_key_expr", &self.pub_key_expr)
134            .field("encoding", &self.encoding)
135            .field("destination", &self.destination)
136            .field("reliability", &self.reliability)
137            .field("congestion_control", &self.congestion_control)
138            .field("priority", &self.priority)
139            .field("is_express", &self.is_express)
140            .field("meta_key_expr", &self.meta_key_expr)
141            .field("sequencing", &self.sequencing)
142            .field("miss_config", &self.miss_config)
143            .field("liveliness", &self.liveliness)
144            .field("history", &self.history)
145            .finish()
146    }
147}
148
149#[zenoh_macros::unstable]
150impl<'a, 'b, 'c> AdvancedPublisherBuilder<'a, 'b, 'c> {
151    #[zenoh_macros::unstable]
152    pub(crate) fn new(builder: PublisherBuilder<'a, 'b>) -> AdvancedPublisherBuilder<'a, 'b, 'c> {
153        AdvancedPublisherBuilder {
154            session: builder.session,
155            pub_key_expr: builder.key_expr,
156            encoding: builder.encoding,
157            destination: builder.destination,
158            reliability: builder.reliability,
159            congestion_control: builder.congestion_control,
160            priority: builder.priority,
161            is_express: builder.is_express,
162            meta_key_expr: None,
163            sequencing: Sequencing::None,
164            miss_config: None,
165            liveliness: false,
166            history: None,
167        }
168    }
169
170    /// Changes the [`zenoh::sample::Locality`] applied when routing the data.
171    ///
172    /// This restricts the matching subscribers that will receive the published data to the ones
173    /// that have the given [`zenoh::sample::Locality`].
174    #[zenoh_macros::unstable]
175    #[inline]
176    pub fn allowed_destination(mut self, destination: Locality) -> Self {
177        self.destination = destination;
178        self
179    }
180
181    /// Changes the [`zenoh::qos::Reliability`] to apply when routing the data.
182    ///
183    /// **NOTE**: Currently `reliability` does not trigger any data retransmission on the wire. It
184    ///   is rather used as a marker on the wire and it may be used to select the best link
185    ///   available (e.g. TCP for reliable data and UDP for best effort data).
186    #[zenoh_macros::unstable]
187    #[inline]
188    pub fn reliability(self, reliability: Reliability) -> Self {
189        Self {
190            reliability,
191            ..self
192        }
193    }
194
195    /// Allow matching [`AdvancedSubscribers`](crate::AdvancedSubscriber) to detect lost samples and optionally ask for retransmission.
196    ///
197    /// Retransmission can only be achieved if [`cache`](crate::AdvancedPublisherBuilder::cache) is enabled.
198    #[zenoh_macros::unstable]
199    pub fn sample_miss_detection(mut self, config: MissDetectionConfig) -> Self {
200        self.sequencing = Sequencing::SequenceNumber;
201        self.miss_config = Some(config);
202        self
203    }
204
205    /// Attach a cache to this [`AdvancedPublisher`].
206    ///
207    /// The cache can be used for history and/or recovery.
208    #[zenoh_macros::unstable]
209    pub fn cache(mut self, config: CacheConfig) -> Self {
210        if self.sequencing == Sequencing::None {
211            self.sequencing = Sequencing::Timestamp;
212        }
213        self.history = Some(config);
214        self
215    }
216
217    /// Allow this [`AdvancedPublisher`] to be detected by [`AdvancedSubscribers`](crate::AdvancedSubscriber).
218    ///
219    /// This allows [`AdvancedSubscribers`](crate::AdvancedSubscriber) to retrieve the local history.
220    #[zenoh_macros::unstable]
221    pub fn publisher_detection(mut self) -> Self {
222        self.liveliness = true;
223        self
224    }
225
226    /// A key expression added to the liveliness token key expression and to the cache queryable key expression.
227    ///
228    /// It can be used to convey meta data.
229    #[zenoh_macros::unstable]
230    pub fn publisher_detection_metadata<TryIntoKeyExpr>(mut self, meta: TryIntoKeyExpr) -> Self
231    where
232        TryIntoKeyExpr: TryInto<KeyExpr<'c>>,
233        <TryIntoKeyExpr as TryInto<KeyExpr<'c>>>::Error: Into<zenoh::Error>,
234    {
235        self.meta_key_expr = Some(meta.try_into().map_err(Into::into));
236        self
237    }
238}
239
240#[zenoh_macros::internal_trait]
241#[zenoh_macros::unstable]
242impl EncodingBuilderTrait for AdvancedPublisherBuilder<'_, '_, '_> {
243    /// Set the [`Encoding`]
244    #[zenoh_macros::unstable]
245    fn encoding<T: Into<Encoding>>(self, encoding: T) -> Self {
246        Self {
247            encoding: encoding.into(),
248            ..self
249        }
250    }
251}
252
253#[zenoh_macros::internal_trait]
254#[zenoh_macros::unstable]
255impl QoSBuilderTrait for AdvancedPublisherBuilder<'_, '_, '_> {
256    /// Changes the [`CongestionControl`] to apply when routing the data.
257    #[inline]
258    #[zenoh_macros::unstable]
259    fn congestion_control(self, congestion_control: CongestionControl) -> Self {
260        Self {
261            congestion_control,
262            ..self
263        }
264    }
265
266    /// Changes the [`Priority`] to apply when routing the data.
267    #[inline]
268    #[zenoh_macros::unstable]
269    fn priority(self, priority: Priority) -> Self {
270        Self { priority, ..self }
271    }
272
273    /// Changes the Express policy to apply when routing the data.
274    ///
275    /// When express is set to `true`, then the message will not be batched.
276    /// This usually has a positive impact on latency but negative impact on throughput.
277    #[inline]
278    #[zenoh_macros::unstable]
279    fn express(self, is_express: bool) -> Self {
280        Self { is_express, ..self }
281    }
282}
283
284#[zenoh_macros::unstable]
285impl<'b> Resolvable for AdvancedPublisherBuilder<'_, 'b, '_> {
286    type To = ZResult<AdvancedPublisher<'b>>;
287}
288
289#[zenoh_macros::unstable]
290impl Wait for AdvancedPublisherBuilder<'_, '_, '_> {
291    #[zenoh_macros::unstable]
292    fn wait(self) -> <Self as Resolvable>::To {
293        AdvancedPublisher::new(self)
294    }
295}
296
297#[zenoh_macros::unstable]
298impl IntoFuture for AdvancedPublisherBuilder<'_, '_, '_> {
299    type Output = <Self as Resolvable>::To;
300    type IntoFuture = Ready<<Self as Resolvable>::To>;
301
302    #[zenoh_macros::unstable]
303    fn into_future(self) -> Self::IntoFuture {
304        std::future::ready(self.wait())
305    }
306}
307
308/// The extension to a [`Publisher`](zenoh::pubsub::Publisher) providing advanced functionalities.
309///
310/// The `AdvancedPublisher` is constructed over a regular [`Publisher`](zenoh::pubsub::Publisher) through
311/// [`advanced`](crate::AdvancedPublisherBuilderExt::advanced) method or by using
312/// any other method of [`AdvancedPublisherBuilder`](crate::AdvancedPublisherBuilder).
313///
314/// The `AdvancedPublisher` works with [`AdvancedSubscriber`](crate::AdvancedSubscriber) to provide additional functionalities such as:
315/// - [`cache`](crate::AdvancedPublisherBuilderExt::cache) last published samples to be retrieved by
316///   [`AdvancedSubscriber`](crate::AdvancedSubscriber)'s [`history`](crate::AdvancedSubscriberBuilderExt::history) mechanism
317/// - [`sample_miss_detection`](crate::AdvancedPublisherBuilderExt::sample_miss_detection) to allow detecting missed samples
318///   using [`AdvancedSubscriber`](crate::AdvancedSubscriber)'s [`sample_miss_listener`](crate::AdvancedSubscriber::sample_miss_listener)
319/// - [`publisher_detection`](crate::AdvancedPublisherBuilderExt::publisher_detection) to create a Liveliness token to assert its presence and
320///   allow it to be requested for missed samples if [`detect_late_publishers`](crate::HistoryConfig::detect_late_publishers) is enabled
321///
322/// # Example
323/// ```no_run
324/// # #[tokio::main]
325/// # async fn main() {
326/// use zenoh_ext::{AdvancedPublisherBuilderExt, CacheConfig, MissDetectionConfig};
327/// let session = zenoh::open(zenoh::Config::default()).await.unwrap();
328/// let publisher = session
329///     .declare_publisher("key/expression")
330///     .cache(CacheConfig::default().max_samples(10))
331///     .sample_miss_detection(
332///         MissDetectionConfig::default().heartbeat(std::time::Duration::from_secs(1))
333///     )
334///     .publisher_detection()
335///     .await
336///     .unwrap();
337/// publisher.put("Value").await.unwrap();
338/// # }
339/// ```
340#[zenoh_macros::unstable]
341pub struct AdvancedPublisher<'a> {
342    publisher: Publisher<'a>,
343    seqnum: Option<Arc<AtomicU32>>,
344    cache: Option<AdvancedCache>,
345    _token: Option<LivelinessToken>,
346    _state_publisher: Option<TerminatableTask>,
347}
348
349#[zenoh_macros::unstable]
350impl fmt::Debug for AdvancedPublisher<'_> {
351    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
352        f.debug_struct("AdvancedPublisher")
353            .field("publisher", &self.publisher)
354            .field(
355                "seqnum",
356                &self
357                    .seqnum
358                    .as_ref()
359                    .map(|seqnum| seqnum.load(Ordering::Relaxed)),
360            )
361            .field("cache", &self.cache.as_ref().map(|_| ".."))
362            .field("token", &self._token.as_ref().map(|_| ".."))
363            .field("state_publisher", &self._state_publisher)
364            .finish()
365    }
366}
367
368#[zenoh_macros::unstable]
369impl<'a> AdvancedPublisher<'a> {
370    #[zenoh_macros::unstable]
371    fn new(conf: AdvancedPublisherBuilder<'_, 'a, '_>) -> ZResult<Self> {
372        let key_expr = conf.pub_key_expr?;
373        let meta = match conf.meta_key_expr {
374            Some(meta) => Some(meta?),
375            None => None,
376        };
377        tracing::debug!("Create AdvancedPublisher{{key_expr: {}}}", &key_expr);
378
379        let publisher = conf
380            .session
381            .declare_publisher(key_expr.clone())
382            .encoding(conf.encoding)
383            .allowed_destination(conf.destination)
384            .reliability(conf.reliability)
385            .congestion_control(conf.congestion_control)
386            .priority(conf.priority)
387            .express(conf.is_express)
388            .wait()?;
389        let id = publisher.id();
390        let suffix = KE_ADV_PREFIX / KE_PUB / &id.zid().into_keyexpr();
391        let suffix = match conf.sequencing {
392            Sequencing::SequenceNumber => {
393                suffix / &KeyExpr::try_from(id.eid().to_string()).unwrap()
394            }
395            _ => suffix / KE_UHLC,
396        };
397        let suffix = match meta {
398            Some(meta) => suffix / &meta,
399            // We need this empty chunk because of a routing matching bug
400            _ => suffix / KE_EMPTY,
401        };
402
403        let seqnum = match conf.sequencing {
404            Sequencing::SequenceNumber => Some(Arc::new(AtomicU32::new(0))),
405            Sequencing::Timestamp => {
406                if conf.session.hlc().is_none() {
407                    bail!(
408                        "Cannot create AdvancedPublisher {} with Sequencing::Timestamp: \
409                            the 'timestamping' setting must be enabled in the Zenoh configuration.",
410                        key_expr,
411                    )
412                }
413                None
414            }
415            _ => None,
416        };
417
418        let cache = conf
419            .history
420            .map(|h| {
421                AdvancedCacheBuilder::new(conf.session, Ok(key_expr.clone()))
422                    .history(h)
423                    .queryable_suffix(&suffix)
424                    .wait()
425            })
426            .transpose()?;
427
428        let token = if conf.liveliness {
429            tracing::debug!(
430                "AdvancedPublisher{{key_expr: {}}}: Declare liveliness token {}",
431                key_expr,
432                &key_expr / &suffix,
433            );
434            Some(
435                conf.session
436                    .liveliness()
437                    .declare_token(&key_expr / &suffix)
438                    .wait()?,
439            )
440        } else {
441            None
442        };
443
444        let state_publisher = if let Some((period, sporadic)) =
445            conf.miss_config.as_ref().and_then(|c| c.state_publisher)
446        {
447            if let Some(seqnum) = seqnum.as_ref() {
448                tracing::debug!(
449                    "AdvancedPublisher{{key_expr: {}}}: Enable {}heartbeat on {} with period {:?}",
450                    key_expr,
451                    if sporadic { "sporadic " } else { "" },
452                    &key_expr / &suffix,
453                    period
454                );
455                let seqnum = seqnum.clone();
456                if !sporadic {
457                    let publisher = conf.session.declare_publisher(&key_expr / &suffix).wait()?;
458                    Some(TerminatableTask::spawn_abortable(
459                        ZRuntime::Net,
460                        async move {
461                            loop {
462                                tokio::time::sleep(period).await;
463                                let seqnum = seqnum.load(Ordering::Relaxed);
464                                if seqnum > 0 {
465                                    let _ = publisher.put(z_serialize(&(seqnum - 1))).await;
466                                }
467                            }
468                        },
469                    ))
470                } else {
471                    let mut last_seqnum = 0;
472                    let publisher = conf
473                        .session
474                        .declare_publisher(&key_expr / &suffix)
475                        .congestion_control(CongestionControl::Block)
476                        .wait()?;
477                    Some(TerminatableTask::spawn_abortable(
478                        ZRuntime::Net,
479                        async move {
480                            loop {
481                                tokio::time::sleep(period).await;
482                                let seqnum = seqnum.load(Ordering::Relaxed);
483                                if seqnum > last_seqnum {
484                                    let _ = publisher.put(z_serialize(&(seqnum - 1))).await;
485                                    last_seqnum = seqnum;
486                                }
487                            }
488                        },
489                    ))
490                }
491            } else {
492                None
493            }
494        } else {
495            None
496        };
497
498        Ok(AdvancedPublisher {
499            publisher,
500            seqnum,
501            cache,
502            _token: token,
503            _state_publisher: state_publisher,
504        })
505    }
506
507    /// Returns the [`EntityGlobalId`] of this Publisher.
508    ///
509    /// Wraps [`Publisher::id`](zenoh::pubsub::Publisher::id) method
510    #[zenoh_macros::unstable]
511    pub fn id(&self) -> EntityGlobalId {
512        self.publisher.id()
513    }
514
515    /// Returns the [`KeyExpr`] of this Publisher.
516    ///
517    /// Wraps [`Publisher::key_expr`](zenoh::pubsub::Publisher::key_expr) method
518    #[inline]
519    #[zenoh_macros::unstable]
520    pub fn key_expr(&self) -> &KeyExpr<'a> {
521        self.publisher.key_expr()
522    }
523
524    /// Get the [`Encoding`] used when publishing data.
525    ///
526    /// Wraps [`Publisher::encoding`](zenoh::pubsub::Publisher::encoding) method
527    #[inline]
528    #[zenoh_macros::unstable]
529    pub fn encoding(&self) -> &Encoding {
530        self.publisher.encoding()
531    }
532
533    /// Get the `congestion_control` applied when routing the data.
534    ///
535    /// Wraps [`Publisher::congestion_control`](zenoh::pubsub::Publisher::congestion_control) method
536    #[inline]
537    #[zenoh_macros::unstable]
538    pub fn congestion_control(&self) -> CongestionControl {
539        self.publisher.congestion_control()
540    }
541
542    /// Get the priority of the written data.
543    ///
544    /// Wraps [`Publisher::priority`](zenoh::pubsub::Publisher::priority) method
545    #[inline]
546    #[zenoh_macros::unstable]
547    pub fn priority(&self) -> Priority {
548        self.publisher.priority()
549    }
550
551    /// Put data.
552    ///
553    /// Wraps [`Publisher::put`](zenoh::pubsub::Publisher::put) method
554    ///
555    /// # Examples
556    /// ```
557    /// # #[tokio::main]
558    /// # async fn main() {
559    /// use zenoh_ext::AdvancedPublisherBuilderExt;
560    ///
561    /// let session = zenoh::open(zenoh::Config::default()).await.unwrap();
562    /// let publisher = session.declare_publisher("key/expression").advanced().await.unwrap();
563    /// publisher.put("value").await.unwrap();
564    /// # }
565    /// ```
566    #[inline]
567    #[zenoh_macros::unstable]
568    pub fn put<IntoZBytes>(&self, payload: IntoZBytes) -> AdvancedPublisherPutBuilder<'_>
569    where
570        IntoZBytes: Into<ZBytes>,
571    {
572        let mut builder = self.publisher.put(payload);
573        if let Some(seqnum) = &self.seqnum {
574            let info = Some(SourceInfo::new(
575                self.publisher.id(),
576                seqnum.fetch_add(1, Ordering::Relaxed),
577            ));
578            tracing::trace!(
579                "AdvancedPublisher{{key_expr: {}}}: Put data with {:?}",
580                self.publisher.key_expr(),
581                info
582            );
583            builder = builder.source_info(info);
584        }
585        if let Some(hlc) = self.publisher.session().hlc() {
586            builder = builder.timestamp(hlc.new_timestamp());
587        }
588        AdvancedPublisherPutBuilder {
589            builder,
590            cache: self.cache.as_ref(),
591        }
592    }
593
594    /// Delete data.
595    ///
596    /// Wraps [`Publisher::delete`](zenoh::pubsub::Publisher::delete) method
597    ///
598    /// # Examples
599    /// ```
600    /// # #[tokio::main]
601    /// # async fn main() {
602    /// use zenoh_ext::AdvancedPublisherBuilderExt;
603    ///
604    /// let session = zenoh::open(zenoh::Config::default()).await.unwrap();
605    /// let publisher = session.declare_publisher("key/expression").advanced().await.unwrap();
606    /// publisher.delete().await.unwrap();
607    /// # }
608    /// ```
609    #[zenoh_macros::unstable]
610    pub fn delete(&self) -> AdvancedPublisherDeleteBuilder<'_> {
611        let mut builder = self.publisher.delete();
612        if let Some(seqnum) = &self.seqnum {
613            builder = builder.source_info(Some(SourceInfo::new(
614                self.publisher.id(),
615                seqnum.fetch_add(1, Ordering::Relaxed),
616            )));
617        }
618        if let Some(hlc) = self.publisher.session().hlc() {
619            builder = builder.timestamp(hlc.new_timestamp());
620        }
621        AdvancedPublisherDeleteBuilder {
622            builder,
623            cache: self.cache.as_ref(),
624        }
625    }
626
627    /// Return the [`MatchingStatus`](zenoh::matching::MatchingStatus) of the publisher.
628    ///
629    /// Wraps [`Publisher::matching_status`](zenoh::pubsub::Publisher::matching_status) method.
630    ///
631    /// [`MatchingStatus::matching`](zenoh::matching::MatchingStatus::matching)
632    /// will return true if there exist Subscribers matching the Publisher's key expression and false otherwise.
633    ///
634    /// # Examples
635    /// ```
636    /// # #[tokio::main]
637    /// # async fn main() {
638    /// use zenoh_ext::AdvancedPublisherBuilderExt;
639    ///
640    /// let session = zenoh::open(zenoh::Config::default()).await.unwrap();
641    /// let publisher = session.declare_publisher("key/expression").advanced().await.unwrap();
642    /// let matching_subscribers: bool = publisher
643    ///     .matching_status()
644    ///     .await
645    ///     .unwrap()
646    ///     .matching();
647    /// # }
648    /// ```
649    #[zenoh_macros::unstable]
650    pub fn matching_status(&self) -> impl Resolve<ZResult<zenoh::matching::MatchingStatus>> + '_ {
651        self.publisher.matching_status()
652    }
653
654    /// Return a [`MatchingListener`](zenoh::matching::MatchingStatus) for this Publisher.
655    ///
656    /// Wraps [`Publisher::matching_listener`](zenoh::pubsub::Publisher::matching_listener) method.
657    ///
658    /// The [`MatchingListener`](zenoh::matching::MatchingStatus) that will send a notification each time
659    /// the [`MatchingStatus`](zenoh::matching::MatchingStatus) of the Publisher changes.
660    ///
661    /// # Examples
662    /// ```no_run
663    /// # #[tokio::main]
664    /// # async fn main() {
665    /// use zenoh_ext::AdvancedPublisherBuilderExt;
666    ///
667    /// let session = zenoh::open(zenoh::Config::default()).await.unwrap();
668    /// let publisher = session.declare_publisher("key/expression").advanced().await.unwrap();
669    /// let matching_listener = publisher.matching_listener().await.unwrap();
670    /// while let Ok(matching_status) = matching_listener.recv_async().await {
671    ///     if matching_status.matching() {
672    ///         println!("Publisher has matching subscribers.");
673    ///     } else {
674    ///         println!("Publisher has NO MORE matching subscribers.");
675    ///     }
676    /// }
677    /// # }
678    /// ```
679    #[zenoh_macros::unstable]
680    pub fn matching_listener(
681        &self,
682    ) -> zenoh::matching::MatchingListenerBuilder<'_, zenoh::handlers::DefaultHandler> {
683        self.publisher.matching_listener()
684    }
685
686    /// Undeclares the [`Publisher`], informing the network that it needn't optimize publications for its key expression anymore.
687    ///
688    /// Wraps [`Publisher::undeclare`](zenoh::pubsub::Publisher::undeclare) method
689    ///
690    /// # Examples
691    /// ```
692    /// # #[tokio::main]
693    /// # async fn main() {
694    /// use zenoh_ext::AdvancedPublisherBuilderExt;
695    ///
696    /// let session = zenoh::open(zenoh::Config::default()).await.unwrap();
697    /// let publisher = session.declare_publisher("key/expression").advanced().await.unwrap();
698    /// publisher.undeclare().await.unwrap();
699    /// # }
700    /// ```
701    #[zenoh_macros::unstable]
702    pub fn undeclare(self) -> PublisherUndeclaration<'a> {
703        tracing::debug!(
704            "AdvancedPublisher{{key_expr: {}}}: Undeclare",
705            self.key_expr()
706        );
707        self.publisher.undeclare()
708    }
709}
710
711#[zenoh_macros::unstable]
712pub type AdvancedPublisherPutBuilder<'a> = AdvancedPublicationBuilder<'a, PublicationBuilderPut>;
713#[zenoh_macros::unstable]
714pub type AdvancedPublisherDeleteBuilder<'a> =
715    AdvancedPublicationBuilder<'a, PublicationBuilderDelete>;
716
717#[must_use = "Resolvables do nothing unless you resolve them using `.await` or `zenoh::Wait::wait`"]
718#[derive(Clone)]
719#[zenoh_macros::unstable]
720pub struct AdvancedPublicationBuilder<'a, P> {
721    pub(crate) builder: PublicationBuilder<&'a Publisher<'a>, P>,
722    pub(crate) cache: Option<&'a AdvancedCache>,
723}
724
725#[zenoh_macros::unstable]
726impl<P> fmt::Debug for AdvancedPublicationBuilder<'_, P> {
727    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
728        f.debug_struct("AdvancedPublicationBuilder")
729            .field("builder", &"..")
730            .field("has_cache", &self.cache.is_some())
731            .finish()
732    }
733}
734
735#[zenoh_macros::internal_trait]
736#[zenoh_macros::unstable]
737impl EncodingBuilderTrait for AdvancedPublicationBuilder<'_, PublicationBuilderPut> {
738    /// Set the [`Encoding`]
739    #[zenoh_macros::unstable]
740    fn encoding<T: Into<Encoding>>(self, encoding: T) -> Self {
741        Self {
742            builder: self.builder.encoding(encoding),
743            ..self
744        }
745    }
746}
747
748#[zenoh_macros::unstable]
749impl<P> AdvancedPublicationBuilder<'_, P> {
750    /// Sets an optional attachment to be sent along with the publication.
751    ///
752    /// The argument is converted via [`OptionZBytes`], which supports both `T: Into<ZBytes>`
753    /// and `Option<T>` where `T: Into<ZBytes>`.
754    pub fn attachment<TA: Into<OptionZBytes>>(self, attachment: TA) -> Self {
755        let attachment: OptionZBytes = attachment.into();
756        Self {
757            builder: self.builder.attachment(attachment),
758            ..self
759        }
760    }
761}
762
763#[zenoh_macros::internal_trait]
764#[zenoh_macros::unstable]
765impl<P> TimestampBuilderTrait for AdvancedPublicationBuilder<'_, P> {
766    /// Sets an optional timestamp to be sent along with the publication.
767    #[zenoh_macros::unstable]
768    fn timestamp<TS: Into<Option<uhlc::Timestamp>>>(self, timestamp: TS) -> Self {
769        Self {
770            builder: self.builder.timestamp(timestamp),
771            ..self
772        }
773    }
774}
775
776#[zenoh_macros::internal_trait]
777#[zenoh_macros::unstable]
778impl<P> TimestampInstrumentationBuilderTrait for AdvancedPublicationBuilder<'_, P> {
779    #[zenoh_macros::unstable]
780    fn timestamp_instrumentation<
781        TS: Into<Option<zenoh::timestamp_stack::TimestampInstrumentation>>,
782    >(
783        self,
784        instrumentation: TS,
785    ) -> Self {
786        Self {
787            builder: self.builder.timestamp_instrumentation(instrumentation),
788            ..self
789        }
790    }
791}
792
793#[zenoh_macros::unstable]
794impl<P> Resolvable for AdvancedPublicationBuilder<'_, P> {
795    type To = ZResult<()>;
796}
797
798#[zenoh_macros::unstable]
799impl Wait for AdvancedPublisherPutBuilder<'_> {
800    #[inline]
801    #[zenoh_macros::unstable]
802    fn wait(self) -> <Self as Resolvable>::To {
803        if let Some(cache) = self.cache {
804            cache.cache_sample(zenoh::sample::Sample::from(&self.builder));
805        }
806        self.builder.wait()
807    }
808}
809
810#[zenoh_macros::unstable]
811impl Wait for AdvancedPublisherDeleteBuilder<'_> {
812    #[inline]
813    #[zenoh_macros::unstable]
814    fn wait(self) -> <Self as Resolvable>::To {
815        if let Some(cache) = self.cache {
816            cache.cache_sample(zenoh::sample::Sample::from(&self.builder));
817        }
818        self.builder.wait()
819    }
820}
821
822#[zenoh_macros::unstable]
823impl IntoFuture for AdvancedPublisherPutBuilder<'_> {
824    type Output = <Self as Resolvable>::To;
825    type IntoFuture = Ready<<Self as Resolvable>::To>;
826
827    #[zenoh_macros::unstable]
828    fn into_future(self) -> Self::IntoFuture {
829        std::future::ready(self.wait())
830    }
831}
832
833#[zenoh_macros::unstable]
834impl IntoFuture for AdvancedPublisherDeleteBuilder<'_> {
835    type Output = <Self as Resolvable>::To;
836    type IntoFuture = Ready<<Self as Resolvable>::To>;
837
838    #[zenoh_macros::unstable]
839    fn into_future(self) -> Self::IntoFuture {
840        std::future::ready(self.wait())
841    }
842}
zenoh_ext/advanced_publisher.rs

zenoh_ext/
advanced_publisher.rs
