library: don't pass around a Reactor, just get Reactor::current() when needed

Author: pchickey

src/http/client.rs

@@ -7,14 +7,12 @@ use crate::runtime::Reactor;
 
 /// An HTTP client.
 #[derive(Debug)]
-pub struct Client<'a> {
-    reactor: &'a Reactor,
-}
+pub struct Client {}
 
-impl<'a> Client<'a> {
+impl Client {
     /// Create a new instance of `Client`
-    pub fn new(reactor: &'a Reactor) -> Self {
-        Self { reactor }
+    pub fn new() -> Self {
+        Self {}
     }
 
     /// Send an HTTP request.
@@ -27,7 +25,7 @@ impl<'a> Client<'a> {
         let res = wasi::http::outgoing_handler::handle(wasi_req, None).unwrap();
 
         // 2. Start sending the request body
-        io::copy(body, OutputStream::new(&self.reactor, body_stream))
+        io::copy(body, OutputStream::new(body_stream))
             .await
             .expect("io::copy broke oh no");
 
@@ -36,42 +34,38 @@ impl<'a> Client<'a> {
         OutgoingBody::finish(wasi_body, trailers).unwrap();
 
         // 4. Receive the response
-        self.reactor.wait_for(res.subscribe()).await;
+        Reactor::current().wait_for(res.subscribe()).await;
         // NOTE: the first `unwrap` is to ensure readiness, the second `unwrap`
         // is to trap if we try and get the response more than once. The final
         // `?` is to raise the actual error if there is one.
         let res = res.get().unwrap().unwrap()?;
-        Ok(Response::try_from_incoming_response(
-            res,
-            self.reactor.clone(),
-        )?)
+        Ok(Response::try_from_incoming_response(res)?)
     }
 }
 
-struct OutputStream<'a> {
-    reactor: &'a Reactor,
+struct OutputStream {
     stream: wasi::http::types::OutputStream,
 }
 
-impl<'a> OutputStream<'a> {
-    fn new(reactor: &'a Reactor, stream: wasi::http::types::OutputStream) -> Self {
-        Self { reactor, stream }
+impl OutputStream {
+    fn new(stream: wasi::http::types::OutputStream) -> Self {
+        Self { stream }
     }
 }
 
-impl<'a> AsyncWrite for OutputStream<'a> {
+impl AsyncWrite for OutputStream {
     async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         let max = self.stream.check_write().unwrap() as usize;
         let max = max.min(buf.len());
         let buf = &buf[0..max];
         self.stream.write(buf).unwrap();
-        self.reactor.wait_for(self.stream.subscribe()).await;
+        Reactor::current().wait_for(self.stream.subscribe()).await;
         Ok(max)
     }
 
     async fn flush(&mut self) -> io::Result<()> {
         self.stream.flush().unwrap();
-        self.reactor.wait_for(self.stream.subscribe()).await;
+        Reactor::current().wait_for(self.stream.subscribe()).await;
         Ok(())
     }
 }

src/http/response.rs

@@ -45,10 +45,7 @@ pub struct Response<B: Body> {
 // }
 
 impl Response<IncomingBody> {
-    pub(crate) fn try_from_incoming_response(
-        incoming: IncomingResponse,
-        reactor: Reactor,
-    ) -> super::Result<Self> {
+    pub(crate) fn try_from_incoming_response(incoming: IncomingResponse) -> super::Result<Self> {
         let headers: Headers = incoming.headers().into();
         let status = incoming.status().into();
 
@@ -64,7 +61,6 @@ impl Response<IncomingBody> {
         let body = IncomingBody {
             buf_offset: 0,
             buf: None,
-            reactor,
             body_stream,
             _incoming_body: incoming_body,
         };
@@ -101,7 +97,6 @@ impl<B: Body> Response<B> {
 /// An incoming HTTP body
 #[derive(Debug)]
 pub struct IncomingBody {
-    reactor: Reactor,
     buf: Option<Vec<u8>>,
     // How many bytes have we already read from the buf?
     buf_offset: usize,
@@ -119,7 +114,7 @@ impl AsyncRead for IncomingBody {
             None => {
                 // Wait for an event to be ready
                 let pollable = self.body_stream.subscribe();
-                self.reactor.wait_for(pollable).await;
+                Reactor::current().wait_for(pollable).await;
 
                 // Read the bytes from the body stream
                 let buf = self.body_stream.read(CHUNK_SIZE).map_err(|err| match err {

src/lib.rs

@@ -19,8 +19,8 @@
 //! use wstd::runtime::block_on;
 //!
 //! fn main() -> io::Result<()> {
-//!     block_on(|reactor| async move {
-//!         let listener = TcpListener::bind(&reactor, "127.0.0.1:8080").await?;
+//!     block_on(async move {
+//!         let listener = TcpListener::bind("127.0.0.1:8080").await?;
 //!         println!("Listening on {}", listener.local_addr()?);
 //!         println!("type `nc localhost 8080` to create a TCP client");
 //!
@@ -56,12 +56,6 @@
 //! bytes. And `wstd::runtime` provides access to async runtime primitives.
 //! These are unique capabilities provided by WASI 0.2, and because this library
 //! is specific to that are exposed from here.
-//!
-//! Finally, this library does not implicitly thread through a
-//! [`Reactor`][runtime::Reactor] handle. Rather than using a `thread_local!`
-//! async resource APIs in `wstd` will borrow an instance of `Reactor`. This is
-//! a little more verbose, but in turn is a little simpler to implement,
-//! maintain, and extend.
 
 pub mod http;
 pub mod io;

src/net/tcp_listener.rs

@@ -11,16 +11,15 @@ use super::TcpStream;
 
 /// A TCP socket server, listening for connections.
 #[derive(Debug)]
-pub struct TcpListener<'a> {
+pub struct TcpListener {
     socket: TcpSocket,
-    reactor: &'a Reactor,
 }
 
-impl<'a> TcpListener<'a> {
+impl TcpListener {
     /// Creates a new TcpListener which will be bound to the specified address.
     ///
     /// The returned listener is ready for accepting connections.
-    pub async fn bind(reactor: &'a Reactor, addr: &str) -> io::Result<Self> {
+    pub async fn bind(addr: &str) -> io::Result<Self> {
         let addr: SocketAddr = addr
             .parse()
             .map_err(|_| io::Error::other("failed to parse string to socket addr"))?;
@@ -41,6 +40,7 @@ impl<'a> TcpListener<'a> {
             }
             SocketAddr::V6(_) => todo!("IPv6 not yet supported in `wstd::net::TcpListener`"),
         };
+        let reactor = Reactor::current();
 
         socket
             .start_bind(&network, local_address)
@@ -51,7 +51,7 @@ impl<'a> TcpListener<'a> {
         socket.start_listen().map_err(to_io_err)?;
         reactor.wait_for(socket.subscribe()).await;
         socket.finish_listen().map_err(to_io_err)?;
-        Ok(Self { socket, reactor })
+        Ok(Self { socket })
     }
 
     /// Returns the local socket address of this listener.
@@ -70,15 +70,14 @@ impl<'a> TcpListener<'a> {
 /// An iterator that infinitely accepts connections on a TcpListener.
 #[derive(Debug)]
 pub struct Incoming<'a> {
-    listener: &'a TcpListener<'a>,
+    listener: &'a TcpListener,
 }
 
 impl<'a> AsyncIterator for Incoming<'a> {
-    type Item = io::Result<TcpStream<'a>>;
+    type Item = io::Result<TcpStream>;
 
     async fn next(&mut self) -> Option<Self::Item> {
-        self.listener
-            .reactor
+        Reactor::current()
             .wait_for(self.listener.socket.subscribe())
             .await;
         let (socket, input, output) = match self.listener.socket.accept().map_err(to_io_err) {
@@ -89,7 +88,6 @@ impl<'a> AsyncIterator for Incoming<'a> {
             socket,
             input,
             output,
-            reactor: self.listener.reactor,
         }))
     }
 }

src/net/tcp_stream.rs

@@ -11,14 +11,13 @@ use crate::{
 };
 
 /// A TCP stream between a local and a remote socket.
-pub struct TcpStream<'a> {
-    pub(super) reactor: &'a Reactor,
+pub struct TcpStream {
     pub(super) input: InputStream,
     pub(super) output: OutputStream,
     pub(super) socket: TcpSocket,
 }
 
-impl<'a> TcpStream<'a> {
+impl TcpStream {
     /// Returns the socket address of the remote peer of this TCP connection.
     pub fn peer_addr(&self) -> io::Result<String> {
         let addr = self
@@ -29,29 +28,29 @@ impl<'a> TcpStream<'a> {
     }
 }
 
-impl<'a> AsyncRead for TcpStream<'a> {
+impl AsyncRead for TcpStream {
     async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        self.reactor.wait_for(self.input.subscribe()).await;
+        Reactor::current().wait_for(self.input.subscribe()).await;
         let slice = self.input.read(buf.len() as u64).map_err(to_io_err)?;
         let bytes_read = slice.len();
         buf[..bytes_read].clone_from_slice(&slice);
         Ok(bytes_read)
     }
 }
 
-impl<'a> AsyncRead for &TcpStream<'a> {
+impl AsyncRead for &TcpStream {
     async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        self.reactor.wait_for(self.input.subscribe()).await;
+        Reactor::current().wait_for(self.input.subscribe()).await;
         let slice = self.input.read(buf.len() as u64).map_err(to_io_err)?;
         let bytes_read = slice.len();
         buf[..bytes_read].clone_from_slice(&slice);
         Ok(bytes_read)
     }
 }
 
-impl<'a> AsyncWrite for TcpStream<'a> {
+impl AsyncWrite for TcpStream {
     async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        self.reactor.wait_for(self.output.subscribe()).await;
+        Reactor::current().wait_for(self.output.subscribe()).await;
         self.output.write(buf).map_err(to_io_err)?;
         Ok(buf.len())
     }
@@ -61,9 +60,9 @@ impl<'a> AsyncWrite for TcpStream<'a> {
     }
 }
 
-impl<'a> AsyncWrite for &TcpStream<'a> {
+impl AsyncWrite for &TcpStream {
     async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        self.reactor.wait_for(self.output.subscribe()).await;
+        Reactor::current().wait_for(self.output.subscribe()).await;
         self.output.write(buf).map_err(to_io_err)?;
         Ok(buf.len())
     }

src/time/mod.rs

@@ -36,32 +36,35 @@ impl SystemTime {
 }
 
 /// An async iterator representing notifications at fixed interval.
-pub fn interval(reactor: &Reactor, duration: Duration) -> Interval {
-    Interval { reactor, duration }
+pub fn interval(duration: Duration) -> Interval {
+    Interval { duration }
 }
 
 /// An async iterator representing notifications at fixed interval.
 ///
 /// See the [`interval`] function for more.
-pub struct Interval<'a> {
+pub struct Interval {
     duration: Duration,
-    reactor: &'a Reactor,
 }
-impl<'a> AsyncIterator for Interval<'a> {
+impl AsyncIterator for Interval {
     type Item = Instant;
 
     async fn next(&mut self) -> Option<Self::Item> {
-        wait_for(self.reactor, self.duration).await;
+        wait_for(self.duration).await;
         Some(Instant(wasi::clocks::monotonic_clock::now()))
     }
 }
 
 /// Wait until the passed duration has elapsed.
-pub async fn wait_for(reactor: &Reactor, duration: Duration) {
-    reactor.wait_for(subscribe_duration(duration.0)).await;
+pub async fn wait_for(duration: Duration) {
+    Reactor::current()
+        .wait_for(subscribe_duration(duration.0))
+        .await;
 }
 
 /// Wait until the passed instant.
-pub async fn wait_until(reactor: &Reactor, deadline: Instant) {
-    reactor.wait_for(subscribe_instant(deadline.0)).await;
+pub async fn wait_until(deadline: Instant) {
+    Reactor::current()
+        .wait_for(subscribe_instant(deadline.0))
+        .await;
 }