Make async-std the default

- Change cfg directives to use async-std channels and executor if
  nothing is specified via rust flags.
- Add new empty RUSTFLAGS matrix variant to CI jobs.
- Add a justfile for local testing.
- Update README.

Author: sveitser

.github/workflows/build.yml

@@ -24,7 +24,10 @@ jobs:
     strategy:
       matrix:
         flags:
-          - --cfg async_executor_impl=\"async-std\" --cfg async_channel_impl=\"async-std\" 
+          # no flags set, default to async-std
+          - ""
+          # crates that explicitly choose async-std should keep working
+          - --cfg async_executor_impl=\"async-std\" --cfg async_channel_impl=\"async-std\"
           - --cfg async_executor_impl=\"async-std\" --cfg async_channel_impl=\"flume\"
           - --cfg async_executor_impl=\"tokio\" --cfg async_channel_impl=\"tokio\"
           - --cfg async_executor_impl=\"tokio\" --cfg async_channel_impl=\"flume\"
@@ -59,7 +62,6 @@ jobs:
     runs-on: ubuntu-latest
     env:
         RUST_LOG: info
-        RUSTFLAGS: "--cfg async_executor_impl=\"async-std\" --cfg async_channel_impl=\"async-std\""
     steps:
       - uses: actions/checkout@v4
       - uses: katyo/publish-crates@v2

.github/workflows/build_nix.yml

@@ -10,6 +10,9 @@ jobs:
     strategy:
       matrix:
         flags:
+          # no flags set, default to async-std
+          - ""
+          # crates that explicitly choose async-std should keep working
           - --cfg async_executor_impl=\"async-std\" --cfg async_channel_impl=\"async-std\"
           - --cfg async_executor_impl=\"async-std\" --cfg async_channel_impl=\"flume\"
           - --cfg async_executor_impl=\"tokio\" --cfg async_channel_impl=\"tokio\"

Cargo.lock

@@ -2,7 +2,7 @@
 name = "async-compatibility-layer"
 description = "an abstraction layer for using both async-std and tokio"
 authors = ["Espresso Systems <[email protected]>"]
-version = "1.1.0"
+version = "1.2.0"
 edition = "2021"
 license = "MIT"
 
@@ -41,7 +41,7 @@ opentelemetry-jaeger = { version = "0.19.0", features = [
 ], optional = true }
 opentelemetry-aws = { version = "0.8.0", features = ["trace"], optional = true }
 
-[target.'cfg(all(async_executor_impl = "tokio"))'.dependencies]
+[target.'cfg(async_executor_impl = "tokio")'.dependencies]
 console-subscriber = { version = "0.2.0" }
 tokio = { version = "1", features = [
     "fs",
@@ -60,13 +60,13 @@ tokio = { version = "1", features = [
 ] }
 tokio-stream = { version = "0.1.14" }
 
-[target.'cfg(all(async_executor_impl = "async-std"))'.dependencies]
+[target.'cfg(not(async_executor_impl = "tokio"))'.dependencies]
 async-std = { version = "1.12", features = [
     "attributes",
     "unstable",
 ]}
 
-[target.'cfg(all(async_channel_impl = "tokio"))'.dependencies]
+[target.'cfg(async_channel_impl = "tokio")'.dependencies]
 tokio = { version = "1", features = [
     "fs",
     "io-util",
@@ -84,7 +84,7 @@ tokio = { version = "1", features = [
 ] }
 tokio-stream = { version = "0.1.14" }
 
-[target.'cfg(all(async_channel_impl = "async-std"))'.dependencies]
+[target.'cfg(not(any(async_channel_impl = "tokio", async_channel_impl = "flume")))'.dependencies]
 async-std = { version = "1.12", features = [
     "attributes",
     "unstable",

Cargo.toml

@@ -2,17 +2,25 @@
 
 This crate exports four things:
 
-- A compatibility/abstraction layer for writing async-executor agnostic code. We support two async executors: async-std and tokio. Each may be toggled with a configuration flag.
-- A compatibility/abstraction layer for writing async channel agnostic code. We support three async channel implementations: async-std-channels. Each may be toggled with a configuration flag.
+- A compatibility/abstraction layer for writing async-executor agnostic code. We
+  support two async executors: async-std and tokio. Each may be toggled with a
+  configuration flag.
+- A compatibility/abstraction layer for writing async channel agnostic code. We
+  support three async channel implementations: async-std, tokio and flume. Each
+  may be toggled with a configuration flag.
 - A library exporting a bunch of useful async primitives.
 - A tracing configuration layer optionally supporting console and opentelemetry integration.
 
 # Example usage
+By default the `async-std` executor and channels are used.
+
+To use tokio:
 
 ```bash
 RUSTFLAGS='--cfg async_executor_impl="tokio" --cfg async_channel_impl="tokio"' cargo build
 ```
 
-`async_executor_impl` may be either `tokio` or `async-std`. `async_channel_impl` may be either `tokio`, `async-std`, or `flume`. Note that using `tokio` channels requires `tokio` to be the runtime. Note that the async executor impl and async channel impl must be set in order for this crate to compile successfully.
-
+`async_executor_impl` may be either `tokio` or `async-std`. `async_channel_impl`
+may be either `tokio`, `async-std`, or `flume`.
 
+Note that using `tokio` channels requires `tokio` to be the runtime.

README.md

@@ -0,0 +1,27 @@
+async_std := '--cfg async_executor_impl=\"async-std\" --cfg async_channel_impl=\"async-std\"'
+tokio := '--cfg async_executor_impl=\"tokio\" --cfg async_channel_impl=\"tokio\"'
+async_std_flume := '--cfg async_executor_impl=\"async-std\" --cfg async_channel_impl=\"flume\"'
+tokio_flume := '--cfg async_executor_impl=\"tokio\" --cfg async_channel_impl=\"flume\"'
+
+
+matrix *args:
+    {{args}}
+    env RUSTFLAGS="{{async_std}}" RUSTDOCFLAGS="{{async_std}}" {{args}}
+    env RUSTFLAGS="{{tokio}}" RUSTDOCFLAGS="{{tokio}}" {{args}}
+    env RUSTFLAGS="{{async_std_flume}}" RUSTDOCFLAGS="{{async_std_flume}}" {{args}}
+    env RUSTFLAGS="{{tokio_flume}}" RUSTDOCFLAGS="{{tokio_flume}}" {{args}}
+
+test *args:
+    cargo test {{args}}
+
+test-all *args:
+    just matrix just test {{args}}
+
+test-all-logging-utils *args:
+    just test-all --features logging-utils {{args}}
+
+clippy:
+    cargo clippy --all-targets --workspace --release --bins --tests --examples --features="logging-utils" -- -D warnings
+
+clippy-all *args:
+    just matrix just clippy

justfile

@@ -5,12 +5,10 @@ use futures::{stream::FuturesOrdered, Future, FutureExt};
 use std::{fmt, time::Duration};
 use tracing::warn;
 
-#[cfg(async_executor_impl = "async-std")]
+#[cfg(not(async_executor_impl = "tokio"))]
 use async_std::prelude::StreamExt;
 #[cfg(async_executor_impl = "tokio")]
 use tokio_stream::StreamExt;
-#[cfg(not(any(async_executor_impl = "async-std", async_executor_impl = "tokio")))]
-std::compile_error! {"The cfg flag async_executor_impl must be set in rustflags to either \"async-std\" or \"tokio\" for this crate. Try adding `--cfg async_executor_impl=\"tokio\""}
 
 /// A mutex that can register subscribers to be notified. This works in the same way as [`Mutex`], but has some additional functions:
 ///
@@ -254,7 +252,7 @@ mod tests {
         async_executor_impl = "tokio",
         tokio::test(flavor = "multi_thread", worker_threads = 2)
     )]
-    #[cfg_attr(async_executor_impl = "async-std", async_std::test)]
+    #[cfg_attr(not(async_executor_impl = "tokio"), async_std::test)]
     async fn test_wait_timeout_until() {
         let mutex: Arc<SubscribableMutex<usize>> = Arc::default();
         {
@@ -279,7 +277,7 @@ mod tests {
         async_executor_impl = "tokio",
         tokio::test(flavor = "multi_thread", worker_threads = 2)
     )]
-    #[cfg_attr(async_executor_impl = "async-std", async_std::test)]
+    #[cfg_attr(not(async_executor_impl = "tokio"), async_std::test)]
     async fn test_wait_timeout_until_fail() {
         let mutex: Arc<SubscribableMutex<usize>> = Arc::default();
         {
@@ -303,7 +301,7 @@ mod tests {
         async_executor_impl = "tokio",
         tokio::test(flavor = "multi_thread", worker_threads = 2)
     )]
-    #[cfg_attr(async_executor_impl = "async-std", async_std::test)]
+    #[cfg_attr(not(async_executor_impl = "tokio"), async_std::test)]
     async fn test_compare_and_set() {
         let mutex = SubscribableMutex::new(5usize);
         let subscriber = mutex.subscribe().await;
@@ -325,7 +323,7 @@ mod tests {
         async_executor_impl = "tokio",
         tokio::test(flavor = "multi_thread", worker_threads = 2)
     )]
-    #[cfg_attr(async_executor_impl = "async-std", async_std::test)]
+    #[cfg_attr(not(async_executor_impl = "tokio"), async_std::test)]
     async fn test_subscriber() {
         let mutex = SubscribableMutex::new(5usize);
         let subscriber = mutex.subscribe().await;

src/async_primitives/subscribable_mutex.rs

@@ -21,8 +21,10 @@ mod oneshot;
 /// Unbounded channels
 mod unbounded;
 
-#[cfg(all(async_executor_impl = "async-std", async_channel_impl = "tokio"))]
-compile_error!("async_executor_impl = 'async-std-executor' and async_channel_impl = 'channel-tokio' cannot be used together; 'channel-tokio' needs the tokio runtime");
+#[cfg(all(not(async_executor_impl = "tokio"), async_channel_impl = "tokio"))]
+compile_error!(
+    "async_channel_impl = 'tokio' requires tokio runtime, e. g. async_executor_impl = 'tokio'"
+);
 
 pub use bounded::{bounded, BoundedStream, Receiver, RecvError, SendError, Sender, TryRecvError};
 pub use oneshot::{oneshot, OneShotReceiver, OneShotRecvError, OneShotSender, OneShotTryRecvError};

src/channel.rs

@@ -75,7 +75,7 @@ mod inner {
 }
 
 /// inner module, used to group feature-specific imports
-#[cfg(async_channel_impl = "async-std")]
+#[cfg(not(any(async_channel_impl = "flume", async_channel_impl = "tokio")))]
 mod inner {
     pub use async_std::channel::{RecvError, SendError, TryRecvError};
 
@@ -134,7 +134,7 @@ impl<T> Receiver<T> {
         let result = self.0.recv_async().await;
         #[cfg(async_channel_impl = "tokio")]
         let result = self.0.recv().await.ok_or(RecvError);
-        #[cfg(async_channel_impl = "async-std")]
+        #[cfg(not(any(async_channel_impl = "flume", async_channel_impl = "tokio")))]
         let result = self.0.recv().await;
 
         result
@@ -144,7 +144,7 @@ impl<T> Receiver<T> {
     where
         T: 'static,
     {
-        #[cfg(async_channel_impl = "async-std")]
+        #[cfg(not(any(async_channel_impl = "flume", async_channel_impl = "tokio")))]
         let result = self.0;
         #[cfg(async_channel_impl = "tokio")]
         let result = tokio_stream::wrappers::ReceiverStream::new(self.0);
@@ -226,7 +226,7 @@ impl<T> Stream for BoundedStream<T> {
             Pin::new(&mut self.0),
             cx,
         );
-        #[cfg(async_channel_impl = "async-std")]
+        #[cfg(not(any(async_channel_impl = "flume", async_channel_impl = "tokio")))]
         return <async_std::channel::Receiver<T> as Stream>::poll_next(Pin::new(&mut self.0), cx);
     }
 }

src/channel/bounded.rs

@@ -61,7 +61,7 @@ mod inner {
 }
 
 /// inner module, used to group feature-specific imports
-#[cfg(async_channel_impl = "async-std")]
+#[cfg(not(any(async_channel_impl = "flume", async_channel_impl = "tokio")))]
 mod inner {
     use async_std::channel::{Receiver, Sender};
     pub use async_std::channel::{
@@ -90,11 +90,11 @@ impl<T> OneShotSender<T> {
     ///
     /// If this fails because the receiver is dropped, a warning will be printed.
     pub fn send(self, msg: T) {
-        #[cfg(async_channel_impl = "async-std")]
+        #[cfg(not(any(async_channel_impl = "flume", async_channel_impl = "tokio")))]
         if self.0.try_send(msg).is_err() {
             tracing::warn!("Could not send msg on OneShotSender, did the receiver drop?");
         }
-        #[cfg(not(all(async_channel_impl = "async-std")))]
+        #[cfg(any(async_channel_impl = "flume", async_channel_impl = "tokio"))]
         if self.0.send(msg).is_err() {
             tracing::warn!("Could not send msg on OneShotSender, did the receiver drop?");
         }
@@ -112,7 +112,7 @@ impl<T> OneShotReceiver<T> {
         let result = self.0.await.map_err(Into::into);
         #[cfg(async_channel_impl = "flume")]
         let result = self.0.recv_async().await;
-        #[cfg(async_channel_impl = "async-std")]
+        #[cfg(not(any(async_channel_impl = "flume", async_channel_impl = "tokio")))]
         let result = self.0.recv().await;
 
         result