Merge branch 'main' into improve/remove_fxhash

Author: Ggiggle

Cargo.lock

@@ -9,10 +9,6 @@ refused! We are always happy to receive patches for new cool features we haven't
 or didn't judge to be a priority. Please however understand that such patches might take longer
 for us to review.
 
-## Runtime
-
-- [ ] #2 Use `monoio` as an opt-in runtime
-
 ## Service Governence
 
 - [ ] Support OpenSergo
@@ -28,16 +24,16 @@ for us to review.
 
 ### Load Balancer
 
-- [ ] #7 Support consistent hash load balancing
+- [x] #7 Support consistent hash load balancing
 
 ### Proxyless
 
 - [ ] Support Proxyless mode
 
 ## TLS
 
-- [ ] #6 Support TLS for `volo-grpc`
+- [x] #6 Support TLS for `volo-grpc`
 
 ## Cli
 
-- [ ] #5 Support auto generate service code in lib.rs
+- [x] #5 Support auto generate service code in lib.rs

ROADMAP.md

@@ -114,6 +114,11 @@ async fn conn_show(conn: ConnectionInfo) -> String {
     format!("{conn:?}\n")
 }
 
+struct State {
+    foo: String,
+    bar: usize,
+}
+
 async fn extension(Extension(state): Extension<Arc<State>>) -> String {
     format!("State {{ foo: {}, bar: {} }}\n", state.foo, state.bar)
 }
@@ -264,9 +269,15 @@ async fn headers_map_response(response: Response) -> impl IntoResponse {
     )
 }
 
-struct State {
-    foo: String,
-    bar: usize,
+fn tracer(cx: &ServerContext) {
+    tracing::info!(
+        "process start at {:?}, end at {:?}, req size: {:?}, resp size: {:?}, resp status: {:?}",
+        cx.stats.process_start_at().unwrap(),
+        cx.stats.process_end_at().unwrap(),
+        cx.common_stats.req_size().unwrap(),
+        cx.common_stats.resp_size().unwrap(),
+        cx.stats.status_code().unwrap(),
+    );
 }
 
 #[tokio::main(flavor = "multi_thread")]
@@ -297,5 +308,9 @@ async fn main() {
 
     println!("Listening on {addr}");
 
-    Server::new(app).run(addr).await.unwrap();
+    Server::new(app)
+        .stat_tracer(tracer)
+        .run(addr)
+        .await
+        .unwrap();
 }

examples/src/http/simple.rs

@@ -8,6 +8,7 @@ use std::{
 use futures_util::future::BoxFuture;
 use hyper::rt::ReadBufCursor;
 use hyper_util::client::legacy::connect::{Connected, Connection};
+use motore::{make::MakeConnection, service::UnaryService};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use volo::net::{
     conn::Conn,
@@ -55,6 +56,19 @@ impl Default for Connector {
     }
 }
 
+impl UnaryService<Address> for Connector {
+    type Response = Conn;
+    type Error = io::Error;
+
+    async fn call(&self, addr: Address) -> Result<Self::Response, Self::Error> {
+        match self {
+            Self::Default(mkt) => mkt.make_connection(addr).await,
+            #[cfg(any(feature = "rustls", feature = "native-tls"))]
+            Self::Tls(mkt) => mkt.make_connection(addr).await,
+        }
+    }
+}
+
 impl tower::Service<hyper::Uri> for Connector {
     type Response = ConnectionWrapper;
 
@@ -67,61 +81,44 @@ impl tower::Service<hyper::Uri> for Connector {
     }
 
     fn call(&mut self, uri: hyper::Uri) -> Self::Future {
-        macro_rules! box_pin_call {
-            ($mk_conn:ident) => {
-                Box::pin(async move {
-                    let authority = uri.authority().expect("authority required").as_str();
-                    let target: Address = match uri.scheme_str() {
-                        Some("http") => {
-                            Address::Ip(authority.parse::<SocketAddr>().map_err(|_| {
-                                io::Error::new(
-                                    io::ErrorKind::InvalidInput,
-                                    "authority must be valid SocketAddr",
-                                )
-                            })?)
-                        }
-                        #[cfg(target_family = "unix")]
-                        Some("http+unix") => {
-                            use hex::FromHex;
-
-                            let bytes = Vec::from_hex(authority).map_err(|_| {
+        let connector = self.clone();
+        Box::pin(async move {
+            let authority = uri.authority().expect("authority required").as_str();
+            let target: Address = match uri.scheme_str() {
+                Some("http") => Address::Ip(authority.parse::<SocketAddr>().map_err(|_| {
+                    io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "authority must be valid SocketAddr",
+                    )
+                })?),
+                #[cfg(target_family = "unix")]
+                Some("http+unix") => {
+                    use hex::FromHex;
+
+                    let bytes = Vec::from_hex(authority).map_err(|_| {
+                        io::Error::new(
+                            io::ErrorKind::InvalidInput,
+                            "authority must be hex-encoded path",
+                        )
+                    })?;
+                    Address::Unix(std::borrow::Cow::Owned(
+                        String::from_utf8(bytes)
+                            .map_err(|_| {
                                 io::Error::new(
                                     io::ErrorKind::InvalidInput,
-                                    "authority must be hex-encoded path",
+                                    "authority must be valid UTF-8",
                                 )
-                            })?;
-                            Address::Unix(std::borrow::Cow::Owned(
-                                String::from_utf8(bytes)
-                                    .map_err(|_| {
-                                        io::Error::new(
-                                            io::ErrorKind::InvalidInput,
-                                            "authority must be valid UTF-8",
-                                        )
-                                    })?
-                                    .into(),
-                            ))
-                        }
-                        _ => unimplemented!(),
-                    };
-
-                    Ok(ConnectionWrapper {
-                        inner: $mk_conn.make_connection(target).await?,
-                    })
-                })
+                            })?
+                            .into(),
+                    ))
+                }
+                _ => unimplemented!(),
             };
-        }
 
-        match self {
-            Self::Default(mk_conn) => {
-                let mk_conn = *mk_conn;
-                box_pin_call!(mk_conn)
-            }
-            #[cfg(any(feature = "rustls", feature = "native-tls"))]
-            Self::Tls(mk_conn) => {
-                let mk_conn = mk_conn.clone();
-                box_pin_call!(mk_conn)
-            }
-        }
+            Ok(ConnectionWrapper {
+                inner: connector.make_connection(target).await?,
+            })
+        })
     }
 }
 

examples/src/http/state.rs

@@ -1,6 +1,6 @@
 [package]
 name = "volo-http"
-version = "0.1.14"
+version = "0.1.17"
 edition.workspace = true
 homepage.workspace = true
 repository.workspace = true
@@ -20,18 +20,20 @@ maintenance = { status = "actively-developed" }
 [dependencies]
 volo = { version = "0.9", path = "../volo" }
 
-http-body-util = "0.1"
-hyper = { version = "1", features = ["server", "http1", "http2"] }
-hyper-util = { version = "0.1", features = ["tokio"] }
-
 ahash.workspace = true
 bytes.workspace = true
 chrono.workspace = true
 cookie = { workspace = true, optional = true, features = ["percent-encode"] }
 faststr.workspace = true
 futures.workspace = true
 futures-util.workspace = true
+http.workspace = true
+http-body.workspace = true
+http-body-util.workspace = true
+hyper = { workspace = true, features = ["server", "http1", "http2"] }
+hyper-util = { workspace = true, features = ["tokio"] }
 matchit.workspace = true
+metainfo.workspace = true
 mime.workspace = true
 motore.workspace = true
 parking_lot.workspace = true

volo-grpc/src/transport/connect.rs

@@ -0,0 +1,93 @@
+use std::{
+    convert::Infallible,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use bytes::Bytes;
+use faststr::FastStr;
+use futures_util::ready;
+use http_body::{Frame, SizeHint};
+use http_body_util::Full;
+use pin_project::pin_project;
+
+#[pin_project]
+pub struct Body {
+    #[pin]
+    inner: Full<Bytes>,
+}
+
+impl http_body::Body for Body {
+    type Data = Bytes;
+
+    type Error = Infallible;
+
+    fn poll_frame(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        Poll::Ready(ready!(self.project().inner.poll_frame(cx)).map(|result| Ok(result.unwrap())))
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+
+    fn size_hint(&self) -> SizeHint {
+        self.inner.size_hint()
+    }
+}
+
+impl From<()> for Body {
+    fn from(_: ()) -> Self {
+        Self {
+            inner: Full::new(Bytes::new()),
+        }
+    }
+}
+
+impl From<&'static str> for Body {
+    fn from(value: &'static str) -> Self {
+        Self {
+            inner: Full::new(value.into()),
+        }
+    }
+}
+
+impl From<Vec<u8>> for Body {
+    fn from(value: Vec<u8>) -> Self {
+        Self {
+            inner: Full::new(value.into()),
+        }
+    }
+}
+
+impl From<Bytes> for Body {
+    fn from(value: Bytes) -> Self {
+        Self {
+            inner: Full::new(value),
+        }
+    }
+}
+
+impl From<FastStr> for Body {
+    fn from(value: FastStr) -> Self {
+        Self {
+            inner: Full::new(value.into()),
+        }
+    }
+}
+
+impl From<String> for Body {
+    fn from(value: String) -> Self {
+        Self {
+            inner: Full::new(value.into()),
+        }
+    }
+}
+
+impl From<Full<Bytes>> for Body {
+    fn from(value: Full<Bytes>) -> Self {
+        Self { inner: value }
+    }
+}