remove async-std

.github/workflows/ci.yml

@@ -31,7 +31,7 @@ jobs:
         run: |
           Start-Process cmd -Args /c,"http-server --tls --tls-key certs/test-certs/server.key --tls-cert certs/test-certs/server.crt --tls-key-algorithm pkcs8"
           cargo install --force --locked bindgen-cli
-          cargo test --features task,subscriber,fixture,task_unstable,io,sync,future,net,tls,rust_tls,timer,fs,zero_copy,mmap,retry
+          cargo test --features task,subscriber,fixture,io,sync,future,net,tls,rust_tls,timer,fs,zero_copy,mmap,retry
   test:
     name: Check ${{ matrix.check }} on (${{ matrix.os }})
     runs-on: ${{ matrix.os }}

Cargo.toml

@@ -12,11 +12,10 @@ resolver = "2"
 all-features = true
 
 [features]
-task = ["async-std/default", "timer", "cfg-if"]
+task = ["async-global-executor", "timer", "cfg-if"]
 subscriber = ["tracing-subscriber", "tracing-subscriber/std", "tracing-subscriber/env-filter"]
 fixture = ["subscriber", "task", "fluvio-future-derive"]
-task_unstable = ["task", "async-std/unstable"]
-io = ["async-std/default"]
+io = ["async-io", "async-global-executor/async-io"]
 sync = ["async-std/default"]
 future = ["async-std/default"]
 net = ["futures-lite", "async-net", "async-trait", "cfg-if", "futures-util/io", "socket2"]
@@ -28,11 +27,11 @@ rust_tls = [
     "pin-project",
     "futures-util/io",
 ]
-native2_tls = [
+native_tls = [
     "net",
     "pin-project",
     "async-native-tls",
-    "native-tls",
+    "dep:native-tls",
     "openssl/vendored",
     "futures-util/io",
 ]
@@ -45,15 +44,16 @@ openssl_tls = [
 ]
 timer = ["async-io", "pin-project", "futures-lite"]
 fs = ["async-fs", "futures-lite", "pin-utils", "async-trait"]
-zero_copy = ["nix", "task_unstable"]
-mmap = ["fs", "memmap2", "task_unstable"]
+zero_copy = ["nix", "task"]
+mmap = ["fs", "memmap2", "task"]
 retry = []
 doomsday = ["task", "sync"]
 tokio1 = ["async-std/tokio1"]
 attributes = []
 
 [dependencies]
 anyhow = { version = "1.0" }
+async-global-executor = { version = "2.4.1", default-features = false, optional = true }
 async-trait = { version = "0.1.80", optional = true }
 cfg-if = { version = "1.0", optional = true }
 futures-lite = { version = "2.3", optional = true }

Makefile

@@ -30,7 +30,7 @@ test-macos: PFX_OPTS=""
 test-macos: certs cert-patch-macos test-derive setup-http-server run-test-macos
 run-test-macos:
 	TEST_PORT=$$(cat tmp-PORT) cargo test \
-		--features "task,subscriber,fixture,task_unstable,io,sync,future,net,tls,timer,fs,zero_copy,retry,doomsday,tokio1"
+		--features "task,subscriber,fixture,io,sync,future,net,tls,timer,fs,zero_copy,retry,doomsday,tokio1"
 	$(MAKE) teardown-http-server
 
 install-wasm-pack:

fluvio-future-derive/src/lib.rs

@@ -1,6 +1,6 @@
 use proc_macro::TokenStream;
 use quote::{quote, quote_spanned};
-use syn::spanned::Spanned;
+use syn::{spanned::Spanned, AttributeArgs};
 
 
 #[proc_macro_attribute]
@@ -47,9 +47,11 @@ pub fn main_async(_attr: TokenStream, item: TokenStream) -> TokenStream {
 }
 
 #[proc_macro_attribute]
-pub fn test_async(_attr: TokenStream, item: TokenStream) -> TokenStream {
-    let input = syn::parse_macro_input!(item as syn::ItemFn);
+pub fn test_async(args: TokenStream, item: TokenStream) -> TokenStream {
+    let attribute_args = syn::parse_macro_input!(args as AttributeArgs);
+    let test_attributes = generate::generate_test_attributes(&attribute_args);
 
+    let input = syn::parse_macro_input!(item as syn::ItemFn);
     let ret = &input.sig.output;
     let name = &input.sig.ident;
     let body = &input.block;
@@ -64,6 +66,7 @@ pub fn test_async(_attr: TokenStream, item: TokenStream) -> TokenStream {
 
     let result = quote! {
         #[::core::prelude::v1::test]
+        #test_attributes
         #(#attrs)*
         #vis fn #name() #ret {
             ::fluvio_future::subscriber::init_logger();
@@ -74,3 +77,24 @@ pub fn test_async(_attr: TokenStream, item: TokenStream) -> TokenStream {
 
     result.into()
 }
+
+mod generate {
+
+    use proc_macro2::TokenStream;
+    use quote::quote;
+    use syn::NestedMeta;
+
+    pub fn generate_test_attributes(attributes: &Vec<NestedMeta>) -> TokenStream {
+        let args = attributes.iter().map(|meta| {
+            quote! {
+                #[#meta]
+            }
+        });
+
+        quote! {
+
+            #(#args)*
+
+        }
+    }
+}

src/doomsday.rs

@@ -10,11 +10,7 @@ use log::info;
 use tracing::{debug, error};
 
 use crate::sync::Mutex;
-#[deprecated(
-    since = "0.5.1",
-    note = "please use `fluvio_future::task::JoinHandle` instead"
-)]
-pub use crate::task::JoinHandle;
+use crate::task::Task;
 
 #[derive(Clone)]
 /// DoomsdayTimer will configurably panic or exit if it is not
@@ -45,7 +41,7 @@ impl DoomsdayTimer {
     /// Spawn a new doomsday timer.
     /// If `exit_on_explode` is true, it will terminate process with `exit(1)` if it explodes.
     /// Otherwise it will call `panic()`. Note that `awaiting` on the jh will panic if the `DoomsdayTimer` panicked
-    pub fn spawn(duration: Duration, exit_on_explode: bool) -> (Self, JoinHandle<()>) {
+    pub fn spawn(duration: Duration, exit_on_explode: bool) -> (Self, Task<()>) {
         let s = Self {
             time_to_explode: Arc::new(Mutex::new(Instant::now() + duration)),
             duration,
@@ -117,11 +113,10 @@ mod tests {
     use std::io::Error;
 
     #[test_async(should_panic)]
-    async fn test_explode() -> Result<(), Error> {
+    async fn test_explode() {
         let (_, jh) = DoomsdayTimer::spawn(Duration::from_millis(1), false);
         crate::timer::sleep(Duration::from_millis(2)).await;
         jh.await;
-        Ok(())
     }
 
     #[test_async]

src/fs/extension.rs

@@ -88,14 +88,13 @@ mod tests {
     use std::io::SeekFrom;
     use std::io::Write;
 
-    use async_std::io::prelude::SeekExt;
-
     use flv_util::fixture::ensure_clean_file;
 
     use super::AsyncFileExtension;
     use crate::fs::util as file_util;
     use crate::test_async;
     use futures_lite::AsyncReadExt;
+    use futures_lite::AsyncSeekExt;
     use futures_lite::AsyncWriteExt;
 
     // sync seek write and read

src/io.rs

@@ -1,2 +1 @@
-pub use async_std::io::*;
-pub use async_std::prelude::*;
+pub use async_io::*;

src/lib.rs

@@ -32,13 +32,13 @@ pub mod net;
 #[cfg(all(any(unix, windows), feature = "rust_tls"))]
 pub mod rust_tls;
 
-#[cfg(all(any(unix, windows), feature = "rust_tls", not(feature = "native2_tls")))]
+#[cfg(all(any(unix, windows), feature = "rust_tls", not(feature = "native_tls")))]
 pub use rust_tls as tls;
 
-#[cfg(all(any(unix, windows), feature = "native2_tls"))]
+#[cfg(all(any(unix, windows), feature = "native_tls"))]
 pub mod native_tls;
 
-#[cfg(all(any(unix, windows), feature = "native2_tls", not(feature = "rust_tls")))]
+#[cfg(all(any(unix, windows), feature = "native_tls", not(feature = "rust_tls")))]
 pub use crate::native_tls as tls;
 
 #[cfg(feature = "openssl_tls")]
@@ -71,10 +71,9 @@ pub mod subscriber {
 #[cfg(feature = "doomsday")]
 pub mod doomsday;
 
-#[cfg(any(feature = "attributes"))]
+#[cfg(feature = "attributes")]
 pub use fluvio_future_derive::main_async;
 
-
 /// re-export tracing
 pub mod tracing {
 

src/retry.rs

@@ -371,10 +371,10 @@ mod delay {
 #[cfg(test)]
 mod test {
     use super::*;
+    use fluvio_future_derive::test_async;
     use std::io::ErrorKind;
     use std::ops::AddAssign;
     use std::time::Duration;
-    use fluvio_future_derive::test_async;
     use tracing::debug;
 
     #[test_async]

src/rust_tls.rs

@@ -46,7 +46,7 @@ mod cert {
     use futures_rustls::rustls::pki_types::PrivateKeyDer;
     use futures_rustls::rustls::RootCertStore;
     use rustls_pemfile::certs;
-    use rustls_pemfile::rsa_private_keys;
+    use rustls_pemfile::pkcs8_private_keys;
 
     pub fn load_certs<P: AsRef<Path>>(path: P) -> Result<Vec<CertificateDer<'static>>, IoError> {
         load_certs_from_reader(&mut BufReader::new(File::open(path)?))
@@ -68,7 +68,7 @@ mod cert {
     pub fn load_keys_from_reader(
         rd: &mut dyn BufRead,
     ) -> Result<Vec<PrivateKeyDer<'static>>, IoError> {
-        rsa_private_keys(rd)
+        pkcs8_private_keys(rd)
             .map(|r| {
                 r.map(|p| p.into())
                     .map_err(|_| IoError::new(ErrorKind::InvalidInput, "invalid key"))
@@ -453,21 +453,10 @@ mod builder {
         }
 
         fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
-            vec![
-                SignatureScheme::RSA_PKCS1_SHA1,
-                SignatureScheme::ECDSA_SHA1_Legacy,
-                SignatureScheme::RSA_PKCS1_SHA256,
-                SignatureScheme::ECDSA_NISTP256_SHA256,
-                SignatureScheme::RSA_PKCS1_SHA384,
-                SignatureScheme::ECDSA_NISTP384_SHA384,
-                SignatureScheme::RSA_PKCS1_SHA512,
-                SignatureScheme::ECDSA_NISTP521_SHA512,
-                SignatureScheme::RSA_PSS_SHA256,
-                SignatureScheme::RSA_PSS_SHA384,
-                SignatureScheme::RSA_PSS_SHA512,
-                SignatureScheme::ED25519,
-                SignatureScheme::ED448,
-            ]
+            let provider = futures_rustls::rustls::crypto::aws_lc_rs::default_provider();
+            provider
+                .signature_verification_algorithms
+                .supported_schemes()
         }
     }
 }
@@ -494,8 +483,8 @@ mod test {
 
     use fluvio_future::net::tcp_stream::stream;
     use fluvio_future::net::TcpListener;
-    use fluvio_future::timer::sleep;
     use fluvio_future::test_async;
+    use fluvio_future::timer::sleep;
 
     use super::{AcceptorBuilder, ConnectorBuilder};
 

src/task.rs

@@ -1,19 +1,16 @@
 use std::future::Future;
 
-use async_std::task;
-
 use crate::timer::sleep;
 
-#[cfg(feature = "task_unstable")]
-pub use async_std::task::spawn_local;
+pub use async_global_executor::spawn_local;
 
 /// run future and wait forever
 /// this is typically used in the server
 pub fn run<F>(spawn_closure: F)
 where
     F: Future<Output = ()> + Send + 'static,
 {
-    task::block_on(spawn_closure);
+    async_global_executor::block_on(spawn_closure);
 }
 
 /// run future and wait forever
@@ -24,7 +21,7 @@ where
 {
     use std::time::Duration;
 
-    task::block_on(async {
+    async_global_executor::block_on(async {
         spawn_closure.await;
         // do infinite loop for now
         loop {
@@ -41,15 +38,14 @@ where
 
 cfg_if::cfg_if! {
     if #[cfg(target_arch = "wasm32")] {
-        pub use async_std::task::spawn_local as spawn;
+        pub use async_global_executor::spawn_local as spawn;
     } else {
-        pub use async_std::task::spawn;
+        pub use async_global_executor::spawn;
     }
 }
 
-#[cfg(feature = "task_unstable")]
 #[cfg(not(target_arch = "wasm32"))]
-pub use async_std::task::spawn_blocking;
+pub use async_global_executor::spawn_blocking;
 
 cfg_if::cfg_if! {
     if #[cfg(target_arch = "wasm32")] {
@@ -58,14 +54,14 @@ cfg_if::cfg_if! {
                 F: Future<Output = T> + 'static,
                 T: 'static,
             {
-                task::block_on(f)
+                block_on(f)
             }
     } else {
-        pub use async_std::task::block_on as run_block_on;
+        pub use async_global_executor::block_on as run_block_on;
     }
 }
 
-pub use async_std::task::JoinHandle;
+pub use async_global_executor::Task;
 
 #[cfg(test)]
 mod basic_test {
@@ -133,8 +129,8 @@ mod basic_test {
         let core_threads = num_cpus::get().max(1);
         debug!("num threads: {}", core_threads);
 
-        spawn(ft1);
-        spawn(ft2);
+        spawn(ft1).detach();
+        spawn(ft2).detach();
         // wait for all futures complete
         thread::sleep(time::Duration::from_millis(2000));
 

src/timer.rs

@@ -85,8 +85,8 @@ mod test {
     use log::debug;
     use tokio::select;
 
-    use crate::timer::sleep;
     use crate::test_async;
+    use crate::timer::sleep;
 
     /// test timer loop
     #[test_async]

src/zero_copy.rs

@@ -75,8 +75,8 @@ impl ZeroCopy {
                     );
 
                     match sendfile(
-                        &target_fd,
-                        &source_fd,
+                        target_fd,
+                        source_fd,
                         Some(&mut current_offset),
                         to_be_transfer,
                     ) {
@@ -206,9 +206,9 @@ mod tests {
     use crate::fs::AsyncFileExtension;
     use crate::net::tcp_stream::stream;
     use crate::net::TcpListener;
+    use crate::test_async;
     use crate::timer::sleep;
     use crate::{fs::util as file_util, zero_copy::ZeroCopy};
-    use crate::test_async;
     use futures_lite::AsyncReadExt;
 
     use super::SendFileError;