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