Try for each concurrent (#24)

* use try-for-each-concurrent

Signed-off-by: Yoshua Wuyts <[email protected]>

* move proxy example over

Signed-off-by: Yoshua Wuyts <[email protected]>

* fmt and guessing

Signed-off-by: Yoshua Wuyts <[email protected]>

* improve upper bound

Signed-off-by: Yoshua Wuyts <[email protected]>

Author: yoshuawuyts

examples/guessing.rs

@@ -61,9 +61,12 @@ async fn main() -> Result<(), failure::Error> {
     let mut listener = TcpListener::bind("127.0.0.1:8080")?;
     println!("Listening on {}", &listener.local_addr()?);
 
-    let mut incoming = listener.incoming();
-    while let Some(stream) = incoming.next().await {
-        runtime::spawn(play(stream?)).await?;
-    }
+    let incoming = listener.incoming().map_err(|e| e.into());
+    incoming
+        .try_for_each_concurrent(None, async move |stream| {
+            runtime::spawn(play(stream)).await?;
+            Ok::<(), failure::Error>(())
+        })
+        .await?;
     Ok(())
 }

examples/tcp-echo.rs

@@ -14,17 +14,18 @@ async fn main() -> std::io::Result<()> {
     println!("Listening on {}", listener.local_addr()?);
 
     // accept connections and process them in parallel
-    let mut incoming = listener.incoming();
-    while let Some(stream) = incoming.next().await {
-        runtime::spawn(async move {
-            let stream = stream?;
-            println!("Accepting from: {}", stream.peer_addr()?);
+    listener
+        .incoming()
+        .try_for_each_concurrent(None, async move |stream| {
+            runtime::spawn(async move {
+                println!("Accepting from: {}", stream.peer_addr()?);
 
-            let (reader, writer) = &mut stream.split();
-            reader.copy_into(writer).await?;
-            Ok::<(), std::io::Error>(())
+                let (reader, writer) = &mut stream.split();
+                reader.copy_into(writer).await?;
+                Ok::<(), std::io::Error>(())
+            })
+            .await
         })
         .await?;
-    }
     Ok(())
 }

examples/tcp-proxy.rs

@@ -11,28 +11,28 @@ async fn main() -> std::io::Result<()> {
     let mut listener = TcpListener::bind("127.0.0.1:8081")?;
     println!("Listening on {}", listener.local_addr()?);
 
-    // accept connections and process them serially
-    let mut incoming = listener.incoming();
-    while let Some(client) = incoming.next().await {
-        let handle = runtime::spawn(async move {
-            let client = client?;
-            let server = TcpStream::connect("127.0.0.1:8080").await?;
-            println!(
-                "Proxying {} to {}",
-                client.peer_addr()?,
-                server.peer_addr()?
-            );
+    // accept connections and process them in parallel
+    listener
+        .incoming()
+        .try_for_each_concurrent(None, async move |client| {
+            runtime::spawn(async move {
+                let server = TcpStream::connect("127.0.0.1:8080").await?;
+                println!(
+                    "Proxying {} to {}",
+                    client.peer_addr()?,
+                    server.peer_addr()?
+                );
 
-            let (cr, cw) = &mut client.split();
-            let (sr, sw) = &mut server.split();
-            let a = cr.copy_into(sw);
-            let b = sr.copy_into(cw);
-            try_join!(a, b)?;
+                let (cr, cw) = &mut client.split();
+                let (sr, sw) = &mut server.split();
+                let a = cr.copy_into(sw);
+                let b = sr.copy_into(cw);
+                try_join!(a, b)?;
 
-            Ok::<(), std::io::Error>(())
-        });
-
-        handle.await?;
-    }
+                Ok::<(), std::io::Error>(())
+            })
+            .await
+        })
+        .await?;
     Ok(())
 }