zerodds_dcps_async/
writer.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3//! AsyncDataWriter — write/dispose/unregister/wait_for_matched async.
4
5use alloc::sync::Arc;
6use core::time::Duration;
7
8use zerodds_dcps::{DataWriter, DataWriterQos, DdsType, InstanceHandle, Result};
9
10/// Async wrapper around `DataWriter<T>`.
11///
12/// Hot path: `write()` is a future form over the sync path with
13/// a yield-based retry loop for
14/// `OutOfResources` backpressure (Spec §5.1
15/// `zerodds-async-1.0`). Instead of a thread-blocking `Condvar::wait_timeout`,
16/// caller tasks yield out of the executor via `yield_for` and stay
17/// cancelable. Other DCPS methods delegate synchronously — they are
18/// non-blocking anyway.
19pub struct AsyncDataWriter<T: DdsType + Send + Sync + 'static> {
20    inner: Arc<DataWriter<T>>,
21}
22
23impl<T: DdsType + Send + Sync + 'static> Clone for AsyncDataWriter<T> {
24    fn clone(&self) -> Self {
25        Self {
26            inner: Arc::clone(&self.inner),
27        }
28    }
29}
30
31impl<T: DdsType + Send + Sync + 'static> AsyncDataWriter<T> {
32    pub(crate) fn from_sync(inner: DataWriter<T>) -> Self {
33        Self {
34            inner: Arc::new(inner),
35        }
36    }
37
38    /// Writes a sample. Spec §2.1.1.
39    ///
40    /// # Errors
41    /// Same as `DataWriter::write` — `OutOfResources` after the
42    /// `max_blocking_time` timeout, otherwise all other errors are
43    /// passed through transparently.
44    ///
45    /// Spec §5.1 zerodds-async-1.0: on `OutOfResources` the future
46    /// suspends via `yield_for` and retries until either a drain
47    /// happens or `reliability.max_blocking_time` has elapsed.
48    /// In the sync path a `Condvar::wait_timeout` would block here —
49    /// the async path uses a yield-retry loop without a
50    /// thread block.
51    pub async fn write(&self, sample: &T) -> Result<()>
52    where
53        T: Clone,
54    {
55        let max_block = self.inner.qos().reliability.max_blocking_time;
56        let max_block_nanos = max_block.to_nanos();
57        // INFINITE → our retry loop still has a safety cap
58        // (~1 s polling) so the caller sees caller-side cancellation.
59        // The spec permits this.
60        let safety_cap = core::time::Duration::from_secs(1);
61        let deadline = if max_block_nanos == u128::MAX {
62            None
63        } else {
64            #[allow(clippy::cast_possible_truncation)]
65            let secs = (max_block_nanos / 1_000_000_000) as u64;
66            #[allow(clippy::cast_possible_truncation)]
67            let nanos = (max_block_nanos % 1_000_000_000) as u32;
68            Some(std::time::Instant::now() + core::time::Duration::new(secs, nanos))
69        };
70
71        let s = sample.clone();
72        loop {
73            match self.inner.write(&s) {
74                Ok(()) => return Ok(()),
75                Err(zerodds_dcps::DdsError::OutOfResources { .. }) => {
76                    // Wait for a drain.
77                    if let Some(d) = deadline {
78                        if std::time::Instant::now() >= d {
79                            return Err(zerodds_dcps::DdsError::Timeout);
80                        }
81                    }
82                    crate::yield_for(core::time::Duration::from_millis(2)).await;
83                }
84                Err(other) => return Err(other),
85            }
86            if deadline.is_none() {
87                // INFINITE: a safety yield after 1 s so the caller
88                // sees at least one await point and can cancel.
89                let _ = safety_cap;
90            }
91        }
92    }
93
94    /// Spec §2.1.2 register_instance.
95    ///
96    /// # Errors
97    /// Same as sync.
98    pub async fn register_instance(&self, sample: &T) -> Result<InstanceHandle> {
99        self.inner.register_instance(sample)
100    }
101
102    /// Spec §2.1.3 dispose. Triggers the DISPOSED wire lifecycle.
103    ///
104    /// # Errors
105    /// Same as sync.
106    pub async fn dispose(&self, sample: &T, handle: InstanceHandle) -> Result<()> {
107        self.inner.dispose(sample, handle)
108    }
109
110    /// Spec §2.1.4 unregister_instance.
111    ///
112    /// # Errors
113    /// Same as sync.
114    pub async fn unregister_instance(&self, sample: &T, handle: InstanceHandle) -> Result<()> {
115        self.inner.unregister_instance(sample, handle)
116    }
117
118    /// Spec §2.1.5 wait_for_matched_subscription. Async polling
119    /// loop with a 10 ms tick.
120    ///
121    /// # Errors
122    /// Same as sync — `Timeout` if `min_count` is not reached within `timeout`.
123    pub async fn wait_for_matched_subscription(
124        &self,
125        min_count: usize,
126        timeout: Duration,
127    ) -> Result<()> {
128        let deadline = std::time::Instant::now() + timeout;
129        loop {
130            if self.inner.matched_subscription_count() >= min_count {
131                return Ok(());
132            }
133            if std::time::Instant::now() >= deadline {
134                return Err(zerodds_dcps::DdsError::Timeout);
135            }
136            // Async sleep without a hard tokio dependency: yield via a futures helper.
137            crate::yield_for(Duration::from_millis(10)).await;
138        }
139    }
140
141    /// Spec §2.1.6 matched_subscription_count (synchronous).
142    #[must_use]
143    pub fn matched_subscription_count(&self) -> usize {
144        self.inner.matched_subscription_count()
145    }
146
147    /// Returns the underlying sync variant.
148    #[must_use]
149    pub fn as_sync(&self) -> &DataWriter<T> {
150        &self.inner
151    }
152
153    /// Returns the DataWriterQos.
154    #[must_use]
155    pub fn qos(&self) -> DataWriterQos {
156        self.inner.qos()
157    }
158}
zerodds_dcps_async/writer.rs

zerodds_dcps_async/
writer.rs
