Refactor AsyncWrite implementation and enhance web crawler error handling

examples/hyper-server.rs

@@ -149,8 +149,8 @@ impl AsyncWrite for SmolStream {
 
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         match &mut *self {
-            Self::Plain(s) => Pin::new(s).poll_close(cx),
-            Self::Tls(s) => Pin::new(s).poll_close(cx),
+            Self::Plain(s) => Pin::new(s).poll_flush(cx),
+            Self::Tls(s) => Pin::new(s).poll_flush(cx),
         }
     }
 }

examples/web-crawler.rs

@@ -14,11 +14,30 @@ use scraper::{Html, Selector};
 
 const ROOT: &str = "https://www.rust-lang.org";
 
+/// A guard that sends an empty string when dropped so the main loop never deadlocks.
+struct FetchGuard {
+    sender: Sender<String>,
+    sent: bool,
+}
+
+impl Drop for FetchGuard {
+    fn drop(&mut self) {
+        if !self.sent {
+            let _ = self.sender.try_send(String::new());
+        }
+    }
+}
+
 /// Fetches the HTML contents of a web page.
 async fn fetch(url: String, sender: Sender<String>) {
+    let mut guard = FetchGuard {
+        sender: sender.clone(),
+        sent: false,
+    };
     let body = surf::get(&url).recv_string().await;
     let body = body.unwrap_or_default();
-    sender.send(body).await.ok();
+    let _ = sender.send(body).await;
+    guard.sent = true;
 }
 
 /// Extracts links from an HTML body.

examples/websocket-server.rs

@@ -16,7 +16,7 @@ use std::net::{TcpListener, TcpStream};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use anyhow::{Context as _, Result};
+use anyhow::Result;
 use async_native_tls::{Identity, TlsAcceptor, TlsStream};
 use async_tungstenite::{tungstenite, WebSocketStream};
 use futures::sink::{Sink, SinkExt};
@@ -25,8 +25,10 @@ use tungstenite::Message;
 
 /// Echoes messages from the client back to it.
 async fn echo(mut stream: WsStream) -> Result<()> {
-    let msg = stream.next().await.context("expected a message")??;
-    stream.send(Message::text(msg.to_string())).await?;
+    while let Some(msg) = stream.next().await {
+        let msg = msg?;
+        stream.send(Message::text(msg.to_string())).await?;
+    }
     Ok(())
 }
 

src/spawn.rs

@@ -41,6 +41,7 @@ pub fn spawn<T: Send + 'static>(future: impl Future<Output = T> + Send + 'static
                     .ok()
                     .and_then(|s| s.parse().ok())
                     .unwrap_or(1)
+                    .max(1)
             };
 
             for n in 1..=num_threads {