diff --git a/examples/udp_echo_server.rs b/examples/udp_echo_server.rs new file mode 100644 index 0000000..f840107 --- /dev/null +++ b/examples/udp_echo_server.rs @@ -0,0 +1,18 @@ +use wstd::io; +use wstd::net::UdpSocket; + +#[wstd::main] +async fn main() -> io::Result<()> { + let socket = UdpSocket::bind("127.0.0.1:8080").await?; + println!("Listening on {}", socket.local_addr()?); + println!("type `nc -u localhost 8080` to create a UDP client"); + + let mut buf = vec![0; 65535]; + loop { + let (len, peer) = socket.recv_from(&mut buf).await?; + println!("Received {len} bytes from: {peer}"); + // If the echo send fails, we can ignore it: one socket serves every + // peer here, so a failure for one must not end the loop for the rest. + let _ = socket.send_to(&buf[..len], peer).await; + } +} diff --git a/examples/udp_stream_client.rs b/examples/udp_stream_client.rs new file mode 100644 index 0000000..013caee --- /dev/null +++ b/examples/udp_stream_client.rs @@ -0,0 +1,42 @@ +use wstd::io; +use wstd::net::{UdpSocket, UdpStream}; + +async fn ping(stream: &UdpStream) -> io::Result<()> { + assert_eq!(stream.send(b"ping\n").await?, 5); + + let mut reply = [0; 5]; + let len = stream.recv(&mut reply).await?; + assert_eq!(&reply[..len], b"pong\n"); + + Ok(()) +} + +#[wstd::main] +async fn main() -> io::Result<()> { + let mut args = std::env::args(); + + let _ = args.next(); + + let addr = args.next().ok_or_else(|| { + io::Error::new( + std::io::ErrorKind::InvalidInput, + "address argument required", + ) + })?; + + let stream = UdpStream::connect(addr).await?; + ping(&stream).await?; + + let peer = stream.peer_addr()?; + drop(stream); + + let local_addr = if peer.is_ipv4() { + "0.0.0.0:0" + } else { + "[::]:0" + }; + let stream = UdpSocket::bind(local_addr).await?.connect(peer)?; + ping(&stream).await?; + + Ok(()) +} diff --git a/src/net/mod.rs b/src/net/mod.rs index 1600edc..dc1dc41 100644 --- a/src/net/mod.rs +++ b/src/net/mod.rs @@ -1,13 +1,15 @@ //! Async network abstractions. use std::io::{self, ErrorKind}; -use wasip2::sockets::network::ErrorCode; +use wasip2::sockets::network::{ErrorCode, IpSocketAddress, Ipv4SocketAddress}; mod tcp_listener; mod tcp_stream; +mod udp; pub use tcp_listener::*; pub use tcp_stream::*; +pub use udp::*; fn to_io_err(err: ErrorCode) -> io::Error { match err { @@ -24,6 +26,55 @@ fn to_io_err(err: ErrorCode) -> io::Error { ErrorCode::ConnectionReset => ErrorKind::ConnectionReset.into(), ErrorCode::ConnectionAborted => ErrorKind::ConnectionAborted.into(), ErrorCode::ConcurrencyConflict => ErrorKind::AlreadyExists.into(), + ErrorCode::DatagramTooLarge => ErrorKind::InvalidInput.into(), _ => ErrorKind::Other.into(), } } + +fn sockaddr_from_wasi(addr: IpSocketAddress) -> std::net::SocketAddr { + use wasip2::sockets::network::Ipv6SocketAddress; + match addr { + IpSocketAddress::Ipv4(Ipv4SocketAddress { address, port }) => { + std::net::SocketAddr::V4(std::net::SocketAddrV4::new( + std::net::Ipv4Addr::new(address.0, address.1, address.2, address.3), + port, + )) + } + IpSocketAddress::Ipv6(Ipv6SocketAddress { + address, + port, + flow_info, + scope_id, + }) => std::net::SocketAddr::V6(std::net::SocketAddrV6::new( + std::net::Ipv6Addr::new( + address.0, address.1, address.2, address.3, address.4, address.5, address.6, + address.7, + ), + port, + flow_info, + scope_id, + )), + } +} + +fn sockaddr_to_wasi(addr: std::net::SocketAddr) -> IpSocketAddress { + use wasip2::sockets::network::Ipv6SocketAddress; + match addr { + std::net::SocketAddr::V4(addr) => { + let ip = addr.ip().octets(); + IpSocketAddress::Ipv4(Ipv4SocketAddress { + address: (ip[0], ip[1], ip[2], ip[3]), + port: addr.port(), + }) + } + std::net::SocketAddr::V6(addr) => { + let ip = addr.ip().segments(); + IpSocketAddress::Ipv6(Ipv6SocketAddress { + address: (ip[0], ip[1], ip[2], ip[3], ip[4], ip[5], ip[6], ip[7]), + port: addr.port(), + flow_info: addr.flowinfo(), + scope_id: addr.scope_id(), + }) + } + } +} diff --git a/src/net/tcp_listener.rs b/src/net/tcp_listener.rs index 9a1f57a..3ee9007 100644 --- a/src/net/tcp_listener.rs +++ b/src/net/tcp_listener.rs @@ -1,11 +1,10 @@ -use wasip2::sockets::network::Ipv4SocketAddress; -use wasip2::sockets::tcp::{IpAddressFamily, IpSocketAddress, TcpSocket}; +use wasip2::sockets::tcp::{IpAddressFamily, TcpSocket}; use crate::io; use crate::iter::AsyncIterator; use std::net::SocketAddr; -use super::{TcpStream, to_io_err}; +use super::{TcpStream, sockaddr_from_wasi, sockaddr_to_wasi, to_io_err}; use crate::runtime::AsyncPollable; /// A TCP socket server, listening for connections. @@ -79,51 +78,3 @@ impl<'a> AsyncIterator for Incoming<'a> { Some(Ok(TcpStream::new(input, output, socket))) } } - -fn sockaddr_from_wasi(addr: IpSocketAddress) -> std::net::SocketAddr { - use wasip2::sockets::network::Ipv6SocketAddress; - match addr { - IpSocketAddress::Ipv4(Ipv4SocketAddress { address, port }) => { - std::net::SocketAddr::V4(std::net::SocketAddrV4::new( - std::net::Ipv4Addr::new(address.0, address.1, address.2, address.3), - port, - )) - } - IpSocketAddress::Ipv6(Ipv6SocketAddress { - address, - port, - flow_info, - scope_id, - }) => std::net::SocketAddr::V6(std::net::SocketAddrV6::new( - std::net::Ipv6Addr::new( - address.0, address.1, address.2, address.3, address.4, address.5, address.6, - address.7, - ), - port, - flow_info, - scope_id, - )), - } -} - -fn sockaddr_to_wasi(addr: std::net::SocketAddr) -> IpSocketAddress { - use wasip2::sockets::network::Ipv6SocketAddress; - match addr { - std::net::SocketAddr::V4(addr) => { - let ip = addr.ip().octets(); - IpSocketAddress::Ipv4(Ipv4SocketAddress { - address: (ip[0], ip[1], ip[2], ip[3]), - port: addr.port(), - }) - } - std::net::SocketAddr::V6(addr) => { - let ip = addr.ip().segments(); - IpSocketAddress::Ipv6(Ipv6SocketAddress { - address: (ip[0], ip[1], ip[2], ip[3], ip[4], ip[5], ip[6], ip[7]), - port: addr.port(), - flow_info: addr.flowinfo(), - scope_id: addr.scope_id(), - }) - } - } -} diff --git a/src/net/udp.rs b/src/net/udp.rs new file mode 100644 index 0000000..59afb0d --- /dev/null +++ b/src/net/udp.rs @@ -0,0 +1,355 @@ +use std::io::ErrorKind; +use std::net::{SocketAddr, ToSocketAddrs}; +use std::sync::OnceLock; + +use wasip2::sockets::instance_network::instance_network; +use wasip2::sockets::udp::{ + IncomingDatagramStream, IpAddressFamily, IpSocketAddress, OutgoingDatagram, + OutgoingDatagramStream, +}; +use wasip2::sockets::udp_create_socket::create_udp_socket; + +use super::{sockaddr_from_wasi, sockaddr_to_wasi, to_io_err}; +use crate::io; +use crate::runtime::AsyncPollable; + +/// A UDP socket, bound to a local address. +/// +/// A `UdpSocket` is not associated with any remote address, so datagrams can be +/// sent to, and received from, any address, using [`UdpSocket::send_to`] and +/// [`UdpSocket::recv_from`]. Use [`UdpSocket::connect`] to associate it with a +/// single remote address instead, giving a [`UdpStream`]. +#[derive(Debug)] +pub struct UdpSocket { + incoming: AsyncIncomingDatagramStream, + outgoing: AsyncOutgoingDatagramStream, + socket: wasip2::sockets::udp::UdpSocket, +} + +impl UdpSocket { + /// Creates a new UdpSocket bound to the specified local address. + pub async fn bind(addr: &str) -> io::Result { + let addr: SocketAddr = addr + .parse() + .map_err(|_| io::Error::other("failed to parse string to socket addr"))?; + let socket = bind_socket(addr).await?; + + // Datagram streams without a remote address may send to, and receive + // from, any address. + let (incoming, outgoing) = socket.stream(None).map_err(to_io_err)?; + Ok(Self { + incoming: AsyncIncomingDatagramStream::new(incoming), + outgoing: AsyncOutgoingDatagramStream::new(outgoing), + socket, + }) + } + + /// Returns the local socket address of this socket. + pub fn local_addr(&self) -> io::Result { + self.socket + .local_address() + .map_err(to_io_err) + .map(sockaddr_from_wasi) + } + + /// Sends a datagram to the given address. + pub async fn send_to(&self, buf: &[u8], addr: SocketAddr) -> io::Result { + self.outgoing + .send_to(buf, Some(sockaddr_to_wasi(addr))) + .await + } + + /// Receives a single datagram. On success, returns the number of bytes + /// received and the address the datagram was sent from. + /// + /// If `buf` is shorter than the datagram, the excess bytes are discarded. + pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { + self.incoming.recv_from(buf).await + } + + /// Associates this socket with a remote address, giving a [`UdpStream`] + /// which sends to, and receives from, only that address. + /// + /// This only changes the local socket configuration, and does not generate + /// any network traffic. + pub fn connect(self, addr: SocketAddr) -> io::Result { + // WASI may trap if streams from a previous call to `stream` are still + // live, so drop the unconnected streams before creating connected ones. + let Self { + incoming, + outgoing, + socket, + } = self; + drop((incoming, outgoing)); + + let (incoming, outgoing) = socket + .stream(Some(sockaddr_to_wasi(addr))) + .map_err(to_io_err)?; + Ok(UdpStream::new(incoming, outgoing, socket)) + } + + /// Returns the unicast hop limit ("time to live") of this socket. + pub fn unicast_hop_limit(&self) -> io::Result { + self.socket.unicast_hop_limit().map_err(to_io_err) + } + + /// Sets the unicast hop limit ("time to live") of this socket. + pub fn set_unicast_hop_limit(&self, value: u8) -> io::Result<()> { + self.socket.set_unicast_hop_limit(value).map_err(to_io_err) + } + + /// Returns the size of the receive buffer of this socket. + pub fn receive_buffer_size(&self) -> io::Result { + self.socket.receive_buffer_size().map_err(to_io_err) + } + + /// Sets the size of the receive buffer of this socket. This is a hint: the + /// size reported by [`UdpSocket::receive_buffer_size`] may differ. + pub fn set_receive_buffer_size(&self, value: u64) -> io::Result<()> { + self.socket + .set_receive_buffer_size(value) + .map_err(to_io_err) + } + + /// Returns the size of the send buffer of this socket. + pub fn send_buffer_size(&self) -> io::Result { + self.socket.send_buffer_size().map_err(to_io_err) + } + + /// Sets the size of the send buffer of this socket. This is a hint: the + /// size reported by [`UdpSocket::send_buffer_size`] may differ. + pub fn set_send_buffer_size(&self, value: u64) -> io::Result<()> { + self.socket.set_send_buffer_size(value).map_err(to_io_err) + } +} + +/// A UDP socket associated with a remote address. +/// +/// A `UdpStream` sends to, and receives from, only the address it was connected +/// to, using [`UdpStream::send`] and [`UdpStream::recv`]. Datagrams sent from +/// any other address are not received. +#[derive(Debug)] +pub struct UdpStream { + incoming: AsyncIncomingDatagramStream, + outgoing: AsyncOutgoingDatagramStream, + socket: wasip2::sockets::udp::UdpSocket, +} + +impl UdpStream { + fn new( + incoming: IncomingDatagramStream, + outgoing: OutgoingDatagramStream, + socket: wasip2::sockets::udp::UdpSocket, + ) -> Self { + Self { + incoming: AsyncIncomingDatagramStream::new(incoming), + outgoing: AsyncOutgoingDatagramStream::new(outgoing), + socket, + } + } + + /// Associates a UDP socket with a remote host. + pub async fn connect(addr: impl ToSocketAddrs) -> io::Result { + let addrs = addr.to_socket_addrs()?; + let mut last_err = None; + for addr in addrs { + match UdpStream::connect_addr(addr).await { + Ok(stream) => return Ok(stream), + Err(e) => last_err = Some(e), + } + } + + Err(last_err.unwrap_or_else(|| { + io::Error::new(ErrorKind::InvalidInput, "could not resolve to any address") + })) + } + + /// Establishes an association with the specified `addr`. + pub async fn connect_addr(addr: SocketAddr) -> io::Result { + // Unlike in POSIX, WASI requires a UDP socket be explicitly bound + // before it can be associated with a remote address. Bind to the + // unspecified address of the same family, and let the host choose a + // port. + let local_addr = match addr { + SocketAddr::V4(_) => SocketAddr::from((std::net::Ipv4Addr::UNSPECIFIED, 0)), + SocketAddr::V6(_) => SocketAddr::from((std::net::Ipv6Addr::UNSPECIFIED, 0)), + }; + let socket = bind_socket(local_addr).await?; + + let (incoming, outgoing) = socket + .stream(Some(sockaddr_to_wasi(addr))) + .map_err(to_io_err)?; + Ok(Self::new(incoming, outgoing, socket)) + } + + /// Returns the local socket address of this socket. + pub fn local_addr(&self) -> io::Result { + self.socket + .local_address() + .map_err(to_io_err) + .map(sockaddr_from_wasi) + } + + /// Returns the socket address of the remote peer of this UDP association. + pub fn peer_addr(&self) -> io::Result { + self.socket + .remote_address() + .map_err(to_io_err) + .map(sockaddr_from_wasi) + } + + /// Sends a datagram to the remote peer. + pub async fn send(&self, buf: &[u8]) -> io::Result { + self.outgoing.send_to(buf, None).await + } + + /// Receives a single datagram from the remote peer. On success, returns the + /// number of bytes received. + /// + /// If `buf` is shorter than the datagram, the excess bytes are discarded. + pub async fn recv(&self, buf: &mut [u8]) -> io::Result { + self.incoming.recv_from(buf).await.map(|(len, _addr)| len) + } + + /// Returns the unicast hop limit ("time to live") of this socket. + pub fn unicast_hop_limit(&self) -> io::Result { + self.socket.unicast_hop_limit().map_err(to_io_err) + } + + /// Sets the unicast hop limit ("time to live") of this socket. + pub fn set_unicast_hop_limit(&self, value: u8) -> io::Result<()> { + self.socket.set_unicast_hop_limit(value).map_err(to_io_err) + } + + /// Returns the size of the receive buffer of this socket. + pub fn receive_buffer_size(&self) -> io::Result { + self.socket.receive_buffer_size().map_err(to_io_err) + } + + /// Sets the size of the receive buffer of this socket. This is a hint: the + /// size reported by [`UdpStream::receive_buffer_size`] may differ. + pub fn set_receive_buffer_size(&self, value: u64) -> io::Result<()> { + self.socket + .set_receive_buffer_size(value) + .map_err(to_io_err) + } + + /// Returns the size of the send buffer of this socket. + pub fn send_buffer_size(&self) -> io::Result { + self.socket.send_buffer_size().map_err(to_io_err) + } + + /// Sets the size of the send buffer of this socket. This is a hint: the + /// size reported by [`UdpStream::send_buffer_size`] may differ. + pub fn set_send_buffer_size(&self, value: u64) -> io::Result<()> { + self.socket.set_send_buffer_size(value).map_err(to_io_err) + } +} + +async fn bind_socket(addr: SocketAddr) -> io::Result { + let family = match addr { + SocketAddr::V4(_) => IpAddressFamily::Ipv4, + SocketAddr::V6(_) => IpAddressFamily::Ipv6, + }; + let socket = create_udp_socket(family).map_err(to_io_err)?; + let network = instance_network(); + let local_address = sockaddr_to_wasi(addr); + + socket + .start_bind(&network, local_address) + .map_err(to_io_err)?; + let pollable = AsyncPollable::new(socket.subscribe()); + pollable.wait_for().await; + socket.finish_bind().map_err(to_io_err)?; + + Ok(socket) +} + +#[derive(Debug)] +struct AsyncIncomingDatagramStream { + subscription: OnceLock, + stream: IncomingDatagramStream, +} + +impl AsyncIncomingDatagramStream { + fn new(stream: IncomingDatagramStream) -> Self { + Self { + subscription: OnceLock::new(), + stream, + } + } + + /// Await receive readiness. + async fn ready(&self) { + let subscription = self + .subscription + .get_or_init(|| AsyncPollable::new(self.stream.subscribe())); + subscription.wait_for().await; + } + + /// Asynchronous receive. + async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { + let datagram = loop { + self.ready().await; + match self + .stream + .receive(1) + .map_err(to_io_err)? + .into_iter() + .next() + { + Some(datagram) => break datagram, + // `self.ready()` cannot guarantee that a datagram is ready to + // receive, so try again if WASI returns an empty list. + None => continue, + } + }; + let len = datagram.data.len().min(buf.len()); + buf[0..len].copy_from_slice(&datagram.data[0..len]); + Ok((len, sockaddr_from_wasi(datagram.remote_address))) + } +} + +#[derive(Debug)] +struct AsyncOutgoingDatagramStream { + subscription: OnceLock, + stream: OutgoingDatagramStream, +} + +impl AsyncOutgoingDatagramStream { + fn new(stream: OutgoingDatagramStream) -> Self { + Self { + subscription: OnceLock::new(), + stream, + } + } + + /// Await send readiness. + async fn ready(&self) { + let subscription = self + .subscription + .get_or_init(|| AsyncPollable::new(self.stream.subscribe())); + subscription.wait_for().await; + } + + /// Asynchronous send. + async fn send_to( + &self, + buf: &[u8], + remote_address: Option, + ) -> io::Result { + let datagrams = [OutgoingDatagram { + data: buf.to_vec(), + remote_address, + }]; + loop { + if self.stream.check_send().map_err(to_io_err)? == 0 { + self.ready().await; + continue; + } + if self.stream.send(&datagrams).map_err(to_io_err)? == 1 { + return Ok(buf.len()); + } + } + } +} diff --git a/test-programs/Cargo.toml b/test-programs/Cargo.toml index f280352..9eddb52 100644 --- a/test-programs/Cargo.toml +++ b/test-programs/Cargo.toml @@ -7,7 +7,6 @@ rust-version.workspace = true publish = false [dev-dependencies] -anyhow.workspace = true test-log.workspace = true serde_json.workspace = true ureq.workspace = true @@ -17,6 +16,7 @@ cargo_metadata.workspace = true heck.workspace = true [dependencies] +anyhow.workspace = true fs2 = "0.4" [features] diff --git a/test-programs/src/lib.rs b/test-programs/src/lib.rs index a62d166..bb017da 100644 --- a/test-programs/src/lib.rs +++ b/test-programs/src/lib.rs @@ -1,5 +1,6 @@ include!(concat!(env!("OUT_DIR"), "/gen.rs")); +use anyhow::{Context, Result, bail}; use std::fs::File; use std::net::TcpStream; use std::process::{Child, Command}; @@ -70,3 +71,41 @@ impl Drop for WasmtimeServe { let _ = self.process.kill(); } } + +/// Read a guest's stdout until it reports where it is listening, and return +/// that address. +/// +/// Guest programs which bind a socket print `Listening on {addr}`, so that a +/// test can discover the address even when the guest picked the port. +pub fn get_listening_address( + mut wasmtime_stdout: std::process::ChildStdout, +) -> Result { + use std::io::Read; + + let mut stdout_contents = String::new(); + let mut buf = [0; 4096]; + loop { + let len = wasmtime_stdout + .read(&mut buf) + .context("reading wasmtime stdout")?; + if len == 0 { + bail!("wasmtime exited before reporting its listening address"); + } + stdout_contents.push_str( + std::str::from_utf8(&buf[..len]).context("wasmtime stdout should be string")?, + ); + + // Parse out the line where guest program says where it is listening + for line in stdout_contents.lines() { + if let Some(rest) = line.strip_prefix("Listening on ") { + // Forget wasmtime_stdout, rather than drop it, so that any + // subsequent stdout from wasmtime doesn't panic on a broken + // pipe. + std::mem::forget(wasmtime_stdout); + return rest + .parse() + .with_context(|| format!("parsing socket addr from line: {line:?}")); + } + } + } +} diff --git a/test-programs/tests/tcp_echo_server.rs b/test-programs/tests/tcp_echo_server.rs index 1e949cc..bb007dd 100644 --- a/test-programs/tests/tcp_echo_server.rs +++ b/test-programs/tests/tcp_echo_server.rs @@ -5,6 +5,7 @@ use std::process::Command; fn tcp_echo_server() -> Result<()> { use std::io::{Read, Write}; use std::net::{Shutdown, TcpStream}; + use test_programs::get_listening_address; println!("testing {}", test_programs::TCP_ECHO_SERVER); @@ -82,40 +83,3 @@ fn tcp_echo_server() -> Result<()> { Ok(()) } - -fn get_listening_address( - mut wasmtime_stdout: std::process::ChildStdout, -) -> Result { - use std::io::Read; - use std::thread::sleep; - use std::time::Duration; - - // Gather complete contents of stdout here - let mut stdout_contents = String::new(); - loop { - // Wait for process to print - sleep(Duration::from_millis(100)); - - // Read more that the process printed, append to contents - let mut buf = vec![0; 4096]; - let len = wasmtime_stdout - .read(&mut buf) - .context("reading wasmtime stdout")?; - buf.truncate(len); - stdout_contents - .push_str(std::str::from_utf8(&buf).context("wasmtime stdout should be string")?); - - // Parse out the line where guest program says where it is listening - for line in stdout_contents.lines() { - if let Some(rest) = line.strip_prefix("Listening on ") { - // Forget wasmtime_stdout, rather than drop it, so that any - // subsequent stdout from wasmtime doesn't panic on a broken - // pipe. - std::mem::forget(wasmtime_stdout); - return rest - .parse() - .with_context(|| format!("parsing socket addr from line: {line:?}")); - } - } - } -} diff --git a/test-programs/tests/udp_echo_server.rs b/test-programs/tests/udp_echo_server.rs new file mode 100644 index 0000000..8ad8c8a --- /dev/null +++ b/test-programs/tests/udp_echo_server.rs @@ -0,0 +1,66 @@ +use anyhow::{Context, Result}; +use std::process::Command; + +#[test_log::test] +fn udp_echo_server() -> Result<()> { + use std::net::{SocketAddr, UdpSocket}; + use std::time::Duration; + + println!("testing {}", test_programs::UDP_ECHO_SERVER); + + // Run the component in wasmtime + // -Sinherit-network required for sockets to work + let mut wasmtime_process = Command::new("wasmtime") + .arg("run") + .arg("-Sinherit-network") + .arg(test_programs::UDP_ECHO_SERVER) + .stdout(std::process::Stdio::piped()) + .spawn()?; + + let addr = test_programs::get_listening_address( + wasmtime_process.stdout.take().expect("stdout is piped"), + )?; + + println!("udp echo server is listening on {addr:?}"); + + // Connect each client so that it only receives its own echo, and give + // every receive a deadline: a datagram may be dropped, and without a + // timeout a lost echo would hang the test suite rather than fail it. + fn client(addr: SocketAddr) -> Result { + let sock = UdpSocket::bind("127.0.0.1:0").context("binding client socket")?; + sock.set_read_timeout(Some(Duration::from_secs(10))) + .context("setting client read timeout")?; + sock.connect(addr).context("connecting client socket")?; + Ok(sock) + } + + let sock1 = client(addr).context("client sock1")?; + println!("sock1 bound to {}", sock1.local_addr()?); + + let sock2 = client(addr).context("client sock2")?; + println!("sock2 bound to {}", sock2.local_addr()?); + + const MESSAGE1: &[u8] = b"hello, echoserver!\n"; + // Exercise the datagram copy path with a larger second payload. + const MESSAGE2: &[u8] = &[0xa5; 4096]; + + sock1.send(MESSAGE1).context("send from sock1")?; + println!("sock1 sent to echo server"); + + sock2.send(MESSAGE2).context("send from sock2")?; + println!("sock2 sent to echo server"); + + let mut buf = vec![0; 65535]; + + let len = sock1.recv(&mut buf).context("recv on sock1")?; + println!("read from sock1"); + assert_eq!(MESSAGE1, &buf[..len], "readback of sock1"); + + let len = sock2.recv(&mut buf).context("recv on sock2")?; + println!("read from sock2"); + assert_eq!(MESSAGE2, &buf[..len], "readback of sock2"); + + wasmtime_process.kill()?; + + Ok(()) +} diff --git a/test-programs/tests/udp_stream_client.rs b/test-programs/tests/udp_stream_client.rs new file mode 100644 index 0000000..ebc4027 --- /dev/null +++ b/test-programs/tests/udp_stream_client.rs @@ -0,0 +1,56 @@ +use anyhow::{Context, Result}; +use std::net::UdpSocket; +use std::process::{Command, Stdio}; +use std::time::Duration; + +#[test_log::test] +fn udp_stream_client() -> Result<()> { + // Port 0: the host picks a free port, which the component is told about + // by argument, so this test can't collide with anything else running. + let server = UdpSocket::bind("127.0.0.1:0").context("binding temporary test server")?; + // Without a deadline, a dropped datagram would hang the test suite rather + // than fail it. + server + .set_read_timeout(Some(Duration::from_secs(10))) + .context("setting server read timeout")?; + let addr = server + .local_addr() + .context("getting local server address")?; + + let child = Command::new("wasmtime") + .arg("run") + .arg("-Sinherit-network") + .arg(test_programs::UDP_STREAM_CLIENT) + .arg(addr.to_string()) + .stderr(Stdio::piped()) + .spawn() + .context("spawning wasmtime component")?; + + // The component exercises both `UdpStream::connect` and converting a bound + // `UdpSocket` with `UdpSocket::connect`. + for _ in 0..2 { + let mut buf = [0; 5]; + let (len, component_addr) = server + .recv_from(&mut buf) + .context("receiving ping datagram from component")?; + assert_eq!(&buf[..len], b"ping\n", "expected ping from component"); + + // The component uses a five-byte buffer, so the remaining bytes should + // be discarded with the rest of this datagram. + server + .send_to(b"pong\nignored", component_addr) + .context("writing reply")?; + } + + let output = child + .wait_with_output() + .context("waiting for component exit")?; + + assert!( + output.status.success(), + "\nComponent exited abnormally (stderr:\n{})", + String::from_utf8_lossy(&output.stderr) + ); + + Ok(()) +}