mctx_core/
tokio_adapter.rs

1use crate::{MctxError, Publication, SendReport};
2use std::io;
3#[cfg(not(unix))]
4use std::time::Duration;
5use thiserror::Error;
6
7/// Errors returned by the Tokio adapter.
8#[derive(Debug, Error)]
9pub enum TokioSendError {
10    /// Waiting for Tokio readiness failed.
11    #[error("MCTX: tokio readiness failed: {0}")]
12    Readiness(io::Error),
13
14    /// The underlying multicast sender returned an error.
15    #[error(transparent)]
16    Send(#[from] MctxError),
17}
18
19/// Thin Tokio wrapper around an owned publication.
20///
21/// On Unix this uses `tokio::io::unix::AsyncFd` to wait for write readiness.
22/// On other platforms it registers a cloned socket handle with Tokio while the
23/// original publication remains available for metadata and metrics.
24#[derive(Debug)]
25pub struct TokioPublication {
26    #[cfg(unix)]
27    inner: tokio::io::unix::AsyncFd<Publication>,
28    #[cfg(not(unix))]
29    inner: Publication,
30    #[cfg(not(unix))]
31    readiness: tokio::net::UdpSocket,
32}
33
34impl TokioPublication {
35    /// Wraps an owned publication for use with Tokio.
36    pub fn new(publication: Publication) -> io::Result<Self> {
37        #[cfg(unix)]
38        {
39            Ok(Self {
40                inner: tokio::io::unix::AsyncFd::new(publication)?,
41            })
42        }
43
44        #[cfg(not(unix))]
45        {
46            let readiness_socket = publication.socket().try_clone()?;
47            let readiness_socket: std::net::UdpSocket = readiness_socket.into();
48            readiness_socket.set_nonblocking(true)?;
49            Ok(Self {
50                inner: publication,
51                readiness: tokio::net::UdpSocket::from_std(readiness_socket)?,
52            })
53        }
54    }
55
56    /// Returns a shared reference to the wrapped publication.
57    pub fn publication(&self) -> &Publication {
58        #[cfg(unix)]
59        {
60            self.inner.get_ref()
61        }
62
63        #[cfg(not(unix))]
64        {
65            &self.inner
66        }
67    }
68
69    /// Consumes the adapter and returns the wrapped publication.
70    pub fn into_publication(self) -> Publication {
71        #[cfg(unix)]
72        {
73            self.inner.into_inner()
74        }
75
76        #[cfg(not(unix))]
77        {
78            self.inner
79        }
80    }
81
82    /// Compatibility no-op retained now that non-Unix platforms use socket
83    /// readiness instead of polling.
84    #[cfg(not(unix))]
85    pub fn with_poll_interval(self, _poll_interval: Duration) -> Self {
86        self
87    }
88
89    /// Waits for socket readiness and sends one payload.
90    pub async fn send(&self, payload: &[u8]) -> Result<SendReport, TokioSendError> {
91        #[cfg(unix)]
92        {
93            loop {
94                let mut readiness = self
95                    .inner
96                    .writable()
97                    .await
98                    .map_err(TokioSendError::Readiness)?;
99
100                match self.inner.get_ref().send(payload) {
101                    Ok(report) => return Ok(report),
102                    Err(error) if error.is_would_block() => readiness.clear_ready(),
103                    Err(error) => return Err(TokioSendError::Send(error)),
104                }
105            }
106        }
107
108        #[cfg(not(unix))]
109        {
110            loop {
111                match self.readiness.try_send(payload) {
112                    Ok(bytes_sent) => {
113                        return self
114                            .inner
115                            .finish_send(Ok(bytes_sent))
116                            .map_err(TokioSendError::Send);
117                    }
118                    Err(error) if error.kind() == io::ErrorKind::WouldBlock => self
119                        .readiness
120                        .writable()
121                        .await
122                        .map_err(TokioSendError::Readiness)?,
123                    Err(error) => {
124                        return self
125                            .inner
126                            .finish_send(Err(error))
127                            .map_err(TokioSendError::Send);
128                    }
129                }
130            }
131        }
132    }
133}
134
135#[cfg(all(test, feature = "tokio"))]
136mod tests {
137    use super::*;
138    use crate::test_support::{
139        TEST_GROUP, is_multicast_test_network_unavailable, recv_payload, test_multicast_receiver,
140    };
141    use crate::{Context, PublicationConfig};
142
143    #[tokio::test]
144    async fn tokio_publication_sends_a_packet() {
145        let (receiver, port) = test_multicast_receiver();
146        let mut context = Context::new();
147        let id = context
148            .add_publication(PublicationConfig::new(TEST_GROUP, port))
149            .unwrap();
150
151        let publication = context.take_publication(id).unwrap();
152        let publication = TokioPublication::new(publication).unwrap();
153
154        match publication.send(b"tokio hello").await {
155            Ok(_) => {}
156            Err(TokioSendError::Send(error)) if is_multicast_test_network_unavailable(&error) => {
157                eprintln!("skipping multicast integration test: {error}");
158                return;
159            }
160            Err(error) => panic!("tokio multicast integration test failed: {error}"),
161        }
162        let payload = recv_payload(&receiver);
163
164        assert_eq!(payload, b"tokio hello");
165    }
166}
mctx_core/tokio_adapter.rs

mctx_core/
tokio_adapter.rs
