1 #![warn(rust_2018_idioms)]
2 #![cfg(feature = "full")]
3 #![cfg(unix)]
4 
5 use futures::future::poll_fn;
6 use tokio::io::ReadBuf;
7 use tokio::net::UnixDatagram;
8 use tokio::try_join;
9 
10 use std::io;
11 use std::sync::Arc;
12 
echo_server(socket: UnixDatagram) -> io::Result<()>13 async fn echo_server(socket: UnixDatagram) -> io::Result<()> {
14     let mut recv_buf = vec![0u8; 1024];
15     loop {
16         let (len, peer_addr) = socket.recv_from(&mut recv_buf[..]).await?;
17         if let Some(path) = peer_addr.as_pathname() {
18             socket.send_to(&recv_buf[..len], path).await?;
19         }
20     }
21 }
22 
23 #[tokio::test]
echo() -> io::Result<()>24 async fn echo() -> io::Result<()> {
25     let dir = tempfile::tempdir().unwrap();
26     let server_path = dir.path().join("server.sock");
27     let client_path = dir.path().join("client.sock");
28 
29     let server_socket = UnixDatagram::bind(server_path.clone())?;
30 
31     tokio::spawn(async move {
32         if let Err(e) = echo_server(server_socket).await {
33             eprintln!("Error in echo server: {}", e);
34         }
35     });
36 
37     {
38         let socket = UnixDatagram::bind(&client_path).unwrap();
39         socket.connect(server_path)?;
40         socket.send(b"ECHO").await?;
41         let mut recv_buf = [0u8; 16];
42         let len = socket.recv(&mut recv_buf[..]).await?;
43         assert_eq!(&recv_buf[..len], b"ECHO");
44     }
45 
46     Ok(())
47 }
48 
49 #[tokio::test]
echo_from() -> io::Result<()>50 async fn echo_from() -> io::Result<()> {
51     let dir = tempfile::tempdir().unwrap();
52     let server_path = dir.path().join("server.sock");
53     let client_path = dir.path().join("client.sock");
54 
55     let server_socket = UnixDatagram::bind(server_path.clone())?;
56 
57     tokio::spawn(async move {
58         if let Err(e) = echo_server(server_socket).await {
59             eprintln!("Error in echo server: {}", e);
60         }
61     });
62 
63     {
64         let socket = UnixDatagram::bind(&client_path).unwrap();
65         socket.connect(&server_path)?;
66         socket.send(b"ECHO").await?;
67         let mut recv_buf = [0u8; 16];
68         let (len, addr) = socket.recv_from(&mut recv_buf[..]).await?;
69         assert_eq!(&recv_buf[..len], b"ECHO");
70         assert_eq!(addr.as_pathname(), Some(server_path.as_path()));
71     }
72 
73     Ok(())
74 }
75 
76 // Even though we use sync non-blocking io we still need a reactor.
77 #[tokio::test]
try_send_recv_never_block() -> io::Result<()>78 async fn try_send_recv_never_block() -> io::Result<()> {
79     let mut recv_buf = [0u8; 16];
80     let payload = b"PAYLOAD";
81     let mut count = 0;
82 
83     let (dgram1, dgram2) = UnixDatagram::pair()?;
84 
85     // Send until we hit the OS `net.unix.max_dgram_qlen`.
86     loop {
87         dgram1.writable().await.unwrap();
88 
89         match dgram1.try_send(payload) {
90             Err(err) => match (err.kind(), err.raw_os_error()) {
91                 (io::ErrorKind::WouldBlock, _) => break,
92                 (_, Some(libc::ENOBUFS)) => break,
93                 _ => {
94                     panic!("unexpected error {:?}", err);
95                 }
96             },
97             Ok(len) => {
98                 assert_eq!(len, payload.len());
99             }
100         }
101         count += 1;
102     }
103 
104     // Read every dgram we sent.
105     while count > 0 {
106         dgram2.readable().await.unwrap();
107         let len = dgram2.try_recv(&mut recv_buf[..])?;
108         assert_eq!(len, payload.len());
109         assert_eq!(payload, &recv_buf[..len]);
110         count -= 1;
111     }
112 
113     let err = dgram2.try_recv(&mut recv_buf[..]).unwrap_err();
114     match err.kind() {
115         io::ErrorKind::WouldBlock => (),
116         _ => unreachable!("unexpected error {:?}", err),
117     }
118 
119     Ok(())
120 }
121 
122 #[tokio::test]
split() -> std::io::Result<()>123 async fn split() -> std::io::Result<()> {
124     let dir = tempfile::tempdir().unwrap();
125     let path = dir.path().join("split.sock");
126     let s = Arc::new(UnixDatagram::bind(path.clone())?);
127     let r = s.clone();
128 
129     let msg = b"hello";
130     let ((), ()) = try_join! {
131         async {
132             s.send_to(msg, path).await?;
133             io::Result::Ok(())
134         },
135         async {
136             let mut recv_buf = [0u8; 32];
137             let (len, _) = r.recv_from(&mut recv_buf[..]).await?;
138             assert_eq!(&recv_buf[..len], msg);
139             Ok(())
140         },
141     }?;
142 
143     Ok(())
144 }
145 
146 #[tokio::test]
send_to_recv_from_poll() -> std::io::Result<()>147 async fn send_to_recv_from_poll() -> std::io::Result<()> {
148     let dir = tempfile::tempdir().unwrap();
149     let sender_path = dir.path().join("sender.sock");
150     let receiver_path = dir.path().join("receiver.sock");
151 
152     let sender = UnixDatagram::bind(&sender_path)?;
153     let receiver = UnixDatagram::bind(&receiver_path)?;
154 
155     let msg = b"hello";
156     poll_fn(|cx| sender.poll_send_to(cx, msg, &receiver_path)).await?;
157 
158     let mut recv_buf = [0u8; 32];
159     let mut read = ReadBuf::new(&mut recv_buf);
160     let addr = poll_fn(|cx| receiver.poll_recv_from(cx, &mut read)).await?;
161 
162     assert_eq!(read.filled(), msg);
163     assert_eq!(addr.as_pathname(), Some(sender_path.as_ref()));
164     Ok(())
165 }
166 
167 #[tokio::test]
send_recv_poll() -> std::io::Result<()>168 async fn send_recv_poll() -> std::io::Result<()> {
169     let dir = tempfile::tempdir().unwrap();
170     let sender_path = dir.path().join("sender.sock");
171     let receiver_path = dir.path().join("receiver.sock");
172 
173     let sender = UnixDatagram::bind(&sender_path)?;
174     let receiver = UnixDatagram::bind(&receiver_path)?;
175 
176     sender.connect(&receiver_path)?;
177     receiver.connect(&sender_path)?;
178 
179     let msg = b"hello";
180     poll_fn(|cx| sender.poll_send(cx, msg)).await?;
181 
182     let mut recv_buf = [0u8; 32];
183     let mut read = ReadBuf::new(&mut recv_buf);
184     let _len = poll_fn(|cx| receiver.poll_recv(cx, &mut read)).await?;
185 
186     assert_eq!(read.filled(), msg);
187     Ok(())
188 }
189 
190 #[tokio::test]
try_send_to_recv_from() -> std::io::Result<()>191 async fn try_send_to_recv_from() -> std::io::Result<()> {
192     let dir = tempfile::tempdir().unwrap();
193     let server_path = dir.path().join("server.sock");
194     let client_path = dir.path().join("client.sock");
195 
196     // Create listener
197     let server = UnixDatagram::bind(&server_path)?;
198 
199     // Create socket pair
200     let client = UnixDatagram::bind(&client_path)?;
201 
202     for _ in 0..5 {
203         loop {
204             client.writable().await?;
205 
206             match client.try_send_to(b"hello world", &server_path) {
207                 Ok(n) => {
208                     assert_eq!(n, 11);
209                     break;
210                 }
211                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
212                 Err(e) => panic!("{:?}", e),
213             }
214         }
215 
216         loop {
217             server.readable().await?;
218 
219             let mut buf = [0; 512];
220 
221             match server.try_recv_from(&mut buf) {
222                 Ok((n, addr)) => {
223                     assert_eq!(n, 11);
224                     assert_eq!(addr.as_pathname(), Some(client_path.as_ref()));
225                     assert_eq!(&buf[0..11], &b"hello world"[..]);
226                     break;
227                 }
228                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
229                 Err(e) => panic!("{:?}", e),
230             }
231         }
232     }
233 
234     Ok(())
235 }
236 
237 #[tokio::test]
try_recv_buf_from() -> std::io::Result<()>238 async fn try_recv_buf_from() -> std::io::Result<()> {
239     let dir = tempfile::tempdir().unwrap();
240     let server_path = dir.path().join("server.sock");
241     let client_path = dir.path().join("client.sock");
242 
243     // Create listener
244     let server = UnixDatagram::bind(&server_path)?;
245 
246     // Create socket pair
247     let client = UnixDatagram::bind(&client_path)?;
248 
249     for _ in 0..5 {
250         loop {
251             client.writable().await?;
252 
253             match client.try_send_to(b"hello world", &server_path) {
254                 Ok(n) => {
255                     assert_eq!(n, 11);
256                     break;
257                 }
258                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
259                 Err(e) => panic!("{:?}", e),
260             }
261         }
262 
263         loop {
264             server.readable().await?;
265 
266             let mut buf = Vec::with_capacity(512);
267 
268             match server.try_recv_buf_from(&mut buf) {
269                 Ok((n, addr)) => {
270                     assert_eq!(n, 11);
271                     assert_eq!(addr.as_pathname(), Some(client_path.as_ref()));
272                     assert_eq!(&buf[0..11], &b"hello world"[..]);
273                     break;
274                 }
275                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
276                 Err(e) => panic!("{:?}", e),
277             }
278         }
279     }
280 
281     Ok(())
282 }
283 
284 // Even though we use sync non-blocking io we still need a reactor.
285 #[tokio::test]
try_recv_buf_never_block() -> io::Result<()>286 async fn try_recv_buf_never_block() -> io::Result<()> {
287     let payload = b"PAYLOAD";
288     let mut count = 0;
289 
290     let (dgram1, dgram2) = UnixDatagram::pair()?;
291 
292     // Send until we hit the OS `net.unix.max_dgram_qlen`.
293     loop {
294         dgram1.writable().await.unwrap();
295 
296         match dgram1.try_send(payload) {
297             Err(err) => match (err.kind(), err.raw_os_error()) {
298                 (io::ErrorKind::WouldBlock, _) => break,
299                 (_, Some(libc::ENOBUFS)) => break,
300                 _ => {
301                     panic!("unexpected error {:?}", err);
302                 }
303             },
304             Ok(len) => {
305                 assert_eq!(len, payload.len());
306             }
307         }
308         count += 1;
309     }
310 
311     // Read every dgram we sent.
312     while count > 0 {
313         let mut recv_buf = Vec::with_capacity(16);
314 
315         dgram2.readable().await.unwrap();
316         let len = dgram2.try_recv_buf(&mut recv_buf)?;
317         assert_eq!(len, payload.len());
318         assert_eq!(payload, &recv_buf[..len]);
319         count -= 1;
320     }
321 
322     let mut recv_buf = vec![0; 16];
323     let err = dgram2.try_recv_from(&mut recv_buf).unwrap_err();
324     match err.kind() {
325         io::ErrorKind::WouldBlock => (),
326         _ => unreachable!("unexpected error {:?}", err),
327     }
328 
329     Ok(())
330 }
331 
332 #[tokio::test]
poll_ready() -> io::Result<()>333 async fn poll_ready() -> io::Result<()> {
334     let dir = tempfile::tempdir().unwrap();
335     let server_path = dir.path().join("server.sock");
336     let client_path = dir.path().join("client.sock");
337 
338     // Create listener
339     let server = UnixDatagram::bind(&server_path)?;
340 
341     // Create socket pair
342     let client = UnixDatagram::bind(&client_path)?;
343 
344     for _ in 0..5 {
345         loop {
346             poll_fn(|cx| client.poll_send_ready(cx)).await?;
347 
348             match client.try_send_to(b"hello world", &server_path) {
349                 Ok(n) => {
350                     assert_eq!(n, 11);
351                     break;
352                 }
353                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
354                 Err(e) => panic!("{:?}", e),
355             }
356         }
357 
358         loop {
359             poll_fn(|cx| server.poll_recv_ready(cx)).await?;
360 
361             let mut buf = Vec::with_capacity(512);
362 
363             match server.try_recv_buf_from(&mut buf) {
364                 Ok((n, addr)) => {
365                     assert_eq!(n, 11);
366                     assert_eq!(addr.as_pathname(), Some(client_path.as_ref()));
367                     assert_eq!(&buf[0..11], &b"hello world"[..]);
368                     break;
369                 }
370                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
371                 Err(e) => panic!("{:?}", e),
372             }
373         }
374     }
375 
376     Ok(())
377 }
378