1//! Unix-specific networking extensions.
2
3use std::fmt;
4use std::os::unix::net::UnixListener as StdUnixListener;
5use std::os::unix::net::UnixStream as StdUnixStream;
6use std::pin::Pin;
7
8use async_io::Async;
9
10use super::SocketAddr;
11use super::UnixStream;
12use crate::io;
13use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
14use crate::path::Path;
15use crate::stream::Stream;
16use crate::sync::Arc;
17use crate::task::{ready, Context, Poll};
18
19/// A Unix domain socket server, listening for connections.
20///
21/// After creating a `UnixListener` by [`bind`]ing it to a socket address, it listens for incoming
22/// connections. These can be accepted by awaiting elements from the async stream of [`incoming`]
23/// connections.
24///
25/// The socket will be closed when the value is dropped.
26///
27/// This type is an async version of [`std::os::unix::net::UnixListener`].
28///
29/// [`std::os::unix::net::UnixListener`]:
30/// https://doc.rust-lang.org/std/os/unix/net/struct.UnixListener.html
31/// [`bind`]: #method.bind
32/// [`incoming`]: #method.incoming
33///
34/// # Examples
35///
36/// ```no_run
37/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
38/// #
39/// use async_std::os::unix::net::UnixListener;
40/// use async_std::prelude::*;
41///
42/// let listener = UnixListener::bind("/tmp/socket").await?;
43/// let mut incoming = listener.incoming();
44///
45/// while let Some(stream) = incoming.next().await {
46/// let mut stream = stream?;
47/// stream.write_all(b"hello world").await?;
48/// }
49/// #
50/// # Ok(()) }) }
51/// ```
52pub struct UnixListener {
53 watcher: Async<StdUnixListener>,
54}
55
56impl UnixListener {
57 /// Creates a Unix datagram listener bound to the given path.
58 ///
59 /// # Examples
60 ///
61 /// ```no_run
62 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
63 /// #
64 /// use async_std::os::unix::net::UnixListener;
65 ///
66 /// let listener = UnixListener::bind("/tmp/socket").await?;
67 /// #
68 /// # Ok(()) }) }
69 /// ```
70 pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<UnixListener> {
71 let path = path.as_ref().to_owned();
72 let listener = Async::<StdUnixListener>::bind(path)?;
73
74 Ok(UnixListener { watcher: listener })
75 }
76
77 /// Accepts a new incoming connection to this listener.
78 ///
79 /// When a connection is established, the corresponding stream and address will be returned.
80 ///
81 /// # Examples
82 ///
83 /// ```no_run
84 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
85 /// #
86 /// use async_std::os::unix::net::UnixListener;
87 ///
88 /// let listener = UnixListener::bind("/tmp/socket").await?;
89 /// let (socket, addr) = listener.accept().await?;
90 /// #
91 /// # Ok(()) }) }
92 /// ```
93 pub async fn accept(&self) -> io::Result<(UnixStream, SocketAddr)> {
94 let (stream, addr) = self.watcher.accept().await?;
95
96 Ok((
97 UnixStream {
98 watcher: Arc::new(stream),
99 },
100 addr,
101 ))
102 }
103
104 /// Returns a stream of incoming connections.
105 ///
106 /// Iterating over this stream is equivalent to calling [`accept`] in a loop. The stream of
107 /// connections is infinite, i.e awaiting the next connection will never result in [`None`].
108 ///
109 /// [`accept`]: #method.accept
110 /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None
111 ///
112 /// # Examples
113 ///
114 /// ```no_run
115 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
116 /// #
117 /// use async_std::os::unix::net::UnixListener;
118 /// use async_std::prelude::*;
119 ///
120 /// let listener = UnixListener::bind("/tmp/socket").await?;
121 /// let mut incoming = listener.incoming();
122 ///
123 /// while let Some(stream) = incoming.next().await {
124 /// let mut stream = stream?;
125 /// stream.write_all(b"hello world").await?;
126 /// }
127 /// #
128 /// # Ok(()) }) }
129 /// ```
130 pub fn incoming(&self) -> Incoming<'_> {
131 Incoming {
132 incoming: Box::pin(self.watcher.incoming()),
133 }
134 }
135
136 /// Returns the local socket address of this listener.
137 ///
138 /// # Examples
139 ///
140 /// ```no_run
141 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
142 /// #
143 /// use async_std::os::unix::net::UnixListener;
144 ///
145 /// let listener = UnixListener::bind("/tmp/socket").await?;
146 /// let addr = listener.local_addr()?;
147 /// #
148 /// # Ok(()) }) }
149 /// ```
150 pub fn local_addr(&self) -> io::Result<SocketAddr> {
151 self.watcher.get_ref().local_addr()
152 }
153}
154
155impl fmt::Debug for UnixListener {
156 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
157 let mut builder: DebugStruct<'_, '_> = f.debug_struct(name:"UnixListener");
158 builder.field(name:"fd", &self.as_raw_fd());
159
160 if let Ok(addr: SocketAddr) = self.local_addr() {
161 builder.field(name:"local", &addr);
162 }
163
164 builder.finish()
165 }
166}
167
168/// A stream of incoming Unix domain socket connections.
169///
170/// This stream is infinite, i.e awaiting the next connection will never result in [`None`]. It is
171/// created by the [`incoming`] method on [`UnixListener`].
172///
173/// This type is an async version of [`std::os::unix::net::Incoming`].
174///
175/// [`std::os::unix::net::Incoming`]: https://doc.rust-lang.org/std/os/unix/net/struct.Incoming.html
176/// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None
177/// [`incoming`]: struct.UnixListener.html#method.incoming
178/// [`UnixListener`]: struct.UnixListener.html
179pub struct Incoming<'a> {
180 incoming: Pin<Box<dyn Stream<Item = io::Result<Async<StdUnixStream>>> + Send + Sync + 'a>>,
181}
182
183impl Stream for Incoming<'_> {
184 type Item = io::Result<UnixStream>;
185
186 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
187 let res: Option, …>> = ready!(Pin::new(&mut self.incoming).poll_next(cx));
188 Poll::Ready(res.map(|res: Result, …>| res.map(|stream: Async| UnixStream { watcher: Arc::new(data:stream) })))
189 }
190}
191
192impl fmt::Debug for Incoming<'_> {
193 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
194 fDebugStruct<'_, '_>.debug_struct(name:"Incoming")
195 .finish()
196 }
197}
198
199impl From<StdUnixListener> for UnixListener {
200 /// Converts a `std::os::unix::net::UnixListener` into its asynchronous equivalent.
201 fn from(listener: StdUnixListener) -> UnixListener {
202 UnixListener {
203 watcher: Async::new(listener).expect(msg:"UnixListener is known to be good"),
204 }
205 }
206}
207
208impl std::convert::TryFrom<UnixListener> for StdUnixListener {
209 type Error = io::Error;
210 /// Converts a `UnixListener` into its synchronous equivalent.
211 fn try_from(listener: UnixListener) -> io::Result<StdUnixListener> {
212 let inner: UnixListener = listener.watcher.into_inner()?;
213 inner.set_nonblocking(false)?;
214 Ok(inner)
215 }
216}
217
218impl AsRawFd for UnixListener {
219 fn as_raw_fd(&self) -> RawFd {
220 self.watcher.as_raw_fd()
221 }
222}
223
224impl FromRawFd for UnixListener {
225 unsafe fn from_raw_fd(fd: RawFd) -> UnixListener {
226 let listener: UnixListener = std::os::unix::net::UnixListener::from_raw_fd(fd);
227 listener.into()
228 }
229}
230
231impl IntoRawFd for UnixListener {
232 fn into_raw_fd(self) -> RawFd {
233 self.watcher.into_inner().unwrap().into_raw_fd()
234 }
235}
236
237cfg_io_safety! {
238 use crate::os::unix::io::{AsFd, BorrowedFd, OwnedFd};
239
240 impl AsFd for UnixListener {
241 fn as_fd(&self) -> BorrowedFd<'_> {
242 self.watcher.get_ref().as_fd()
243 }
244 }
245
246 impl From<OwnedFd> for UnixListener {
247 fn from(fd: OwnedFd) -> UnixListener {
248 std::os::unix::net::UnixListener::from(fd).into()
249 }
250 }
251
252 impl From<UnixListener> for OwnedFd {
253 fn from(stream: UnixListener) -> OwnedFd {
254 stream.watcher.into_inner().unwrap().into()
255 }
256 }
257}
258