Upgrade to the v1 of smol building blocks

Author: oblique

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Changed
+
+- Upgrade to the v1 of [smol] building blocks.
+
 ## [0.3.2] - 2020-08-31
 
 ### Changed

Cargo.toml

@@ -22,11 +22,11 @@ num-derive = "0.3.2"
 num-traits = "0.2.12"
 log = "0.4.11"
 
-async-io = "0.2.6"
-async-mutex = "1.2.0"
 async-trait = "0.1.40"
-blocking = "0.6.1"
-futures-lite = "1.0.0"
+async-io = "1.0.2"
+async-mutex = "1.3.0"
+blocking = "1.0.0"
+futures-lite = "1.4.0"
 futures-util = { version = "0.3.5", default-features = false, features = ["alloc"] }
 
 [dev-dependencies]
@@ -36,8 +36,8 @@ rand = { version = "0.7.3", features = ["small_rng"] }
 md5 = "0.7.0"
 tempfile = "3.1.0"
 fern = "0.6.0"
-async-executor = "0.2.1"
-async-channel = "1.4.1"
+async-executor = "1.1.1"
+async-channel = "1.4.2"
 
 # deps for tftpd-targz.rs
 async-std = "1.6.3"

src/tests/rrq.rs

@@ -3,43 +3,59 @@
 
 use async_executor::Executor;
 use blocking::Unblock;
+use futures_lite::future::block_on;
+use std::cell::Cell;
+use std::rc::Rc;
+use std::sync::Arc;
 
 use super::external_client::*;
 use super::handlers::*;
 use crate::server::TftpServerBuilder;
 
 fn transfer(file_size: usize, block_size: Option<u16>) {
-    let ex = Executor::new();
-
-    ex.run(async {
-        let (md5_tx, md5_rx) = async_channel::bounded(1);
-        let handler = RandomHandler::new(file_size, md5_tx);
-
-        // bind
-        let tftpd = TftpServerBuilder::with_handler(handler)
-            .bind("127.0.0.1:0".parse().unwrap())
-            .build()
-            .await
-            .unwrap();
-        let addr = tftpd.listen_addr().unwrap();
-
-        // start client
-        let mut tftp_recv = Unblock::new(());
-        let tftp_recv = tftp_recv
-            .with_mut(move |_| external_tftp_recv("test", addr, block_size));
-
-        // start server
-        ex.spawn(async move {
-            tftpd.serve().await.unwrap();
-        })
-        .detach();
-
-        // check md5
-        let client_md5 = tftp_recv.await.expect("failed to run tftp client");
-        let server_md5 =
-            md5_rx.recv().await.expect("failed to receive server md5");
-        assert_eq!(client_md5, server_md5);
-    });
+    let ex = Arc::new(Executor::new());
+    let transfered = Rc::new(Cell::new(false));
+
+    block_on(ex.run({
+        let ex = ex.clone();
+        let transfered = transfered.clone();
+
+        async move {
+            let (md5_tx, md5_rx) = async_channel::bounded(1);
+            let handler = RandomHandler::new(file_size, md5_tx);
+
+            // bind
+            let tftpd = TftpServerBuilder::with_handler(handler)
+                .bind("127.0.0.1:0".parse().unwrap())
+                .build()
+                .await
+                .unwrap();
+            let addr = tftpd.listen_addr().unwrap();
+
+            // start client
+            let mut tftp_recv = Unblock::new(());
+            let tftp_recv = tftp_recv.with_mut(move |_| {
+                external_tftp_recv("test", addr, block_size)
+            });
+
+            // start server
+            ex.spawn(async move {
+                tftpd.serve().await.unwrap();
+            })
+            .detach();
+
+            // check md5
+            let client_md5 =
+                tftp_recv.await.expect("failed to run tftp client");
+            let server_md5 =
+                md5_rx.recv().await.expect("failed to receive server md5");
+            assert_eq!(client_md5, server_md5);
+
+            transfered.set(true);
+        }
+    }));
+
+    assert!(transfered.get());
 }
 
 #[test]

src/utils.rs

@@ -1,5 +1,5 @@
 use async_io::Timer;
-use futures_util::future::{select, Either};
+use futures_lite::future;
 use std::future::Future;
 use std::io;
 use std::time::Duration;
@@ -8,10 +8,9 @@ pub async fn io_timeout<T>(
     dur: Duration,
     f: impl Future<Output = io::Result<T>>,
 ) -> io::Result<T> {
-    futures_lite::pin!(f);
-
-    match select(f, Timer::after(dur)).await {
-        Either::Left((out, _)) => out,
-        Either::Right(_) => Err(io::ErrorKind::TimedOut.into()),
-    }
+    future::race(f, async move {
+        Timer::after(dur).await;
+        Err(io::ErrorKind::TimedOut.into())
+    })
+    .await
 }