1 #![warn(rust_2018_idioms)]
2 #![cfg(feature = "full")]
3 #![cfg(unix)]
4
5 use tokio::io::ReadBuf;
6 use tokio::net::UnixDatagram;
7 use tokio::try_join;
8
9 use std::future::poll_fn;
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]
24 #[cfg_attr(miri, ignore)] // No `socket` on miri.
echo() -> io::Result<()>25 async fn echo() -> io::Result<()> {
26 let dir = tempfile::tempdir().unwrap();
27 let server_path = dir.path().join("server.sock");
28 let client_path = dir.path().join("client.sock");
29
30 let server_socket = UnixDatagram::bind(server_path.clone())?;
31
32 tokio::spawn(async move {
33 let _ = echo_server(server_socket).await;
34 });
35
36 {
37 let socket = UnixDatagram::bind(&client_path).unwrap();
38 socket.connect(server_path)?;
39 socket.send(b"ECHO").await?;
40 let mut recv_buf = [0u8; 16];
41 let len = socket.recv(&mut recv_buf[..]).await?;
42 assert_eq!(&recv_buf[..len], b"ECHO");
43 }
44
45 Ok(())
46 }
47
48 #[tokio::test]
49 #[cfg_attr(miri, ignore)] // No `socket` on miri.
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 let _ = echo_server(server_socket).await;
59 });
60
61 {
62 let socket = UnixDatagram::bind(&client_path).unwrap();
63 socket.connect(&server_path)?;
64 socket.send(b"ECHO").await?;
65 let mut recv_buf = [0u8; 16];
66 let (len, addr) = socket.recv_from(&mut recv_buf[..]).await?;
67 assert_eq!(&recv_buf[..len], b"ECHO");
68 assert_eq!(addr.as_pathname(), Some(server_path.as_path()));
69 }
70
71 Ok(())
72 }
73
74 // Even though we use sync non-blocking io we still need a reactor.
75 #[tokio::test]
76 #[cfg_attr(miri, ignore)] // No SOCK_DGRAM for `socketpair` in miri.
try_send_recv_never_block() -> io::Result<()>77 async fn try_send_recv_never_block() -> io::Result<()> {
78 let mut recv_buf = [0u8; 16];
79 let payload = b"PAYLOAD";
80 let mut count = 0;
81
82 let (dgram1, dgram2) = UnixDatagram::pair()?;
83
84 // Send until we hit the OS `net.unix.max_dgram_qlen`.
85 loop {
86 dgram1.writable().await.unwrap();
87
88 match dgram1.try_send(payload) {
89 Err(err) => match (err.kind(), err.raw_os_error()) {
90 (io::ErrorKind::WouldBlock, _) => break,
91 (_, Some(libc::ENOBUFS)) => break,
92 _ => {
93 panic!("unexpected error {err:?}");
94 }
95 },
96 Ok(len) => {
97 assert_eq!(len, payload.len());
98 }
99 }
100 count += 1;
101 }
102
103 // Read every dgram we sent.
104 while count > 0 {
105 dgram2.readable().await.unwrap();
106 let len = dgram2.try_recv(&mut recv_buf[..])?;
107 assert_eq!(len, payload.len());
108 assert_eq!(payload, &recv_buf[..len]);
109 count -= 1;
110 }
111
112 let err = dgram2.try_recv(&mut recv_buf[..]).unwrap_err();
113 match err.kind() {
114 io::ErrorKind::WouldBlock => (),
115 _ => unreachable!("unexpected error {:?}", err),
116 }
117
118 Ok(())
119 }
120
121 #[tokio::test]
122 #[cfg_attr(miri, ignore)] // No `socket` on miri.
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]
147 #[cfg_attr(miri, ignore)] // No `socket` on miri.
send_to_recv_from_poll() -> std::io::Result<()>148 async fn send_to_recv_from_poll() -> std::io::Result<()> {
149 let dir = tempfile::tempdir().unwrap();
150 let sender_path = dir.path().join("sender.sock");
151 let receiver_path = dir.path().join("receiver.sock");
152
153 let sender = UnixDatagram::bind(&sender_path)?;
154 let receiver = UnixDatagram::bind(&receiver_path)?;
155
156 let msg = b"hello";
157 poll_fn(|cx| sender.poll_send_to(cx, msg, &receiver_path)).await?;
158
159 let mut recv_buf = [0u8; 32];
160 let mut read = ReadBuf::new(&mut recv_buf);
161 let addr = poll_fn(|cx| receiver.poll_recv_from(cx, &mut read)).await?;
162
163 assert_eq!(read.filled(), msg);
164 assert_eq!(addr.as_pathname(), Some(sender_path.as_ref()));
165 Ok(())
166 }
167
168 #[tokio::test]
169 #[cfg_attr(miri, ignore)] // No `socket` on miri.
send_recv_poll() -> std::io::Result<()>170 async fn send_recv_poll() -> std::io::Result<()> {
171 let dir = tempfile::tempdir().unwrap();
172 let sender_path = dir.path().join("sender.sock");
173 let receiver_path = dir.path().join("receiver.sock");
174
175 let sender = UnixDatagram::bind(&sender_path)?;
176 let receiver = UnixDatagram::bind(&receiver_path)?;
177
178 sender.connect(&receiver_path)?;
179 receiver.connect(&sender_path)?;
180
181 let msg = b"hello";
182 poll_fn(|cx| sender.poll_send(cx, msg)).await?;
183
184 let mut recv_buf = [0u8; 32];
185 let mut read = ReadBuf::new(&mut recv_buf);
186 poll_fn(|cx| receiver.poll_recv(cx, &mut read)).await?;
187
188 assert_eq!(read.filled(), msg);
189 Ok(())
190 }
191
192 #[tokio::test]
193 #[cfg_attr(miri, ignore)] // No `socket` on miri.
try_send_to_recv_from() -> std::io::Result<()>194 async fn try_send_to_recv_from() -> std::io::Result<()> {
195 let dir = tempfile::tempdir().unwrap();
196 let server_path = dir.path().join("server.sock");
197 let client_path = dir.path().join("client.sock");
198
199 // Create listener
200 let server = UnixDatagram::bind(&server_path)?;
201
202 // Create socket pair
203 let client = UnixDatagram::bind(&client_path)?;
204
205 for _ in 0..5 {
206 loop {
207 client.writable().await?;
208
209 match client.try_send_to(b"hello world", &server_path) {
210 Ok(n) => {
211 assert_eq!(n, 11);
212 break;
213 }
214 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
215 Err(e) => panic!("{e:?}"),
216 }
217 }
218
219 loop {
220 server.readable().await?;
221
222 let mut buf = [0; 512];
223
224 match server.try_recv_from(&mut buf) {
225 Ok((n, addr)) => {
226 assert_eq!(n, 11);
227 assert_eq!(addr.as_pathname(), Some(client_path.as_ref()));
228 assert_eq!(&buf[0..11], &b"hello world"[..]);
229 break;
230 }
231 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
232 Err(e) => panic!("{e:?}"),
233 }
234 }
235 }
236
237 Ok(())
238 }
239
240 #[tokio::test]
241 #[cfg_attr(miri, ignore)] // No `socket` on miri.
try_recv_buf_from() -> std::io::Result<()>242 async fn try_recv_buf_from() -> std::io::Result<()> {
243 let dir = tempfile::tempdir().unwrap();
244 let server_path = dir.path().join("server.sock");
245 let client_path = dir.path().join("client.sock");
246
247 // Create listener
248 let server = UnixDatagram::bind(&server_path)?;
249
250 // Create socket pair
251 let client = UnixDatagram::bind(&client_path)?;
252
253 for _ in 0..5 {
254 loop {
255 client.writable().await?;
256
257 match client.try_send_to(b"hello world", &server_path) {
258 Ok(n) => {
259 assert_eq!(n, 11);
260 break;
261 }
262 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
263 Err(e) => panic!("{e:?}"),
264 }
265 }
266
267 loop {
268 server.readable().await?;
269
270 let mut buf = Vec::with_capacity(512);
271
272 match server.try_recv_buf_from(&mut buf) {
273 Ok((n, addr)) => {
274 assert_eq!(n, 11);
275 assert_eq!(addr.as_pathname(), Some(client_path.as_ref()));
276 assert_eq!(&buf[0..11], &b"hello world"[..]);
277 break;
278 }
279 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
280 Err(e) => panic!("{e:?}"),
281 }
282 }
283 }
284
285 Ok(())
286 }
287
288 #[tokio::test]
289 #[cfg_attr(miri, ignore)] // No `socket` on miri.
recv_buf_from() -> std::io::Result<()>290 async fn recv_buf_from() -> std::io::Result<()> {
291 let tmp = tempfile::tempdir()?;
292
293 // Bind each socket to a filesystem path
294 let tx_path = tmp.path().join("tx");
295 let tx = UnixDatagram::bind(&tx_path)?;
296 let rx_path = tmp.path().join("rx");
297 let rx = UnixDatagram::bind(&rx_path)?;
298
299 let bytes = b"hello world";
300 tx.send_to(bytes, &rx_path).await?;
301
302 let mut buf = Vec::with_capacity(24);
303 let (size, addr) = rx.recv_buf_from(&mut buf).await?;
304
305 let dgram = &buf[..size];
306 assert_eq!(dgram, bytes);
307 assert_eq!(addr.as_pathname().unwrap(), &tx_path);
308 Ok(())
309 }
310
311 // Even though we use sync non-blocking io we still need a reactor.
312 #[tokio::test]
313 #[cfg_attr(miri, ignore)] // No SOCK_DGRAM for `socketpair` in miri.
try_recv_buf_never_block() -> io::Result<()>314 async fn try_recv_buf_never_block() -> io::Result<()> {
315 let payload = b"PAYLOAD";
316 let mut count = 0;
317
318 let (dgram1, dgram2) = UnixDatagram::pair()?;
319
320 // Send until we hit the OS `net.unix.max_dgram_qlen`.
321 loop {
322 dgram1.writable().await.unwrap();
323
324 match dgram1.try_send(payload) {
325 Err(err) => match (err.kind(), err.raw_os_error()) {
326 (io::ErrorKind::WouldBlock, _) => break,
327 (_, Some(libc::ENOBUFS)) => break,
328 _ => {
329 panic!("unexpected error {err:?}");
330 }
331 },
332 Ok(len) => {
333 assert_eq!(len, payload.len());
334 }
335 }
336 count += 1;
337 }
338
339 // Read every dgram we sent.
340 while count > 0 {
341 let mut recv_buf = Vec::with_capacity(16);
342
343 dgram2.readable().await.unwrap();
344 let len = dgram2.try_recv_buf(&mut recv_buf)?;
345 assert_eq!(len, payload.len());
346 assert_eq!(payload, &recv_buf[..len]);
347 count -= 1;
348 }
349
350 let mut recv_buf = vec![0; 16];
351 let err = dgram2.try_recv_from(&mut recv_buf).unwrap_err();
352 match err.kind() {
353 io::ErrorKind::WouldBlock => (),
354 _ => unreachable!("unexpected error {:?}", err),
355 }
356
357 Ok(())
358 }
359
360 #[tokio::test]
361 #[cfg_attr(miri, ignore)] // No SOCK_DGRAM for `socketpair` in miri.
recv_buf() -> std::io::Result<()>362 async fn recv_buf() -> std::io::Result<()> {
363 // Create the pair of sockets
364 let (sock1, sock2) = UnixDatagram::pair()?;
365
366 // Since the sockets are paired, the paired send/recv
367 // functions can be used
368 let bytes = b"hello world";
369 sock1.send(bytes).await?;
370
371 let mut buff = Vec::with_capacity(24);
372 let size = sock2.recv_buf(&mut buff).await?;
373
374 let dgram = &buff[..size];
375 assert_eq!(dgram, bytes);
376 Ok(())
377 }
378
379 #[tokio::test]
380 #[cfg_attr(miri, ignore)] // No `socket` on miri.
poll_ready() -> io::Result<()>381 async fn poll_ready() -> io::Result<()> {
382 let dir = tempfile::tempdir().unwrap();
383 let server_path = dir.path().join("server.sock");
384 let client_path = dir.path().join("client.sock");
385
386 // Create listener
387 let server = UnixDatagram::bind(&server_path)?;
388
389 // Create socket pair
390 let client = UnixDatagram::bind(&client_path)?;
391
392 for _ in 0..5 {
393 loop {
394 poll_fn(|cx| client.poll_send_ready(cx)).await?;
395
396 match client.try_send_to(b"hello world", &server_path) {
397 Ok(n) => {
398 assert_eq!(n, 11);
399 break;
400 }
401 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
402 Err(e) => panic!("{e:?}"),
403 }
404 }
405
406 loop {
407 poll_fn(|cx| server.poll_recv_ready(cx)).await?;
408
409 let mut buf = Vec::with_capacity(512);
410
411 match server.try_recv_buf_from(&mut buf) {
412 Ok((n, addr)) => {
413 assert_eq!(n, 11);
414 assert_eq!(addr.as_pathname(), Some(client_path.as_ref()));
415 assert_eq!(&buf[0..11], &b"hello world"[..]);
416 break;
417 }
418 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
419 Err(e) => panic!("{e:?}"),
420 }
421 }
422 }
423
424 Ok(())
425 }
426