tokio/net/unix/datagram/socket.rs
1use crate::io::{Interest, PollEvented, ReadBuf, Ready};
2use crate::net::unix::SocketAddr;
3use crate::util::check_socket_for_blocking;
4
5use std::fmt;
6use std::io;
7use std::net::Shutdown;
8use std::os::unix::io::{AsFd, AsRawFd, BorrowedFd, FromRawFd, IntoRawFd, RawFd};
9use std::os::unix::net;
10use std::path::Path;
11use std::task::{ready, Context, Poll};
12
13cfg_io_util! {
14 use bytes::BufMut;
15}
16
17cfg_net_unix! {
18 /// An I/O object representing a Unix datagram socket.
19 ///
20 /// A socket can be either named (associated with a filesystem path) or
21 /// unnamed.
22 ///
23 /// This type does not provide a `split` method, because this functionality
24 /// can be achieved by wrapping the socket in an [`Arc`]. Note that you do
25 /// not need a `Mutex` to share the `UnixDatagram` — an `Arc<UnixDatagram>`
26 /// is enough. This is because all of the methods take `&self` instead of
27 /// `&mut self`.
28 ///
29 /// **Note:** named sockets are persisted even after the object is dropped
30 /// and the program has exited, and cannot be reconnected. It is advised
31 /// that you either check for and unlink the existing socket if it exists,
32 /// or use a temporary file that is guaranteed to not already exist.
33 ///
34 /// [`Arc`]: std::sync::Arc
35 ///
36 /// # Examples
37 /// Using named sockets, associated with a filesystem path:
38 /// ```
39 /// # use std::error::Error;
40 /// # #[tokio::main]
41 /// # async fn main() -> Result<(), Box<dyn Error>> {
42 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
43 /// use tokio::net::UnixDatagram;
44 /// use tempfile::tempdir;
45 ///
46 /// // We use a temporary directory so that the socket
47 /// // files left by the bound sockets will get cleaned up.
48 /// let tmp = tempdir()?;
49 ///
50 /// // Bind each socket to a filesystem path
51 /// let tx_path = tmp.path().join("tx");
52 /// let tx = UnixDatagram::bind(&tx_path)?;
53 /// let rx_path = tmp.path().join("rx");
54 /// let rx = UnixDatagram::bind(&rx_path)?;
55 ///
56 /// let bytes = b"hello world";
57 /// tx.send_to(bytes, &rx_path).await?;
58 ///
59 /// let mut buf = vec![0u8; 24];
60 /// let (size, addr) = rx.recv_from(&mut buf).await?;
61 ///
62 /// let dgram = &buf[..size];
63 /// assert_eq!(dgram, bytes);
64 /// assert_eq!(addr.as_pathname().unwrap(), &tx_path);
65 ///
66 /// # Ok(())
67 /// # }
68 /// ```
69 ///
70 /// Using unnamed sockets, created as a pair
71 /// ```
72 /// # use std::error::Error;
73 /// # #[tokio::main]
74 /// # async fn main() -> Result<(), Box<dyn Error>> {
75 /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
76 /// use tokio::net::UnixDatagram;
77 ///
78 /// // Create the pair of sockets
79 /// let (sock1, sock2) = UnixDatagram::pair()?;
80 ///
81 /// // Since the sockets are paired, the paired send/recv
82 /// // functions can be used
83 /// let bytes = b"hello world";
84 /// sock1.send(bytes).await?;
85 ///
86 /// let mut buff = vec![0u8; 24];
87 /// let size = sock2.recv(&mut buff).await?;
88 ///
89 /// let dgram = &buff[..size];
90 /// assert_eq!(dgram, bytes);
91 ///
92 /// # Ok(())
93 /// # }
94 /// ```
95 #[cfg_attr(docsrs, doc(alias = "uds"))]
96 pub struct UnixDatagram {
97 io: PollEvented<mio::net::UnixDatagram>,
98 }
99}
100
101impl UnixDatagram {
102 pub(crate) fn from_mio(sys: mio::net::UnixDatagram) -> io::Result<UnixDatagram> {
103 let datagram = UnixDatagram::new(sys)?;
104
105 if let Some(e) = datagram.io.take_error()? {
106 return Err(e);
107 }
108
109 Ok(datagram)
110 }
111
112 /// Waits for any of the requested ready states.
113 ///
114 /// This function is usually paired with `try_recv()` or `try_send()`. It
115 /// can be used to concurrently `recv` / `send` to the same socket on a single
116 /// task without splitting the socket.
117 ///
118 /// The function may complete without the socket being ready. This is a
119 /// false-positive and attempting an operation will return with
120 /// `io::ErrorKind::WouldBlock`. The function can also return with an empty
121 /// [`Ready`] set, so you should always check the returned value and possibly
122 /// wait again if the requested states are not set.
123 ///
124 /// # Cancel safety
125 ///
126 /// This method is cancel safe. Once a readiness event occurs, the method
127 /// will continue to return immediately until the readiness event is
128 /// consumed by an attempt to read or write that fails with `WouldBlock` or
129 /// `Poll::Pending`.
130 ///
131 /// # Examples
132 ///
133 /// Concurrently receive from and send to the socket on the same task
134 /// without splitting.
135 ///
136 /// ```no_run
137 /// use tokio::io::Interest;
138 /// use tokio::net::UnixDatagram;
139 /// use std::io;
140 ///
141 /// #[tokio::main]
142 /// async fn main() -> io::Result<()> {
143 /// let dir = tempfile::tempdir().unwrap();
144 /// let client_path = dir.path().join("client.sock");
145 /// let server_path = dir.path().join("server.sock");
146 /// let socket = UnixDatagram::bind(&client_path)?;
147 /// socket.connect(&server_path)?;
148 ///
149 /// loop {
150 /// let ready = socket.ready(Interest::READABLE | Interest::WRITABLE).await?;
151 ///
152 /// if ready.is_readable() {
153 /// let mut data = [0; 1024];
154 /// match socket.try_recv(&mut data[..]) {
155 /// Ok(n) => {
156 /// println!("received {:?}", &data[..n]);
157 /// }
158 /// // False-positive, continue
159 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
160 /// Err(e) => {
161 /// return Err(e);
162 /// }
163 /// }
164 /// }
165 ///
166 /// if ready.is_writable() {
167 /// // Write some data
168 /// match socket.try_send(b"hello world") {
169 /// Ok(n) => {
170 /// println!("sent {} bytes", n);
171 /// }
172 /// // False-positive, continue
173 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
174 /// Err(e) => {
175 /// return Err(e);
176 /// }
177 /// }
178 /// }
179 /// }
180 /// }
181 /// ```
182 pub async fn ready(&self, interest: Interest) -> io::Result<Ready> {
183 let event = self.io.registration().readiness(interest).await?;
184 Ok(event.ready)
185 }
186
187 /// Waits for the socket to become writable.
188 ///
189 /// This function is equivalent to `ready(Interest::WRITABLE)` and is
190 /// usually paired with `try_send()` or `try_send_to()`.
191 ///
192 /// The function may complete without the socket being writable. This is a
193 /// false-positive and attempting a `try_send()` will return with
194 /// `io::ErrorKind::WouldBlock`.
195 ///
196 /// # Cancel safety
197 ///
198 /// This method is cancel safe. Once a readiness event occurs, the method
199 /// will continue to return immediately until the readiness event is
200 /// consumed by an attempt to write that fails with `WouldBlock` or
201 /// `Poll::Pending`.
202 ///
203 /// # Examples
204 ///
205 /// ```no_run
206 /// use tokio::net::UnixDatagram;
207 /// use std::io;
208 ///
209 /// #[tokio::main]
210 /// async fn main() -> io::Result<()> {
211 /// let dir = tempfile::tempdir().unwrap();
212 /// let client_path = dir.path().join("client.sock");
213 /// let server_path = dir.path().join("server.sock");
214 /// let socket = UnixDatagram::bind(&client_path)?;
215 /// socket.connect(&server_path)?;
216 ///
217 /// loop {
218 /// // Wait for the socket to be writable
219 /// socket.writable().await?;
220 ///
221 /// // Try to send data, this may still fail with `WouldBlock`
222 /// // if the readiness event is a false positive.
223 /// match socket.try_send(b"hello world") {
224 /// Ok(n) => {
225 /// break;
226 /// }
227 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
228 /// continue;
229 /// }
230 /// Err(e) => {
231 /// return Err(e);
232 /// }
233 /// }
234 /// }
235 ///
236 /// Ok(())
237 /// }
238 /// ```
239 pub async fn writable(&self) -> io::Result<()> {
240 self.ready(Interest::WRITABLE).await?;
241 Ok(())
242 }
243
244 /// Polls for write/send readiness.
245 ///
246 /// If the socket is not currently ready for sending, this method will
247 /// store a clone of the `Waker` from the provided `Context`. When the socket
248 /// becomes ready for sending, `Waker::wake` will be called on the
249 /// waker.
250 ///
251 /// Note that on multiple calls to `poll_send_ready` or `poll_send`, only
252 /// the `Waker` from the `Context` passed to the most recent call is
253 /// scheduled to receive a wakeup. (However, `poll_recv_ready` retains a
254 /// second, independent waker.)
255 ///
256 /// This function is intended for cases where creating and pinning a future
257 /// via [`writable`] is not feasible. Where possible, using [`writable`] is
258 /// preferred, as this supports polling from multiple tasks at once.
259 ///
260 /// # Return value
261 ///
262 /// The function returns:
263 ///
264 /// * `Poll::Pending` if the socket is not ready for writing.
265 /// * `Poll::Ready(Ok(()))` if the socket is ready for writing.
266 /// * `Poll::Ready(Err(e))` if an error is encountered.
267 ///
268 /// # Errors
269 ///
270 /// This function may encounter any standard I/O error except `WouldBlock`.
271 ///
272 /// [`writable`]: method@Self::writable
273 pub fn poll_send_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
274 self.io.registration().poll_write_ready(cx).map_ok(|_| ())
275 }
276
277 /// Waits for the socket to become readable.
278 ///
279 /// This function is equivalent to `ready(Interest::READABLE)` and is usually
280 /// paired with `try_recv()`.
281 ///
282 /// The function may complete without the socket being readable. This is a
283 /// false-positive and attempting a `try_recv()` will return with
284 /// `io::ErrorKind::WouldBlock`.
285 ///
286 /// # Cancel safety
287 ///
288 /// This method is cancel safe. Once a readiness event occurs, the method
289 /// will continue to return immediately until the readiness event is
290 /// consumed by an attempt to read that fails with `WouldBlock` or
291 /// `Poll::Pending`.
292 ///
293 /// # Examples
294 ///
295 /// ```no_run
296 /// use tokio::net::UnixDatagram;
297 /// use std::io;
298 ///
299 /// #[tokio::main]
300 /// async fn main() -> io::Result<()> {
301 /// // Connect to a peer
302 /// let dir = tempfile::tempdir().unwrap();
303 /// let client_path = dir.path().join("client.sock");
304 /// let server_path = dir.path().join("server.sock");
305 /// let socket = UnixDatagram::bind(&client_path)?;
306 /// socket.connect(&server_path)?;
307 ///
308 /// loop {
309 /// // Wait for the socket to be readable
310 /// socket.readable().await?;
311 ///
312 /// // The buffer is **not** included in the async task and will
313 /// // only exist on the stack.
314 /// let mut buf = [0; 1024];
315 ///
316 /// // Try to recv data, this may still fail with `WouldBlock`
317 /// // if the readiness event is a false positive.
318 /// match socket.try_recv(&mut buf) {
319 /// Ok(n) => {
320 /// println!("GOT {:?}", &buf[..n]);
321 /// break;
322 /// }
323 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
324 /// continue;
325 /// }
326 /// Err(e) => {
327 /// return Err(e);
328 /// }
329 /// }
330 /// }
331 ///
332 /// Ok(())
333 /// }
334 /// ```
335 pub async fn readable(&self) -> io::Result<()> {
336 self.ready(Interest::READABLE).await?;
337 Ok(())
338 }
339
340 /// Polls for read/receive readiness.
341 ///
342 /// If the socket is not currently ready for receiving, this method will
343 /// store a clone of the `Waker` from the provided `Context`. When the
344 /// socket becomes ready for reading, `Waker::wake` will be called on the
345 /// waker.
346 ///
347 /// Note that on multiple calls to `poll_recv_ready`, `poll_recv` or
348 /// `poll_peek`, only the `Waker` from the `Context` passed to the most
349 /// recent call is scheduled to receive a wakeup. (However,
350 /// `poll_send_ready` retains a second, independent waker.)
351 ///
352 /// This function is intended for cases where creating and pinning a future
353 /// via [`readable`] is not feasible. Where possible, using [`readable`] is
354 /// preferred, as this supports polling from multiple tasks at once.
355 ///
356 /// # Return value
357 ///
358 /// The function returns:
359 ///
360 /// * `Poll::Pending` if the socket is not ready for reading.
361 /// * `Poll::Ready(Ok(()))` if the socket is ready for reading.
362 /// * `Poll::Ready(Err(e))` if an error is encountered.
363 ///
364 /// # Errors
365 ///
366 /// This function may encounter any standard I/O error except `WouldBlock`.
367 ///
368 /// [`readable`]: method@Self::readable
369 pub fn poll_recv_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
370 self.io.registration().poll_read_ready(cx).map_ok(|_| ())
371 }
372
373 /// Creates a new `UnixDatagram` bound to the specified path.
374 ///
375 /// # Examples
376 /// ```
377 /// # use std::error::Error;
378 /// # #[tokio::main]
379 /// # async fn main() -> Result<(), Box<dyn Error>> {
380 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
381 /// use tokio::net::UnixDatagram;
382 /// use tempfile::tempdir;
383 ///
384 /// // We use a temporary directory so that the socket
385 /// // files left by the bound sockets will get cleaned up.
386 /// let tmp = tempdir()?;
387 ///
388 /// // Bind the socket to a filesystem path
389 /// let socket_path = tmp.path().join("socket");
390 /// let socket = UnixDatagram::bind(&socket_path)?;
391 ///
392 /// # Ok(())
393 /// # }
394 /// ```
395 pub fn bind<P>(path: P) -> io::Result<UnixDatagram>
396 where
397 P: AsRef<Path>,
398 {
399 let socket = mio::net::UnixDatagram::bind(path)?;
400 UnixDatagram::new(socket)
401 }
402
403 /// Creates an unnamed pair of connected sockets.
404 ///
405 /// This function will create a pair of interconnected Unix sockets for
406 /// communicating back and forth between one another.
407 ///
408 /// # Examples
409 /// ```
410 /// # use std::error::Error;
411 /// # #[tokio::main]
412 /// # async fn main() -> Result<(), Box<dyn Error>> {
413 /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
414 /// use tokio::net::UnixDatagram;
415 ///
416 /// // Create the pair of sockets
417 /// let (sock1, sock2) = UnixDatagram::pair()?;
418 ///
419 /// // Since the sockets are paired, the paired send/recv
420 /// // functions can be used
421 /// let bytes = b"hail eris";
422 /// sock1.send(bytes).await?;
423 ///
424 /// let mut buff = vec![0u8; 24];
425 /// let size = sock2.recv(&mut buff).await?;
426 ///
427 /// let dgram = &buff[..size];
428 /// assert_eq!(dgram, bytes);
429 ///
430 /// # Ok(())
431 /// # }
432 /// ```
433 pub fn pair() -> io::Result<(UnixDatagram, UnixDatagram)> {
434 let (a, b) = mio::net::UnixDatagram::pair()?;
435 let a = UnixDatagram::new(a)?;
436 let b = UnixDatagram::new(b)?;
437 // A fresh pair has empty buffers: writable now, readable only once
438 // the peer sends.
439 a.io.registration().assume_ready(Ready::WRITABLE);
440 b.io.registration().assume_ready(Ready::WRITABLE);
441
442 Ok((a, b))
443 }
444
445 /// Creates new [`UnixDatagram`] from a [`std::os::unix::net::UnixDatagram`].
446 ///
447 /// This function is intended to be used to wrap a `UnixDatagram` from the
448 /// standard library in the Tokio equivalent.
449 ///
450 /// # Notes
451 ///
452 /// The caller is responsible for ensuring that the socket is in
453 /// non-blocking mode. Otherwise all I/O operations on the socket
454 /// will block the thread, which will cause unexpected behavior.
455 /// Non-blocking mode can be set using [`set_nonblocking`].
456 ///
457 /// Passing a listener in blocking mode is always erroneous,
458 /// and the behavior in that case may change in the future.
459 /// For example, it could panic.
460 ///
461 /// [`set_nonblocking`]: std::os::unix::net::UnixDatagram::set_nonblocking
462 ///
463 /// # Panics
464 ///
465 /// This function panics if it is not called from within a runtime with
466 /// IO enabled.
467 ///
468 /// The runtime is usually set implicitly when this function is called
469 /// from a future driven by a Tokio runtime, otherwise runtime can be set
470 /// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
471 /// # Examples
472 /// ```
473 /// # use std::error::Error;
474 /// # #[tokio::main]
475 /// # async fn main() -> Result<(), Box<dyn Error>> {
476 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
477 /// use tokio::net::UnixDatagram;
478 /// use std::os::unix::net::UnixDatagram as StdUDS;
479 /// use tempfile::tempdir;
480 ///
481 /// // We use a temporary directory so that the socket
482 /// // files left by the bound sockets will get cleaned up.
483 /// let tmp = tempdir()?;
484 ///
485 /// // Bind the socket to a filesystem path
486 /// let socket_path = tmp.path().join("socket");
487 /// let std_socket = StdUDS::bind(&socket_path)?;
488 /// std_socket.set_nonblocking(true)?;
489 /// let tokio_socket = UnixDatagram::from_std(std_socket)?;
490 ///
491 /// # Ok(())
492 /// # }
493 /// ```
494 #[track_caller]
495 pub fn from_std(datagram: net::UnixDatagram) -> io::Result<UnixDatagram> {
496 check_socket_for_blocking(&datagram)?;
497
498 let socket = mio::net::UnixDatagram::from_std(datagram);
499 let io = PollEvented::new(socket)?;
500 Ok(UnixDatagram { io })
501 }
502
503 /// Turns a [`tokio::net::UnixDatagram`] into a [`std::os::unix::net::UnixDatagram`].
504 ///
505 /// The returned [`std::os::unix::net::UnixDatagram`] will have nonblocking
506 /// mode set as `true`. Use [`set_nonblocking`] to change the blocking mode
507 /// if needed.
508 ///
509 /// # Examples
510 ///
511 /// ```rust,no_run
512 /// # use std::error::Error;
513 /// # async fn dox() -> Result<(), Box<dyn Error>> {
514 /// let tokio_socket = tokio::net::UnixDatagram::bind("/path/to/the/socket")?;
515 /// let std_socket = tokio_socket.into_std()?;
516 /// std_socket.set_nonblocking(false)?;
517 /// # Ok(())
518 /// # }
519 /// ```
520 ///
521 /// [`tokio::net::UnixDatagram`]: UnixDatagram
522 /// [`std::os::unix::net::UnixDatagram`]: std::os::unix::net::UnixDatagram
523 /// [`set_nonblocking`]: fn@std::os::unix::net::UnixDatagram::set_nonblocking
524 pub fn into_std(self) -> io::Result<std::os::unix::net::UnixDatagram> {
525 self.io
526 .into_inner()
527 .map(IntoRawFd::into_raw_fd)
528 .map(|raw_fd| unsafe { std::os::unix::net::UnixDatagram::from_raw_fd(raw_fd) })
529 }
530
531 fn new(socket: mio::net::UnixDatagram) -> io::Result<UnixDatagram> {
532 let io = PollEvented::new(socket)?;
533 Ok(UnixDatagram { io })
534 }
535
536 /// Creates a new `UnixDatagram` which is not bound to any address.
537 ///
538 /// # Examples
539 /// ```
540 /// # use std::error::Error;
541 /// # #[tokio::main]
542 /// # async fn main() -> Result<(), Box<dyn Error>> {
543 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
544 /// use tokio::net::UnixDatagram;
545 /// use tempfile::tempdir;
546 ///
547 /// // Create an unbound socket
548 /// let tx = UnixDatagram::unbound()?;
549 ///
550 /// // Create another, bound socket
551 /// let tmp = tempdir()?;
552 /// let rx_path = tmp.path().join("rx");
553 /// let rx = UnixDatagram::bind(&rx_path)?;
554 ///
555 /// // Send to the bound socket
556 /// let bytes = b"hello world";
557 /// tx.send_to(bytes, &rx_path).await?;
558 ///
559 /// let mut buf = vec![0u8; 24];
560 /// let (size, addr) = rx.recv_from(&mut buf).await?;
561 ///
562 /// let dgram = &buf[..size];
563 /// assert_eq!(dgram, bytes);
564 ///
565 /// # Ok(())
566 /// # }
567 /// ```
568 pub fn unbound() -> io::Result<UnixDatagram> {
569 let socket = mio::net::UnixDatagram::unbound()?;
570 UnixDatagram::new(socket)
571 }
572
573 /// Connects the socket to the specified address.
574 ///
575 /// The `send` method may be used to send data to the specified address.
576 /// `recv` and `recv_from` will only receive data from that address.
577 ///
578 /// # Examples
579 /// ```
580 /// # use std::error::Error;
581 /// # #[tokio::main]
582 /// # async fn main() -> Result<(), Box<dyn Error>> {
583 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
584 /// use tokio::net::UnixDatagram;
585 /// use tempfile::tempdir;
586 ///
587 /// // Create an unbound socket
588 /// let tx = UnixDatagram::unbound()?;
589 ///
590 /// // Create another, bound socket
591 /// let tmp = tempdir()?;
592 /// let rx_path = tmp.path().join("rx");
593 /// let rx = UnixDatagram::bind(&rx_path)?;
594 ///
595 /// // Connect to the bound socket
596 /// tx.connect(&rx_path)?;
597 ///
598 /// // Send to the bound socket
599 /// let bytes = b"hello world";
600 /// tx.send(bytes).await?;
601 ///
602 /// let mut buf = vec![0u8; 24];
603 /// let (size, addr) = rx.recv_from(&mut buf).await?;
604 ///
605 /// let dgram = &buf[..size];
606 /// assert_eq!(dgram, bytes);
607 ///
608 /// # Ok(())
609 /// # }
610 /// ```
611 pub fn connect<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
612 self.io.connect(path)
613 }
614
615 /// Sends data on the socket to the socket's peer.
616 ///
617 /// # Cancel safety
618 ///
619 /// This method is cancel safe. If `send` is used as a branch in
620 /// [`tokio::select!`](crate::select) and another branch
621 /// completes first, then it is guaranteed that the message was not sent.
622 ///
623 /// # Examples
624 /// ```
625 /// # use std::error::Error;
626 /// # #[tokio::main]
627 /// # async fn main() -> Result<(), Box<dyn Error>> {
628 /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
629 /// use tokio::net::UnixDatagram;
630 ///
631 /// // Create the pair of sockets
632 /// let (sock1, sock2) = UnixDatagram::pair()?;
633 ///
634 /// // Since the sockets are paired, the paired send/recv
635 /// // functions can be used
636 /// let bytes = b"hello world";
637 /// sock1.send(bytes).await?;
638 ///
639 /// let mut buff = vec![0u8; 24];
640 /// let size = sock2.recv(&mut buff).await?;
641 ///
642 /// let dgram = &buff[..size];
643 /// assert_eq!(dgram, bytes);
644 ///
645 /// # Ok(())
646 /// # }
647 /// ```
648 pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
649 self.io
650 .registration()
651 .async_io(Interest::WRITABLE, || self.io.send(buf))
652 .await
653 }
654
655 /// Tries to send a datagram to the peer without waiting.
656 ///
657 /// # Examples
658 ///
659 /// ```no_run
660 /// use tokio::net::UnixDatagram;
661 /// use std::io;
662 ///
663 /// #[tokio::main]
664 /// async fn main() -> io::Result<()> {
665 /// let dir = tempfile::tempdir().unwrap();
666 /// let client_path = dir.path().join("client.sock");
667 /// let server_path = dir.path().join("server.sock");
668 /// let socket = UnixDatagram::bind(&client_path)?;
669 /// socket.connect(&server_path)?;
670 ///
671 /// loop {
672 /// // Wait for the socket to be writable
673 /// socket.writable().await?;
674 ///
675 /// // Try to send data, this may still fail with `WouldBlock`
676 /// // if the readiness event is a false positive.
677 /// match socket.try_send(b"hello world") {
678 /// Ok(n) => {
679 /// break;
680 /// }
681 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
682 /// continue;
683 /// }
684 /// Err(e) => {
685 /// return Err(e);
686 /// }
687 /// }
688 /// }
689 ///
690 /// Ok(())
691 /// }
692 /// ```
693 pub fn try_send(&self, buf: &[u8]) -> io::Result<usize> {
694 self.io
695 .registration()
696 .try_io(Interest::WRITABLE, || self.io.send(buf))
697 }
698
699 /// Tries to send a datagram to the peer without waiting.
700 ///
701 /// # Examples
702 ///
703 /// ```no_run
704 /// use tokio::net::UnixDatagram;
705 /// use std::io;
706 ///
707 /// #[tokio::main]
708 /// async fn main() -> io::Result<()> {
709 /// let dir = tempfile::tempdir().unwrap();
710 /// let client_path = dir.path().join("client.sock");
711 /// let server_path = dir.path().join("server.sock");
712 /// let socket = UnixDatagram::bind(&client_path)?;
713 ///
714 /// loop {
715 /// // Wait for the socket to be writable
716 /// socket.writable().await?;
717 ///
718 /// // Try to send data, this may still fail with `WouldBlock`
719 /// // if the readiness event is a false positive.
720 /// match socket.try_send_to(b"hello world", &server_path) {
721 /// Ok(n) => {
722 /// break;
723 /// }
724 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
725 /// continue;
726 /// }
727 /// Err(e) => {
728 /// return Err(e);
729 /// }
730 /// }
731 /// }
732 ///
733 /// Ok(())
734 /// }
735 /// ```
736 pub fn try_send_to<P>(&self, buf: &[u8], target: P) -> io::Result<usize>
737 where
738 P: AsRef<Path>,
739 {
740 self.io
741 .registration()
742 .try_io(Interest::WRITABLE, || self.io.send_to(buf, target))
743 }
744
745 /// Receives data from the socket.
746 ///
747 /// # Cancel safety
748 ///
749 /// This method is cancel safe. If `recv` is used as a branch in
750 /// [`tokio::select!`](crate::select) and another branch
751 /// completes first, it is guaranteed that no messages were received on this
752 /// socket.
753 ///
754 /// # Examples
755 /// ```
756 /// # use std::error::Error;
757 /// # #[tokio::main]
758 /// # async fn main() -> Result<(), Box<dyn Error>> {
759 /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
760 /// use tokio::net::UnixDatagram;
761 ///
762 /// // Create the pair of sockets
763 /// let (sock1, sock2) = UnixDatagram::pair()?;
764 ///
765 /// // Since the sockets are paired, the paired send/recv
766 /// // functions can be used
767 /// let bytes = b"hello world";
768 /// sock1.send(bytes).await?;
769 ///
770 /// let mut buff = vec![0u8; 24];
771 /// let size = sock2.recv(&mut buff).await?;
772 ///
773 /// let dgram = &buff[..size];
774 /// assert_eq!(dgram, bytes);
775 ///
776 /// # Ok(())
777 /// # }
778 /// ```
779 pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
780 self.io
781 .registration()
782 .async_io(Interest::READABLE, || self.io.recv(buf))
783 .await
784 }
785
786 /// Tries to receive a datagram from the peer without waiting.
787 ///
788 /// # Examples
789 ///
790 /// ```no_run
791 /// use tokio::net::UnixDatagram;
792 /// use std::io;
793 ///
794 /// #[tokio::main]
795 /// async fn main() -> io::Result<()> {
796 /// // Connect to a peer
797 /// let dir = tempfile::tempdir().unwrap();
798 /// let client_path = dir.path().join("client.sock");
799 /// let server_path = dir.path().join("server.sock");
800 /// let socket = UnixDatagram::bind(&client_path)?;
801 /// socket.connect(&server_path)?;
802 ///
803 /// loop {
804 /// // Wait for the socket to be readable
805 /// socket.readable().await?;
806 ///
807 /// // The buffer is **not** included in the async task and will
808 /// // only exist on the stack.
809 /// let mut buf = [0; 1024];
810 ///
811 /// // Try to recv data, this may still fail with `WouldBlock`
812 /// // if the readiness event is a false positive.
813 /// match socket.try_recv(&mut buf) {
814 /// Ok(n) => {
815 /// println!("GOT {:?}", &buf[..n]);
816 /// break;
817 /// }
818 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
819 /// continue;
820 /// }
821 /// Err(e) => {
822 /// return Err(e);
823 /// }
824 /// }
825 /// }
826 ///
827 /// Ok(())
828 /// }
829 /// ```
830 pub fn try_recv(&self, buf: &mut [u8]) -> io::Result<usize> {
831 self.io
832 .registration()
833 .try_io(Interest::READABLE, || self.io.recv(buf))
834 }
835
836 cfg_io_util! {
837 /// Tries to receive data from the socket without waiting.
838 ///
839 /// This method can be used even if `buf` is uninitialized.
840 ///
841 /// # Examples
842 ///
843 /// ```no_run
844 /// use tokio::net::UnixDatagram;
845 /// use std::io;
846 ///
847 /// #[tokio::main]
848 /// async fn main() -> io::Result<()> {
849 /// // Connect to a peer
850 /// let dir = tempfile::tempdir().unwrap();
851 /// let client_path = dir.path().join("client.sock");
852 /// let server_path = dir.path().join("server.sock");
853 /// let socket = UnixDatagram::bind(&client_path)?;
854 ///
855 /// loop {
856 /// // Wait for the socket to be readable
857 /// socket.readable().await?;
858 ///
859 /// let mut buf = Vec::with_capacity(1024);
860 ///
861 /// // Try to recv data, this may still fail with `WouldBlock`
862 /// // if the readiness event is a false positive.
863 /// match socket.try_recv_buf_from(&mut buf) {
864 /// Ok((n, _addr)) => {
865 /// println!("GOT {:?}", &buf[..n]);
866 /// break;
867 /// }
868 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
869 /// continue;
870 /// }
871 /// Err(e) => {
872 /// return Err(e);
873 /// }
874 /// }
875 /// }
876 ///
877 /// Ok(())
878 /// }
879 /// ```
880 pub fn try_recv_buf_from<B: BufMut>(&self, buf: &mut B) -> io::Result<(usize, SocketAddr)> {
881 let (n, addr) = self.io.registration().try_io(Interest::READABLE, || {
882 let dst = buf.chunk_mut();
883 let dst =
884 unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
885
886 // Safety: We trust `UnixDatagram::recv_from` to have filled up `n` bytes in the
887 // buffer.
888 let (n, addr) = (*self.io).recv_from(dst)?;
889
890 unsafe {
891 buf.advance_mut(n);
892 }
893
894 Ok((n, addr))
895 })?;
896
897 Ok((n, SocketAddr(addr)))
898 }
899
900 /// Receives from the socket, advances the
901 /// buffer's internal cursor and returns how many bytes were read and the origin.
902 ///
903 /// This method can be used even if `buf` is uninitialized.
904 ///
905 /// # Examples
906 /// ```
907 /// # use std::error::Error;
908 /// # #[tokio::main]
909 /// # async fn main() -> Result<(), Box<dyn Error>> {
910 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
911 /// use tokio::net::UnixDatagram;
912 /// use tempfile::tempdir;
913 ///
914 /// // We use a temporary directory so that the socket
915 /// // files left by the bound sockets will get cleaned up.
916 /// let tmp = tempdir()?;
917 ///
918 /// // Bind each socket to a filesystem path
919 /// let tx_path = tmp.path().join("tx");
920 /// let tx = UnixDatagram::bind(&tx_path)?;
921 /// let rx_path = tmp.path().join("rx");
922 /// let rx = UnixDatagram::bind(&rx_path)?;
923 ///
924 /// let bytes = b"hello world";
925 /// tx.send_to(bytes, &rx_path).await?;
926 ///
927 /// let mut buf = Vec::with_capacity(24);
928 /// let (size, addr) = rx.recv_buf_from(&mut buf).await?;
929 ///
930 /// let dgram = &buf[..size];
931 /// assert_eq!(dgram, bytes);
932 /// assert_eq!(addr.as_pathname().unwrap(), &tx_path);
933 ///
934 /// # Ok(())
935 /// # }
936 /// ```
937 pub async fn recv_buf_from<B: BufMut>(&self, buf: &mut B) -> io::Result<(usize, SocketAddr)> {
938 self.io.registration().async_io(Interest::READABLE, || {
939 let dst = buf.chunk_mut();
940 let dst =
941 unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
942
943 // Safety: We trust `UnixDatagram::recv_from` to have filled up `n` bytes in the
944 // buffer.
945 let (n, addr) = (*self.io).recv_from(dst)?;
946
947 unsafe {
948 buf.advance_mut(n);
949 }
950 Ok((n,SocketAddr(addr)))
951 }).await
952 }
953
954 /// Tries to read data from the stream into the provided buffer, advancing the
955 /// buffer's internal cursor, returning how many bytes were read.
956 ///
957 /// This method can be used even if `buf` is uninitialized.
958 ///
959 /// # Examples
960 ///
961 /// ```no_run
962 /// use tokio::net::UnixDatagram;
963 /// use std::io;
964 ///
965 /// #[tokio::main]
966 /// async fn main() -> io::Result<()> {
967 /// // Connect to a peer
968 /// let dir = tempfile::tempdir().unwrap();
969 /// let client_path = dir.path().join("client.sock");
970 /// let server_path = dir.path().join("server.sock");
971 /// let socket = UnixDatagram::bind(&client_path)?;
972 /// socket.connect(&server_path)?;
973 ///
974 /// loop {
975 /// // Wait for the socket to be readable
976 /// socket.readable().await?;
977 ///
978 /// let mut buf = Vec::with_capacity(1024);
979 ///
980 /// // Try to recv data, this may still fail with `WouldBlock`
981 /// // if the readiness event is a false positive.
982 /// match socket.try_recv_buf(&mut buf) {
983 /// Ok(n) => {
984 /// println!("GOT {:?}", &buf[..n]);
985 /// break;
986 /// }
987 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
988 /// continue;
989 /// }
990 /// Err(e) => {
991 /// return Err(e);
992 /// }
993 /// }
994 /// }
995 ///
996 /// Ok(())
997 /// }
998 /// ```
999 pub fn try_recv_buf<B: BufMut>(&self, buf: &mut B) -> io::Result<usize> {
1000 self.io.registration().try_io(Interest::READABLE, || {
1001 let dst = buf.chunk_mut();
1002 let dst =
1003 unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
1004
1005 // Safety: We trust `UnixDatagram::recv` to have filled up `n` bytes in the
1006 // buffer.
1007 let n = (*self.io).recv(dst)?;
1008
1009 unsafe {
1010 buf.advance_mut(n);
1011 }
1012
1013 Ok(n)
1014 })
1015 }
1016
1017 /// Receives data from the socket from the address to which it is connected,
1018 /// advancing the buffer's internal cursor, returning how many bytes were read.
1019 ///
1020 /// This method can be used even if `buf` is uninitialized.
1021 ///
1022 /// # Examples
1023 /// ```
1024 /// # use std::error::Error;
1025 /// # #[tokio::main]
1026 /// # async fn main() -> Result<(), Box<dyn Error>> {
1027 /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
1028 /// use tokio::net::UnixDatagram;
1029 ///
1030 /// // Create the pair of sockets
1031 /// let (sock1, sock2) = UnixDatagram::pair()?;
1032 ///
1033 /// // Since the sockets are paired, the paired send/recv
1034 /// // functions can be used
1035 /// let bytes = b"hello world";
1036 /// sock1.send(bytes).await?;
1037 ///
1038 /// let mut buff = Vec::with_capacity(24);
1039 /// let size = sock2.recv_buf(&mut buff).await?;
1040 ///
1041 /// let dgram = &buff[..size];
1042 /// assert_eq!(dgram, bytes);
1043 ///
1044 /// # Ok(())
1045 /// # }
1046 /// ```
1047 pub async fn recv_buf<B: BufMut>(&self, buf: &mut B) -> io::Result<usize> {
1048 self.io.registration().async_io(Interest::READABLE, || {
1049 let dst = buf.chunk_mut();
1050 let dst =
1051 unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
1052
1053 // Safety: We trust `UnixDatagram::recv_from` to have filled up `n` bytes in the
1054 // buffer.
1055 let n = (*self.io).recv(dst)?;
1056
1057 unsafe {
1058 buf.advance_mut(n);
1059 }
1060 Ok(n)
1061 }).await
1062 }
1063 }
1064
1065 /// Sends data on the socket to the specified address.
1066 ///
1067 /// # Cancel safety
1068 ///
1069 /// This method is cancel safe. If `send_to` is used as a branch in
1070 /// [`tokio::select!`](crate::select) and another branch
1071 /// completes first, then it is guaranteed that the message was not sent.
1072 ///
1073 /// # Examples
1074 /// ```
1075 /// # use std::error::Error;
1076 /// # #[tokio::main]
1077 /// # async fn main() -> Result<(), Box<dyn Error>> {
1078 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1079 /// use tokio::net::UnixDatagram;
1080 /// use tempfile::tempdir;
1081 ///
1082 /// // We use a temporary directory so that the socket
1083 /// // files left by the bound sockets will get cleaned up.
1084 /// let tmp = tempdir()?;
1085 ///
1086 /// // Bind each socket to a filesystem path
1087 /// let tx_path = tmp.path().join("tx");
1088 /// let tx = UnixDatagram::bind(&tx_path)?;
1089 /// let rx_path = tmp.path().join("rx");
1090 /// let rx = UnixDatagram::bind(&rx_path)?;
1091 ///
1092 /// let bytes = b"hello world";
1093 /// tx.send_to(bytes, &rx_path).await?;
1094 ///
1095 /// let mut buf = vec![0u8; 24];
1096 /// let (size, addr) = rx.recv_from(&mut buf).await?;
1097 ///
1098 /// let dgram = &buf[..size];
1099 /// assert_eq!(dgram, bytes);
1100 /// assert_eq!(addr.as_pathname().unwrap(), &tx_path);
1101 ///
1102 /// # Ok(())
1103 /// # }
1104 /// ```
1105 pub async fn send_to<P>(&self, buf: &[u8], target: P) -> io::Result<usize>
1106 where
1107 P: AsRef<Path>,
1108 {
1109 self.io
1110 .registration()
1111 .async_io(Interest::WRITABLE, || self.io.send_to(buf, target.as_ref()))
1112 .await
1113 }
1114
1115 /// Receives data from the socket.
1116 ///
1117 /// # Cancel safety
1118 ///
1119 /// This method is cancel safe. If `recv_from` is used as a branch in
1120 /// [`tokio::select!`](crate::select) and another branch
1121 /// completes first, it is guaranteed that no messages were received on this
1122 /// socket.
1123 ///
1124 /// # Examples
1125 /// ```
1126 /// # use std::error::Error;
1127 /// # #[tokio::main]
1128 /// # async fn main() -> Result<(), Box<dyn Error>> {
1129 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1130 /// use tokio::net::UnixDatagram;
1131 /// use tempfile::tempdir;
1132 ///
1133 /// // We use a temporary directory so that the socket
1134 /// // files left by the bound sockets will get cleaned up.
1135 /// let tmp = tempdir()?;
1136 ///
1137 /// // Bind each socket to a filesystem path
1138 /// let tx_path = tmp.path().join("tx");
1139 /// let tx = UnixDatagram::bind(&tx_path)?;
1140 /// let rx_path = tmp.path().join("rx");
1141 /// let rx = UnixDatagram::bind(&rx_path)?;
1142 ///
1143 /// let bytes = b"hello world";
1144 /// tx.send_to(bytes, &rx_path).await?;
1145 ///
1146 /// let mut buf = vec![0u8; 24];
1147 /// let (size, addr) = rx.recv_from(&mut buf).await?;
1148 ///
1149 /// let dgram = &buf[..size];
1150 /// assert_eq!(dgram, bytes);
1151 /// assert_eq!(addr.as_pathname().unwrap(), &tx_path);
1152 ///
1153 /// # Ok(())
1154 /// # }
1155 /// ```
1156 pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
1157 let (n, addr) = self
1158 .io
1159 .registration()
1160 .async_io(Interest::READABLE, || self.io.recv_from(buf))
1161 .await?;
1162
1163 Ok((n, SocketAddr(addr)))
1164 }
1165
1166 /// Attempts to receive a single datagram on the specified address.
1167 ///
1168 /// Note that on multiple calls to a `poll_*` method in the `recv` direction, only the
1169 /// `Waker` from the `Context` passed to the most recent call will be scheduled to
1170 /// receive a wakeup.
1171 ///
1172 /// # Return value
1173 ///
1174 /// The function returns:
1175 ///
1176 /// * `Poll::Pending` if the socket is not ready to read
1177 /// * `Poll::Ready(Ok(addr))` reads data from `addr` into `ReadBuf` if the socket is ready
1178 /// * `Poll::Ready(Err(e))` if an error is encountered.
1179 ///
1180 /// # Errors
1181 ///
1182 /// This function may encounter any standard I/O error except `WouldBlock`.
1183 pub fn poll_recv_from(
1184 &self,
1185 cx: &mut Context<'_>,
1186 buf: &mut ReadBuf<'_>,
1187 ) -> Poll<io::Result<SocketAddr>> {
1188 #[allow(clippy::blocks_in_conditions)]
1189 let (n, addr) = ready!(self.io.registration().poll_read_io(cx, || {
1190 // Safety: will not read the maybe uninitialized bytes.
1191 let b = unsafe {
1192 &mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
1193 };
1194
1195 self.io.recv_from(b)
1196 }))?;
1197
1198 // Safety: We trust `recv` to have filled up `n` bytes in the buffer.
1199 unsafe {
1200 buf.assume_init(n);
1201 }
1202 buf.advance(n);
1203 Poll::Ready(Ok(SocketAddr(addr)))
1204 }
1205
1206 /// Attempts to send data to the specified address.
1207 ///
1208 /// Note that on multiple calls to a `poll_*` method in the send direction, only the
1209 /// `Waker` from the `Context` passed to the most recent call will be scheduled to
1210 /// receive a wakeup.
1211 ///
1212 /// # Return value
1213 ///
1214 /// The function returns:
1215 ///
1216 /// * `Poll::Pending` if the socket is not ready to write
1217 /// * `Poll::Ready(Ok(n))` `n` is the number of bytes sent.
1218 /// * `Poll::Ready(Err(e))` if an error is encountered.
1219 ///
1220 /// # Errors
1221 ///
1222 /// This function may encounter any standard I/O error except `WouldBlock`.
1223 pub fn poll_send_to<P>(
1224 &self,
1225 cx: &mut Context<'_>,
1226 buf: &[u8],
1227 target: P,
1228 ) -> Poll<io::Result<usize>>
1229 where
1230 P: AsRef<Path>,
1231 {
1232 self.io
1233 .registration()
1234 .poll_write_io(cx, || self.io.send_to(buf, target.as_ref()))
1235 }
1236
1237 /// Attempts to send data on the socket to the remote address to which it
1238 /// was previously `connect`ed.
1239 ///
1240 /// The [`connect`] method will connect this socket to a remote address.
1241 /// This method will fail if the socket is not connected.
1242 ///
1243 /// Note that on multiple calls to a `poll_*` method in the send direction,
1244 /// only the `Waker` from the `Context` passed to the most recent call will
1245 /// be scheduled to receive a wakeup.
1246 ///
1247 /// # Return value
1248 ///
1249 /// The function returns:
1250 ///
1251 /// * `Poll::Pending` if the socket is not available to write
1252 /// * `Poll::Ready(Ok(n))` `n` is the number of bytes sent
1253 /// * `Poll::Ready(Err(e))` if an error is encountered.
1254 ///
1255 /// # Errors
1256 ///
1257 /// This function may encounter any standard I/O error except `WouldBlock`.
1258 ///
1259 /// [`connect`]: method@Self::connect
1260 pub fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
1261 self.io
1262 .registration()
1263 .poll_write_io(cx, || self.io.send(buf))
1264 }
1265
1266 /// Attempts to receive a single datagram message on the socket from the remote
1267 /// address to which it is `connect`ed.
1268 ///
1269 /// The [`connect`] method will connect this socket to a remote address. This method
1270 /// resolves to an error if the socket is not connected.
1271 ///
1272 /// Note that on multiple calls to a `poll_*` method in the `recv` direction, only the
1273 /// `Waker` from the `Context` passed to the most recent call will be scheduled to
1274 /// receive a wakeup.
1275 ///
1276 /// # Return value
1277 ///
1278 /// The function returns:
1279 ///
1280 /// * `Poll::Pending` if the socket is not ready to read
1281 /// * `Poll::Ready(Ok(()))` reads data `ReadBuf` if the socket is ready
1282 /// * `Poll::Ready(Err(e))` if an error is encountered.
1283 ///
1284 /// # Errors
1285 ///
1286 /// This function may encounter any standard I/O error except `WouldBlock`.
1287 ///
1288 /// [`connect`]: method@Self::connect
1289 pub fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
1290 #[allow(clippy::blocks_in_conditions)]
1291 let n = ready!(self.io.registration().poll_read_io(cx, || {
1292 // Safety: will not read the maybe uninitialized bytes.
1293 let b = unsafe {
1294 &mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
1295 };
1296
1297 self.io.recv(b)
1298 }))?;
1299
1300 // Safety: We trust `recv` to have filled up `n` bytes in the buffer.
1301 unsafe {
1302 buf.assume_init(n);
1303 }
1304 buf.advance(n);
1305 Poll::Ready(Ok(()))
1306 }
1307
1308 /// Tries to receive data from the socket without waiting.
1309 ///
1310 /// # Examples
1311 ///
1312 /// ```no_run
1313 /// use tokio::net::UnixDatagram;
1314 /// use std::io;
1315 ///
1316 /// #[tokio::main]
1317 /// async fn main() -> io::Result<()> {
1318 /// // Connect to a peer
1319 /// let dir = tempfile::tempdir().unwrap();
1320 /// let client_path = dir.path().join("client.sock");
1321 /// let server_path = dir.path().join("server.sock");
1322 /// let socket = UnixDatagram::bind(&client_path)?;
1323 ///
1324 /// loop {
1325 /// // Wait for the socket to be readable
1326 /// socket.readable().await?;
1327 ///
1328 /// // The buffer is **not** included in the async task and will
1329 /// // only exist on the stack.
1330 /// let mut buf = [0; 1024];
1331 ///
1332 /// // Try to recv data, this may still fail with `WouldBlock`
1333 /// // if the readiness event is a false positive.
1334 /// match socket.try_recv_from(&mut buf) {
1335 /// Ok((n, _addr)) => {
1336 /// println!("GOT {:?}", &buf[..n]);
1337 /// break;
1338 /// }
1339 /// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
1340 /// continue;
1341 /// }
1342 /// Err(e) => {
1343 /// return Err(e);
1344 /// }
1345 /// }
1346 /// }
1347 ///
1348 /// Ok(())
1349 /// }
1350 /// ```
1351 pub fn try_recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
1352 let (n, addr) = self
1353 .io
1354 .registration()
1355 .try_io(Interest::READABLE, || self.io.recv_from(buf))?;
1356
1357 Ok((n, SocketAddr(addr)))
1358 }
1359
1360 /// Tries to read or write from the socket using a user-provided IO operation.
1361 ///
1362 /// If the socket is ready, the provided closure is called. The closure
1363 /// should attempt to perform IO operation on the socket by manually
1364 /// calling the appropriate syscall. If the operation fails because the
1365 /// socket is not actually ready, then the closure should return a
1366 /// `WouldBlock` error and the readiness flag is cleared. The return value
1367 /// of the closure is then returned by `try_io`.
1368 ///
1369 /// If the socket is not ready, then the closure is not called
1370 /// and a `WouldBlock` error is returned.
1371 ///
1372 /// The closure should only return a `WouldBlock` error if it has performed
1373 /// an IO operation on the socket that failed due to the socket not being
1374 /// ready. Returning a `WouldBlock` error in any other situation will
1375 /// incorrectly clear the readiness flag, which can cause the socket to
1376 /// behave incorrectly.
1377 ///
1378 /// The closure should not perform the IO operation using any of the methods
1379 /// defined on the Tokio `UnixDatagram` type, as this will mess with the
1380 /// readiness flag and can cause the socket to behave incorrectly.
1381 ///
1382 /// This method is not intended to be used with combined interests.
1383 /// The closure should perform only one type of IO operation, so it should not
1384 /// require more than one ready state. This method may panic or sleep forever
1385 /// if it is called with a combined interest.
1386 ///
1387 /// Usually, [`readable()`], [`writable()`] or [`ready()`] is used with this function.
1388 ///
1389 /// [`readable()`]: UnixDatagram::readable()
1390 /// [`writable()`]: UnixDatagram::writable()
1391 /// [`ready()`]: UnixDatagram::ready()
1392 pub fn try_io<R>(
1393 &self,
1394 interest: Interest,
1395 f: impl FnOnce() -> io::Result<R>,
1396 ) -> io::Result<R> {
1397 self.io
1398 .registration()
1399 .try_io(interest, || self.io.try_io(f))
1400 }
1401
1402 /// Reads or writes from the socket using a user-provided IO operation.
1403 ///
1404 /// The readiness of the socket is awaited and when the socket is ready,
1405 /// the provided closure is called. The closure should attempt to perform
1406 /// IO operation on the socket by manually calling the appropriate syscall.
1407 /// If the operation fails because the socket is not actually ready,
1408 /// then the closure should return a `WouldBlock` error. In such case the
1409 /// readiness flag is cleared and the socket readiness is awaited again.
1410 /// This loop is repeated until the closure returns an `Ok` or an error
1411 /// other than `WouldBlock`.
1412 ///
1413 /// The closure should only return a `WouldBlock` error if it has performed
1414 /// an IO operation on the socket that failed due to the socket not being
1415 /// ready. Returning a `WouldBlock` error in any other situation will
1416 /// incorrectly clear the readiness flag, which can cause the socket to
1417 /// behave incorrectly.
1418 ///
1419 /// The closure should not perform the IO operation using any of the methods
1420 /// defined on the Tokio `UnixDatagram` type, as this will mess with the
1421 /// readiness flag and can cause the socket to behave incorrectly.
1422 ///
1423 /// This method is not intended to be used with combined interests.
1424 /// The closure should perform only one type of IO operation, so it should not
1425 /// require more than one ready state. This method may panic or sleep forever
1426 /// if it is called with a combined interest.
1427 pub async fn async_io<R>(
1428 &self,
1429 interest: Interest,
1430 mut f: impl FnMut() -> io::Result<R>,
1431 ) -> io::Result<R> {
1432 self.io
1433 .registration()
1434 .async_io(interest, || self.io.try_io(&mut f))
1435 .await
1436 }
1437
1438 /// Returns the local address that this socket is bound to.
1439 ///
1440 /// # Examples
1441 /// For a socket bound to a local path
1442 /// ```
1443 /// # use std::error::Error;
1444 /// # #[tokio::main]
1445 /// # async fn main() -> Result<(), Box<dyn Error>> {
1446 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1447 /// use tokio::net::UnixDatagram;
1448 /// use tempfile::tempdir;
1449 ///
1450 /// // We use a temporary directory so that the socket
1451 /// // files left by the bound sockets will get cleaned up.
1452 /// let tmp = tempdir()?;
1453 ///
1454 /// // Bind socket to a filesystem path
1455 /// let socket_path = tmp.path().join("socket");
1456 /// let socket = UnixDatagram::bind(&socket_path)?;
1457 ///
1458 /// assert_eq!(socket.local_addr()?.as_pathname().unwrap(), &socket_path);
1459 ///
1460 /// # Ok(())
1461 /// # }
1462 /// ```
1463 ///
1464 /// For an unbound socket
1465 /// ```
1466 /// # use std::error::Error;
1467 /// # #[tokio::main]
1468 /// # async fn main() -> Result<(), Box<dyn Error>> {
1469 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1470 /// use tokio::net::UnixDatagram;
1471 ///
1472 /// // Create an unbound socket
1473 /// let socket = UnixDatagram::unbound()?;
1474 ///
1475 /// assert!(socket.local_addr()?.is_unnamed());
1476 ///
1477 /// # Ok(())
1478 /// # }
1479 /// ```
1480 pub fn local_addr(&self) -> io::Result<SocketAddr> {
1481 self.io.local_addr().map(SocketAddr)
1482 }
1483
1484 /// Returns the address of this socket's peer.
1485 ///
1486 /// The `connect` method will connect the socket to a peer.
1487 ///
1488 /// # Examples
1489 /// For a peer with a local path
1490 /// ```
1491 /// # use std::error::Error;
1492 /// # #[tokio::main]
1493 /// # async fn main() -> Result<(), Box<dyn Error>> {
1494 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1495 /// use tokio::net::UnixDatagram;
1496 /// use tempfile::tempdir;
1497 ///
1498 /// // Create an unbound socket
1499 /// let tx = UnixDatagram::unbound()?;
1500 ///
1501 /// // Create another, bound socket
1502 /// let tmp = tempdir()?;
1503 /// let rx_path = tmp.path().join("rx");
1504 /// let rx = UnixDatagram::bind(&rx_path)?;
1505 ///
1506 /// // Connect to the bound socket
1507 /// tx.connect(&rx_path)?;
1508 ///
1509 /// assert_eq!(tx.peer_addr()?.as_pathname().unwrap(), &rx_path);
1510 ///
1511 /// # Ok(())
1512 /// # }
1513 /// ```
1514 ///
1515 /// For an unbound peer
1516 /// ```
1517 /// # use std::error::Error;
1518 /// # #[tokio::main]
1519 /// # async fn main() -> Result<(), Box<dyn Error>> {
1520 /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
1521 /// use tokio::net::UnixDatagram;
1522 ///
1523 /// // Create the pair of sockets
1524 /// let (sock1, sock2) = UnixDatagram::pair()?;
1525 ///
1526 /// assert!(sock1.peer_addr()?.is_unnamed());
1527 ///
1528 /// # Ok(())
1529 /// # }
1530 /// ```
1531 pub fn peer_addr(&self) -> io::Result<SocketAddr> {
1532 self.io.peer_addr().map(SocketAddr)
1533 }
1534
1535 /// Returns the value of the `SO_ERROR` option.
1536 ///
1537 /// # Examples
1538 /// ```
1539 /// # use std::error::Error;
1540 /// # #[tokio::main]
1541 /// # async fn main() -> Result<(), Box<dyn Error>> {
1542 /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1543 /// use tokio::net::UnixDatagram;
1544 ///
1545 /// // Create an unbound socket
1546 /// let socket = UnixDatagram::unbound()?;
1547 ///
1548 /// if let Ok(Some(err)) = socket.take_error() {
1549 /// println!("Got error: {:?}", err);
1550 /// }
1551 ///
1552 /// # Ok(())
1553 /// # }
1554 /// ```
1555 pub fn take_error(&self) -> io::Result<Option<io::Error>> {
1556 self.io.take_error()
1557 }
1558
1559 /// Shuts down the read, write, or both halves of this connection.
1560 ///
1561 /// This function will cause all pending and future I/O calls on the
1562 /// specified portions to immediately return with an appropriate value
1563 /// (see the documentation of `Shutdown`).
1564 ///
1565 /// # Examples
1566 /// ```
1567 /// # use std::error::Error;
1568 /// # #[tokio::main]
1569 /// # async fn main() -> Result<(), Box<dyn Error>> {
1570 /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
1571 /// use tokio::net::UnixDatagram;
1572 /// use std::net::Shutdown;
1573 ///
1574 /// // Create an unbound socket
1575 /// let (socket, other) = UnixDatagram::pair()?;
1576 ///
1577 /// socket.shutdown(Shutdown::Both)?;
1578 ///
1579 /// // NOTE: the following commented out code does NOT work as expected.
1580 /// // Due to an underlying issue, the recv call will block indefinitely.
1581 /// // See: https://github.com/tokio-rs/tokio/issues/1679
1582 /// //let mut buff = vec![0u8; 24];
1583 /// //let size = socket.recv(&mut buff).await?;
1584 /// //assert_eq!(size, 0);
1585 ///
1586 /// let send_result = socket.send(b"hello world").await;
1587 /// assert!(send_result.is_err());
1588 ///
1589 /// # Ok(())
1590 /// # }
1591 /// ```
1592 pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
1593 self.io.shutdown(how)
1594 }
1595}
1596
1597impl TryFrom<std::os::unix::net::UnixDatagram> for UnixDatagram {
1598 type Error = io::Error;
1599
1600 /// Consumes stream, returning the Tokio I/O object.
1601 ///
1602 /// This is equivalent to
1603 /// [`UnixDatagram::from_std(stream)`](UnixDatagram::from_std).
1604 fn try_from(stream: std::os::unix::net::UnixDatagram) -> Result<Self, Self::Error> {
1605 Self::from_std(stream)
1606 }
1607}
1608
1609impl fmt::Debug for UnixDatagram {
1610 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1611 (*self.io).fmt(f)
1612 }
1613}
1614
1615impl AsRawFd for UnixDatagram {
1616 fn as_raw_fd(&self) -> RawFd {
1617 self.io.as_raw_fd()
1618 }
1619}
1620
1621impl AsFd for UnixDatagram {
1622 fn as_fd(&self) -> BorrowedFd<'_> {
1623 unsafe { BorrowedFd::borrow_raw(self.as_raw_fd()) }
1624 }
1625}