1 files changed, 330 insertions, 0 deletions
diff --git a/vendor/tokio/tests/uds_datagram.rs b/vendor/tokio/tests/uds_datagram.rs
new file mode 100644
index 000000000..4d2846865
--- /dev/null
+++ b/vendor/tokio/tests/uds_datagram.rs
@@ -0,0 +1,330 @@
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+#![cfg(unix)]
+
+use futures::future::poll_fn;
+use tokio::io::ReadBuf;
+use tokio::net::UnixDatagram;
+use tokio::try_join;
+
+use std::io;
+use std::sync::Arc;
+
+async fn echo_server(socket: UnixDatagram) -> io::Result<()> {
+    let mut recv_buf = vec![0u8; 1024];
+    loop {
+        let (len, peer_addr) = socket.recv_from(&mut recv_buf[..]).await?;
+        if let Some(path) = peer_addr.as_pathname() {
+            socket.send_to(&recv_buf[..len], path).await?;
+        }
+    }
+}
+
+#[tokio::test]
+async fn echo() -> io::Result<()> {
+    let dir = tempfile::tempdir().unwrap();
+    let server_path = dir.path().join("server.sock");
+    let client_path = dir.path().join("client.sock");
+
+    let server_socket = UnixDatagram::bind(server_path.clone())?;
+
+    tokio::spawn(async move {
+        if let Err(e) = echo_server(server_socket).await {
+            eprintln!("Error in echo server: {}", e);
+        }
+    });
+
+    {
+        let socket = UnixDatagram::bind(&client_path).unwrap();
+        socket.connect(server_path)?;
+        socket.send(b"ECHO").await?;
+        let mut recv_buf = [0u8; 16];
+        let len = socket.recv(&mut recv_buf[..]).await?;
+        assert_eq!(&recv_buf[..len], b"ECHO");
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn echo_from() -> io::Result<()> {
+    let dir = tempfile::tempdir().unwrap();
+    let server_path = dir.path().join("server.sock");
+    let client_path = dir.path().join("client.sock");
+
+    let server_socket = UnixDatagram::bind(server_path.clone())?;
+
+    tokio::spawn(async move {
+        if let Err(e) = echo_server(server_socket).await {
+            eprintln!("Error in echo server: {}", e);
+        }
+    });
+
+    {
+        let socket = UnixDatagram::bind(&client_path).unwrap();
+        socket.connect(&server_path)?;
+        socket.send(b"ECHO").await?;
+        let mut recv_buf = [0u8; 16];
+        let (len, addr) = socket.recv_from(&mut recv_buf[..]).await?;
+        assert_eq!(&recv_buf[..len], b"ECHO");
+        assert_eq!(addr.as_pathname(), Some(server_path.as_path()));
+    }
+
+    Ok(())
+}
+
+// Even though we use sync non-blocking io we still need a reactor.
+#[tokio::test]
+async fn try_send_recv_never_block() -> io::Result<()> {
+    let mut recv_buf = [0u8; 16];
+    let payload = b"PAYLOAD";
+    let mut count = 0;
+
+    let (dgram1, dgram2) = UnixDatagram::pair()?;
+
+    // Send until we hit the OS `net.unix.max_dgram_qlen`.
+    loop {
+        dgram1.writable().await.unwrap();
+
+        match dgram1.try_send(payload) {
+            Err(err) => match (err.kind(), err.raw_os_error()) {
+                (io::ErrorKind::WouldBlock, _) => break,
+                (_, Some(libc::ENOBUFS)) => break,
+                _ => {
+                    panic!("unexpected error {:?}", err);
+                }
+            },
+            Ok(len) => {
+                assert_eq!(len, payload.len());
+            }
+        }
+        count += 1;
+    }
+
+    // Read every dgram we sent.
+    while count > 0 {
+        dgram2.readable().await.unwrap();
+        let len = dgram2.try_recv(&mut recv_buf[..])?;
+        assert_eq!(len, payload.len());
+        assert_eq!(payload, &recv_buf[..len]);
+        count -= 1;
+    }
+
+    let err = dgram2.try_recv(&mut recv_buf[..]).unwrap_err();
+    match err.kind() {
+        io::ErrorKind::WouldBlock => (),
+        _ => unreachable!("unexpected error {:?}", err),
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn split() -> std::io::Result<()> {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("split.sock");
+    let s = Arc::new(UnixDatagram::bind(path.clone())?);
+    let r = s.clone();
+
+    let msg = b"hello";
+    let ((), ()) = try_join! {
+        async {
+            s.send_to(msg, path).await?;
+            io::Result::Ok(())
+        },
+        async {
+            let mut recv_buf = [0u8; 32];
+            let (len, _) = r.recv_from(&mut recv_buf[..]).await?;
+            assert_eq!(&recv_buf[..len], msg);
+            Ok(())
+        },
+    }?;
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn send_to_recv_from_poll() -> std::io::Result<()> {
+    let dir = tempfile::tempdir().unwrap();
+    let sender_path = dir.path().join("sender.sock");
+    let receiver_path = dir.path().join("receiver.sock");
+
+    let sender = UnixDatagram::bind(&sender_path)?;
+    let receiver = UnixDatagram::bind(&receiver_path)?;
+
+    let msg = b"hello";
+    poll_fn(|cx| sender.poll_send_to(cx, msg, &receiver_path)).await?;
+
+    let mut recv_buf = [0u8; 32];
+    let mut read = ReadBuf::new(&mut recv_buf);
+    let addr = poll_fn(|cx| receiver.poll_recv_from(cx, &mut read)).await?;
+
+    assert_eq!(read.filled(), msg);
+    assert_eq!(addr.as_pathname(), Some(sender_path.as_ref()));
+    Ok(())
+}
+
+#[tokio::test]
+async fn send_recv_poll() -> std::io::Result<()> {
+    let dir = tempfile::tempdir().unwrap();
+    let sender_path = dir.path().join("sender.sock");
+    let receiver_path = dir.path().join("receiver.sock");
+
+    let sender = UnixDatagram::bind(&sender_path)?;
+    let receiver = UnixDatagram::bind(&receiver_path)?;
+
+    sender.connect(&receiver_path)?;
+    receiver.connect(&sender_path)?;
+
+    let msg = b"hello";
+    poll_fn(|cx| sender.poll_send(cx, msg)).await?;
+
+    let mut recv_buf = [0u8; 32];
+    let mut read = ReadBuf::new(&mut recv_buf);
+    let _len = poll_fn(|cx| receiver.poll_recv(cx, &mut read)).await?;
+
+    assert_eq!(read.filled(), msg);
+    Ok(())
+}
+
+#[tokio::test]
+async fn try_send_to_recv_from() -> std::io::Result<()> {
+    let dir = tempfile::tempdir().unwrap();
+    let server_path = dir.path().join("server.sock");
+    let client_path = dir.path().join("client.sock");
+
+    // Create listener
+    let server = UnixDatagram::bind(&server_path)?;
+
+    // Create socket pair
+    let client = UnixDatagram::bind(&client_path)?;
+
+    for _ in 0..5 {
+        loop {
+            client.writable().await?;
+
+            match client.try_send_to(b"hello world", &server_path) {
+                Ok(n) => {
+                    assert_eq!(n, 11);
+                    break;
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
+                Err(e) => panic!("{:?}", e),
+            }
+        }
+
+        loop {
+            server.readable().await?;
+
+            let mut buf = [0; 512];
+
+            match server.try_recv_from(&mut buf) {
+                Ok((n, addr)) => {
+                    assert_eq!(n, 11);
+                    assert_eq!(addr.as_pathname(), Some(client_path.as_ref()));
+                    assert_eq!(&buf[0..11], &b"hello world"[..]);
+                    break;
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
+                Err(e) => panic!("{:?}", e),
+            }
+        }
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn try_recv_buf_from() -> std::io::Result<()> {
+    let dir = tempfile::tempdir().unwrap();
+    let server_path = dir.path().join("server.sock");
+    let client_path = dir.path().join("client.sock");
+
+    // Create listener
+    let server = UnixDatagram::bind(&server_path)?;
+
+    // Create socket pair
+    let client = UnixDatagram::bind(&client_path)?;
+
+    for _ in 0..5 {
+        loop {
+            client.writable().await?;
+
+            match client.try_send_to(b"hello world", &server_path) {
+                Ok(n) => {
+                    assert_eq!(n, 11);
+                    break;
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
+                Err(e) => panic!("{:?}", e),
+            }
+        }
+
+        loop {
+            server.readable().await?;
+
+            let mut buf = Vec::with_capacity(512);
+
+            match server.try_recv_buf_from(&mut buf) {
+                Ok((n, addr)) => {
+                    assert_eq!(n, 11);
+                    assert_eq!(addr.as_pathname(), Some(client_path.as_ref()));
+                    assert_eq!(&buf[0..11], &b"hello world"[..]);
+                    break;
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
+                Err(e) => panic!("{:?}", e),
+            }
+        }
+    }
+
+    Ok(())
+}
+
+// Even though we use sync non-blocking io we still need a reactor.
+#[tokio::test]
+async fn try_recv_buf_never_block() -> io::Result<()> {
+    let payload = b"PAYLOAD";
+    let mut count = 0;
+
+    let (dgram1, dgram2) = UnixDatagram::pair()?;
+
+    // Send until we hit the OS `net.unix.max_dgram_qlen`.
+    loop {
+        dgram1.writable().await.unwrap();
+
+        match dgram1.try_send(payload) {
+            Err(err) => match (err.kind(), err.raw_os_error()) {
+                (io::ErrorKind::WouldBlock, _) => break,
+                (_, Some(libc::ENOBUFS)) => break,
+                _ => {
+                    panic!("unexpected error {:?}", err);
+                }
+            },
+            Ok(len) => {
+                assert_eq!(len, payload.len());
+            }
+        }
+        count += 1;
+    }
+
+    // Read every dgram we sent.
+    while count > 0 {
+        let mut recv_buf = Vec::with_capacity(16);
+
+        dgram2.readable().await.unwrap();
+        let len = dgram2.try_recv_buf(&mut recv_buf)?;
+        assert_eq!(len, payload.len());
+        assert_eq!(payload, &recv_buf[..len]);
+        count -= 1;
+    }
+
+    let mut recv_buf = vec![0; 16];
+    let err = dgram2.try_recv_from(&mut recv_buf).unwrap_err();
+    match err.kind() {
+        io::ErrorKind::WouldBlock => (),
+        _ => unreachable!("unexpected error {:?}", err),
+    }
+
+    Ok(())
+}