compio_net/
unix.rs

1use std::{
2    future::Future,
3    io,
4    path::Path,
5    pin::Pin,
6    task::{Context, Poll},
7};
8
9use compio_buf::{BufResult, IoBuf, IoBufMut, IoVectoredBuf, IoVectoredBufMut};
10use compio_driver::{
11    BufferRef, SharedFd, impl_raw_fd,
12    op::{RecvFlags, RecvMsgMultiResult, SendMsgZc, SendVectoredZc, SendZc},
13};
14use compio_io::{
15    AsyncRead, AsyncReadManaged, AsyncReadMulti, AsyncWrite, AsyncWriteZerocopy,
16    ancillary::{
17        AsyncReadAncillary, AsyncReadAncillaryManaged, AsyncReadAncillaryMulti,
18        AsyncWriteAncillary, AsyncWriteAncillaryZerocopy, ReturnFlags,
19    },
20    util::Splittable,
21};
22#[cfg(unix)]
23use compio_runtime::Runtime;
24use compio_runtime::fd::PollFd;
25use futures_util::{Stream, StreamExt, stream::FusedStream};
26use socket2::{Domain, SockAddr, Socket as Socket2, Type};
27
28use crate::{Extract, Incoming, MSG_NOSIGNAL, ReadHalf, Socket, WriteHalf, Zerocopy};
29
30/// A Unix socket server, listening for connections.
31///
32/// You can accept a new connection by using the [`UnixListener::accept`]
33/// method.
34///
35/// # Examples
36///
37/// ```
38/// use compio_io::{AsyncReadExt, AsyncWriteExt};
39/// use compio_net::{UnixListener, UnixStream};
40/// use tempfile::tempdir;
41///
42/// let dir = tempdir().unwrap();
43/// let sock_file = dir.path().join("unix-server.sock");
44///
45/// # compio_runtime::Runtime::new().unwrap().block_on(async move {
46/// let listener = UnixListener::bind(&sock_file).await.unwrap();
47///
48/// let (mut tx, (mut rx, _)) =
49///     futures_util::try_join!(UnixStream::connect(&sock_file), listener.accept()).unwrap();
50///
51/// tx.write_all("test").await.0.unwrap();
52///
53/// let (_, buf) = rx.read_exact(Vec::with_capacity(4)).await.unwrap();
54///
55/// assert_eq!(buf, b"test");
56/// # });
57/// ```
58#[derive(Debug, Clone)]
59pub struct UnixListener {
60    inner: Socket,
61}
62
63impl UnixListener {
64    /// Creates a new [`UnixListener`], which will be bound to the specified
65    /// file path. See [`UnixListener::bind_addr`] for more details.
66    pub async fn bind(path: impl AsRef<Path>) -> io::Result<Self> {
67        Self::bind_addr(&SockAddr::unix(path)?).await
68    }
69
70    /// Creates a new [`UnixListener`] with [`SockAddr`], which will be bound to
71    /// the specified file path. The file path cannot yet exist.
72    ///
73    /// To configure the socket before binding, you can use the [`UnixSocket`]
74    /// type.
75    pub async fn bind_addr(addr: &SockAddr) -> io::Result<Self> {
76        if !addr.is_unix() {
77            return Err(io::Error::new(
78                io::ErrorKind::InvalidInput,
79                "addr is not unix socket address",
80            ));
81        }
82
83        let socket = Socket::new(addr.domain(), Type::STREAM, None).await?;
84        socket.bind(addr).await?;
85        socket.listen(1024).await?;
86        Ok(UnixListener { inner: socket })
87    }
88
89    #[cfg(unix)]
90    /// Creates new UnixListener from a [`std::os::unix::net::UnixListener`].
91    pub fn from_std(stream: std::os::unix::net::UnixListener) -> io::Result<Self> {
92        if Runtime::with_current(|r| r.driver_type().is_polling()) {
93            stream.set_nonblocking(true)?;
94        }
95
96        Ok(Self {
97            inner: Socket::from_socket2(Socket2::from(stream))?,
98        })
99    }
100
101    /// Close the socket. If the returned future is dropped before polling, the
102    /// socket won't be closed.
103    ///
104    /// See [`TcpStream::close`] for more details.
105    ///
106    /// [`TcpStream::close`]: crate::tcp::TcpStream::close
107    pub fn close(self) -> impl Future<Output = io::Result<()>> {
108        self.inner.close()
109    }
110
111    /// Accepts a new incoming connection from this listener.
112    ///
113    /// This function will yield once a new Unix domain socket connection
114    /// is established. When established, the corresponding [`UnixStream`] and
115    /// the remote peer’s address will be returned.
116    pub async fn accept(&self) -> io::Result<(UnixStream, SockAddr)> {
117        let (socket, addr) = self.inner.accept().await?;
118        let stream = UnixStream { inner: socket };
119        Ok((stream, addr))
120    }
121
122    /// Accepts a new incoming connection from this listener using the provided
123    /// socket.
124    #[cfg(windows)]
125    pub async fn accept_with(&self, sock: UnixSocket) -> io::Result<(UnixStream, SockAddr)> {
126        let (socket, addr) = self.inner.accept_with(sock.inner).await?;
127        let stream = UnixStream { inner: socket };
128        Ok((stream, addr))
129    }
130
131    /// Returns a stream of incoming connections to this listener.
132    pub fn incoming(&self) -> UnixIncoming<'_> {
133        UnixIncoming {
134            inner: self.inner.incoming(),
135        }
136    }
137
138    /// Returns the local address that this listener is bound to.
139    pub fn local_addr(&self) -> io::Result<SockAddr> {
140        self.inner.local_addr()
141    }
142
143    /// Returns the value of the `SO_ERROR` option.
144    pub fn take_error(&self) -> io::Result<Option<io::Error>> {
145        self.inner.socket.take_error()
146    }
147}
148
149impl_raw_fd!(UnixListener, socket2::Socket, inner, socket);
150
151/// A stream of incoming Unix connections.
152pub struct UnixIncoming<'a> {
153    inner: Incoming<'a>,
154}
155
156impl Stream for UnixIncoming<'_> {
157    type Item = io::Result<UnixStream>;
158
159    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
160        let this = self.get_mut();
161        this.inner.poll_next_unpin(cx).map(|res| {
162            res.map(|res| {
163                let socket = res?;
164                Ok(UnixStream { inner: socket })
165            })
166        })
167    }
168}
169
170impl FusedStream for UnixIncoming<'_> {
171    fn is_terminated(&self) -> bool {
172        self.inner.is_terminated()
173    }
174}
175
176/// A Unix stream between two local sockets on Windows & WSL.
177///
178/// A Unix stream can either be created by connecting to an endpoint, via the
179/// `connect` method, or by accepting a connection from a listener.
180///
181/// # Examples
182///
183/// ```no_run
184/// use compio_io::AsyncWrite;
185/// use compio_net::UnixStream;
186///
187/// # compio_runtime::Runtime::new().unwrap().block_on(async {
188/// // Connect to a peer
189/// let mut stream = UnixStream::connect("unix-server.sock").await.unwrap();
190///
191/// // Write some data.
192/// stream.write("hello world!").await.unwrap();
193/// # })
194/// ```
195#[derive(Debug, Clone)]
196pub struct UnixStream {
197    inner: Socket,
198}
199
200impl UnixStream {
201    /// Opens a Unix connection to the specified file path. See
202    /// [`UnixStream::connect_addr`] for more details.
203    pub async fn connect(path: impl AsRef<Path>) -> io::Result<Self> {
204        Self::connect_addr(&SockAddr::unix(path)?).await
205    }
206
207    /// Opens a Unix connection to the specified address. There must be a
208    /// [`UnixListener`] or equivalent listening on the corresponding Unix
209    /// domain socket to successfully connect and return a [`UnixStream`].
210    ///
211    /// To configure the socket before connecting, you can use the
212    /// [`UnixSocket`] type.
213    pub async fn connect_addr(addr: &SockAddr) -> io::Result<Self> {
214        if !addr.is_unix() {
215            return Err(io::Error::new(
216                io::ErrorKind::InvalidInput,
217                "addr is not unix socket address",
218            ));
219        }
220        let socket = Socket::new(Domain::UNIX, Type::STREAM, None).await?;
221        #[cfg(windows)]
222        {
223            let new_addr = empty_unix_socket();
224            socket.bind(&new_addr).await?
225        }
226        socket.connect_async(addr).await?;
227        let unix_stream = UnixStream { inner: socket };
228        Ok(unix_stream)
229    }
230
231    /// Creates new UnixStream from a [`std::os::unix::net::UnixStream`].
232    #[cfg(unix)]
233    pub fn from_std(stream: std::os::unix::net::UnixStream) -> io::Result<Self> {
234        if Runtime::with_current(|r| r.driver_type().is_polling()) {
235            stream.set_nonblocking(true)?;
236        }
237
238        Ok(Self {
239            inner: Socket::from_socket2(Socket2::from(stream))?,
240        })
241    }
242
243    /// Close the socket. If the returned future is dropped before polling, the
244    /// socket won't be closed.
245    ///
246    /// See [`TcpStream::close`] for more details.
247    ///
248    /// [`TcpStream::close`]: crate::tcp::TcpStream::close
249    pub fn close(self) -> impl Future<Output = io::Result<()>> {
250        self.inner.close()
251    }
252
253    /// Returns the socket path of the remote peer of this connection.
254    pub fn peer_addr(&self) -> io::Result<SockAddr> {
255        #[allow(unused_mut)]
256        let mut addr = self.inner.peer_addr()?;
257        #[cfg(windows)]
258        {
259            fix_unix_socket_length(&mut addr);
260        }
261        Ok(addr)
262    }
263
264    /// Returns the socket path of the local half of this connection.
265    pub fn local_addr(&self) -> io::Result<SockAddr> {
266        self.inner.local_addr()
267    }
268
269    /// Returns the value of the `SO_ERROR` option.
270    pub fn take_error(&self) -> io::Result<Option<io::Error>> {
271        self.inner.socket.take_error()
272    }
273
274    /// Splits a [`UnixStream`] into a read half and a write half, which can be
275    /// used to read and write the stream concurrently.
276    ///
277    /// This method is more efficient than
278    /// [`into_split`](UnixStream::into_split), but the halves cannot
279    /// be moved into independently spawned tasks.
280    pub fn split(&self) -> (ReadHalf<'_, Self>, WriteHalf<'_, Self>) {
281        crate::split(self)
282    }
283
284    /// Splits a [`UnixStream`] into a read half and a write half, which can be
285    /// used to read and write the stream concurrently.
286    ///
287    /// Unlike [`split`](UnixStream::split), the owned halves can be moved to
288    /// separate tasks.
289    pub fn into_split(self) -> (Self, Self) {
290        (self.clone(), self)
291    }
292
293    /// Create [`PollFd`] from inner socket.
294    pub fn to_poll_fd(&self) -> io::Result<PollFd<Socket2>> {
295        self.inner.to_poll_fd()
296    }
297
298    /// Create [`PollFd`] from inner socket.
299    pub fn into_poll_fd(self) -> io::Result<PollFd<Socket2>> {
300        self.inner.into_poll_fd()
301    }
302
303    /// Close the connection of the socket, and reuse it to create a new
304    /// connection. This method is useful when the socket is created by
305    /// [`UnixListener::accept`], and will be reused in
306    /// [`UnixListener::accept_with`] to accept a new connection.
307    #[cfg(windows)]
308    pub async fn disconnect(self) -> io::Result<UnixSocket> {
309        self.inner.disconnect().await?;
310        Ok(UnixSocket { inner: self.inner })
311    }
312}
313
314impl AsyncRead for UnixStream {
315    #[inline]
316    async fn read<B: IoBufMut>(&mut self, buf: B) -> BufResult<usize, B> {
317        (&*self).read(buf).await
318    }
319
320    #[inline]
321    async fn read_vectored<V: IoVectoredBufMut>(&mut self, buf: V) -> BufResult<usize, V> {
322        (&*self).read_vectored(buf).await
323    }
324}
325
326impl AsyncRead for &UnixStream {
327    #[inline]
328    async fn read<B: IoBufMut>(&mut self, buf: B) -> BufResult<usize, B> {
329        self.inner.recv(buf, RecvFlags::empty()).await
330    }
331
332    #[inline]
333    async fn read_vectored<V: IoVectoredBufMut>(&mut self, buf: V) -> BufResult<usize, V> {
334        self.inner.recv_vectored(buf, RecvFlags::empty()).await
335    }
336}
337
338impl AsyncReadManaged for UnixStream {
339    type Buffer = BufferRef;
340
341    async fn read_managed(&mut self, len: usize) -> io::Result<Option<Self::Buffer>> {
342        (&*self).read_managed(len).await
343    }
344}
345
346impl AsyncReadManaged for &UnixStream {
347    type Buffer = BufferRef;
348
349    async fn read_managed(&mut self, len: usize) -> io::Result<Option<Self::Buffer>> {
350        self.inner.recv_managed(len, RecvFlags::empty()).await
351    }
352}
353
354impl AsyncReadMulti for UnixStream {
355    fn read_multi(&mut self, len: usize) -> impl Stream<Item = io::Result<Self::Buffer>> {
356        self.inner.recv_multi(len, RecvFlags::empty())
357    }
358}
359
360impl AsyncReadMulti for &UnixStream {
361    fn read_multi(&mut self, len: usize) -> impl Stream<Item = io::Result<Self::Buffer>> {
362        self.inner.recv_multi(len, RecvFlags::empty())
363    }
364}
365
366impl AsyncReadAncillary for UnixStream {
367    #[inline]
368    async fn read_with_ancillary<T: IoBufMut, C: IoBufMut>(
369        &mut self,
370        buffer: T,
371        control: C,
372    ) -> BufResult<(usize, usize, ReturnFlags), (T, C)> {
373        (&*self).read_with_ancillary(buffer, control).await
374    }
375
376    #[inline]
377    async fn read_vectored_with_ancillary<T: IoVectoredBufMut, C: IoBufMut>(
378        &mut self,
379        buffer: T,
380        control: C,
381    ) -> BufResult<(usize, usize, ReturnFlags), (T, C)> {
382        (&*self).read_vectored_with_ancillary(buffer, control).await
383    }
384}
385
386impl AsyncReadAncillary for &UnixStream {
387    #[inline]
388    async fn read_with_ancillary<T: IoBufMut, C: IoBufMut>(
389        &mut self,
390        buffer: T,
391        control: C,
392    ) -> BufResult<(usize, usize, ReturnFlags), (T, C)> {
393        self.inner
394            .recv_msg(buffer, control, RecvFlags::empty())
395            .await
396            .map_res(|(res, len, _addr, flags)| (res, len, flags))
397    }
398
399    #[inline]
400    async fn read_vectored_with_ancillary<T: IoVectoredBufMut, C: IoBufMut>(
401        &mut self,
402        buffer: T,
403        control: C,
404    ) -> BufResult<(usize, usize, ReturnFlags), (T, C)> {
405        self.inner
406            .recv_msg_vectored(buffer, control, RecvFlags::empty())
407            .await
408            .map_res(|(res, len, _addr, flags)| (res, len, flags))
409    }
410}
411
412impl AsyncReadAncillaryManaged for UnixStream {
413    #[inline]
414    async fn read_managed_with_ancillary<C: IoBufMut>(
415        &mut self,
416        len: usize,
417        control: C,
418    ) -> io::Result<Option<(Self::Buffer, C, ReturnFlags)>> {
419        (&*self).read_managed_with_ancillary(len, control).await
420    }
421}
422
423impl AsyncReadAncillaryManaged for &UnixStream {
424    #[inline]
425    async fn read_managed_with_ancillary<C: IoBufMut>(
426        &mut self,
427        len: usize,
428        control: C,
429    ) -> io::Result<Option<(Self::Buffer, C, ReturnFlags)>> {
430        self.inner
431            .recv_msg_managed(len, control, RecvFlags::empty())
432            .await
433            .map(|res| res.map(|(res, len, _addr, flags)| (res, len, flags)))
434    }
435}
436
437impl AsyncReadAncillaryMulti for UnixStream {
438    type Return = RecvMsgMultiResult;
439
440    #[inline]
441    fn read_multi_with_ancillary(
442        &mut self,
443        control_len: usize,
444    ) -> impl Stream<Item = io::Result<Self::Return>> {
445        self.inner.recv_msg_multi(control_len, RecvFlags::empty())
446    }
447}
448
449impl AsyncReadAncillaryMulti for &UnixStream {
450    type Return = RecvMsgMultiResult;
451
452    #[inline]
453    fn read_multi_with_ancillary(
454        &mut self,
455        control_len: usize,
456    ) -> impl Stream<Item = io::Result<Self::Return>> {
457        self.inner.recv_msg_multi(control_len, RecvFlags::empty())
458    }
459}
460
461impl AsyncWrite for UnixStream {
462    #[inline]
463    async fn write<T: IoBuf>(&mut self, buf: T) -> BufResult<usize, T> {
464        (&*self).write(buf).await
465    }
466
467    #[inline]
468    async fn write_vectored<T: IoVectoredBuf>(&mut self, buf: T) -> BufResult<usize, T> {
469        (&*self).write_vectored(buf).await
470    }
471
472    #[inline]
473    async fn flush(&mut self) -> io::Result<()> {
474        (&*self).flush().await
475    }
476
477    #[inline]
478    async fn shutdown(&mut self) -> io::Result<()> {
479        (&*self).shutdown().await
480    }
481}
482
483impl AsyncWrite for &UnixStream {
484    #[inline]
485    async fn write<T: IoBuf>(&mut self, buf: T) -> BufResult<usize, T> {
486        self.inner.send(buf, MSG_NOSIGNAL).await
487    }
488
489    #[inline]
490    async fn write_vectored<T: IoVectoredBuf>(&mut self, buf: T) -> BufResult<usize, T> {
491        self.inner.send_vectored(buf, MSG_NOSIGNAL).await
492    }
493
494    #[inline]
495    async fn flush(&mut self) -> io::Result<()> {
496        Ok(())
497    }
498
499    #[inline]
500    async fn shutdown(&mut self) -> io::Result<()> {
501        self.inner.shutdown().await
502    }
503}
504
505impl AsyncWriteZerocopy for UnixStream {
506    type BufferReadyFuture<T: IoBuf> = Zerocopy<SendZc<T, SharedFd<Socket2>>>;
507    type VectoredBufferReadyFuture<T: IoVectoredBuf> =
508        Zerocopy<SendVectoredZc<T, SharedFd<Socket2>>>;
509
510    async fn write_zerocopy<T: IoBuf>(
511        &mut self,
512        buf: T,
513    ) -> BufResult<usize, Self::BufferReadyFuture<T>> {
514        self.inner.send_zerocopy(buf, MSG_NOSIGNAL).await
515    }
516
517    async fn write_zerocopy_vectored<T: IoVectoredBuf>(
518        &mut self,
519        buf: T,
520    ) -> BufResult<usize, Self::VectoredBufferReadyFuture<T>> {
521        self.inner.send_zerocopy_vectored(buf, MSG_NOSIGNAL).await
522    }
523}
524
525impl AsyncWriteZerocopy for &UnixStream {
526    type BufferReadyFuture<T: IoBuf> = Zerocopy<SendZc<T, SharedFd<Socket2>>>;
527    type VectoredBufferReadyFuture<T: IoVectoredBuf> =
528        Zerocopy<SendVectoredZc<T, SharedFd<Socket2>>>;
529
530    async fn write_zerocopy<T: IoBuf>(
531        &mut self,
532        buf: T,
533    ) -> BufResult<usize, Self::BufferReadyFuture<T>> {
534        self.inner.send_zerocopy(buf, MSG_NOSIGNAL).await
535    }
536
537    async fn write_zerocopy_vectored<T: IoVectoredBuf>(
538        &mut self,
539        buf: T,
540    ) -> BufResult<usize, Self::VectoredBufferReadyFuture<T>> {
541        self.inner.send_zerocopy_vectored(buf, MSG_NOSIGNAL).await
542    }
543}
544
545impl AsyncWriteAncillaryZerocopy for UnixStream {
546    type BufferReadyFuture<T: IoBuf, C: IoBuf> =
547        Extract<Zerocopy<SendMsgZc<[T; 1], C, SharedFd<Socket2>>>, T, C>;
548    type VectoredBufferReadyFuture<T: IoVectoredBuf, C: IoBuf> =
549        Zerocopy<SendMsgZc<T, C, SharedFd<Socket2>>>;
550
551    async fn write_zerocopy_with_ancillary<T: IoBuf, C: IoBuf>(
552        &mut self,
553        buf: T,
554        control: C,
555    ) -> BufResult<usize, Self::BufferReadyFuture<T, C>> {
556        self.inner
557            .send_msg_zerocopy(buf, control, None, MSG_NOSIGNAL)
558            .await
559    }
560
561    async fn write_zerocopy_vectored_with_ancillary<T: IoVectoredBuf, C: IoBuf>(
562        &mut self,
563        buf: T,
564        control: C,
565    ) -> BufResult<usize, Self::VectoredBufferReadyFuture<T, C>> {
566        self.inner
567            .send_msg_zerocopy_vectored(buf, control, None, MSG_NOSIGNAL)
568            .await
569    }
570}
571
572impl AsyncWriteAncillaryZerocopy for &UnixStream {
573    type BufferReadyFuture<T: IoBuf, C: IoBuf> =
574        Extract<Zerocopy<SendMsgZc<[T; 1], C, SharedFd<Socket2>>>, T, C>;
575    type VectoredBufferReadyFuture<T: IoVectoredBuf, C: IoBuf> =
576        Zerocopy<SendMsgZc<T, C, SharedFd<Socket2>>>;
577
578    async fn write_zerocopy_with_ancillary<T: IoBuf, C: IoBuf>(
579        &mut self,
580        buf: T,
581        control: C,
582    ) -> BufResult<usize, Self::BufferReadyFuture<T, C>> {
583        self.inner
584            .send_msg_zerocopy(buf, control, None, MSG_NOSIGNAL)
585            .await
586    }
587
588    async fn write_zerocopy_vectored_with_ancillary<T: IoVectoredBuf, C: IoBuf>(
589        &mut self,
590        buf: T,
591        control: C,
592    ) -> BufResult<usize, Self::VectoredBufferReadyFuture<T, C>> {
593        self.inner
594            .send_msg_zerocopy_vectored(buf, control, None, MSG_NOSIGNAL)
595            .await
596    }
597}
598
599/// # Example
600///
601/// Send and receive a file descriptor over a Unix socket pair using
602/// `SCM_RIGHTS`:
603///
604/// ```
605/// # #[cfg(unix)] {
606/// use std::os::unix::io::RawFd;
607///
608/// use compio_io::ancillary::*;
609/// use compio_net::UnixStream;
610///
611/// const BUF_SIZE: usize = ancillary_space::<RawFd>();
612///
613/// # compio_runtime::Runtime::new().unwrap().block_on(async {
614/// // Create a socket pair.
615/// let (std_a, std_b) = std::os::unix::net::UnixStream::pair().unwrap();
616/// let mut a = UnixStream::from_std(std_a).unwrap();
617/// let mut b = UnixStream::from_std(std_b).unwrap();
618///
619/// // Pass fd 0 (stdin) as ancillary data via SCM_RIGHTS.
620/// let mut ctrl_send = AncillaryBuf::<BUF_SIZE>::new();
621/// let mut builder = ctrl_send.builder();
622/// builder
623///     .push(libc::SOL_SOCKET, libc::SCM_RIGHTS, &(0 as RawFd))
624///     .unwrap();
625///
626/// // Send the payload together with the ancillary data.
627/// a.write_with_ancillary(b"hello", ctrl_send).await.0.unwrap();
628///
629/// // Receive on the other end.
630/// let payload = Vec::with_capacity(5);
631/// let ctrl_recv = AncillaryBuf::<BUF_SIZE>::new();
632/// let ((_, ctrl_len, _flags), (payload, ctrl_recv)) =
633///     b.read_with_ancillary(payload, ctrl_recv).await.unwrap();
634///
635/// assert_eq!(&payload[..], b"hello");
636///
637/// // Parse the received ancillary messages.
638/// let mut iter = unsafe { AncillaryIter::new(&ctrl_recv[..ctrl_len]) };
639/// let msg = iter.next().unwrap();
640/// assert_eq!(msg.level(), libc::SOL_SOCKET);
641/// assert_eq!(msg.ty(), libc::SCM_RIGHTS);
642/// // The kernel duplicates the fd, so the received value may differ.
643/// let _received_fd = unsafe { msg.data::<RawFd>() };
644/// assert!(iter.next().is_none());
645/// # });
646/// # }
647/// ```
648impl AsyncWriteAncillary for UnixStream {
649    #[inline]
650    async fn write_with_ancillary<T: IoBuf, C: IoBuf>(
651        &mut self,
652        buffer: T,
653        control: C,
654    ) -> BufResult<usize, (T, C)> {
655        (&*self).write_with_ancillary(buffer, control).await
656    }
657
658    #[inline]
659    async fn write_vectored_with_ancillary<T: IoVectoredBuf, C: IoBuf>(
660        &mut self,
661        buffer: T,
662        control: C,
663    ) -> BufResult<usize, (T, C)> {
664        (&*self)
665            .write_vectored_with_ancillary(buffer, control)
666            .await
667    }
668}
669
670impl AsyncWriteAncillary for &UnixStream {
671    #[inline]
672    async fn write_with_ancillary<T: IoBuf, C: IoBuf>(
673        &mut self,
674        buffer: T,
675        control: C,
676    ) -> BufResult<usize, (T, C)> {
677        self.inner
678            .send_msg(buffer, control, None, MSG_NOSIGNAL)
679            .await
680    }
681
682    #[inline]
683    async fn write_vectored_with_ancillary<T: IoVectoredBuf, C: IoBuf>(
684        &mut self,
685        buffer: T,
686        control: C,
687    ) -> BufResult<usize, (T, C)> {
688        self.inner
689            .send_msg_vectored(buffer, control, None, MSG_NOSIGNAL)
690            .await
691    }
692}
693
694impl Splittable for UnixStream {
695    type ReadHalf = Self;
696    type WriteHalf = Self;
697
698    fn split(self) -> (Self::ReadHalf, Self::WriteHalf) {
699        self.into_split()
700    }
701}
702
703impl<'a> Splittable for &'a UnixStream {
704    type ReadHalf = ReadHalf<'a, UnixStream>;
705    type WriteHalf = WriteHalf<'a, UnixStream>;
706
707    fn split(self) -> (Self::ReadHalf, Self::WriteHalf) {
708        crate::split(self)
709    }
710}
711
712impl<'a> Splittable for &'a mut UnixStream {
713    type ReadHalf = ReadHalf<'a, UnixStream>;
714    type WriteHalf = WriteHalf<'a, UnixStream>;
715
716    fn split(self) -> (Self::ReadHalf, Self::WriteHalf) {
717        crate::split(self)
718    }
719}
720
721impl_raw_fd!(UnixStream, socket2::Socket, inner, socket);
722
723/// A Unix socket that has not yet been converted to a [`UnixStream`] or
724/// [`UnixListener`].
725#[derive(Debug)]
726pub struct UnixSocket {
727    inner: Socket,
728}
729
730impl UnixSocket {
731    /// Creates a new Unix stream socket.
732    pub async fn new_stream() -> io::Result<UnixSocket> {
733        UnixSocket::new(socket2::Type::STREAM).await
734    }
735
736    async fn new(ty: socket2::Type) -> io::Result<UnixSocket> {
737        let inner = Socket::new(socket2::Domain::UNIX, ty, None).await?;
738        Ok(UnixSocket { inner })
739    }
740
741    /// Returns the local address that this socket is bound to.
742    pub fn local_addr(&self) -> io::Result<SockAddr> {
743        self.inner.local_addr()
744    }
745
746    /// Returns the value of the `SO_ERROR` option.
747    pub fn take_error(&self) -> io::Result<Option<io::Error>> {
748        self.inner.socket.take_error()
749    }
750
751    /// Binds the socket to the given address.
752    pub async fn bind(&self, path: impl AsRef<Path>) -> io::Result<()> {
753        self.bind_addr(&SockAddr::unix(path)?).await
754    }
755
756    /// Binds the socket to the given address.
757    pub async fn bind_addr(&self, addr: &SockAddr) -> io::Result<()> {
758        if !addr.is_unix() {
759            return Err(io::Error::new(
760                io::ErrorKind::InvalidInput,
761                "addr is not unix socket address",
762            ));
763        }
764        self.inner.bind(addr).await
765    }
766
767    /// Converts the socket into a `UnixListener`.
768    ///
769    /// `backlog` defines the maximum number of pending connections are queued
770    /// by the operating system at any given time. Connections are removed from
771    /// the queue with [`UnixListener::accept`]. When the queue is full, the
772    /// operating-system will start rejecting connections.
773    pub async fn listen(self, backlog: i32) -> io::Result<UnixListener> {
774        self.inner.listen(backlog).await?;
775        Ok(UnixListener { inner: self.inner })
776    }
777
778    /// Establishes a Unix connection with a peer at the specified socket
779    /// address.
780    ///
781    /// See [`UnixSocket::connect_addr`] for more details.
782    pub async fn connect(self, path: impl AsRef<Path>) -> io::Result<UnixStream> {
783        self.connect_addr(&SockAddr::unix(path)?).await
784    }
785
786    /// Establishes a Unix connection with a peer at the specified socket
787    /// address.
788    ///
789    /// The [`UnixSocket`] is consumed. Once the connection is established, a
790    /// connected [`UnixStream`] is returned. If the connection fails, the
791    /// encountered error is returned.
792    ///
793    /// On Windows, the socket should be bound to an empty address before
794    /// connecting.
795    pub async fn connect_addr(self, addr: &SockAddr) -> io::Result<UnixStream> {
796        if !addr.is_unix() {
797            return Err(io::Error::new(
798                io::ErrorKind::InvalidInput,
799                "addr is not unix socket address",
800            ));
801        }
802        self.inner.connect_async(addr).await?;
803        Ok(UnixStream { inner: self.inner })
804    }
805}
806
807impl_raw_fd!(UnixSocket, socket2::Socket, inner, socket);
808
809#[cfg(windows)]
810#[inline]
811fn empty_unix_socket() -> SockAddr {
812    use windows_sys::Win32::Networking::WinSock::{AF_UNIX, SOCKADDR_UN};
813
814    // SAFETY: the length is correct
815    unsafe {
816        SockAddr::try_init(|addr, len| {
817            let addr: *mut SOCKADDR_UN = addr.cast();
818            std::ptr::write(
819                addr,
820                SOCKADDR_UN {
821                    sun_family: AF_UNIX,
822                    sun_path: [0; 108],
823                },
824            );
825            std::ptr::write(len, 3);
826            Ok(())
827        })
828    }
829    // it is always Ok
830    .unwrap()
831    .1
832}
833
834// The peer addr returned after ConnectEx is buggy. It contains bytes that
835// should not belong to the address. Luckily a unix path should not contain `\0`
836// until the end. We can determine the path ending by that.
837#[cfg(windows)]
838#[inline]
839fn fix_unix_socket_length(addr: &mut SockAddr) {
840    use windows_sys::Win32::Networking::WinSock::SOCKADDR_UN;
841
842    // SAFETY: cannot construct non-unix socket address in safe way.
843    let unix_addr: &SOCKADDR_UN = unsafe { &*addr.as_ptr().cast() };
844    let sun_path = unsafe {
845        std::slice::from_raw_parts(
846            unix_addr.sun_path.as_ptr() as *const u8,
847            unix_addr.sun_path.len(),
848        )
849    };
850    let addr_len = match std::ffi::CStr::from_bytes_until_nul(sun_path) {
851        Ok(str) => str.to_bytes_with_nul().len() + 2,
852        Err(_) => std::mem::size_of::<SOCKADDR_UN>(),
853    };
854    unsafe {
855        addr.set_length(addr_len as _);
856    }
857}
compio_net/unix.rs

compio_net/
unix.rs
