Rearrange ReadNotifier to remove dependency on async-global-executor.

Cargo.toml

@@ -18,9 +18,7 @@ readme = "README.md"
 edition = "2018"
 
 [dependencies]
-async-channel = "1.5.1"
 async-dup = "1.2.2"
-async-global-executor = "2.3.1"
 async-io = "1.13.0"
 futures-lite = "1.13.0"
 http-types = { version = "2.9.0", default-features = false }
@@ -29,5 +27,6 @@ log = "0.4.11"
 pin-project = "1.0.2"
 
 [dev-dependencies]
+async-channel = "1.5.1"
 async-std = { version = "1.7.0", features = ["attributes"] }
 pretty_assertions = "0.6.1"

src/read_notifier.rs

@@ -2,8 +2,10 @@ use std::fmt;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use async_channel::Sender;
-use futures_lite::io::{self, AsyncBufRead as BufRead, AsyncRead as Read};
+use futures_lite::{
+    io::{self, AsyncBufRead as BufRead, AsyncRead as Read},
+    Future, FutureExt,
+};
 
 /// ReadNotifier forwards [`async_std::io::Read`] and
 /// [`async_std::io::BufRead`] to an inner reader. When the
@@ -14,7 +16,7 @@ use futures_lite::io::{self, AsyncBufRead as BufRead, AsyncRead as Read};
 pub(crate) struct ReadNotifier<B> {
     #[pin]
     reader: B,
-    sender: Sender<()>,
+    before_read: Pin<Box<dyn Future<Output = ()> + Send + Sync>>,
     has_been_read: bool,
 }
 
@@ -27,10 +29,10 @@ impl<B> fmt::Debug for ReadNotifier<B> {
 }
 
 impl<B: Read> ReadNotifier<B> {
-    pub(crate) fn new(reader: B, sender: Sender<()>) -> Self {
+    pub(crate) fn new(reader: B, before_read: Box<dyn Future<Output = ()> + Send + Sync>) -> Self {
         Self {
             reader,
-            sender,
+            before_read: Box::into_pin(before_read),
             has_been_read: false,
         }
     }
@@ -55,9 +57,13 @@ impl<B: Read> Read for ReadNotifier<B> {
         let this = self.project();
 
         if !*this.has_been_read {
-            if let Ok(()) = this.sender.try_send(()) {
-                *this.has_been_read = true;
-            };
+            // execute the before_read future before we start polling the reader
+            match this.before_read.poll(cx) {
+                Poll::Ready(_) => {
+                    *this.has_been_read = true;
+                }
+                Poll::Pending => return Poll::Pending,
+            }
         }
 
         this.reader.poll_read(cx, buf)

src/server/decode.rs

@@ -96,25 +96,16 @@ where
         "Unexpected Content-Length header"
     );
 
-    // Establish a channel to wait for the body to be read. This
-    // allows us to avoid sending 100-continue in situations that
-    // respond without reading the body, saving clients from uploading
-    // their body.
-    let (body_read_sender, body_read_receiver) = async_channel::bounded(1);
-
-    if Some(CONTINUE_HEADER_VALUE) == req.header(EXPECT).map(|h| h.as_str()) {
-        async_global_executor::spawn(async move {
-            // If the client expects a 100-continue header, spawn a
-            // task to wait for the first read attempt on the body.
-            if let Ok(()) = body_read_receiver.recv().await {
+    // Create a future to be polled when the body is to be read. By avoiding
+    // sending a 100-continue until the handler explicitly reads the body,
+    // we can save the client from uploading unnecessary data.
+    let before_read: Box<dyn Future<Output = ()> + Send + Sync> =
+        match req.header(EXPECT).map(|h| h.as_str()) {
+            Some(CONTINUE_HEADER_VALUE) => Box::new(async move {
                 io.write_all(CONTINUE_RESPONSE).await.ok();
-            };
-            // Since the sender is moved into the Body, this task will
-            // finish when the client disconnects, whether or not
-            // 100-continue was sent.
-        })
-        .detach();
-    }
+            }),
+            _ => Box::new(async {}),
+        };
 
     // Check for Transfer-Encoding
     if transfer_encoding
@@ -125,15 +116,15 @@ where
         let reader = ChunkedDecoder::new(reader, trailer_sender);
         let reader = Arc::new(Mutex::new(reader));
         let reader_clone = reader.clone();
-        let reader = ReadNotifier::new(reader, body_read_sender);
+        let reader = ReadNotifier::new(reader, before_read);
         let reader = BufReader::new(reader);
         req.set_body(Body::from_reader(reader, None));
         Ok(Some((req, BodyReader::Chunked(reader_clone))))
     } else if let Some(len) = content_length {
         let len = len.len();
         let reader = Arc::new(Mutex::new(reader.take(len)));
         req.set_body(Body::from_reader(
-            BufReader::new(ReadNotifier::new(reader.clone(), body_read_sender)),
+            BufReader::new(ReadNotifier::new(reader.clone(), before_read)),
             Some(len as usize),
         ));
         Ok(Some((req, BodyReader::Fixed(reader))))