WIP: Patch/bric - build to wasm32-wasi target

Cargo.toml

@@ -21,20 +21,20 @@ repository = "https://github.com/cloudwego/volo"
 license = "MIT OR Apache-2.0"
 
 [workspace.dependencies]
-pilota = "0.9"
-pilota-build = "0.9"
-pilota-thrift-parser = "0.9"
-#pilota = { git = "https://github.com/cloudwego/pilota", branch = "main" }
-#pilota-build = { git = "https://github.com/cloudwego/pilota", branch = "main" }
-#pilota-thrift-parser = { git = "https://github.com/cloudwego/pilota", branch = "main" }
+# pilota = "0.9"
+# pilota-build = "0.9"
+# pilota-thrift-parser = "0.9"
+pilota = { git = "https://github.com/cloudwego/pilota", branch = "main" }
+pilota-build = { git = "https://github.com/cloudwego/pilota", branch = "main" }
+pilota-thrift-parser = { git = "https://github.com/cloudwego/pilota", branch = "main" }
 
 motore = "0.4"
 # motore = { git = "https://github.com/cloudwego/motore", branch = "main" }
 
 metainfo = "0.7"
 
 anyhow = "1"
-async-broadcast = "0.6"
+# async-broadcast = "0.6"
 async-stream = "0.3"
 base64 = "0.13"
 bytes = "1"
@@ -96,9 +96,19 @@ socket2 = "0.5"
 syn = "1"
 tempfile = "3"
 thiserror = "1"
-tokio = "1"
-tokio-stream = "0.1"
-tokio-util = "0.7"
+tokio = { version = "1.29.1", default-features = false, features = [
+  "sync",
+  "macros",
+  "io-util",
+  "rt",
+  "time",
+] }
+tokio-stream = { version = "0.1", default-features = false, features = [
+  "time",
+] }
+
+tokio-util = { version = "0.7", default-features = false, features = [] }
+
 tower = "0.4"
 tracing = "0.1"
 tracing-subscriber = "0.3"

volo-grpc/Cargo.toml

@@ -45,7 +45,7 @@ hyper-util.workspace = true
 matchit.workspace = true
 percent-encoding.workspace = true
 pin-project.workspace = true
-tokio = { workspace = true, features = ["time", "rt", "net", "sync", "signal"] }
+tokio = { workspace = true }
 tokio-stream.workspace = true
 tokio-util = { workspace = true, features = ["codec", "compat"] }
 tower = { workspace = true, features = [

volo-grpc/src/layer/loadbalance/mod.rs

@@ -62,8 +62,8 @@ where
             tokio::spawn(async move {
                 loop {
                     match channel.recv().await {
-                        Ok(recv) => lb.rebalance(recv),
-                        Err(err) => warn!("[VOLO] discovering subscription error {:?}", err),
+                        Some(recv) => lb.rebalance(recv),
+                        None => warn!("[VOLO] discovering subscription error"),
                     }
                 }
             });

volo-thrift/Cargo.toml

@@ -36,17 +36,19 @@ parking_lot.workspace = true
 paste.workspace = true
 pin-project.workspace = true
 thiserror.workspace = true
-tokio = { workspace = true, features = [
+tokio = { workspace = true, default-features = false, features = [
+    "net",
     "time",
-    "macros",
-    "rt",
-    "signal",
-    "parking_lot",
+    "sync",
+    "io-util",
 ] }
 tracing.workspace = true
 
 [features]
+# default = ["async_broadcast"]
+# wasi = []
 default = []
+
 # multiplex is unstable and we don't provide backward compatibility
 multiplex = []
 # unsafe-codec can achieve better performance for thrift binary protocol, but may cause undefined behavior

volo-thrift/src/message_wrapper.rs

@@ -117,7 +117,7 @@ where
     U: EntryMessage + Send,
 {
     #[inline]
-    pub(crate) fn encode<T: TOutputProtocol>(&self, protocol: &mut T) -> Result<(), EncodeError> {
+    pub fn encode<T: TOutputProtocol>(&self, protocol: &mut T) -> Result<(), EncodeError> {
         let ident = TMessageIdentifier::new(
             self.meta.method.clone(),
             self.meta.msg_type,
@@ -151,7 +151,7 @@ where
     }
 
     #[inline]
-    pub(crate) fn decode<Cx: ThriftContext, T: TInputProtocol>(
+    pub fn decode<Cx: ThriftContext, T: TInputProtocol>(
         protocol: &mut T,
         cx: &mut Cx,
     ) -> Result<Self, DecodeError> {

volo/Cargo.toml

@@ -25,7 +25,7 @@ maintenance = { status = "actively-developed" }
 [dependencies]
 motore.workspace = true
 
-async-broadcast.workspace = true
+# async-broadcast.workspace = true
 dashmap.workspace = true
 faststr.workspace = true
 futures.workspace = true
@@ -45,7 +45,12 @@ pin-project.workspace = true
 rand.workspace = true
 socket2 = { workspace = true, features = ["all"] }
 thiserror.workspace = true
-tokio = { workspace = true, features = ["net", "time", "sync", "io-util"] }
+tokio = { workspace = true, default-features = false, features = [
+    "net",
+    "time",
+    "sync",
+    "io-util",
+] }
 tokio-stream = { workspace = true, features = ["net"] }
 tower.workspace = true
 tracing.workspace = true
@@ -59,4 +64,4 @@ tokio-native-tls = { workspace = true, optional = true }
 default = []
 
 rustls = ["tokio-rustls", "librustls"]
-native-tls = ["tokio-native-tls"]
+native-tls = ["tokio-native-tls"]

volo/src/discovery/mod.rs

@@ -13,7 +13,9 @@ use std::{
     sync::Arc,
 };
 
-use async_broadcast::Receiver;
+// #[cfg(not(target_os = "wasi"))]
+// use async_broadcast::Receiver;
+use tokio::sync::mpsc::UnboundedReceiver;
 
 use crate::{context::Endpoint, loadbalance::error::LoadBalanceError, net::Address};
 
@@ -41,7 +43,7 @@ pub trait Discover: Send + Sync + 'static {
     fn key(&self, endpoint: &Endpoint) -> Self::Key;
     /// `watch` should return a [`async_broadcast::Receiver`] which can be used to subscribe
     /// [`Change`].
-    fn watch(&self, keys: Option<&[Self::Key]>) -> Option<Receiver<Change<Self::Key>>>;
+    fn watch(&self, keys: Option<&[Self::Key]>) -> Option<UnboundedReceiver<Change<Self::Key>>>;
 }
 
 /// Change indicates the change of the service discover.
@@ -158,7 +160,7 @@ impl Discover for StaticDiscover {
 
     fn key(&self, _: &Endpoint) -> Self::Key {}
 
-    fn watch(&self, _keys: Option<&[Self::Key]>) -> Option<Receiver<Change<Self::Key>>> {
+    fn watch(&self, _keys: Option<&[Self::Key]>) -> Option<UnboundedReceiver<Change<Self::Key>>> {
         None
     }
 }
@@ -179,7 +181,7 @@ impl Discover for DummyDiscover {
 
     fn key(&self, _: &Endpoint) {}
 
-    fn watch(&self, _keys: Option<&[Self::Key]>) -> Option<Receiver<Change<Self::Key>>> {
+    fn watch(&self, _keys: Option<&[Self::Key]>) -> Option<UnboundedReceiver<Change<Self::Key>>> {
         None
     }
 }

volo/src/loadbalance/layer.rs

@@ -33,8 +33,8 @@ where
             tokio::spawn(async move {
                 loop {
                     match channel.recv().await {
-                        Ok(recv) => lb.rebalance(recv),
-                        Err(err) => warn!("[VOLO] discovering subscription error: {:?}", err),
+                        Some(recv) => lb.rebalance(recv),
+                        None => warn!("[VOLO] discovering subscription error"),
                     }
                 }
             });