Add client feature flag to support connect_with_connector in wasm32 targets

tonic/Cargo.toml

@@ -38,9 +38,19 @@ transport = [
   "channel",
   "dep:h2",
   "dep:hyper",
+  "hyper/full",
   "tokio/net",
   "tokio/time",
   "dep:tower",
+  "tower/balance",
+  "dep:hyper-timeout",
+]
+client = [
+  "channel",
+  "dep:h2",
+  "hyper/client",
+  "hyper/http2",
+  "dep:tower",
   "dep:hyper-timeout",
 ]
 channel = []
@@ -55,7 +65,7 @@ bytes = "1.0"
 http = "0.2"
 tracing = "0.1"
 
-tokio = "1.0.1"
+tokio = { version = "1.0.1" }
 http-body = "0.4.4"
 percent-encoding = "2.1"
 pin-project = "1.0.11"
@@ -70,7 +80,7 @@ async-trait = {version = "0.1.13", optional = true}
 
 # transport
 h2 = {version = "0.3.17", optional = true}
-hyper = {version = "0.14.26", features = ["full"], optional = true}
+hyper = {version = "0.14.26", default-features = false, optional = true}
 hyper-timeout = {version = "0.4", optional = true}
 tokio-stream = "0.1"
 tower = {version = "0.4.7", default-features = false, features = ["balance", "buffer", "discover", "limit", "load", "make", "timeout", "util"], optional = true}
@@ -88,6 +98,9 @@ webpki-roots = { version = "0.25.0", optional = true }
 flate2 = {version = "1.0", optional = true}
 zstd = { version = "0.12.3", optional = true }
 
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+wasm-bindgen-futures = "0.4.38"
+
 [dev-dependencies]
 bencher = "0.1.5"
 quickcheck = "1.0"

tonic/src/lib.rs

@@ -101,7 +101,7 @@ pub mod metadata;
 pub mod server;
 pub mod service;
 
-#[cfg(feature = "transport")]
+#[cfg(any(feature = "transport", feature = "client"))]
 #[cfg_attr(docsrs, doc(cfg(feature = "transport")))]
 pub mod transport;
 

tonic/src/transport/channel/endpoint.rs

@@ -27,10 +27,15 @@ pub struct Endpoint {
     pub(crate) buffer_size: Option<usize>,
     pub(crate) init_stream_window_size: Option<u32>,
     pub(crate) init_connection_window_size: Option<u32>,
+    #[cfg(feature = "transport")]
     pub(crate) tcp_keepalive: Option<Duration>,
+    #[cfg(feature = "transport")]
     pub(crate) tcp_nodelay: bool,
+    #[cfg(feature = "transport")]
     pub(crate) http2_keep_alive_interval: Option<Duration>,
+    #[cfg(feature = "transport")]
     pub(crate) http2_keep_alive_timeout: Option<Duration>,
+    #[cfg(feature = "transport")]
     pub(crate) http2_keep_alive_while_idle: Option<bool>,
     pub(crate) connect_timeout: Option<Duration>,
     pub(crate) http2_adaptive_window: Option<bool>,
@@ -167,6 +172,7 @@ impl Endpoint {
     ///
     /// Default is no keepalive (`None`)
     ///
+    #[cfg(feature = "transport")]
     pub fn tcp_keepalive(self, tcp_keepalive: Option<Duration>) -> Self {
         Endpoint {
             tcp_keepalive,
@@ -251,6 +257,7 @@ impl Endpoint {
     }
 
     /// Set the value of `TCP_NODELAY` option for accepted connections. Enabled by default.
+    #[cfg(feature = "transport")]
     pub fn tcp_nodelay(self, enabled: bool) -> Self {
         Endpoint {
             tcp_nodelay: enabled,
@@ -259,6 +266,7 @@ impl Endpoint {
     }
 
     /// Set http2 KEEP_ALIVE_INTERVAL. Uses `hyper`'s default otherwise.
+    #[cfg(feature = "transport")]
     pub fn http2_keep_alive_interval(self, interval: Duration) -> Self {
         Endpoint {
             http2_keep_alive_interval: Some(interval),
@@ -267,6 +275,7 @@ impl Endpoint {
     }
 
     /// Set http2 KEEP_ALIVE_TIMEOUT. Uses `hyper`'s default otherwise.
+    #[cfg(feature = "transport")]
     pub fn keep_alive_timeout(self, duration: Duration) -> Self {
         Endpoint {
             http2_keep_alive_timeout: Some(duration),
@@ -275,6 +284,7 @@ impl Endpoint {
     }
 
     /// Set http2 KEEP_ALIVE_WHILE_IDLE. Uses `hyper`'s default otherwise.
+    #[cfg(feature = "transport")]
     pub fn keep_alive_while_idle(self, enabled: bool) -> Self {
         Endpoint {
             http2_keep_alive_while_idle: Some(enabled),
@@ -312,6 +322,7 @@ impl Endpoint {
     }
 
     /// Create a channel from this config.
+    #[cfg(feature = "transport")]
     pub async fn connect(&self) -> Result<Channel, Error> {
         let mut http = hyper::client::connect::HttpConnector::new();
         http.enforce_http(false);
@@ -333,6 +344,7 @@ impl Endpoint {
     ///
     /// The channel returned by this method does not attempt to connect to the endpoint until first
     /// use.
+    #[cfg(feature = "transport")]
     pub fn connect_lazy(&self) -> Channel {
         let mut http = hyper::client::connect::HttpConnector::new();
         http.enforce_http(false);
@@ -421,14 +433,19 @@ impl From<Uri> for Endpoint {
             buffer_size: None,
             init_stream_window_size: None,
             init_connection_window_size: None,
+            #[cfg(feature = "transport")]
             tcp_keepalive: None,
+            #[cfg(feature = "transport")]
             tcp_nodelay: true,
+            #[cfg(feature = "transport")]
             http2_keep_alive_interval: None,
+            #[cfg(feature = "transport")]
             http2_keep_alive_timeout: None,
+            #[cfg(feature = "transport")]
             http2_keep_alive_while_idle: None,
             connect_timeout: None,
             http2_adaptive_window: None,
-            executor: SharedExec::tokio(),
+            executor: SharedExec::default_exec(),
         }
     }
 }

tonic/src/transport/channel/mod.rs

@@ -9,7 +9,9 @@ pub use endpoint::Endpoint;
 #[cfg(feature = "tls")]
 pub use tls::ClientTlsConfig;
 
-use super::service::{Connection, DynamicServiceStream, SharedExec};
+use super::service::Connection;
+#[cfg(feature = "transport")]
+use super::service::{DynamicServiceStream, SharedExec};
 use crate::body::BoxBody;
 use crate::transport::Executor;
 use bytes::Bytes;
@@ -18,22 +20,26 @@ use http::{
     Request, Response,
 };
 use hyper::client::connect::Connection as HyperConnection;
+#[cfg(feature = "transport")]
+use std::hash::Hash;
 use std::{
     fmt,
     future::Future,
-    hash::Hash,
     pin::Pin,
     task::{ready, Context, Poll},
 };
-use tokio::{
-    io::{AsyncRead, AsyncWrite},
-    sync::mpsc::{channel, Sender},
+use tokio::io::{AsyncRead, AsyncWrite};
+
+#[cfg(feature = "transport")]
+use tokio::sync::mpsc::{channel, Sender};
+#[cfg(feature = "transport")]
+use tower::{
+    balance::p2c::Balance,
+    discover::{Change, Discover},
 };
 
-use tower::balance::p2c::Balance;
 use tower::{
     buffer::{self, Buffer},
-    discover::{Change, Discover},
     util::{BoxService, Either},
     Service,
 };
@@ -109,6 +115,7 @@ impl Channel {
     ///
     /// This creates a [`Channel`] that will load balance across all the
     /// provided endpoints.
+    #[cfg(feature = "transport")]
     pub fn balance_list(list: impl Iterator<Item = Endpoint>) -> Self {
         let (channel, tx) = Self::balance_channel(DEFAULT_BUFFER_SIZE);
         list.for_each(|endpoint| {
@@ -122,18 +129,20 @@ impl Channel {
     /// Balance a list of [`Endpoint`]'s.
     ///
     /// This creates a [`Channel`] that will listen to a stream of change events and will add or remove provided endpoints.
+    #[cfg(feature = "transport")]
     pub fn balance_channel<K>(capacity: usize) -> (Self, Sender<Change<K, Endpoint>>)
     where
         K: Hash + Eq + Send + Clone + 'static,
     {
-        Self::balance_channel_with_executor(capacity, SharedExec::tokio())
+        Self::balance_channel_with_executor(capacity, SharedExec::default_exec())
     }
 
     /// Balance a list of [`Endpoint`]'s.
     ///
     /// This creates a [`Channel`] that will listen to a stream of change events and will add or remove provided endpoints.
     ///
     /// The [`Channel`] will use the given executor to spawn async tasks.
+    #[cfg(feature = "transport")]
     pub fn balance_channel_with_executor<K, E>(
         capacity: usize,
         executor: E,
@@ -183,6 +192,7 @@ impl Channel {
         Ok(Channel { svc })
     }
 
+    #[cfg(feature = "transport")]
     pub(crate) fn balance<D, E>(discover: D, buffer_size: usize, executor: E) -> Self
     where
         D: Discover<Service = Connection> + Unpin + Send + 'static,

tonic/src/transport/mod.rs

@@ -88,6 +88,7 @@
 //! [rustls]: https://docs.rs/rustls/0.16.0/rustls/
 
 pub mod channel;
+#[cfg(feature = "transport")]
 pub mod server;
 
 mod error;
@@ -100,13 +101,15 @@ mod tls;
 #[cfg_attr(docsrs, doc(cfg(feature = "channel")))]
 pub use self::channel::{Channel, Endpoint};
 pub use self::error::Error;
+#[cfg(feature = "transport")]
 #[doc(inline)]
 pub use self::server::Server;
 #[doc(inline)]
 pub use self::service::grpc_timeout::TimeoutExpired;
 #[cfg(feature = "tls")]
 #[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
 pub use self::tls::Certificate;
+#[cfg(feature = "transport")]
 pub use axum::{body::BoxBody as AxumBoxBody, Router as AxumRouter};
 pub use hyper::{Body, Uri};
 

tonic/src/transport/service/connection.rs

@@ -1,16 +1,12 @@
-use super::{grpc_timeout::GrpcTimeout, reconnect::Reconnect, AddOrigin, UserAgent};
-use crate::{
-    body::BoxBody,
-    transport::{BoxFuture, Endpoint},
+use std::{
+    fmt,
+    task::{Context, Poll},
 };
+
 use http::Uri;
 use hyper::client::conn::Builder;
 use hyper::client::connect::Connection as HyperConnection;
 use hyper::client::service::Connect as HyperConnect;
-use std::{
-    fmt,
-    task::{Context, Poll},
-};
 use tokio::io::{AsyncRead, AsyncWrite};
 use tower::load::Load;
 use tower::{
@@ -21,6 +17,13 @@ use tower::{
 };
 use tower_service::Service;
 
+use crate::{
+    body::BoxBody,
+    transport::{BoxFuture, Endpoint},
+};
+
+use super::{grpc_timeout::GrpcTimeout, reconnect::Reconnect, AddOrigin, UserAgent};
+
 pub(crate) type Request = http::Request<BoxBody>;
 pub(crate) type Response = http::Response<hyper::Body>;
 
@@ -40,20 +43,32 @@ impl Connection {
             .http2_initial_stream_window_size(endpoint.init_stream_window_size)
             .http2_initial_connection_window_size(endpoint.init_connection_window_size)
             .http2_only(true)
-            .http2_keep_alive_interval(endpoint.http2_keep_alive_interval)
             .executor(endpoint.executor.clone())
             .clone();
 
-        if let Some(val) = endpoint.http2_keep_alive_timeout {
-            settings.http2_keep_alive_timeout(val);
+        if let Some(val) = endpoint.http2_adaptive_window {
+            settings.http2_adaptive_window(val);
         }
 
-        if let Some(val) = endpoint.http2_keep_alive_while_idle {
-            settings.http2_keep_alive_while_idle(val);
+        #[cfg(feature = "transport")]
+        {
+            settings.http2_keep_alive_interval(endpoint.http2_keep_alive_interval);
+
+            if let Some(val) = endpoint.http2_keep_alive_timeout {
+                settings.http2_keep_alive_timeout(val);
+            }
+
+            if let Some(val) = endpoint.http2_keep_alive_while_idle {
+                settings.http2_keep_alive_while_idle(val);
+            }
         }
 
-        if let Some(val) = endpoint.http2_adaptive_window {
-            settings.http2_adaptive_window(val);
+        #[cfg(target_arch = "wasm32")]
+        {
+            settings
+                .executor(wasm::Executor)
+                // reset streams require `Instant::now` which is not available on wasm
+                .http2_max_concurrent_reset_streams(0);
         }
 
         let stack = ServiceBuilder::new()
@@ -126,3 +141,19 @@ impl fmt::Debug for Connection {
         f.debug_struct("Connection").finish()
     }
 }
+
+#[cfg(target_arch = "wasm32")]
+mod wasm {
+    use std::future::Future;
+    use std::pin::Pin;
+
+    type BoxSendFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
+
+    pub(crate) struct Executor;
+
+    impl hyper::rt::Executor<BoxSendFuture> for Executor {
+        fn execute(&self, fut: BoxSendFuture) {
+            wasm_bindgen_futures::spawn_local(fut)
+        }
+    }
+}