refactor(credentials): make ServerCredentials trait dyn-compatible

grpc/src/credentials/local.rs

@@ -45,7 +45,6 @@ use crate::credentials::server;
 use crate::credentials::server::ServerConnectionSecurityInfo;
 use crate::private;
 use crate::rt::BoxEndpoint;
-use crate::rt::GrpcEndpoint;
 use crate::rt::GrpcRuntime;
 
 pub const PROTOCOL_NAME: &str = "local";
@@ -162,15 +161,14 @@ impl LocalServerCredentials {
     }
 }
 
+#[async_trait]
 impl ServerCredentials for LocalServerCredentials {
-    type Output<I> = I;
-
-    async fn accept<Input: GrpcEndpoint>(
+    async fn accept(
         &self,
-        source: Input,
+        source: BoxEndpoint,
         _runtime: GrpcRuntime,
         _token: private::Internal,
-    ) -> Result<server::HandshakeOutput<Self::Output<Input>>, String> {
+    ) -> Result<server::HandshakeOutput, String> {
         let security_level =
             security_level_for_endpoint(source.get_peer_address(), source.get_network_type())?;
         Ok(server::HandshakeOutput {
@@ -204,7 +202,6 @@ mod test {
     use crate::credentials::common::Authority;
     use crate::rt;
     use crate::rt::AsyncIoAdapter;
-    use crate::rt::GrpcEndpoint;
     use crate::rt::TcpOptions;
     use crate::rt::tokio::TokioIoStream;
 
@@ -321,7 +318,7 @@ mod test {
         let server_stream = TokioIoStream::new_from_tcp(stream).unwrap();
 
         let output = creds
-            .accept(server_stream, runtime, private::Internal)
+            .accept(Box::new(server_stream), runtime, private::Internal)
             .await
             .unwrap();
         let endpoint = output.endpoint;
@@ -339,4 +336,56 @@ mod test {
 
         client_handle.abort();
     }
+
+    #[tokio::test]
+    async fn test_dyn_server_credential_dispatch() {
+        let creds = LocalServerCredentials::new();
+        let dyn_creds: Arc<dyn ServerCredentials> = Arc::new(creds);
+
+        let info = dyn_creds.info();
+        assert_eq!(info.security_protocol, "local");
+
+        let addr = "127.0.0.1:0";
+        let runtime = rt::default_runtime();
+        let listener = TcpListener::bind(addr).await.unwrap();
+        let server_addr = listener.local_addr().unwrap();
+
+        let client_handle = tokio::spawn(async move {
+            let mut stream = TcpStream::connect(server_addr).await.unwrap();
+            let data = b"hello dynamic grpc server";
+            stream.write_all(data).await.unwrap();
+
+            // Keep the connection alive for a bit so server can read
+            let mut buf = vec![0u8; 1];
+            let _ = stream.read(&mut buf).await;
+        });
+
+        let (stream, _) = listener.accept().await.unwrap();
+        let server_stream = TokioIoStream::new_from_tcp(stream).unwrap();
+
+        let result = dyn_creds
+            .accept(
+                Box::new(server_stream) as BoxEndpoint,
+                runtime,
+                private::Internal,
+            )
+            .await;
+
+        assert!(result.is_ok());
+        let output = result.unwrap();
+        let endpoint = output.endpoint;
+        let security_info = output.security;
+
+        assert_eq!(security_info.security_protocol(), "local");
+        assert_eq!(security_info.security_level(), SecurityLevel::NoSecurity);
+
+        let mut buf = vec![0u8; 25];
+        AsyncIoAdapter::new(endpoint)
+            .read_exact(&mut buf)
+            .await
+            .unwrap();
+        assert_eq!(&buf[..], b"hello dynamic grpc server");
+
+        client_handle.abort();
+    }
 }

grpc/src/credentials/mod.rs

@@ -37,7 +37,6 @@
 
 pub mod call;
 pub(crate) mod client;
-pub(crate) mod dyn_wrapper;
 mod local;
 #[cfg(feature = "tls-rustls")]
 pub mod rustls;
@@ -56,7 +55,6 @@ use crate::credentials::client::HandshakeOutput;
 use crate::credentials::common::Authority;
 use crate::private;
 use crate::rt::BoxEndpoint;
-use crate::rt::GrpcEndpoint;
 use crate::rt::GrpcRuntime;
 
 /// Client-side trait for all live gRPC wire protocols and supported transport
@@ -100,11 +98,8 @@ pub trait ChannelCredentials: Send + Sync + 'static {
 
 /// Server-side trait for all live gRPC wire protocols and supported
 /// transport security protocols (e.g., TLS, ALTS).
-#[trait_variant::make(Send)]
-pub trait ServerCredentials: Sync + 'static {
-    #[doc(hidden)]
-    type Output<I>;
-
+#[async_trait]
+pub trait ServerCredentials: Send + Sync + 'static {
     /// Provides the ProtocolInfo of these credentials.
     fn info(&self) -> &ProtocolInfo;
 
@@ -113,12 +108,12 @@ pub trait ServerCredentials: Sync + 'static {
     /// This method wraps the incoming raw `source` connection with the configured
     /// security protocol (e.g., TLS).
     #[doc(hidden)]
-    async fn accept<Input: GrpcEndpoint>(
+    async fn accept(
         &self,
-        source: Input,
+        source: BoxEndpoint,
         runtime: GrpcRuntime,
         token: private::Internal,
-    ) -> Result<server::HandshakeOutput<Self::Output<Input>>, String>;
+    ) -> Result<server::HandshakeOutput, String>;
 }
 
 /// Defines the level of protection provided by an established connection.

grpc/src/credentials/rustls/server/mod.rs

@@ -56,7 +56,7 @@ use crate::credentials::server::HandshakeOutput;
 use crate::credentials::server::ServerConnectionSecurityInfo;
 use crate::private;
 use crate::rt::AsyncIoAdapter;
-use crate::rt::GrpcEndpoint;
+use crate::rt::BoxEndpoint;
 use crate::rt::GrpcRuntime;
 
 #[cfg(test)]
@@ -338,15 +338,14 @@ impl ProducesTickets for NoTicketer {
     }
 }
 
+#[tonic::async_trait]
 impl ServerCredentials for RustlsServerCredendials {
-    type Output<Input> = TlsStream<Input>;
-
-    async fn accept<Input: GrpcEndpoint>(
+    async fn accept(
         &self,
-        source: Input,
+        source: BoxEndpoint,
         _runtime: GrpcRuntime,
         _token: private::Internal,
-    ) -> Result<HandshakeOutput<Self::Output<Input>>, String> {
+    ) -> Result<HandshakeOutput, String> {
         let input_io = AsyncIoAdapter::new(source);
         let tls_stream = self
             .acceptor
@@ -366,7 +365,7 @@ impl ServerCredentials for RustlsServerCredendials {
         );
         let endpoint = TlsStream::new(RustlsStream::Server(tls_stream));
         Ok(HandshakeOutput {
-            endpoint,
+            endpoint: Box::new(endpoint),
             security: auth_info,
         })
     }