UdpStream::poll_shutdown usage

Author: ssrlive

examples/echo-dtls.rs

@@ -1,4 +1,4 @@
-use std::{error::Error, net::SocketAddr, pin::Pin, str::FromStr, time::Duration};
+use std::{net::SocketAddr, pin::Pin, str::FromStr, time::Duration};
 use udp_stream::UdpListener;
 
 use openssl::{
@@ -29,28 +29,30 @@ fn ssl_acceptor(certificate: &[u8], private_key: &[u8]) -> std::io::Result<SslCo
 }
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let listener = UdpListener::bind(SocketAddr::from_str("127.0.0.1:8080")?).await?;
     let acceptor = ssl_acceptor(SERVER_CERT, SERVER_KEY)?;
     loop {
         let (socket, _) = listener.accept().await?;
         let acceptor = acceptor.clone();
         tokio::spawn(async move {
-            let ssl = Ssl::new(&acceptor).unwrap();
-            let mut stream = tokio_openssl::SslStream::new(ssl, socket).unwrap();
-            Pin::new(&mut stream).accept().await.unwrap();
+            let ssl = Ssl::new(&acceptor).map_err(std::io::Error::other)?;
+            let mut stream = tokio_openssl::SslStream::new(ssl, socket).map_err(std::io::Error::other)?;
+            Pin::new(&mut stream).accept().await.map_err(std::io::Error::other)?;
             let mut buf = vec![0u8; UDP_BUFFER_SIZE];
             loop {
                 let duration = Duration::from_millis(UDP_TIMEOUT);
-                let n = match timeout(duration, stream.read(&mut buf)).await.unwrap() {
-                    Ok(len) => len,
+                let n = match timeout(duration, stream.read(&mut buf)).await {
+                    Ok(len) => len?,
                     Err(_) => {
-                        return;
+                        stream.shutdown().await?;
+                        break;
                     }
                 };
 
-                stream.write_all(&buf[0..n]).await.unwrap();
+                stream.write_all(&buf[0..n]).await?;
             }
+            Ok::<(), std::io::Error>(())
         });
     }
 }

examples/echo-udp.rs

@@ -1,4 +1,4 @@
-use std::{error::Error, net::SocketAddr, str::FromStr, time::Duration};
+use std::{net::SocketAddr, str::FromStr, time::Duration};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     time::timeout,
@@ -8,7 +8,7 @@ const UDP_BUFFER_SIZE: usize = 17480; // 17kb
 const UDP_TIMEOUT: u64 = 10 * 1000; // 10sec
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let level = format!("{}={}", module_path!(), "trace");
     env_logger::Builder::from_env(env_logger::Env::default().default_filter_or(level)).init();
 
@@ -21,11 +21,17 @@ async fn main() -> Result<(), Box<dyn Error>> {
                 let mut buf = vec![0u8; UDP_BUFFER_SIZE];
                 let duration = Duration::from_millis(UDP_TIMEOUT);
                 loop {
-                    let n = timeout(duration, stream.read(&mut buf)).await??;
+                    let n = match timeout(duration, stream.read(&mut buf)).await {
+                        Err(err) => {
+                            log::debug!("{id:?} {err}");
+                            stream.shutdown().await?;
+                            break;
+                        }
+                        Ok(val) => val?,
+                    };
                     stream.write_all(&buf[0..n]).await?;
                     log::trace!("{:?} echoed {:?} for {} bytes", id, stream.peer_addr(), n);
                 }
-                #[allow(unreachable_code)]
                 Ok::<(), std::io::Error>(())
             };
             if let Err(e) = block.await {