Add rt-tokio-with-wasm feature

Author: brendanhay

Cargo.toml

@@ -25,6 +25,7 @@ criterion = "0.5"
 futures-core = "0.3"
 futures-executor = "0.3"
 futures-util = { version = "0.3", default-features = false }
+futures-timer = "3"
 http = { version = "1.1", default-features = false, features = ["std"] }
 http-body-util = "0.1"
 hyper = { version = "1.3", default-features = false }
@@ -46,6 +47,7 @@ tonic = { version = "0.14.1", default-features = false }
 tonic-prost-build = "0.14.1"
 tonic-prost = "0.14.1"
 tokio = { version = "1", default-features = false }
+tokio_with_wasm = { version = "0.8.7", default-features = false }
 tokio-stream = "0.1"
 # Using `tracing 0.1.40` because 0.1.39 (which is yanked) introduces the ability to set event names in macros,
 # required for OpenTelemetry's internal logging macros.

opentelemetry-sdk/Cargo.toml

@@ -11,20 +11,22 @@ rust-version = "1.75.0"
 autobenches = false
 
 [dependencies]
-opentelemetry = { workspace = true }
-opentelemetry-http = { workspace = true, optional = true }
 futures-channel = { workspace = true }
 futures-executor = { workspace = true }
+futures-timer = { workspace = true, optional = true }
 futures-util = { workspace = true, features = ["std", "sink", "async-await-macro"] }
+http = { workspace = true, optional = true }
+opentelemetry = { workspace = true }
+opentelemetry-http = { workspace = true, optional = true }
 percent-encoding = { workspace = true, optional = true }
 rand = { workspace = true, features = ["std", "std_rng", "small_rng", "os_rng", "thread_rng"], optional = true }
 serde = { workspace = true, features = ["derive", "rc"], optional = true }
 serde_json = { workspace = true, optional = true }
 thiserror = { workspace = true }
-url = { workspace = true, optional = true }
 tokio = { workspace = true, default-features = false, optional = true }
 tokio-stream = { workspace = true, optional = true }
-http = { workspace = true, optional = true }
+tokio_with_wasm = { workspace = true, default-features = false, optional = true }
+url = { workspace = true, optional = true }
 
 [target.'cfg(all(target_arch = "wasm32", not(target_os = "wasi")))'.dependencies]
 getrandom = { workspace = true, features = ["wasm_js"] }
@@ -52,6 +54,7 @@ testing = ["opentelemetry/testing", "trace", "metrics", "logs", "rt-tokio", "rt-
 experimental_async_runtime = []
 rt-tokio = ["tokio/rt", "tokio/time", "tokio-stream", "experimental_async_runtime"]
 rt-tokio-current-thread = ["tokio/rt", "tokio/time", "tokio-stream", "experimental_async_runtime"]
+rt-tokio-with-wasm = ["tokio_with_wasm/rt", "tokio_with_wasm/sync", "tokio_with_wasm/time", "tokio-stream", "experimental_async_runtime", "futures-timer"]
 internal-logs = ["opentelemetry/internal-logs"]
 experimental_metrics_periodicreader_with_async_runtime = ["metrics", "experimental_async_runtime"]
 spec_unstable_metrics_views = ["metrics"]

opentelemetry-sdk/src/runtime.rs

@@ -137,6 +137,35 @@ impl Runtime for TokioCurrentThread {
     }
 }
 
+/// Runtime implementation, which works with Tokio's multi thread runtime.
+#[cfg(all(feature = "experimental_async_runtime", feature = "rt-tokio-with-wasm"))]
+#[cfg_attr(
+    docsrs,
+    doc(cfg(all(feature = "experimental_async_runtime", feature = "rt-tokio-with-wasm")))
+)]
+#[derive(Debug, Clone)]
+pub struct TokioWithWasm;
+
+#[cfg(all(feature = "experimental_async_runtime", feature = "rt-tokio-with-wasm"))]
+#[cfg_attr(
+    docsrs,
+    doc(cfg(all(feature = "experimental_async_runtime", feature = "rt-tokio-with-wasm")))
+)]
+impl Runtime for TokioWithWasm {
+    fn spawn<F>(&self, future: F)
+    where
+        F: Future<Output = ()> + Send + 'static,
+    {
+        #[allow(clippy::let_underscore_future)]
+        // we don't have to await on the returned future to execute
+        let _ = tokio_with_wasm::alias::spawn(future);
+    }
+
+    fn delay(&self, duration: Duration) -> impl Future<Output = ()> + Send + 'static {
+        futures_timer::Delay::new(duration)
+    }
+}
+
 /// `RuntimeChannel` is an extension to [`Runtime`]. Currently, it provides a
 /// channel that is used by the [log] and [span] batch processors.
 ///
@@ -185,7 +214,7 @@ pub trait TrySend: Sync + Send {
 
 #[cfg(all(
     feature = "experimental_async_runtime",
-    any(feature = "rt-tokio", feature = "rt-tokio-current-thread")
+    any(feature = "rt-tokio", feature = "rt-tokio-current-thread",)
 ))]
 impl<T: Send> TrySend for tokio::sync::mpsc::Sender<T> {
     type Message = T;
@@ -198,6 +227,25 @@ impl<T: Send> TrySend for tokio::sync::mpsc::Sender<T> {
     }
 }
 
+#[cfg(all(
+    feature = "experimental_async_runtime",
+    any(feature = "rt-tokio-with-wasm")
+))]
+impl<T: Send> TrySend for tokio_with_wasm::alias::sync::mpsc::Sender<T> {
+    type Message = T;
+
+    fn try_send(&self, item: Self::Message) -> Result<(), TrySendError> {
+        self.try_send(item).map_err(|err| match err {
+            tokio_with_wasm::alias::sync::mpsc::error::TrySendError::Full(_) => {
+                TrySendError::ChannelFull
+            }
+            tokio_with_wasm::alias::sync::mpsc::error::TrySendError::Closed(_) => {
+                TrySendError::ChannelClosed
+            }
+        })
+    }
+}
+
 #[cfg(all(feature = "experimental_async_runtime", feature = "rt-tokio"))]
 #[cfg_attr(
     docsrs,
@@ -245,3 +293,24 @@ impl RuntimeChannel for TokioCurrentThread {
         )
     }
 }
+
+#[cfg(all(feature = "experimental_async_runtime", feature = "rt-tokio-with-wasm"))]
+#[cfg_attr(
+    docsrs,
+    doc(cfg(all(feature = "experimental_async_runtime", feature = "rt-tokio-with-wasm")))
+)]
+impl RuntimeChannel for TokioWithWasm {
+    type Receiver<T: Debug + Send> = tokio_stream::wrappers::ReceiverStream<T>;
+    type Sender<T: Debug + Send> = tokio_with_wasm::alias::sync::mpsc::Sender<T>;
+
+    fn batch_message_channel<T: Debug + Send>(
+        &self,
+        capacity: usize,
+    ) -> (Self::Sender<T>, Self::Receiver<T>) {
+        let (sender, receiver) = tokio_with_wasm::alias::sync::mpsc::channel(capacity);
+        (
+            sender,
+            tokio_stream::wrappers::ReceiverStream::new(receiver),
+        )
+    }
+}